| from clfgui.backend.api import BackendApi |
| from clfgui.backend.context_store import ContextStore |
| from clfgui.backend.hermes_adapter import HermesRunError |
| from clfgui.backend.profiles import ProfileDiscovery |
| from clfgui.backend.service import BackendService |
|
|
|
|
| class FakeHermes: |
| def __init__(self, output: str | None = None, error: Exception | None = None): |
| self.output = output |
| self.error = error |
| self.calls = [] |
|
|
| def run_agent(self, agent_id: str, current_context: str): |
| self.calls.append((agent_id, current_context)) |
| if self.error is not None: |
| raise self.error |
|
|
| class Result: |
| output = self.output or "" |
|
|
| return Result() |
|
|
|
|
| def make_api(tmp_path, hermes): |
| profiles_dir = tmp_path / "Hermes" / "profiles" |
| profiles_dir.mkdir(parents=True) |
| (profiles_dir / "planner").mkdir() |
| service = BackendService( |
| profiles=ProfileDiscovery(profiles_dir), |
| store=ContextStore(tmp_path / "state"), |
| hermes=hermes, |
| ) |
| return BackendApi(service) |
|
|
|
|
| def test_context_endpoint_round_trip(tmp_path): |
| api = make_api(tmp_path, FakeHermes()) |
|
|
| response = api.handle("POST", "/api/context", {"context": "initial context"}) |
| assert response.status == 200 |
|
|
| response = api.handle("GET", "/api/context") |
| assert response.status == 200 |
| assert response.body["context"] == "initial context" |
| assert response.body["updated_at"] is not None |
|
|
|
|
| def test_run_agent_updates_context_and_logs_success(tmp_path): |
| hermes = FakeHermes("new context") |
| api = make_api(tmp_path, hermes) |
| api.handle("POST", "/api/context", {"context": "old context"}) |
|
|
| response = api.handle("POST", "/api/run-agent", {"agent_id": "planner"}) |
|
|
| assert response.status == 200 |
| assert response.body["context"] == "new context" |
| assert hermes.calls == [("planner", "old context")] |
|
|
| context = api.handle("GET", "/api/context") |
| assert context.body["context"] == "new context" |
|
|
| runs = api.handle("GET", "/api/runs") |
| assert len(runs.body["runs"]) == 1 |
| run = runs.body["runs"][0] |
| assert run["status"] == "success" |
| assert run["input_context"] == "old context" |
| assert run["output_context"] == "new context" |
|
|
| run_response = api.handle("GET", f"/api/runs/{run['id']}") |
| assert run_response.status == 200 |
| assert run_response.body["id"] == run["id"] |
|
|
|
|
| def test_run_agent_failure_preserves_old_context_and_logs_failure(tmp_path): |
| hermes = FakeHermes(error=HermesRunError("boom")) |
| api = make_api(tmp_path, hermes) |
| api.handle("POST", "/api/context", {"context": "old context"}) |
|
|
| response = api.handle("POST", "/api/run-agent", {"agent_id": "planner"}) |
|
|
| assert response.status == 502 |
| assert "boom" in response.body["error"] |
| assert api.handle("GET", "/api/context").body["context"] == "old context" |
|
|
| runs = api.handle("GET", "/api/runs").body["runs"] |
| assert len(runs) == 1 |
| assert runs[0]["status"] == "failed" |
| assert runs[0]["input_context"] == "old context" |
| assert runs[0]["output_context"] is None |
|
|
|
|
| def test_api_rejects_invalid_or_unknown_agents(tmp_path): |
| api = make_api(tmp_path, FakeHermes("new")) |
|
|
| invalid = api.handle("POST", "/api/run-agent", {"agent_id": "../planner"}) |
| unknown = api.handle("POST", "/api/run-agent", {"agent_id": "missing"}) |
|
|
| assert invalid.status == 400 |
| assert unknown.status == 404 |
|
|
|
|
| def test_api_supports_health_index_and_trailing_slashes(tmp_path): |
| api = make_api(tmp_path, FakeHermes("new")) |
|
|
| index = api.handle("GET", "/") |
| health = api.handle("GET", "/api/health") |
| agents = api.handle("GET", "/api/agents/") |
| context = api.handle("GET", "/api/context/") |
| runs = api.handle("GET", "/api/runs/") |
|
|
| assert index.status == 200 |
| assert index.body["service"] == "clfgui-api" |
| assert "GET /api/agents" in index.body["endpoints"] |
| assert health.status == 200 |
| assert health.body == {"status": "ok"} |
| assert agents.status == 200 |
| assert agents.body["agents"] == [{"id": "planner", "name": "planner"}] |
| assert context.status == 200 |
| assert context.body["context"] == "" |
| assert runs.status == 200 |
| assert runs.body["runs"] == [] |
|
|
|
|
| def test_api_404_includes_route_metadata(tmp_path): |
| api = make_api(tmp_path, FakeHermes("new")) |
|
|
| response = api.handle("GET", "/missing") |
|
|
| assert response.status == 404 |
| assert response.body["error"] == "Not found." |
| assert response.body["method"] == "GET" |
| assert response.body["path"] == "/missing" |
| assert "POST /api/run-agent" in response.body["endpoints"] |
|
|
