HAL9000
5ede1c018a
The new entity-sync feature introduced A2aRequest with a `method` field (JSON-RPC 2.0 wire format), but several test/helper files used the wrong field name `operation` when constructing requests or logging, and checked non-existent `.status`/`.data` attributes on A2aResponse instead of `.result`/`.error`. Fixes: - asgi_app.py: log `a2a_request.method`, not `.operation` (typecheck error) - server_lifecycle_steps.py: send `method` key in JSON-RPC payload; look up status in `result` dict, not top-level response body - server_lifecycle.feature: expect "healthy" (what the handler returns), not "ok" - entity_sync_steps.py: construct A2aRequest(method=...) not (operation=...); assert on .result/.error instead of .status/.data - robot/helper_entity_sync.py: same method= and result/error fixes across sync_pull, sync_push, sync_status, sync_facade_no_service helpers - robot/helper_server_lifecycle.py: same method= and result path fixes ISSUES CLOSED: #1125
118 lines
3.4 KiB
Python
118 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Integration test helper for server lifecycle tests.
|
|
|
|
Each sub-command exercises a real import / instantiation path and prints
|
|
a unique sentinel on success. Robot tests parse the output for that
|
|
sentinel.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
|
|
|
|
def _test_create_app() -> None:
|
|
"""Verify ``create_asgi_app`` returns a FastAPI instance."""
|
|
from cleveragents.infrastructure.server.asgi_app import create_asgi_app
|
|
|
|
app = create_asgi_app()
|
|
assert app is not None
|
|
print("create-app-ok")
|
|
|
|
|
|
def _test_health_endpoint() -> None:
|
|
"""Exercise the /health endpoint via FastAPI TestClient."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
from cleveragents.infrastructure.server.asgi_app import create_asgi_app
|
|
|
|
app = create_asgi_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/health")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["status"] == "healthy"
|
|
print("health-endpoint-ok")
|
|
|
|
|
|
def _test_agent_card() -> None:
|
|
"""Exercise the /.well-known/agent.json endpoint."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
from cleveragents.infrastructure.server.asgi_app import create_asgi_app
|
|
|
|
app = create_asgi_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/.well-known/agent.json")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["name"] == "CleverAgents"
|
|
assert "url" in data
|
|
print("agent-card-ok")
|
|
|
|
|
|
def _test_a2a_dispatch() -> None:
|
|
"""Exercise the /a2a POST endpoint with a health check operation."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
from cleveragents.infrastructure.server.asgi_app import create_asgi_app
|
|
|
|
app = create_asgi_app()
|
|
client = TestClient(app)
|
|
payload = {"jsonrpc": "2.0", "method": "_cleveragents/health/check", "params": {}}
|
|
resp = client.post("/a2a", json=payload)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["result"]["status"] == "healthy"
|
|
print("a2a-dispatch-ok")
|
|
|
|
|
|
def _test_lifecycle_construction() -> None:
|
|
"""Verify ServerLifecycle can be constructed and inspected."""
|
|
from cleveragents.infrastructure.server.server_lifecycle import ServerLifecycle
|
|
|
|
lc = ServerLifecycle(host="127.0.0.1", port=9123)
|
|
assert lc.host == "127.0.0.1"
|
|
assert lc.port == 9123
|
|
assert not lc.is_started
|
|
assert not lc.is_stopped
|
|
print("lifecycle-construction-ok")
|
|
|
|
|
|
def _test_lifecycle_shutdown() -> None:
|
|
"""Verify request_shutdown sets the should_exit flag on a mock server."""
|
|
from unittest.mock import MagicMock
|
|
|
|
from cleveragents.infrastructure.server.server_lifecycle import ServerLifecycle
|
|
|
|
lc = ServerLifecycle(host="127.0.0.1", port=9124)
|
|
mock_server = MagicMock()
|
|
mock_server.should_exit = False
|
|
lc._server = mock_server
|
|
lc.request_shutdown()
|
|
assert mock_server.should_exit is True
|
|
print("lifecycle-shutdown-ok")
|
|
|
|
|
|
_COMMANDS: dict[str, object] = {
|
|
"create-app": _test_create_app,
|
|
"health-endpoint": _test_health_endpoint,
|
|
"agent-card": _test_agent_card,
|
|
"a2a-dispatch": _test_a2a_dispatch,
|
|
"lifecycle-construction": _test_lifecycle_construction,
|
|
"lifecycle-shutdown": _test_lifecycle_shutdown,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
|
|
print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
handler = _COMMANDS[sys.argv[1]]
|
|
handler() # type: ignore[operator]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|