fix(gitea): expose bundle backup restore readback

apps/api/tests/test_gitea_repo_bundle_backup_readback_api.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from src.api.v1 import agents
+from src.api.v1.agents import router
+from src.services.gitea_repo_bundle_backup_readback import (
+    build_gitea_repo_bundle_backup_readback,
+)
+
+
+def _sample(name: str, value: float, **labels: str) -> dict:
+    merged = {"host": "188", **labels}
+    return {"name": name, "labels": merged, "value": value}
+
+
+def _fresh_repo_samples() -> list[dict]:
+    return [
+        _sample("awoooi_gitea_bundle_expected_repo_count", 2),
+        _sample("awoooi_gitea_bundle_expected_repo_missing_count", 0),
+        _sample("awoooi_gitea_bundle_failed_repo_count", 0),
+        _sample("awoooi_gitea_bundle_checksum_missing_count", 0),
+        _sample("awoooi_gitea_bundle_age_seconds", 600),
+        _sample("awoooi_gitea_bundle_fresh", 1),
+        _sample("awoooi_gitea_bundle_all_expected_ok", 1),
+        _sample("awoooi_gitea_bundle_repo_present", 1, repo="wooo/awoooi"),
+        _sample("awoooi_gitea_bundle_repo_ok", 1, repo="wooo/awoooi"),
+        _sample("awoooi_gitea_bundle_repo_head_count", 196, repo="wooo/awoooi"),
+        _sample("awoooi_gitea_bundle_checksum_digest_present", 1, repo="wooo/awoooi"),
+        _sample("awoooi_gitea_bundle_repo_present", 1, repo="wooo/tsenyang-website"),
+        _sample("awoooi_gitea_bundle_repo_ok", 1, repo="wooo/tsenyang-website"),
+        _sample("awoooi_gitea_bundle_repo_head_count", 2, repo="wooo/tsenyang-website"),
+        _sample(
+            "awoooi_gitea_bundle_checksum_digest_present",
+            1,
+            repo="wooo/tsenyang-website",
+        ),
+    ]
+
+
+def test_gitea_repo_bundle_readback_separates_stale_from_missing_repos() -> None:
+    samples = _fresh_repo_samples()
+    for sample in samples:
+        if sample["name"] == "awoooi_gitea_bundle_age_seconds":
+            sample["value"] = 116_157
+        if sample["name"] == "awoooi_gitea_bundle_fresh":
+            sample["value"] = 0
+        if sample["name"] == "awoooi_gitea_bundle_all_expected_ok":
+            sample["value"] = 0
+
+    payload = build_gitea_repo_bundle_backup_readback(metric_samples=samples)
+
+    assert payload["status"] == "blocked_gitea_bundle_backup_stale"
+    assert payload["summary"]["expected_repo_missing_count"] == 0
+    assert payload["summary"]["failed_repo_count"] == 0
+    assert payload["summary"]["checksum_missing_count"] == 0
+    assert "gitea_bundle_backup_stale" in payload["active_blockers"]
+    assert "gitea_bundle_expected_repo_missing" not in payload["active_blockers"]
+    assert "gitea_bundle_sample_restore_dry_run_missing" in payload["active_blockers"]
+    assert payload["repos"][0]["repo"] == "wooo/awoooi"
+
+
+def test_gitea_repo_bundle_readback_ready_requires_restore_dry_run_metric() -> None:
+    samples = _fresh_repo_samples()
+    samples.append(_sample("awoooi_gitea_bundle_sample_restore_dry_run_ok", 1))
+
+    payload = build_gitea_repo_bundle_backup_readback(metric_samples=samples)
+
+    assert payload["status"] == "gitea_repo_bundle_backup_readback_ready"
+    assert payload["ready"] is True
+    assert payload["active_blockers"] == []
+    assert payload["restore_dry_run"]["production_restore_performed"] is False
+
+
+def test_gitea_repo_bundle_readback_endpoint_returns_payload(monkeypatch) -> None:
+    def fake_loader() -> dict:
+        return build_gitea_repo_bundle_backup_readback(
+            metric_samples=_fresh_repo_samples(),
+            generated_at="2026-07-02T20:00:00+08:00",
+        )
+
+    monkeypatch.setattr(
+        agents,
+        "load_latest_gitea_repo_bundle_backup_readback",
+        fake_loader,
+    )
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+    client = TestClient(app)
+
+    response = client.get("/api/v1/agents/gitea-repo-bundle-backup-readback")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["schema_version"] == "gitea_repo_bundle_backup_readback_v1"
+    assert payload["status"] == "blocked_gitea_bundle_restore_dry_run_missing"
+    assert payload["operation_boundaries"]["restore_to_production_allowed"] is False