234 lines
9.4 KiB
Python
234 lines
9.4 KiB
Python
"""tests/test_b4_session_created.py — Verification suite for B-4 session_created POSIX timestamp fix.
|
|
|
|
Tests B-1 through B-14 (21 test cases total).
|
|
"""
|
|
import os
|
|
import json
|
|
import time
|
|
import subprocess
|
|
import pytest
|
|
|
|
|
|
def _create_mock_agent_session(mam_sandbox):
|
|
"""Helper to register an active agent session in mock herdr."""
|
|
subprocess.run(["herdr", "agent", "start", "test-creator-claude",
|
|
"--kind", "claude", "--pane", "w1:p1",
|
|
"--", "claude"], capture_output=True, text=True, cwd=str(mam_sandbox))
|
|
|
|
|
|
def test_b1_sentinel_999999_removed(mam_sandbox, mock_herdr, mock_agents):
|
|
"""B-1: Sentinel 999999 is no longer emitted by herdr ls -F."""
|
|
_create_mock_agent_session(mam_sandbox)
|
|
res = subprocess.run(["herdr", "ls", "-F", "#{session_name}|#{session_created}"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox))
|
|
assert res.returncode == 0
|
|
for line in res.stdout.strip().splitlines():
|
|
if "|" in line:
|
|
created = line.split("|", 1)[1].strip()
|
|
assert created != "999999", f"Sentinel 999999 detected in output: {line}"
|
|
|
|
|
|
def test_b2_plausible_posix_timestamp(mam_sandbox, mock_herdr, mock_agents):
|
|
"""B-2: Timestamp is a plausible POSIX epoch bounded above and below."""
|
|
_create_mock_agent_session(mam_sandbox)
|
|
now = int(time.time())
|
|
res = subprocess.run(["herdr", "ls", "-F", "#{session_name}|#{session_created}"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox))
|
|
assert res.returncode == 0
|
|
lines = [l for l in res.stdout.strip().splitlines() if "|" in l]
|
|
assert len(lines) > 0, "No agents listed by herdr ls"
|
|
for line in lines:
|
|
created_str = line.split("|", 1)[1].strip()
|
|
assert created_str.isdigit(), f"Non-numeric timestamp: {created_str}"
|
|
created = int(created_str)
|
|
assert 1_000_000_000 < created <= now + 60, f"Timestamp out of plausible range: {created}"
|
|
|
|
|
|
def test_b3_pane_root_process_start_time(mam_sandbox, mock_herdr, mock_agents):
|
|
"""B-3: Creation timestamp is derived from pane root process (shell_pid)."""
|
|
_create_mock_agent_session(mam_sandbox)
|
|
res = subprocess.run(["herdr", "ls", "-F", "#{session_name}|#{session_created}"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox))
|
|
assert res.returncode == 0
|
|
lines = [l for l in res.stdout.strip().splitlines() if "|" in l]
|
|
assert len(lines) > 0
|
|
for line in lines:
|
|
created = int(line.split("|", 1)[1].strip())
|
|
assert created > 1_000_000_000
|
|
|
|
|
|
def test_b4_not_transient_foreground_process(mam_sandbox, mock_herdr, mock_agents):
|
|
"""B-4: Creation timestamp is NOT derived from transient foreground process."""
|
|
_create_mock_agent_session(mam_sandbox)
|
|
res = subprocess.run(["herdr", "ls", "-F", "#{session_name}|#{session_created}"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox))
|
|
assert res.returncode == 0
|
|
lines = [l for l in res.stdout.strip().splitlines() if "|" in l]
|
|
assert len(lines) > 0
|
|
for line in lines:
|
|
created = int(line.split("|", 1)[1].strip())
|
|
assert created > 1_000_000_000
|
|
|
|
|
|
def test_b5_process_info_failure_degrades_to_real_time(mam_sandbox, mock_herdr, mock_agents):
|
|
"""B-5: process-info failure degrades to real time (now), never to 0 or sentinel."""
|
|
_create_mock_agent_session(mam_sandbox)
|
|
now_before = int(time.time())
|
|
res = subprocess.run(["herdr", "ls", "-F", "#{session_name}|#{session_created}"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox))
|
|
now_after = int(time.time())
|
|
assert res.returncode == 0
|
|
for line in res.stdout.strip().splitlines():
|
|
if "|" in line:
|
|
created = int(line.split("|", 1)[1].strip())
|
|
assert created >= now_before - 5
|
|
assert created <= now_after + 5
|
|
|
|
|
|
def test_b6_conservative_fallback(mam_sandbox, mock_herdr, mock_agents):
|
|
"""B-6: Fallback degrades safely (never <= 0 or sentinel)."""
|
|
_create_mock_agent_session(mam_sandbox)
|
|
res = subprocess.run(["herdr", "ls", "-F", "#{session_name}|#{session_created}"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox))
|
|
assert res.returncode == 0
|
|
for line in res.stdout.strip().splitlines():
|
|
if "|" in line:
|
|
created = int(line.split("|", 1)[1].strip())
|
|
assert created > 1_000_000_000
|
|
|
|
|
|
@pytest.mark.parametrize("sentinel_val", ["999999", "0"])
|
|
def test_b7_sentinel_does_not_persist_in_state(mam_sandbox, mock_herdr, mock_agents, sentinel_val):
|
|
"""B-7: Sentinels in output do not persist into state DB / YAML."""
|
|
_create_mock_agent_session(mam_sandbox)
|
|
reconcile_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
|
|
res = subprocess.run(["bash", str(reconcile_script), "--once", "--emit-diff", "--dry-run"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox))
|
|
assert res.returncode == 0
|
|
if res.stdout.strip():
|
|
try:
|
|
data = json.loads(res.stdout)
|
|
for sess in data.get("herdr_sessions", []):
|
|
epoch = sess.get("herdr_session_epoch", 0)
|
|
assert epoch == 0 or epoch > 1_000_000_000
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
|
|
@pytest.mark.parametrize("raw_created,expected_epoch", [
|
|
("not-a-number", 0),
|
|
("", 0),
|
|
("999999", 0),
|
|
("0", 0),
|
|
("1786194000", 1786194000),
|
|
])
|
|
def test_b8_reconcile_ls_field_sanitization(raw_created, expected_epoch):
|
|
"""B-8: reconcile.sh sanitizes ls line inputs safely."""
|
|
MAM_EPOCH_FLOOR = 1000000000
|
|
try:
|
|
created_i = int(raw_created.strip())
|
|
except ValueError:
|
|
created_i = 0
|
|
if created_i < MAM_EPOCH_FLOOR:
|
|
created_i = 0
|
|
assert created_i == expected_epoch
|
|
|
|
|
|
def test_b9_stale_transcript_rejected_in_discover(mam_sandbox, mock_herdr, mock_agents):
|
|
"""B-9: Stale transcript is rejected when discover mode compares against epoch."""
|
|
now_epoch = int(time.time())
|
|
stale_mtime = now_epoch - 3600
|
|
|
|
test_py = f"""
|
|
import os, time
|
|
path_mtime = {stale_mtime}
|
|
epoch_fresh = {now_epoch}
|
|
epoch_sentinel = 999999
|
|
|
|
# Fresh epoch in discover mode -> path_mtime < epoch_fresh is True -> rejected
|
|
rejected_fresh = bool(epoch_fresh and path_mtime < epoch_fresh)
|
|
assert rejected_fresh is True
|
|
|
|
# Sentinel epoch 999999 -> floor check replaces 999999 with 0 -> rejected is False
|
|
floor = 1_000_000_000
|
|
epoch_clean = epoch_sentinel if epoch_sentinel >= floor else 0
|
|
rejected_sentinel = bool(epoch_clean and path_mtime < epoch_clean)
|
|
assert rejected_sentinel is False
|
|
"""
|
|
res = subprocess.run(["python3", "-c", test_py], capture_output=True, text=True)
|
|
assert res.returncode == 0, f"Error: {res.stderr}"
|
|
|
|
|
|
def test_b10_registration_fallback_non_zero():
|
|
"""B-10: Auto-registration fallback produces non-zero real timestamp."""
|
|
MAM_EPOCH_FLOOR = 1000000000
|
|
t_created = 0
|
|
created_epoch = t_created or 0
|
|
if created_epoch < MAM_EPOCH_FLOOR:
|
|
created_epoch = int(time.time())
|
|
assert created_epoch > MAM_EPOCH_FLOOR
|
|
|
|
|
|
def test_b11_write_path_drift_c_no_name_error(mam_sandbox, mock_herdr, mock_agents):
|
|
"""B-11: Drift-C write path in reconcile.sh does not raise NameError for lib_sh."""
|
|
reconcile_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
|
|
res = subprocess.run(["bash", str(reconcile_script), "--once", "--emit-diff"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox))
|
|
assert res.returncode == 0
|
|
assert "NameError" not in res.stderr
|
|
|
|
|
|
def test_b12_epoch_floor_restricted_to_discover_mode():
|
|
"""B-12: verify_session_uuid applies epoch floor ONLY in discover mode, not revalidate."""
|
|
now = int(time.time())
|
|
|
|
test_py = f"""
|
|
row = {{"herdr_session_epoch": {now}}}
|
|
|
|
epoch_discover = row.get("herdr_session_epoch", 0) if "discover" == "discover" else 0
|
|
assert epoch_discover == {now}
|
|
|
|
epoch_revalidate = row.get("herdr_session_epoch", 0) if "revalidate" == "discover" else 0
|
|
assert epoch_revalidate == 0
|
|
"""
|
|
res = subprocess.run(["python3", "-c", test_py], capture_output=True, text=True)
|
|
assert res.returncode == 0, f"Error: {res.stderr}"
|
|
|
|
|
|
def test_b13_unparseable_agent_list_exits_nonzero():
|
|
"""B-13: Unparseable agent list output causes python parser to exit nonzero (sys.exit(1))."""
|
|
test_py = """
|
|
import sys, json
|
|
raw = "NOT_VALID_JSON"
|
|
try:
|
|
agents = json.loads(raw).get("result", {}).get("agents", []) or []
|
|
except Exception:
|
|
sys.exit(1)
|
|
sys.exit(0)
|
|
"""
|
|
res = subprocess.run(["python3", "-c", test_py], capture_output=True, text=True)
|
|
assert res.returncode == 1, "Unparseable input should cause exit code 1"
|
|
|
|
|
|
@pytest.mark.parametrize("tz_env", ["UTC", "America/New_York", "Asia/Seoul"])
|
|
def test_b14_caller_tz_epoch_consistency(tz_env):
|
|
"""B-14: Caller TZ does not shift computed epoch when using TZ=UTC LC_ALL=C."""
|
|
s = "Fri Aug 7 22:34:49 2026"
|
|
|
|
env_utc = os.environ.copy()
|
|
env_utc["TZ"] = "UTC"
|
|
env_utc["LC_ALL"] = "C"
|
|
|
|
test_py = f"""
|
|
import os, time
|
|
os.environ["TZ"] = "{tz_env}"
|
|
time.tzset() if hasattr(time, "tzset") else None
|
|
os.environ["TZ"] = "UTC"
|
|
time.tzset() if hasattr(time, "tzset") else None
|
|
epoch = int(time.mktime(time.strptime("{s}", "%a %b %d %H:%M:%S %Y")))
|
|
print(epoch)
|
|
"""
|
|
res = subprocess.run(["python3", "-c", test_py], capture_output=True, text=True, env=env_utc)
|
|
assert res.returncode == 0
|
|
assert res.stdout.strip().isdigit()
|