prose-contracts/tests/test_daily_infra_report.py

"""
Regression tests for daily-infra-report.py fixes (PR #64).

Tests:
(a) Asserts the nested zulip read feeds the agent-card fields
(b) Asserts an unreachable pve_get renders labelled-unreachable, not "0/0"
"""
import json
import os
import subprocess
import sys
from pathlib import Path
from unittest.mock import patch, MagicMock

# Add scripts to path
sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
import importlib.util

def load_script():
    """Load the daily-infra-report script as a module."""
    script_path = Path(__file__).parent.parent / "scripts" / "daily-infra-report.py"
    spec = importlib.util.spec_from_file_location("daily_infra_report", script_path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_pve_token_is_read_from_the_environment():
    """The PVE token must come from the injected environment, never a literal.

    Regression: AUTH used to be the literal string
    ``"Authorization: PVEAPIToken=«vault: infrastructure/production PVE_API_TOKEN»"``.
    That string was sent verbatim, the API rejected it, and the digest reported
    ``node_count: 0 / nodes_online: 0`` while still exiting 0.
    """
    mod = load_script()
    assert hasattr(mod, "pve_auth"), "pve_auth() must exist to resolve the token at call time"
    with patch.dict("os.environ", {"PVE_TOKEN": "user@pve!tokid=secretvalue"}, clear=False):
        assert mod.pve_auth() == "Authorization: PVEAPIToken=user@pve!tokid=secretvalue"


def test_missing_pve_token_is_degraded_not_a_placeholder():
    """With no PVE_TOKEN, pve_get must return None (probe failure), not send a placeholder."""
    mod = load_script()
    env = {k: v for k, v in os.environ.items() if k != "PVE_TOKEN"}
    with patch.dict("os.environ", env, clear=True):
        assert mod.pve_get("/api2/json/nodes") is None, (
            "a missing PVE_TOKEN must degrade to None so the caller records a probe failure"
        )


def test_unreachable_probe_is_recorded_as_a_failure():
    """An unreachable probe must be recorded, so the run cannot pass silently."""
    mod = load_script()
    assert hasattr(mod, "PROBE_FAILURES"), "PROBE_FAILURES must exist"
    mod.PROBE_FAILURES.clear()
    with patch.object(mod, "pve_get", return_value=None):
        report = mod.collect()
    assert report["pve_probe_status"] == "unreachable"
    assert any("unreachable" in f for f in mod.PROBE_FAILURES), (
        f"unreachable probe must be recorded in PROBE_FAILURES, got {mod.PROBE_FAILURES}"
    )


def test_pve_token_placeholder_is_gone():
    """The literal placeholder must no longer appear anywhere in the script."""
    src = (Path(__file__).parent.parent / "scripts" / "daily-infra-report.py").read_text()
    assert "«vault:" not in src, "the unresolved vault placeholder must not remain in the script"
    assert "AUTH = \"Authorization" not in src, "the hardcoded AUTH literal must be gone"


def test_nested_zulip_read_feeds_agent_card():
    """Test that Zulip state is read from the nested 'zulip' key and feeds agent-card fields."""
    # Mock the http_get_body response with nested structure
    mock_health_response = json.dumps({
        "status": "ok",
        "platform": "pi",
        "agent": "abiba",
        "zulip": {
            "connected": True,
            "queue_id": "test-queue-id",
            "messages_processed": 42,
            "skipped": 5,
            "last_error": None
        }
    })
    
    # Import and patch
    report_mod = load_script()
    
    with patch.object(report_mod, 'http_get_body', return_value=mock_health_response):
        # Simulate the collect() function's Zulip section
        zulip_health = json.loads(report_mod.http_get_body("http://localhost:9200/health"))
        zulip_state = zulip_health.get("zulip", {})
        
        # Assert the nested key is read correctly
        assert zulip_state.get("connected") == True, "Zulip connected should be True from nested key"
        assert zulip_state.get("messages_processed") == 42, "messages_processed should be 42 from nested key"
        assert zulip_state.get("queue_id") == "test-queue-id", "queue_id should be read from nested key"
        
        # Simulate the agent card field population
        agent_card = {
            "zulip_connected": zulip_state.get("connected", False),
            "zulip_processed": zulip_state.get("messages_processed", 0),
        }
        
        assert agent_card["zulip_connected"] == True, "Agent card should show Zulip connected"
        assert agent_card["zulip_processed"] == 42, "Agent card should show 42 processed messages"


def test_unreachable_pve_get_renders_labelled_unreachable():
    """Test that an unreachable PVE API renders 'unreachable' instead of '0/0'."""
    # Import and patch
    report_mod = load_script()
    
    # Test pve_get returns None on error
    with patch.object(report_mod.subprocess, 'run') as mock_run:
        mock_run.return_value.returncode = 7  # Connection failure
        result = report_mod.pve_get("/api2/json/nodes")
        assert result is None, "pve_get should return None on connection failure"
    
    # Test the render logic
    report = {
        "nodes": {},
        "node_count": 0,
        "nodes_online": 0,
        "pve_probe_status": "unreachable",
        "total_vms": 0,
        "running_vms": 0,
    }
    
    # The render should show "unreachable" not "0/0"
    pve_status_label = "unreachable" if report.get('pve_probe_status') == 'unreachable' else f"{report['nodes_online']}/{report['node_count']}"
    
    assert pve_status_label == "unreachable", "PVE status should show 'unreachable' when probe fails, not '0/0'"


def test_unreachable_resources_renders_labelled_unreachable():
    """Test that unreachable resources probe renders 'unreachable' instead of '0/0'."""
    report_mod = load_script()
    
    # Test resources probe returns None
    with patch.object(report_mod.subprocess, 'run') as mock_run:
        mock_run.return_value.returncode = 7
        result = report_mod.pve_get("/api2/json/cluster/resources")
        assert result is None, "pve_get for resources should return None on connection failure"
    
    # Test the render logic
    report = {
        "resources_probe_status": "unreachable",
        "total_vms": 0,
        "running_vms": 0,
    }
    
    resources_label = "unreachable" if report.get('resources_probe_status') == 'unreachable' else f"{report['running_vms']}/{report['total_vms']}"
    
    assert resources_label == "unreachable", "Resources status should show 'unreachable' when probe fails, not '0/0'"


if __name__ == "__main__":
    print("Running tests...")
    try:
        test_nested_zulip_read_feeds_agent_card()
        print("✓ test_nested_zulip_read_feeds_agent_card passed")
    except AssertionError as e:
        print(f"✗ test_nested_zulip_read_feeds_agent_card failed: {e}")
        sys.exit(1)
    
    try:
        test_unreachable_pve_get_renders_labelled_unreachable()
        print("✓ test_unreachable_pve_get_renders_labelled_unreachable passed")
    except AssertionError as e:
        print(f"✗ test_unreachable_pve_get_renders_labelled_unreachable failed: {e}")
        sys.exit(1)
    
    try:
        test_unreachable_resources_renders_labelled_unreachable()
        print("✓ test_unreachable_resources_renders_labelled_unreachable passed")
    except AssertionError as e:
        print(f"✗ test_unreachable_resources_renders_labelled_unreachable failed: {e}")
        sys.exit(1)
    
    try:
        test_pve_token_is_read_from_the_environment()
        print("✓ test_pve_token_is_read_from_the_environment passed")
    except AssertionError as e:
        print(f"✗ test_pve_token_is_read_from_the_environment failed: {e}")
        sys.exit(1)

    try:
        test_missing_pve_token_is_degraded_not_a_placeholder()
        print("✓ test_missing_pve_token_is_degraded_not_a_placeholder passed")
    except AssertionError as e:
        print(f"✗ test_missing_pve_token_is_degraded_not_a_placeholder failed: {e}")
        sys.exit(1)

    try:
        test_unreachable_probe_is_recorded_as_a_failure()
        print("✓ test_unreachable_probe_is_recorded_as_a_failure passed")
    except AssertionError as e:
        print(f"✗ test_unreachable_probe_is_recorded_as_a_failure failed: {e}")
        sys.exit(1)

    try:
        test_pve_token_placeholder_is_gone()
        print("✓ test_pve_token_placeholder_is_gone passed")
    except AssertionError as e:
        print(f"✗ test_pve_token_placeholder_is_gone failed: {e}")
        sys.exit(1)

    print("All tests passed!")