File size: 4,597 Bytes
a4514b4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | 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"]
|