awoooi/scripts/ops/tests/test_backup_health_textfile_exporter.py

from __future__ import annotations

import importlib.util
import sys
from pathlib import Path


SCRIPT_ROOT = Path(__file__).resolve().parents[1]
EXPORTER_PATH = SCRIPT_ROOT / "backup-health-textfile-exporter.py"


def load_exporter():
    spec = importlib.util.spec_from_file_location("backup_health_textfile_exporter", EXPORTER_PATH)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def test_default_gitea_bundle_expected_repos_cover_current_project_set() -> None:
    exporter = load_exporter()

    assert exporter.DEFAULT_GITEA_BUNDLE_EXPECTED_REPOS == [
        "wooo/awoooi",
        "wooo/ewoooc",
        "wooo/2026FIFAWorldCup",
        "wooo/agent-bounty-protocol",
        "wooo/AwoooGo",
        "wooo/stockplatform-v2",
        "wooo/vibework",
        "wooo/momo-pro-system",
        "wooo/tsenyang-website",
        "wooo/clawbot-v5",
        "wooo/bitan-pharmacy",
        "wooo/vtuber",
    ]


def test_gitea_bundle_metrics_require_all_expected_repos(tmp_path: Path, monkeypatch) -> None:
    exporter = load_exporter()
    bundle_root = tmp_path / "latest-private-complete"
    bundle_root.mkdir()
    monkeypatch.setattr(exporter, "GITEA_BUNDLE_ROOT", bundle_root)
    monkeypatch.setattr(exporter, "GITEA_BUNDLE_MAX_AGE_HOURS", 25)
    monkeypatch.setenv("AIOPS_GITEA_BUNDLE_EXPECTED_REPOS", "wooo/awoooi,wooo/tsenyang-website")

    (bundle_root / "manifest.remote.tsv").write_text(
        "\n".join(
            [
                "repo\tstatus\thead_count\tbundle\tchecksum",
                f"wooo/awoooi\tok\t2\t{bundle_root / 'wooo__awoooi.bundle'}\t{bundle_root / 'wooo__awoooi.bundle.sha256'}",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (bundle_root / "wooo__awoooi.bundle").write_text("bundle", encoding="utf-8")
    (bundle_root / "wooo__awoooi.bundle.sha256").write_text(
        "a" * 64 + "  wooo__awoooi.bundle\n",
        encoding="utf-8",
    )
    (bundle_root / "tsenyang-website-local-20260701.bundle").write_text("bundle", encoding="utf-8")
    (bundle_root / "tsenyang-website-local-20260701.bundle.sha256").write_text(
        "b" * 64 + "  tsenyang-website-local-20260701.bundle\n",
        encoding="utf-8",
    )

    metrics, all_ok = exporter._gitea_bundle_metric_lines("188")

    rendered = "\n".join(metrics)
    assert all_ok == 1
    assert 'awoooi_gitea_bundle_expected_repo_count{host="188"' in rendered
    assert 'awoooi_gitea_bundle_expected_repo_missing_count{host="188"' in rendered
    assert 'awoooi_gitea_bundle_failed_repo_count{host="188"' in rendered
    assert 'repo="wooo/tsenyang-website",status="bundle_file_only"' in rendered
    assert 'awoooi_gitea_bundle_all_expected_ok{host="188"' in rendered
    assert rendered.rstrip().endswith(" 1")


def test_gitea_bundle_metrics_fail_when_checksum_missing(tmp_path: Path, monkeypatch) -> None:
    exporter = load_exporter()
    bundle_root = tmp_path / "latest-private-complete"
    bundle_root.mkdir()
    monkeypatch.setattr(exporter, "GITEA_BUNDLE_ROOT", bundle_root)
    monkeypatch.setenv("AIOPS_GITEA_BUNDLE_EXPECTED_REPOS", "wooo/awoooi")

    (bundle_root / "manifest.remote.tsv").write_text(
        "\n".join(
            [
                "repo\tstatus\thead_count\tbundle\tchecksum",
                f"wooo/awoooi\tok\t2\t{bundle_root / 'wooo__awoooi.bundle'}\t{bundle_root / 'wooo__awoooi.bundle.sha256'}",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (bundle_root / "wooo__awoooi.bundle").write_text("bundle", encoding="utf-8")

    metrics, all_ok = exporter._gitea_bundle_metric_lines("188")

    rendered = "\n".join(metrics)
    assert all_ok == 0
    assert 'awoooi_gitea_bundle_checksum_missing_count{host="188"' in rendered
    assert rendered.rstrip().endswith(" 0")


def test_dr_phase_does_not_regress_when_full_offsite_is_fresh_and_partial_is_stale(
    tmp_path: Path, monkeypatch
) -> None:
    exporter = load_exporter()
    offsite_dir = tmp_path / "offsite"
    escrow_dir = tmp_path / "escrow"
    offsite_dir.mkdir()
    escrow_dir.mkdir()
    now = 1_782_900_000

    monkeypatch.setattr(exporter, "OFFSITE_STATUS_DIR", offsite_dir)
    monkeypatch.setattr(exporter, "ESCROW_EVIDENCE_DIR", escrow_dir)
    monkeypatch.setattr(exporter.time, "time", lambda: now)
    monkeypatch.setattr(exporter, "_b2_configured", lambda: False)
    monkeypatch.setattr(exporter, "_rclone_configured", lambda: True)
    (offsite_dir / "rclone-last-success").write_text(str(now - 3600), encoding="utf-8")
    (offsite_dir / "rclone-partial-last-success").write_text(str(now - 72 * 3600), encoding="utf-8")

    metrics = exporter._offsite_and_escrow_metric_lines("110")
    rendered = "\n".join(metrics)

    assert 'awoooi_backup_offsite_fresh{host="110",provider="rclone",max_age_hours="48"} 1' in rendered
    assert (
        'awoooi_backup_offsite_partial_fresh{host="110",provider="rclone",scope="partial",max_age_hours="48"} 0'
        in rendered
    )
    assert 'awoooi_backup_dr_credential_escrow_missing_count{host="110"} 5' in rendered
    assert 'awoooi_backup_dr_phase{host="110",next_step="complete_credential_escrow_review"} 3' in rendered


def test_dr_phase_ready_when_full_offsite_is_fresh_and_escrow_is_complete(
    tmp_path: Path, monkeypatch
) -> None:
    exporter = load_exporter()
    offsite_dir = tmp_path / "offsite"
    escrow_dir = tmp_path / "escrow"
    offsite_dir.mkdir()
    escrow_dir.mkdir()
    now = 1_782_900_000

    monkeypatch.setattr(exporter, "OFFSITE_STATUS_DIR", offsite_dir)
    monkeypatch.setattr(exporter, "ESCROW_EVIDENCE_DIR", escrow_dir)
    monkeypatch.setattr(exporter.time, "time", lambda: now)
    monkeypatch.setattr(exporter, "_b2_configured", lambda: False)
    monkeypatch.setattr(exporter, "_rclone_configured", lambda: True)
    (offsite_dir / "rclone-last-success").write_text(str(now - 3600), encoding="utf-8")
    (offsite_dir / "rclone-partial-last-success").write_text(str(now - 72 * 3600), encoding="utf-8")
    for item in exporter.ESCROW_ITEMS:
        (escrow_dir / f"{item}.last_verified").write_text(str(now - 60), encoding="utf-8")

    metrics = exporter._offsite_and_escrow_metric_lines("110")
    rendered = "\n".join(metrics)

    assert 'awoooi_backup_offsite_fresh{host="110",provider="rclone",max_age_hours="48"} 1' in rendered
    assert (
        'awoooi_backup_offsite_partial_fresh{host="110",provider="rclone",scope="partial",max_age_hours="48"} 0'
        in rendered
    )
    assert 'awoooi_backup_dr_credential_escrow_missing_count{host="110"} 0' in rendered
    assert 'awoooi_backup_dr_phase{host="110",next_step="offsite_and_escrow_ready"} 5' in rendered