awoooi/apps/api/tests/test_generate_monitoring_stabilization.py

from __future__ import annotations

import importlib.util
import sys
import unittest
from pathlib import Path

SCRIPT_PATH = Path(__file__).resolve().parents[3] / "scripts" / "generate_monitoring.py"
SPEC = importlib.util.spec_from_file_location("generate_monitoring", SCRIPT_PATH)
generate_monitoring = importlib.util.module_from_spec(SPEC)
assert SPEC and SPEC.loader
sys.dont_write_bytecode = True
SPEC.loader.exec_module(generate_monitoring)


def targets_payload(
    down_jobs: set[str] | None = None,
    missing_jobs: set[str] | None = None,
    down_targets: set[str] | None = None,
    missing_targets: set[str] | None = None,
    duplicate_targets: set[str] | None = None,
):
    down_jobs = down_jobs or set()
    missing_jobs = missing_jobs or set()
    down_targets = down_targets or set()
    missing_targets = missing_targets or set()
    duplicate_targets = duplicate_targets or set()

    active_targets = []
    for job in generate_monitoring.EXPECTED_JOBS:
        if job in missing_jobs:
            continue
        active_targets.append(
            {
                "labels": {"job": job, "instance": f"{job}:1"},
                "health": "down" if job in down_jobs else "up",
            }
        )

    for target_id, expected in generate_monitoring.EXPECTED_TARGETS.items():
        if target_id in missing_targets:
            continue
        target = {
            "labels": dict(expected["labels"]),
            "health": "down" if target_id in down_targets else "up",
        }
        active_targets.append(target)
        if target_id in duplicate_targets:
            active_targets.append(dict(target))

    return {"activeTargets": active_targets}


class GenerateMonitoringStabilizationTest(unittest.TestCase):
    def test_stabilized_report_uses_later_clean_target_snapshot(self):
        snapshots = [
            targets_payload(down_jobs={"awoooi-api"}),
            targets_payload(),
        ]

        def fetch_targets():
            return snapshots.pop(0)

        report = generate_monitoring.build_stabilized_report(
            fetch_targets,
            attempts=3,
            sleep_seconds=0,
            emit_status=False,
        )

        self.assertEqual(report["summary"]["real_down_jobs"], 0)
        self.assertEqual(report["stabilization"]["attempt"], 2)
        self.assertEqual(report["stabilization"]["status"], "cleared")

    def test_stabilized_report_retries_missing_expected_target(self):
        snapshots = [
            targets_payload(missing_jobs={"awoooi-api"}),
            targets_payload(),
        ]

        def fetch_targets():
            return snapshots.pop(0)

        report = generate_monitoring.build_stabilized_report(
            fetch_targets,
            attempts=3,
            sleep_seconds=0,
            emit_status=False,
        )

        self.assertEqual(report["missing_expected"], [])
        self.assertEqual(report["stabilization"]["attempt"], 2)
        self.assertEqual(report["stabilization"]["status"], "cleared")

    def test_healthy_blackbox_job_cannot_hide_missing_gcp_provider_targets(self):
        report = generate_monitoring.build_stabilized_report(
            lambda: targets_payload(
                missing_targets=set(generate_monitoring.EXPECTED_TARGETS)
            ),
            attempts=1,
            sleep_seconds=0,
            emit_status=False,
        )

        self.assertEqual(report["summary"]["expected_coverage_pct"], 100.0)
        self.assertEqual(
            report["missing_expected_targets"],
            ["ollama_gcp_a_cloud_edge", "ollama_gcp_b_cloud_edge"],
        )
        self.assertEqual(report["stabilization"]["status"], "failed")

    def test_stabilized_report_retries_down_gcp_target(self):
        snapshots = [
            targets_payload(down_targets={"ollama_gcp_a_cloud_edge"}),
            targets_payload(),
        ]

        report = generate_monitoring.build_stabilized_report(
            lambda: snapshots.pop(0),
            attempts=3,
            sleep_seconds=0,
            emit_status=False,
        )

        self.assertEqual(report["down_expected_targets"], [])
        self.assertEqual(report["stabilization"]["attempt"], 2)
        self.assertEqual(report["stabilization"]["status"], "cleared")

    def test_duplicate_exact_provider_target_fails_closed(self):
        report = generate_monitoring.build_stabilized_report(
            lambda: targets_payload(
                duplicate_targets={"ollama_gcp_b_cloud_edge"}
            ),
            attempts=1,
            sleep_seconds=0,
            emit_status=False,
        )

        self.assertEqual(
            report["ambiguous_expected_targets"], ["ollama_gcp_b_cloud_edge"]
        )
        self.assertEqual(report["stabilization"]["status"], "failed")

    def test_retired_github_exporter_is_policy_drift_not_extra_coverage(self):
        payload = targets_payload()
        payload["activeTargets"].append(
            {
                "labels": {
                    "job": "github-actions",
                    "instance": "retired-exporter:9504",
                },
                "health": "up",
            }
        )

        report = generate_monitoring.build_stabilized_report(
            lambda: payload,
            attempts=1,
            sleep_seconds=0,
            emit_status=False,
        )

        self.assertEqual(report["forbidden_jobs_present"], ["github-actions"])
        self.assertEqual(report["summary"]["forbidden_jobs_present"], 1)
        self.assertEqual(report["stabilization"]["status"], "failed")

    def test_stabilized_report_keeps_real_down_after_attempts_exhausted(self):
        def fetch_targets():
            return targets_payload(down_jobs={"awoooi-api"})

        report = generate_monitoring.build_stabilized_report(
            fetch_targets,
            attempts=2,
            sleep_seconds=0,
            emit_status=False,
        )

        self.assertEqual(report["summary"]["real_down_jobs"], 1)
        self.assertEqual(report["real_down_jobs"], ["awoooi-api"])
        self.assertEqual(report["stabilization"]["status"], "failed")

    def test_stabilized_report_does_not_retry_clean_snapshot(self):
        calls = 0

        def fetch_targets():
            nonlocal calls
            calls += 1
            return targets_payload()

        report = generate_monitoring.build_stabilized_report(
            fetch_targets,
            attempts=3,
            sleep_seconds=0,
            emit_status=False,
        )

        self.assertEqual(calls, 1)
        self.assertEqual(report["stabilization"]["status"], "stable")


if __name__ == "__main__":
    unittest.main()