Files
cleveragents-core/features/steps/server_lifecycle_steps.py
T
freemo bb1dc17b08
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 37s
CI / lint (pull_request) Successful in 4m54s
CI / quality (pull_request) Successful in 5m42s
CI / typecheck (pull_request) Successful in 5m46s
CI / security (pull_request) Successful in 5m51s
CI / benchmark-regression (pull_request) Has started running
CI / integration_tests (pull_request) Successful in 8m7s
CI / unit_tests (pull_request) Successful in 9m8s
CI / e2e_tests (pull_request) Successful in 10m12s
CI / docker (pull_request) Successful in 1m8s
CI / coverage (pull_request) Successful in 12m22s
CI / status-check (pull_request) Successful in 2s
feat(server): entity sync (_cleveragents/sync/*)
Implement entity synchronization between client and server using
the _cleveragents/sync/* A2A extension methods per the specification.

Sync models (src/cleveragents/a2a/sync_models.py):
- Pydantic v2 models for pull/push/status requests and responses
- VectorClock with increment, merge, happens-before, and concurrency
  detection for distributed conflict detection
- SyncEntitySnapshot, SyncConflict, SyncState, SyncQueueEntry models
- SyncEntityType enum: actor, skill, action, plan, session
- ConflictResolution enum: last_writer_wins, manual, server_wins,
  client_wins

SyncService (src/cleveragents/application/services/sync_service.py):
- pull(): download server namespace entities to local cache with
  incremental sync (since timestamps) and force-overwrite option
- push(): push local entity definitions to server namespace with
  configurable conflict resolution strategy
- status(): compare local and server entity versions, detect drift,
  report local-ahead, server-ahead, and concurrent conflicts
- resolve_conflict(): manually resolve detected conflicts with
  winner selection (local or server)
- Offline queue: enqueue_offline() and process_offline_queue() for
  retry on reconnect, with max-retries enforcement
- The local/ namespace is never synced per specification

A2A facade (src/cleveragents/a2a/facade.py):
- Replaced sync stub handlers with real _handle_sync_pull,
  _handle_sync_push, _handle_sync_status routing to SyncService
- Added sync_service property and TYPE_CHECKING import
- Facade returns stubs when no SyncService is registered

Tests:
- Behave BDD: features/entity_sync.feature (65 scenarios, 247 steps)
  covering models, vector clocks, pull/push/status, conflict resolution,
  offline queue, facade integration, constructor validation, edge cases
- Robot Framework: robot/entity_sync.robot (8 integration tests)

ISSUES CLOSED: #866
2026-03-23 22:50:56 +00:00

287 lines
9.3 KiB
Python

"""Step definitions for ``server_lifecycle.feature``.
Tests the ASGI application factory, health/agent-card endpoints,
A2A JSON-RPC dispatch, ServerLifecycle construction and shutdown,
and Settings-based server configuration.
"""
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock
from behave import given, then, use_step_matcher, when # type: ignore[import-untyped]
from cleveragents.a2a.facade import A2aLocalFacade
from cleveragents.config.settings import Settings
from cleveragents.infrastructure.server.asgi_app import create_asgi_app
from cleveragents.infrastructure.server.server_lifecycle import ServerLifecycle
# ---------------------------------------------------------------------------
# ASGI application creation
# ---------------------------------------------------------------------------
@when("I create an ASGI application with default settings")
def step_create_asgi_default(context: Any) -> None:
context.asgi_app = create_asgi_app()
@given("an A2aLocalFacade with no services")
def step_given_facade(context: Any) -> None:
context.facade = A2aLocalFacade()
@when("I create an ASGI application with that facade")
def step_create_asgi_with_facade(context: Any) -> None:
context.asgi_app = create_asgi_app(facade=context.facade)
@then("the ASGI app should be a FastAPI instance")
def step_check_fastapi_instance(context: Any) -> None:
from fastapi import FastAPI
assert isinstance(context.asgi_app, FastAPI), (
f"Expected FastAPI, got {type(context.asgi_app)}"
)
@when('I try to create an ASGI application with facade "{value}"')
def step_create_asgi_bad_facade(context: Any, value: str) -> None:
context.caught_exception = None
try:
create_asgi_app(facade=value) # type: ignore[arg-type]
except TypeError as exc:
context.caught_exception = exc
@then("a TypeError should be raised for invalid facade")
def step_check_type_error_facade(context: Any) -> None:
assert context.caught_exception is not None, "Expected TypeError"
assert isinstance(context.caught_exception, TypeError)
@when("I try to create an ASGI application with port {port:d}")
def step_create_asgi_bad_port(context: Any, port: int) -> None:
context.caught_exception = None
try:
create_asgi_app(port=port)
except ValueError as exc:
context.caught_exception = exc
@then("a ValueError should be raised for invalid port")
def step_check_value_error_port(context: Any) -> None:
assert context.caught_exception is not None, "Expected ValueError"
assert isinstance(context.caught_exception, ValueError)
use_step_matcher("re")
@when('I try to create an ASGI application with host "(?P<host>[^"]*)"')
def step_create_asgi_bad_host(context: Any, host: str) -> None:
context.caught_exception = None
try:
create_asgi_app(host=host)
except ValueError as exc:
context.caught_exception = exc
use_step_matcher("parse")
@then("a ValueError should be raised for invalid host")
def step_check_value_error_host(context: Any) -> None:
assert context.caught_exception is not None, "Expected ValueError"
assert isinstance(context.caught_exception, ValueError)
# ---------------------------------------------------------------------------
# Test client helpers
# ---------------------------------------------------------------------------
@given("a running ASGI test client")
def step_given_test_client(context: Any) -> None:
from fastapi.testclient import TestClient
app = create_asgi_app()
context.test_client = TestClient(app)
@when("I request GET /health")
def step_get_health(context: Any) -> None:
context.response = context.test_client.get("/health")
@when("I request GET /.well-known/agent.json")
def step_get_agent_card(context: Any) -> None:
context.response = context.test_client.get("/.well-known/agent.json")
@when('I POST a JSON-RPC request to /a2a with operation "{operation}"')
def step_post_a2a(context: Any, operation: str) -> None:
payload = {"operation": operation, "params": {}}
context.response = context.test_client.post("/a2a", json=payload)
@when("I POST a malformed JSON body to /a2a")
def step_post_a2a_malformed(context: Any) -> None:
# Send a body that cannot be parsed into A2aRequest (missing operation)
context.response = context.test_client.post("/a2a", json={"bad": "data"})
@then("the response status code should be {code:d}")
def step_check_status_code(context: Any, code: int) -> None:
assert context.response.status_code == code, (
f"Expected {code}, got {context.response.status_code}: {context.response.text}"
)
@then('the response body should contain "{text}"')
def step_check_response_body(context: Any, text: str) -> None:
body = context.response.text
assert text in body, f"Expected '{text}' in response body: {body}"
@then('the A2A response status should be "{expected_status}"')
def step_check_a2a_status(context: Any, expected_status: str) -> None:
data = context.response.json()
assert data.get("status") == expected_status, (
f"Expected A2A status '{expected_status}', got {data}"
)
# ---------------------------------------------------------------------------
# ServerLifecycle construction
# ---------------------------------------------------------------------------
@when('I create a ServerLifecycle with host "{host}" and port {port:d}')
def step_create_lifecycle(context: Any, host: str, port: int) -> None:
context.lifecycle = ServerLifecycle(host=host, port=port)
@then('the lifecycle host should be "{expected}"')
def step_check_lifecycle_host(context: Any, expected: str) -> None:
assert context.lifecycle.host == expected
@then("the lifecycle port should be {expected:d}")
def step_check_lifecycle_port(context: Any, expected: int) -> None:
assert context.lifecycle.port == expected
@then("the lifecycle should not be started")
def step_check_not_started(context: Any) -> None:
assert not context.lifecycle.is_started
@then("the lifecycle should not be stopped")
def step_check_not_stopped(context: Any) -> None:
assert not context.lifecycle.is_stopped
use_step_matcher("re")
@when('I try to create a ServerLifecycle with host "(?P<host>[^"]*)"')
def step_create_lifecycle_bad_host(context: Any, host: str) -> None:
context.caught_exception = None
try:
ServerLifecycle(host=host)
except ValueError as exc:
context.caught_exception = exc
use_step_matcher("parse")
@when("I try to create a ServerLifecycle with port {port:d}")
def step_create_lifecycle_bad_port(context: Any, port: int) -> None:
context.caught_exception = None
try:
ServerLifecycle(port=port)
except ValueError as exc:
context.caught_exception = exc
@when("I try to create a ServerLifecycle with empty log_level")
def step_create_lifecycle_bad_log_level(context: Any) -> None:
context.caught_exception = None
try:
ServerLifecycle(log_level="")
except ValueError as exc:
context.caught_exception = exc
@then("a ValueError should be raised for invalid log_level")
def step_check_value_error_log_level(context: Any) -> None:
assert context.caught_exception is not None, "Expected ValueError"
assert isinstance(context.caught_exception, ValueError)
# ---------------------------------------------------------------------------
# Settings
# ---------------------------------------------------------------------------
@then('the default server host from Settings should be "{expected}"')
def step_check_settings_host(context: Any, expected: str) -> None:
settings = Settings()
assert settings.server_host == expected
@then("the default server port from Settings should be {expected:d}")
def step_check_settings_port(context: Any, expected: int) -> None:
settings = Settings()
assert settings.server_port == expected
# ---------------------------------------------------------------------------
# Graceful shutdown
# ---------------------------------------------------------------------------
@given("a ServerLifecycle with a mock uvicorn server")
def step_lifecycle_with_mock_server(context: Any) -> None:
lifecycle = ServerLifecycle(host="127.0.0.1", port=9999)
mock_server = MagicMock()
mock_server.should_exit = False
lifecycle._server = mock_server
context.lifecycle = lifecycle
context.mock_server = mock_server
@when("I call request_shutdown on the lifecycle")
def step_call_request_shutdown(context: Any) -> None:
context.lifecycle.request_shutdown()
@then("the mock server should_exit flag should be true")
def step_check_should_exit(context: Any) -> None:
assert context.mock_server.should_exit is True
@given("a ServerLifecycle that has already been started")
def step_lifecycle_already_started(context: Any) -> None:
lifecycle = ServerLifecycle(host="127.0.0.1", port=9998)
lifecycle._started = True
context.lifecycle = lifecycle
@when("I try to start the lifecycle again")
def step_try_double_start(context: Any) -> None:
context.caught_exception = None
try:
context.lifecycle.start()
except RuntimeError as exc:
context.caught_exception = exc
@then("a RuntimeError should be raised for double start")
def step_check_runtime_error(context: Any) -> None:
assert context.caught_exception is not None, "Expected RuntimeError"
assert isinstance(context.caught_exception, RuntimeError)