XCClaw/tests/test_api.py

import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.models.session import SessionType


@pytest.mark.asyncio
async def test_health_check():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        response = await client.get("/api/xcclaw/health")
        assert response.status_code == 200
        data = response.json()
        assert data["status"] in ["ok", "error"]


@pytest.mark.asyncio
async def test_create_task():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        response = await client.post(
            "/api/xcclaw/task", json={"type": "ephemeral", "prompt": "test task"}
        )
        assert response.status_code == 200
        data = response.json()
        assert "id" in data
        assert data["prompt"] == "test task"


@pytest.mark.asyncio
async def test_list_tasks():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        await client.post("/api/xcclaw/task", json={"prompt": "task 1"})

        response = await client.get("/api/xcclaw/task")
        assert response.status_code == 200
        data = response.json()
        assert isinstance(data, list)


@pytest.mark.asyncio
async def test_get_task():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        create_response = await client.post(
            "/api/xcclaw/task", json={"prompt": "get test"}
        )
        task_id = create_response.json()["id"]

        response = await client.get(f"/api/xcclaw/task/{task_id}")
        assert response.status_code == 200
        data = response.json()
        assert data["id"] == task_id


@pytest.mark.asyncio
async def test_get_task_not_found():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        response = await client.get("/api/xcclaw/task/nonexistent")
        assert response.status_code == 404


@pytest.mark.asyncio
async def test_create_schedule():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        response = await client.post(
            "/api/xcclaw/schedule",
            json={
                "id": "test-schedule",
                "name": "Test Schedule",
                "cron": "0 9 * * *",
                "prompt": "daily task",
                "enabled": True,
            },
        )
        assert response.status_code == 200
        data = response.json()
        assert data["id"] == "test-schedule"


@pytest.mark.asyncio
async def test_list_schedules():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        await client.post(
            "/api/xcclaw/schedule",
            json={
                "id": "s1",
                "name": "Schedule 1",
                "cron": "0 9 * * *",
                "prompt": "t1",
            },
        )

        response = await client.get("/api/xcclaw/schedule")
        assert response.status_code == 200
        data = response.json()
        assert isinstance(data, list)


@pytest.mark.asyncio
async def test_delete_schedule():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        await client.post(
            "/api/xcclaw/schedule",
            json={
                "id": "to-delete",
                "name": "Delete Me",
                "cron": "0 9 * * *",
                "prompt": "t",
            },
        )

        response = await client.delete("/api/xcclaw/schedule/to-delete")
        assert response.status_code == 200


@pytest.mark.asyncio
async def test_get_history():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        response = await client.get("/api/xcclaw/history")
        assert response.status_code == 200
        data = response.json()
        assert isinstance(data, list)


@pytest.mark.asyncio
async def test_get_history_with_limit():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        response = await client.get("/api/xcclaw/history?limit=10")
        assert response.status_code == 200


@pytest.mark.asyncio
async def test_clear_history():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        response = await client.delete("/api/xcclaw/history")
        assert response.status_code == 200
        assert response.json()["cleared"] is True


@pytest.mark.asyncio
@pytest.mark.skip(reason="Test environment issue with event loop")
async def test_create_persistent_session():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        response = await client.post("/api/xcclaw/persistent?name=TestSession")
        assert response.status_code == 200
        data = response.json()
        assert "id" in data
        assert data["name"] == "TestSession"


@pytest.mark.asyncio
@pytest.mark.skip(reason="Test environment issue with event loop")
async def test_list_persistent_sessions():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        await client.post("/api/xcclaw/persistent?name=PS1")

        response = await client.get("/api/xcclaw/persistent")
        assert response.status_code == 200
        data = response.json()
        assert isinstance(data, list)


@pytest.mark.asyncio
@pytest.mark.skip(reason="Test environment issue with event loop")
async def test_delete_persistent_session():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        create_response = await client.post("/api/xcclaw/persistent?name=ToDelete")
        session_id = create_response.json()["id"]

        response = await client.delete(f"/api/xcclaw/persistent/{session_id}")
        assert response.status_code == 200