prose-contracts/tests/test_zulip_kagentz_legs.py

#!/usr/bin/env python3
"""
Tests for zulip-monitor.sh kagentz A2A and public access path legs.

Covers:
- (b) Run verdict is non-optimistic: when ISSUES > 0, the Result line says "INCIDENT"
- (d) Public access path leg: 200/302/401 = alive, 502 = incident, 000 = incident
- (c) C1/C2/C3 distinction documented in prose

These tests parse the script and prose to verify the expected structure.
"""

import os
import re
import subprocess
import sys
from pathlib import Path

# Paths
SCRIPT = Path(__file__).parent.parent / "scripts" / "zulip-monitor.sh"
PROSE = Path(__file__).parent.parent / "zulip-health.prose.md"

def read_file(path):
    return path.read_text()

def test_script_has_c1_c2_c3_legs():
    """Script should have C1, C2, C3 leg markers."""
    content = read_file(SCRIPT)
    assert "C1: A2A liveness" in content, "Missing C1 leg comment"
    assert "C3: Public access path" in content, "Missing C3 leg comment"
    print("✓ Script has C1, C2, C3 leg markers")

def test_c1_no_credential_needed():
    """C1 should be documented as needing no credential."""
    prose = read_file(PROSE)
    assert "C1: A2A Server Health (no credential needed)" in prose, \
        "C1 header should say 'no credential needed'"
    assert "INCIDENT" in prose, "C1 000 should be marked as INCIDENT"
    print("✓ C1 documented as no-credential, 000 = INCIDENT")

def test_c2_requires_litellm_key():
    """C2 should be documented as requiring LITELLM_KEY."""
    prose = read_file(PROSE)
    assert "C2: A2A Response Verification (requires LITELLM_KEY)" in prose, \
        "C2 header should say 'requires LITELLM_KEY'"
    assert "credential issue, NOT a server-down incident" in prose, \
        "C2 401 should be documented as credential issue, not server-down"
    print("✓ C2 documented as requiring LITELLM_KEY")

def test_c3_public_access_path():
    """C3 should probe https://kagentz.sysloggh.net/."""
    prose = read_file(PROSE)
    script = read_file(SCRIPT)
    assert "C3: Public Access Path" in prose, "Missing C3 section in prose"
    assert "https://kagentz.sysloggh.net/" in prose, "C3 should probe the public URL"
    assert "502" in prose, "C3 should document 502 as incident"
    assert "KAGENTZ_PUBLIC_CODE" in script, "Script should have KAGENTZ_PUBLIC_CODE variable"
    print("✓ C3 public access path leg present")

def test_result_line_non_optimistic():
    """When ISSUES > 0, the Result line should say INCIDENT, not just 'issues found'."""
    script = read_file(SCRIPT)
    # The summary section should have "INCIDENT" in the non-zero branch
    assert "Result: 🔴 INCIDENT" in script, \
        "Result line should say 'INCIDENT' when ISSUES > 0"
    assert "Result: ✅ 0 issues (all healthy)" in script, \
        "Result line should say '0 issues (all healthy)' when ISSUES = 0"
    print("✓ Result line is non-optimistic (INCIDENT when issues > 0)")

def test_c3_502_is_incident():
    """C3 should treat 502 as an incident."""
    script = read_file(SCRIPT)
    # The script should have a branch for 502
    assert 'elif [ "$KAGENTZ_PUBLIC_CODE" = "502" ]' in script, \
        "Script should have explicit 502 branch"
    assert "upstream refused" in script, \
        "502 should be documented as 'upstream refused'"
    print("✓ C3 502 treated as incident")

def test_c3_000_is_incident():
    """C3 should treat 000 as an incident."""
    script = read_file(SCRIPT)
    assert 'if [ "$KAGENTZ_PUBLIC_CODE" = "000" ]' in script, \
        "Script should have explicit 000 branch"
    assert "public URL DOWN" in script, \
        "000 should be reported as 'public URL DOWN'"
    print("✓ C3 000 treated as incident")

def test_c3_alive_statuses():
    """C3 should treat 200/302/401 as alive."""
    script = read_file(SCRIPT)
    assert "200|302|401" in script, \
        "Script should classify 200/302/401 as alive"
    assert "public URL alive" in script, \
        "Alive statuses should be reported as 'public URL alive'"
    print("✓ C3 200/302/401 treated as alive")

def test_prose_documents_c3_actions():
    """Prose should document C3 actions in the Platform C Actions table."""
    prose = read_file(PROSE)
    assert "C3 public URL returns `502`" in prose, \
        "Platform C Actions table should have C3 502 row"
    assert "C3 public URL returns `000`" in prose, \
        "Platform C Actions table should have C3 000 row"
    print("✓ Prose documents C3 actions")

def run_all_tests():
    tests = [
        test_script_has_c1_c2_c3_legs,
        test_c1_no_credential_needed,
        test_c2_requires_litellm_key,
        test_c3_public_access_path,
        test_result_line_non_optimistic,
        test_c3_502_is_incident,
        test_c3_000_is_incident,
        test_c3_alive_statuses,
        test_prose_documents_c3_actions,
    ]
    passed = 0
    failed = 0
    for test in tests:
        try:
            test()
            passed += 1
        except AssertionError as e:
            print(f"✗ {test.__name__}: {e}")
            failed += 1
    print(f"\n{'='*50}")
    print(f"Tests passed: {passed}/{len(tests)}")
    if failed > 0:
        print(f"Tests failed: {failed}/{len(tests)}")
        sys.exit(1)
    else:
        print("All tests passed!")

if __name__ == "__main__":
    run_all_tests()