From 1e10d5aa953d2cc98a799b9ad5fa329e3f01cb61 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Sun, 12 Jul 2026 18:40:29 +0800 Subject: [PATCH] test(cloud-api): verify restart recovery --- apps/cloud-api/tests/test_app.py | 157 ++++++++++++++++++ .../cloud-control-plane-integration/tasks.md | 2 +- 2 files changed, 158 insertions(+), 1 deletion(-) diff --git a/apps/cloud-api/tests/test_app.py b/apps/cloud-api/tests/test_app.py index 1c5db6c..5f4ce90 100644 --- a/apps/cloud-api/tests/test_app.py +++ b/apps/cloud-api/tests/test_app.py @@ -1,6 +1,7 @@ from __future__ import annotations import time +from datetime import timedelta import pytest from fastapi.testclient import TestClient @@ -8,6 +9,10 @@ from fastapi.testclient import TestClient import cloud_api.app as app_module from cloud_api.app import create_app from cloud.control_config import CloudConfigurationError, CloudControlConfig +from cloud.database import CloudDatabase +from cloud.pool import PooledDevice +from cloud.scheduler import ScheduledTask, TaskConstraints +from core.models import utc_now def test_create_app_returns_independent_cloud_application() -> None: @@ -292,6 +297,158 @@ def test_request_correlation_id_is_propagated_without_sensitive_headers( assert "never-log-this" not in repr(request_logs) +def test_startup_database_failure_closes_database() -> None: + closed = False + + class FailingRepository: + def health_check(self) -> None: + raise RuntimeError("database unavailable") + + class FailingDatabase: + repository = FailingRepository() + + def close(self) -> None: + nonlocal closed + closed = True + + app = create_app( + config=CloudControlConfig(database_url="sqlite:///:memory:"), + database_factory=lambda _config: FailingDatabase(), # type: ignore[arg-type,return-value] + ) + + with pytest.raises(RuntimeError, match="database unavailable"): + with TestClient(app): + pass + assert closed is True + + +def test_persisted_queue_is_recovered_after_control_plane_restart(tmp_path) -> None: + database_url = f"sqlite:///{(tmp_path / 'queue-recovery.sqlite3').as_posix()}" + seed = CloudDatabase(database_url) + now = utc_now() + seed.repository.upsert_host("host-a", address=None, last_seen_at=now) + seed.repository.replace_host_devices( + "host-a", + [ + PooledDevice( + device_id="device-a", + host_id="host-a", + driver_type="wda", + status="idle", + synced_at=now, + ) + ], + ) + seed.repository.enqueue_task( + ScheduledTask( + id="persisted-task", + goal="resume queued work", + workflow_definition_id=None, + constraints=TaskConstraints(), + created_at=now, + ) + ) + seed.close() + + app = create_app( + config=CloudControlConfig( + database_url=database_url, + scheduler_interval_seconds=0.01, + lease_reaper_interval_seconds=1, + ) + ) + with TestClient(app): + assert _wait_until( + lambda: ( + app.state.cloud_services.repository.get_task("persisted-task").status + == "assigned" + ) + ) + + verification = CloudDatabase(database_url) + try: + task = verification.repository.get_task("persisted-task") + assert task is not None + assert task.status == "assigned" + assert task.attempt_count == 1 + finally: + verification.close() + + +def test_expired_lease_is_recovered_after_control_plane_restart(tmp_path) -> None: + database_url = f"sqlite:///{(tmp_path / 'lease-recovery.sqlite3').as_posix()}" + seed = CloudDatabase(database_url) + now = utc_now() + seed.repository.upsert_host("host-a", address=None, last_seen_at=now) + seed.repository.replace_host_devices( + "host-a", + [ + PooledDevice( + device_id="device-a", + host_id="host-a", + driver_type="wda", + status="idle", + synced_at=now, + ) + ], + ) + seed.repository.enqueue_task( + ScheduledTask( + id="expired-task", + goal="recover expired work", + workflow_definition_id=None, + constraints=TaskConstraints(), + created_at=now, + ) + ) + seed.repository.assign_task( + task_id="expired-task", + host_id="host-a", + device_id="device-a", + lease_id="expired-lease", + lease_expires_at=now - timedelta(seconds=1), + now=now - timedelta(minutes=1), + ) + seed.close() + + app = create_app( + config=CloudControlConfig( + database_url=database_url, + scheduler_interval_seconds=1, + lease_reaper_interval_seconds=0.01, + max_task_attempts=1, + ) + ) + with TestClient(app): + assert _wait_until( + lambda: ( + app.state.cloud_services.repository.get_task("expired-task").status + == "failed" + ) + ) + + verification = CloudDatabase(database_url) + try: + task = verification.repository.get_task("expired-task") + assert task is not None + assert task.status == "failed" + assert task.failure_reason == "lease expired after 1 attempts" + assert verification.repository.list_task_attempts("expired-task")[0].status == ( + "expired" + ) + finally: + verification.close() + + +def _wait_until(predicate, timeout_seconds: float = 1.0) -> bool: + deadline = time.monotonic() + timeout_seconds + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return False + + def test_production_app_rejects_missing_credentials() -> None: with pytest.raises(CloudConfigurationError, match="credential"): create_app( diff --git a/openspec/changes/cloud-control-plane-integration/tasks.md b/openspec/changes/cloud-control-plane-integration/tasks.md index 9194c38..7bd991b 100644 --- a/openspec/changes/cloud-control-plane-integration/tasks.md +++ b/openspec/changes/cloud-control-plane-integration/tasks.md @@ -48,7 +48,7 @@ - [x] 6.3 Ensure lifecycle iteration failures are logged and retried without terminating later iterations. - [x] 6.4 Add `/health/live` and `/health/ready` with separate process, database/schema, and worker-state semantics. - [x] 6.5 Add structured correlation-aware logging for requests and task lifecycle events with sensitive payload redaction. -- [ ] 6.6 Add app-level tests for startup failures, readiness transitions, graceful shutdown, persisted queue recovery, and expired-lease recovery. +- [x] 6.6 Add app-level tests for startup failures, readiness transitions, graceful shutdown, persisted queue recovery, and expired-lease recovery. ## 7. Device Host Agent Execution Loop