194 lines
7.5 KiB
Python
194 lines
7.5 KiB
Python
"""Tests for scripts/collect_telegram_visible_direct_claim_preflight.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 collect_telegram_visible_direct_claim_preflight as preflight # noqa: E402
|
|
|
|
|
|
def args_for(packet: Path, tmp_path: Path) -> argparse.Namespace:
|
|
return argparse.Namespace(
|
|
packet=packet,
|
|
ssh_target="root@example.invalid",
|
|
ssh_key=tmp_path / "key",
|
|
connect_timeout=1,
|
|
timeout=5,
|
|
remote_json_base64=None,
|
|
out=tmp_path / "out.json",
|
|
markdown_out=tmp_path / "out.md",
|
|
remote_command_out=tmp_path / "remote-readback.sh",
|
|
)
|
|
|
|
|
|
def good_remote_payload() -> dict:
|
|
return {
|
|
"deploy_head": {"returncode": 0, "stdout": "abc123", "stderr": ""},
|
|
"last_deploy_sha": {"returncode": 0, "stdout": "abc123", "stderr": ""},
|
|
"packet_present": True,
|
|
"packet_bytes": 1024,
|
|
"service": {
|
|
"returncode": 0,
|
|
"stderr": "",
|
|
"state": {
|
|
"ActiveState": "active",
|
|
"SubState": "running",
|
|
"MainPID": "1908033",
|
|
"NRestarts": "0",
|
|
"ExecMainStartTimestamp": "Fri 2026-07-10 01:04:22 UTC",
|
|
},
|
|
},
|
|
"db_counts": {
|
|
"returncode": 0,
|
|
"stderr": "",
|
|
"counts": {
|
|
"public.claims": 1837,
|
|
"public.sources": 4145,
|
|
"public.claim_evidence": 4670,
|
|
"public.claim_edges": 4916,
|
|
"kb_stage.kb_proposals": 26,
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def good_direct_runner(command: list[str], _input: str, _timeout: int) -> preflight.CommandResult:
|
|
remote_command = command[-1]
|
|
if "git rev-parse HEAD" in remote_command:
|
|
return preflight.CommandResult(
|
|
0,
|
|
"deploy_head=abc123\nlast_deploy_sha=abc123\npacket_present=true\npacket_bytes=1024\n",
|
|
"",
|
|
)
|
|
if "systemctl show leoclean-gateway.service" in remote_command:
|
|
return preflight.CommandResult(
|
|
0,
|
|
"\n".join(
|
|
[
|
|
"ActiveState=active",
|
|
"SubState=running",
|
|
"MainPID=1908033",
|
|
"NRestarts=0",
|
|
"ExecMainStartTimestamp=Fri 2026-07-10 01:04:22 UTC",
|
|
]
|
|
),
|
|
"",
|
|
)
|
|
if "docker exec teleo-pg" in remote_command:
|
|
return preflight.CommandResult(0, json.dumps(good_remote_payload()["db_counts"]["counts"]), "")
|
|
raise AssertionError(f"unexpected command: {remote_command}")
|
|
|
|
|
|
def test_preflight_passes_with_ready_packet_and_remote_readback(tmp_path: Path):
|
|
result = preflight.collect_preflight(args_for(preflight.DEFAULT_PACKET, tmp_path), runner=good_direct_runner)
|
|
|
|
assert result["status"] == "pass"
|
|
assert result["ready_for_authorized_telegram_visible_dc_run"] is True
|
|
assert result["telegram_visible_messages_sent"] is False
|
|
assert result["production_apply_executed"] is False
|
|
assert result["mutates_db"] is False
|
|
assert all(result["packet_checks"].values())
|
|
assert all(result["remote_checks"].values())
|
|
assert result["remote_source"] == "ssh_direct_commands"
|
|
assert result["remote_readback_command_path"].endswith("remote-readback.sh")
|
|
assert len(result["remote_readback_command_sha256"]) == 64
|
|
|
|
|
|
def test_preflight_can_use_supplied_remote_payload_without_subprocess_ssh(tmp_path: Path):
|
|
args = args_for(preflight.DEFAULT_PACKET, tmp_path)
|
|
args.remote_json_base64 = base64.b64encode(json.dumps(good_remote_payload()).encode()).decode()
|
|
|
|
def failing_runner(_command: list[str], _input: str, _timeout: int) -> preflight.CommandResult:
|
|
raise AssertionError("runner should not be called when remote_json_base64 is supplied")
|
|
|
|
result = preflight.collect_preflight(args, runner=failing_runner)
|
|
|
|
assert result["status"] == "pass"
|
|
assert result["remote_source"] == "supplied_remote_json_base64"
|
|
assert result["remote"]["deploy_head"]["stdout"] == "abc123"
|
|
|
|
|
|
def test_remote_readback_command_is_copy_pasteable_shape(tmp_path: Path):
|
|
args = args_for(preflight.DEFAULT_PACKET, tmp_path)
|
|
|
|
digest = preflight.write_remote_readback_command(args.remote_command_out, args)
|
|
|
|
text = args.remote_command_out.read_text(encoding="utf-8")
|
|
assert text.startswith("#!/usr/bin/env bash\nset -euo pipefail\n# Remote packet:")
|
|
assert "root@example.invalid" in text
|
|
assert "DEPLOY_READBACK" in text
|
|
assert "SERVICE_READBACK" in text
|
|
assert "DB_COUNTS_READBACK" in text
|
|
assert "public.claims" in text
|
|
assert "telegram-visible-direct-claim-authorization-packet-current.json" in text
|
|
assert text.endswith("printf '\\n'\n")
|
|
assert len(digest) == 64
|
|
|
|
|
|
def test_preflight_fails_on_invalid_supplied_remote_payload(tmp_path: Path):
|
|
args = args_for(preflight.DEFAULT_PACKET, tmp_path)
|
|
args.remote_json_base64 = "not valid base64"
|
|
|
|
result = preflight.collect_preflight(args)
|
|
|
|
assert result["status"] == "fail"
|
|
assert result["ready_for_authorized_telegram_visible_dc_run"] is False
|
|
assert result["remote_returncode"] == 2
|
|
assert "failed to parse --remote-json-base64" in result["remote_stderr"]
|
|
assert "Pre-send Telegram-visible direct-claim benchmark" in result["blocker_readback"]["current_canary"]
|
|
assert "remote readback" in result["blocker_readback"]["next_non_user_action"]
|
|
|
|
|
|
def test_preflight_fails_if_deploy_stamp_drift_detected(tmp_path: Path):
|
|
def drift_runner(command: list[str], _input: str, _timeout: int) -> preflight.CommandResult:
|
|
remote_command = command[-1]
|
|
if "git rev-parse HEAD" in remote_command:
|
|
return preflight.CommandResult(
|
|
0,
|
|
"deploy_head=abc123\nlast_deploy_sha=different\npacket_present=true\npacket_bytes=1024\n",
|
|
"",
|
|
)
|
|
return good_direct_runner(command, _input, _timeout)
|
|
|
|
result = preflight.collect_preflight(args_for(preflight.DEFAULT_PACKET, tmp_path), runner=drift_runner)
|
|
|
|
assert result["status"] == "fail"
|
|
assert result["ready_for_authorized_telegram_visible_dc_run"] is False
|
|
assert result["remote_checks"]["deploy_head_matches_last_deploy"] is False
|
|
|
|
|
|
def test_markdown_reports_counts_and_claim_ceiling(tmp_path: Path):
|
|
result = preflight.collect_preflight(args_for(preflight.DEFAULT_PACKET, tmp_path), runner=good_direct_runner)
|
|
out = tmp_path / "preflight.md"
|
|
|
|
preflight.write_markdown(out, result)
|
|
|
|
text = out.read_text(encoding="utf-8")
|
|
assert "Telegram-Visible Direct-Claim Preflight" in text
|
|
assert "Ready for authorized Telegram-visible DC run: `True`" in text
|
|
assert "Remote source: `ssh_direct_commands`" in text
|
|
assert "Remote readback command:" in text
|
|
assert "`public.claims`: `1837`" in text
|
|
assert "does not send Telegram messages" in text
|
|
|
|
|
|
def test_markdown_reports_clear_blocker_cta_on_failure(tmp_path: Path):
|
|
def failing_runner(_command: list[str], _input: str, _timeout: int) -> preflight.CommandResult:
|
|
return preflight.CommandResult(255, "", "ssh denied\n")
|
|
|
|
result = preflight.collect_preflight(args_for(preflight.DEFAULT_PACKET, tmp_path), runner=failing_runner)
|
|
out = tmp_path / "preflight-fail.md"
|
|
|
|
preflight.write_markdown(out, result)
|
|
|
|
text = out.read_text(encoding="utf-8")
|
|
assert "## Blocker Readback" in text
|
|
assert "Exact gate: `deploy: ssh denied" in text
|
|
assert "--remote-json-base64 <payload>" in text
|