prose-contracts/tests/test_probe_drift.py

"""Regression tests for the 2026-09-09/10 probe-drift corrections.

WHY THIS FILE EXISTS: the monitoring contracts kept emitting false alarms from
stale expectations rather than live faults.

  * agent-health-check v3 reported 6 failures that were all stale expectations:
    abiba (pi-only since the harness purge) was tested as a Hermes host, koby
    (report-only per the captain's 2026-08-17 ruling) was counted as repairable,
    koby's CT 111 was probed on amdpve where it does not exist (it runs on
    storepve .6), and a .env-based hermes wrapper was FAILed for not mentioning
    infisical.
  * gpu-monitor emitted "DEGRADED — GPU-rtx3090 000, GPU-rtx5070 000" three
    times from probing bare port 80 on GPU hosts while :8080 answered 200.
  * infrastructure-monitoring probed CT 116 for the PVE API (no pveproxy ->
    000) instead of the five real cluster nodes, which answer 401 = alive.

These tests pin the corrections so the false alarms cannot return silently.
They are static/structural over the contracts plus an inert import of the health
script — no live network, vault, or SSH access is required.
"""
from __future__ import annotations

import importlib.util
import pathlib
import re

import pytest

ROOT = pathlib.Path(__file__).resolve().parents[1]
AHC = ROOT / "scripts" / "agent-health-check.py"
GPU = ROOT / "gpu-monitor.prose.md"
INFRA = ROOT / "infrastructure-monitoring.prose.md"
PROXMOX = ROOT / "proxmox-monitor.prose.md"


@pytest.fixture(scope="module")
def ahc():
    """Import agent-health-check.py without live network/SSH side effects."""
    spec = importlib.util.spec_from_file_location("agent_health_check", AHC)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


# ── agent-health-check: stale-expectation legs ───────────────────────

def test_import_does_not_contact_vault(ahc):
    # Keys are loaded in main() via load_agent_keys(); importing must stay inert.
    assert callable(ahc.load_agent_keys)
    assert all(agent.get("key") is None for agent in ahc.AGENTS.values())


def test_abiba_is_pi_only_runtime(ahc):
    # .24 has run pi-only since the harness purge: no Hermes gateway, config, or
    # wrapper. Probing those legs produced false failures.
    assert ahc.AGENTS["abiba"]["runtime"] == "pi"


def test_koby_is_report_only(ahc):
    # Captain's 2026-08-17 ruling (Rule 17): detect and report, never repair.
    assert ahc.AGENTS["koby"]["report_only"] is True


def test_koby_ct111_is_on_storepve(ahc):
    # Live-verified 2026-09-10: `pct status 111` = running on storepve (.6);
    # amdpve has no lxc/111.conf, which is what false-failed before.
    assert ahc.AGENTS["koby"]["pve"] == "storepve"


@pytest.mark.parametrize("agent", ["koby", "koonimo", "tanko"])
def test_report_only_legs_never_count_as_failures(ahc, agent):
    ahc.FAIL.clear()
    try:
        ahc._fail(f"probe:{agent}", agent)
        if ahc.AGENTS[agent].get("report_only"):
            assert ahc.FAIL == []
        else:
            assert ahc.FAIL == [f"probe:{agent}"]
    finally:
        ahc.FAIL.clear()


def test_failure_recording_accepts_agentless_keys(ahc):
    ahc.FAIL.clear()
    try:
        ahc._fail("gpu-no-port:gpu-rtx3090 (.8)")
        assert ahc.FAIL == ["gpu-no-port:gpu-rtx3090 (.8)"]
    finally:
        ahc.FAIL.clear()


def test_script_reports_absolute_execution_provenance(ahc):
    src = AHC.read_text()
    assert "execution_path" in src
    assert "os.getcwd()" in src


def test_wrapper_check_has_no_bare_infisical_expectation(ahc):
    # A wrapper that never mentions infisical (koonimo sources ~/.hermes/.env)
    # must not be FAILed merely for lacking an infisical reference.
    assert "wrapper resolves creds without infisical" in AHC.read_text()


# ── item 4: every contract report carries its execution path ──────────

@pytest.mark.parametrize("contract", [GPU, INFRA, PROXMOX], ids=lambda p: p.name)
def test_report_format_requires_execution_provenance(contract):
    text = contract.read_text()
    assert "**Report format**" in text
    assert re.search(r"absolute path|pwd -P|executed from", text)


# ── item 3: gpu-monitor probes the real GPU endpoints ────────────────

def test_gpu_monitor_probes_gpu_health_on_8080():
    text = GPU.read_text()
    assert "http://192.168.68.8:8080/health" in text
    assert "http://192.168.68.110:8080/health" in text


def test_gpu_monitor_forbids_bare_port_80_on_gpu_hosts():
    text = GPU.read_text()
    assert re.search(r"never .{0,20}bare port 80", text, re.I)
    # The false-DEGRADED symptom must be documented, not just implied.
    assert "DEGRADED" in text


def test_gpu_monitor_documents_router_301_as_alive():
    text = GPU.read_text()
    assert "/gpu/gpu-data" in text
    assert "301" in text


# ── item 2: infrastructure-monitoring PVE API vantage ────────────────

def test_infra_monitoring_does_not_probe_ct116_for_pve_api():
    assert "https://192.168.68.116:8006" not in INFRA.read_text()


@pytest.mark.parametrize(
    "node",
    ["192.168.68.9", "192.168.68.5", "192.168.68.15", "192.168.68.6", "192.168.68.12"],
)
def test_infra_monitoring_probes_every_real_pve_node(node):
    assert node in INFRA.read_text()


def test_infra_monitoring_uses_any_http_liveness_rule():
    text = INFRA.read_text()
    assert "api2/json/version" in text
    assert "ANY HTTP status" in text or "any-HTTP" in text