prose-contracts/tests/test_zulip_kagentz_legs.py

#!/usr/bin/env python3
"""Behavioural tests for zulip-monitor.sh kagentz C1/C3 legs and the Result verdict.

WHY THIS FILE EXISTS: the 2026-09-19 kagentz A2A outage was correctly detected
by the monitor (C1 returned 000, the run logged an issue) but the lane's own
summarization in ops.status wrote "OK" with the note "A2A server DOWN — expected
(no credentials configured)". The optimistic verdict came from the lane, not the
script. The fix adds a C3 public-access-path leg and makes the Result line say
"INCIDENT" when issues are found, so the lane can quote it verbatim.

CONTRACT UNDER TEST:
  * C1 (A2A liveness, no credential): 000 → INCIDENT.
  * C3 (public access path, no credential): 502 → INCIDENT, 000 → INCIDENT,
    200/302/401 → alive.
  * Result verdict: when ISSUES > 0, the log's Result line says "INCIDENT",
    not just "issues found".
  * Healthy control: C1 401 + C3 302 → 0 issues, "all healthy".

HOW: behavioural execution using the sandbox pattern already in this repo
(tests/test_mumuni_monitor_removal.py). The sandbox copies the shipped monitor
verbatim, rewrites only its LOG constant, and runs it with stub ssh/curl on
PATH. Each test asserts from the run's own log/verdict, not from file text.

Usage: python3 -m pytest tests/test_zulip_kagentz_legs.py
"""
from __future__ import annotations

import os
import pathlib
import stat
import subprocess

import pytest

ROOT = pathlib.Path(__file__).resolve().parents[1]
ZULIP_MONITOR = ROOT / "scripts" / "zulip-monitor.sh"
CONNECTED_FIXTURE = ROOT / "tests" / "fixtures" / "zulip-health-connected.json"

TANKO_VANTAGE = "192.168.68.15"   # amdpve — Tanko CT 112 via pct exec
AGENT_ZERO_HOST = "192.168.68.14"  # kagentz host, Agent Zero docker


# ── Stub ssh: answers Tanko and Agent Zero probes by env vars ──────────

SSH_STUB = r"""#!/usr/bin/env bash
# Stub ssh: record the target host, then answer by host + remote command.
printf '%s\n' "$*" >> "$RECORD_DIR/ssh.calls"
host=""
for a in "$@"; do
  case "$a" in
    *@192.168.*) host="${a##*@}" ;;
  esac
done
printf '%s\n' "$host" >> "$RECORD_DIR/ssh.hosts"
cmd="${*: -1}"
case "$host" in
  192.168.68.15)
    case "$cmd" in
      *"systemctl is-active"*) printf '%s' "$TANKO_SVC" ;;
      *curl*)                  printf '%s' "$TANKO_HTTP" ;;
    esac ;;
  192.168.68.14)
    case "$cmd" in
      *"/a2a/"*) printf '%s' "$AZ_A2A_CODE"; exit "${AZ_A2A_EXIT:-0}" ;;
    esac ;;
  *)
    printf 'UNEXPECTED-SSH-HOST %s\n' "$host" >> "$RECORD_DIR/unexpected-ssh" ;;
esac
exit 0
"""


# ── Stub curl: serves Zulip server, Abiba health, and C3 public URL ───

CURL_STUB = r"""#!/usr/bin/env bash
# Stub curl: serve the Abiba health fixture, the Zulip server 200, and the
# C3 public URL probe (https://kagentz.sysloggh.net/). Record every call.
printf '%s\n' "$*" >> "$RECORD_DIR/curl.calls"
case "$*" in
  *:9200/health*)
    case " $* " in
      *" -w "*) printf '%s' "$PI_HTTP" ;;   # -w '%{http_code}' probe
      *)        printf '%s' "$PI_BODY" ;;   # body probe
    esac ;;
  *server_settings*)
    printf '%s' "$SERVER_HTTP" ;;
  *kagentz.sysloggh.net*)
    printf '%s' "$KAGENTZ_PUBLIC_CODE" ;;
esac
exit 0
"""


def _write_exec(path: pathlib.Path, body: str) -> None:
    path.write_text(body)
    path.chmod(path.stat().st_mode
               | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)


def _run_monitor(tmp_path, *, tanko_svc="active", tanko_http="200",
                 az_a2a_code="401", az_a2a_exit=0,
                 kagentz_public_code="302"):
    """Run the shipped monitor in a sandbox; return (proc, record_dir, log_path).

    Only the LOG constant is rewritten (to keep the run inside the worktree).
    Everything else — legs, labels, notify logic — is the shipped script.
    """
    sandbox = tmp_path / "sandbox"
    bindir = sandbox / "bin"
    record = sandbox / "record"
    bindir.mkdir(parents=True)
    record.mkdir()

    _write_exec(bindir / "ssh", SSH_STUB)
    _write_exec(bindir / "curl", CURL_STUB)

    source = ZULIP_MONITOR.read_text()
    log_line = 'LOG="/root/zulip-health-monitor.log"'
    assert log_line in source, "LOG constant moved — update the sandbox harness"
    log_path = sandbox / "zulip-health-monitor.log"
    script = sandbox / "zulip-monitor.sh"
    script.write_text(source.replace(log_line, f'LOG="{log_path}"'))

    env = dict(os.environ)
    env.update({
        "PATH": f"{bindir}:{env['PATH']}",
        "RECORD_DIR": str(record),
        "TANKO_SVC": tanko_svc,
        "TANKO_HTTP": tanko_http,
        "AZ_A2A_CODE": az_a2a_code,
        "AZ_A2A_EXIT": str(az_a2a_exit),
        "PI_HTTP": "200",
        "PI_BODY": CONNECTED_FIXTURE.read_text(),
        "SERVER_HTTP": "200",
        "KAGENTZ_PUBLIC_CODE": kagentz_public_code,
    })
    proc = subprocess.run(["bash", str(script)], cwd=sandbox, env=env,
                          capture_output=True, text=True)
    return proc, record, log_path


# ── Required behavioural cases ─────────────────────────────────────────

def test_c3_502_is_incident(tmp_path):
    """C3 public leg returns 502 → the run's verdict is an INCIDENT,
    the C3 line names the 502, and the run is not summarised as healthy."""
    proc, record, log_path = _run_monitor(tmp_path, kagentz_public_code="502")
    assert proc.returncode == 0, proc.stderr
    log = log_path.read_text()

    # The C3 line names the 502.
    assert "kagentz C3: ❌ public URL 502 (upstream refused)" in log
    # The verdict is an INCIDENT, not healthy.
    assert "Result: 🔴 INCIDENT" in log
    assert "all healthy" not in log
    # The run is not summarised as healthy.
    assert "✅ 0 issues" not in log


def test_c3_000_is_incident(tmp_path):
    """C3 public leg returns 000 → INCIDENT."""
    proc, record, log_path = _run_monitor(tmp_path, kagentz_public_code="000")
    assert proc.returncode == 0, proc.stderr
    log = log_path.read_text()

    # The C3 line reports the connection failure.
    assert "kagentz C3: ❌ public URL down (HTTP 000)" in log
    # The verdict is an INCIDENT.
    assert "Result: 🔴 INCIDENT" in log
    assert "all healthy" not in log
    assert "✅ 0 issues" not in log


def test_healthy_control_c1_401_c3_302(tmp_path):
    """Healthy control: C1 401 plus C3 302 → 0 issues and a healthy verdict,
    proving the new leg cannot cry wolf."""
    proc, record, log_path = _run_monitor(tmp_path,
                                          az_a2a_code="401",
                                          kagentz_public_code="302")
    assert proc.returncode == 0, proc.stderr
    log = log_path.read_text()

    # Both legs report alive.
    assert "kagentz C1: ✅ A2A alive (HTTP 401)" in log
    assert "kagentz C3: ✅ public URL alive (HTTP 302)" in log
    # Zero issues, healthy verdict.
    assert "Result: ✅ 0 issues (all healthy)" in log
    # No INCIDENT.
    assert "INCIDENT" not in log
    # No notify fired for kagentz.
    assert "kagentz public URL" not in proc.stdout
    assert "kagentz A2A server" not in proc.stdout


def test_c1_000_is_incident(tmp_path):
    """C1 returns 000 → INCIDENT, keeping the leg that actually caught this
    outage covered behaviourally."""
    proc, record, log_path = _run_monitor(tmp_path,
                                          az_a2a_code="000",
                                          az_a2a_exit=7,
                                          kagentz_public_code="302")
    assert proc.returncode == 0, proc.stderr
    log = log_path.read_text()

    # The C1 line reports the A2A down.
    assert "kagentz C1: ❌ A2A down (HTTP 000)" in log
    # The verdict is an INCIDENT (even though C3 is healthy).
    assert "Result: 🔴 INCIDENT" in log
    assert "all healthy" not in log
    # The notify fired for the A2A down.
    assert "kagentz A2A server DOWN" in proc.stdout


# ── Prose assertions (kept as secondary, do not replace behavioural) ──

def test_prose_c1_no_credential_needed():
    """C1 header should say 'no credential needed'."""
    prose = (ROOT / "zulip-health.prose.md").read_text()
    assert "C1: A2A Server Health (no credential needed)" in prose


def test_prose_c2_requires_litellm_key():
    """C2 header should say 'requires LITELLM_KEY'."""
    prose = (ROOT / "zulip-health.prose.md").read_text()
    assert "C2: A2A Response Verification (requires LITELLM_KEY)" in prose


def test_prose_c3_public_access_path():
    """C3 section should exist and document 502/000 as incidents."""
    prose = (ROOT / "zulip-health.prose.md").read_text()
    assert "C3: Public Access Path" in prose
    assert "https://kagentz.sysloggh.net/" in prose
    assert "502" in prose