124 lines
4.6 KiB
Python
124 lines
4.6 KiB
Python
"""Tests for scripts/probe_argus_kb_state.py."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
|
|
|
import probe_argus_kb_state as probe # noqa: E402
|
|
|
|
|
|
def args_for(tmp_path: Path) -> argparse.Namespace:
|
|
return argparse.Namespace(
|
|
base_url="http://example.invalid",
|
|
timeout=1,
|
|
kb_proposals_json_base64=None,
|
|
vital_signs_json_base64=None,
|
|
metrics_json_base64=None,
|
|
out=tmp_path / "argus.json",
|
|
markdown_out=tmp_path / "argus.md",
|
|
)
|
|
|
|
|
|
def good_payloads() -> dict[str, dict]:
|
|
return {
|
|
"kb_proposals": {
|
|
"generated_at": "2026-07-10T02:43:40+00:00",
|
|
"filters": {"status": "approved", "limit": 20},
|
|
"total": 3,
|
|
"status_counts": {"approved": 3},
|
|
"review_state_counts": {"approved_needs_apply_payload": 3},
|
|
"worker_applyable_count": 0,
|
|
"read_only": True,
|
|
"packets": [
|
|
{
|
|
"proposal_id": "14fa5ecc-ac7a-41c1-807d-a2e85b936617",
|
|
"review_state": "approved_needs_apply_payload",
|
|
"worker_applyable": False,
|
|
"has_apply_payload": False,
|
|
"next_admin_action": "Normalize first.",
|
|
},
|
|
{
|
|
"proposal_id": "ac036c9d-20a0-4ffe-881f-57d6b7bacf22",
|
|
"review_state": "approved_needs_apply_payload",
|
|
"worker_applyable": False,
|
|
"has_apply_payload": False,
|
|
"next_admin_action": "Normalize first.",
|
|
},
|
|
{
|
|
"proposal_id": "a64df080-8502-42e2-98f4-9bbdecb8da73",
|
|
"review_state": "approved_needs_apply_payload",
|
|
"worker_applyable": False,
|
|
"has_apply_payload": False,
|
|
"next_admin_action": "Normalize first.",
|
|
},
|
|
],
|
|
},
|
|
"vital_signs": {"claim_index_status": "unavailable", "funnel": {"sources_total": 3676}},
|
|
"metrics": {"status_map": {"merged": 4094}, "source_map": {"extracted": 2233}},
|
|
}
|
|
|
|
|
|
def fake_fetch(url: str, _timeout: int) -> dict:
|
|
payloads = good_payloads()
|
|
if url.endswith("/api/kb-proposals"):
|
|
return payloads["kb_proposals"]
|
|
if url.endswith("/api/vital-signs"):
|
|
return payloads["vital_signs"]
|
|
if url.endswith("/api/metrics"):
|
|
return payloads["metrics"]
|
|
raise AssertionError(url)
|
|
|
|
|
|
def test_probe_passes_when_approved_proposals_are_visible_but_not_applyable(tmp_path: Path):
|
|
result = probe.collect(args_for(tmp_path), fetcher=fake_fetch)
|
|
|
|
assert result["status"] == "pass"
|
|
assert all(result["checks"].values())
|
|
assert result["summary"]["proposals"]["total"] == 3
|
|
assert result["summary"]["proposals"]["worker_applyable_count"] == 0
|
|
assert "a64df080-8502-42e2-98f4-9bbdecb8da73" in result["summary"]["proposals"]["proposal_ids"]
|
|
|
|
|
|
def test_probe_can_use_supplied_payloads_without_network_fetch(tmp_path: Path):
|
|
args = args_for(tmp_path)
|
|
payloads = good_payloads()
|
|
args.kb_proposals_json_base64 = base64.b64encode(json.dumps(payloads["kb_proposals"]).encode()).decode()
|
|
args.vital_signs_json_base64 = base64.b64encode(json.dumps(payloads["vital_signs"]).encode()).decode()
|
|
args.metrics_json_base64 = base64.b64encode(json.dumps(payloads["metrics"]).encode()).decode()
|
|
|
|
def failing_fetch(_url: str, _timeout: int) -> dict:
|
|
raise AssertionError("fetcher should not be called when all payloads are supplied")
|
|
|
|
result = probe.collect(args, fetcher=failing_fetch)
|
|
|
|
assert result["status"] == "pass"
|
|
assert set(result["fetch_sources"].values()) == {"supplied_base64_json"}
|
|
|
|
|
|
def test_probe_fails_when_fetch_fails(tmp_path: Path):
|
|
def failing_fetch(_url: str, _timeout: int) -> dict:
|
|
return {"_fetch_error": "network denied"}
|
|
|
|
result = probe.collect(args_for(tmp_path), fetcher=failing_fetch)
|
|
|
|
assert result["status"] == "fail"
|
|
assert result["checks"]["http_readback_ok"] is False
|
|
|
|
|
|
def test_markdown_summarizes_applyability_ceiling(tmp_path: Path):
|
|
result = probe.collect(args_for(tmp_path), fetcher=fake_fetch)
|
|
out = tmp_path / "argus.md"
|
|
|
|
probe.write_markdown(out, result)
|
|
|
|
text = out.read_text(encoding="utf-8")
|
|
assert "Argus KB State Probe" in text
|
|
assert "Worker-applyable count: `0`" in text
|
|
assert "This is public Argus HTTP diagnostics evidence only" in text
|