feat(cloud-api): expose health readiness

2026-07-12 18:33:45 +08:00
parent b8d02abf87
commit a5de7399f8
3 changed files with 112 additions and 2 deletions
@@ -182,6 +182,79 @@ def test_lifecycle_workers_log_failures_and_continue(monkeypatch) -> None:
    }


+def test_liveness_stays_up_when_database_readiness_fails() -> None:
+    database_healthy = True
+
+    class FakeRepository:
+        def health_check(self) -> None:
+            if not database_healthy:
+                raise RuntimeError("database unavailable with secret credentials")
+
+        def list_queued_tasks(self) -> list[object]:
+            return []
+
+        def reap_expired_leases(self, *, now, max_attempts: int) -> list[str]:
+            return []
+
+    class FakeDatabase:
+        repository = FakeRepository()
+
+        def close(self) -> None:
+            return None
+
+    app = create_app(
+        config=CloudControlConfig(database_url="sqlite:///:memory:"),
+        database_factory=lambda _config: FakeDatabase(),  # type: ignore[arg-type,return-value]
+    )
+
+    with TestClient(app) as client:
+        assert client.get("/health/live").json() == {"status": "live"}
+        ready = client.get("/health/ready")
+        assert ready.status_code == 200
+        assert all(ready.json()["checks"].values())
+
+        database_healthy = False
+        assert client.get("/health/live").status_code == 200
+        not_ready = client.get("/health/ready")
+        assert not_ready.status_code == 503
+        assert not_ready.json()["checks"]["database"] is False
+        assert "secret credentials" not in not_ready.text
+
+
+def test_readiness_reports_stopped_worker() -> None:
+    class FakeRepository:
+        def health_check(self) -> None:
+            return None
+
+        def list_queued_tasks(self) -> list[object]:
+            return []
+
+        def reap_expired_leases(self, *, now, max_attempts: int) -> list[str]:
+            return []
+
+    class FakeDatabase:
+        repository = FakeRepository()
+
+        def close(self) -> None:
+            return None
+
+    class StoppedWorker:
+        def done(self) -> bool:
+            return True
+
+    app = create_app(
+        config=CloudControlConfig(database_url="sqlite:///:memory:"),
+        database_factory=lambda _config: FakeDatabase(),  # type: ignore[arg-type,return-value]
+    )
+
+    with TestClient(app) as client:
+        app.state.worker_tasks = (StoppedWorker(),)
+        response = client.get("/health/ready")
+
+        assert response.status_code == 503
+        assert response.json()["checks"]["workers"] is False
+
+
 def test_production_app_rejects_missing_credentials() -> None:
    with pytest.raises(CloudConfigurationError, match="credential"):
        create_app(