HAL9000
9c3b0331a2
- Fix asgi_app.py: use a2a_request.method instead of a2a_request.operation (A2aRequest model was updated in master to use JSON-RPC 2.0 field names) - Fix server_lifecycle_steps.py: send JSON-RPC 2.0 wire format with "method" field; check result.status instead of top-level status - Fix server_lifecycle.feature: update step name to match new step definition (A2A response result status) - Fix helper_server_lifecycle.py: use "method" field in JSON-RPC 2.0 payload; check result.status in response
126 lines
3.7 KiB
Python
126 lines
3.7 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
|
|
from collections.abc import Callable
|
|
|
|
|
|
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)
|
|
# JSON-RPC 2.0 wire format: use "method" field
|
|
payload = {
|
|
"jsonrpc": "2.0",
|
|
"method": "_cleveragents/health/check",
|
|
"params": {},
|
|
}
|
|
resp = client.post("/a2a", json=payload)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
# JSON-RPC 2.0 response: result is nested under "result"
|
|
result = data.get("result") or {}
|
|
assert result.get("status") == "healthy", f"Unexpected response: {data}"
|
|
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, Callable[[], None]] = {
|
|
"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()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|