"""v2.4 tests for design review V1-V8 (beyond the original repro tests in test_review_r1.py)."""
import json, os, shutil, sys
import pytest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import aicep_sub
from aicep_common import Ledger, Reject, canonical, b64
from aicep_pub import publish
from aicep_sub import Held, ingest
from test_delivery import w, send, OBJ # noqa: F401
from test_tamper import env, pub, sub, resign, NOW # noqa: F401
def skid_seq(d):
m = json.load(open(os.path.join(d, "manifest.json"))); return m["sender_key_id"], m["seq"]
# V1: a forged manifest sitting in holding/ is dead-lettered, never alerted on
def test_forged_batch_in_holding_goes_to_dlq(w):
res = send(w); d = res["batch_dir"]
files = {n: open(os.path.join(d, n), "rb").read() for n in os.listdir(d)}
m = json.loads(files["manifest.json"]); m["seq"] = 50; files["manifest.json"] = canonical(m)
hp = os.path.join(w["bus"].root, "c1", "holding", "forged-h"); os.makedirs(hp)
for n, b in files.items():
open(os.path.join(hp, n), "wb").write(b)
r = w["recv"](); w["clk"].t += 4000
out = r.run_once()
assert out["forged-h"] == "dlq"
assert not any(n["event"] == "gap_timeout" for n in r.notices)
# V2b: commit merges, never rolls back last_seq written by another receiver
def test_commit_state_never_rolls_back_last_seq(w):
r = w["recv"]()
aicep_sub.atomic_write(r.sub_state, json.dumps({"last_seq": {"k": 9, "o": 1}}).encode())
tmp = os.path.join(r.dir, "sub_state.x.tmp")
aicep_sub.atomic_write(tmp, json.dumps({"last_seq": {"k": 3, "o": 2}}).encode())
r._commit_state(tmp)
assert json.load(open(r.sub_state))["last_seq"] == {"k": 9, "o": 2}
# V2c: losing the pub->processing rename race skips the batch instead of aborting fetch
def test_fetch_rename_race_skips(w, monkeypatch):
send(w); send(w)
real = os.rename; calls = []
def flaky(a, b):
calls.append(a)
if len(calls) == 1:
raise FileNotFoundError(a)
return real(a, b)
monkeypatch.setattr(os, "rename", flaky)
got = w["bus"].fetch("c1", 10, 300, "r1", folders={"agent-a"})
assert len(got) == 1
# V3: ingest-level downgrade guard (CLI path too)
def test_ingest_rejects_v1_after_chain(env):
d1 = pub(env, prev_batch_id="genesis"); sub(env, d1)
d2 = pub(env) # no prev_batch_id
with pytest.raises(Reject, match="chain downgrade"):
sub(env, d2)
def test_ingest_rejects_wrong_prev(env):
d1 = pub(env, prev_batch_id="genesis"); sub(env, d1)
d2 = pub(env, prev_batch_id="not-the-last")
with pytest.raises(Reject, match="chain break"):
sub(env, d2)
# V4: missing size_padded is a schema reject
def test_missing_size_padded_rejected(env):
d = pub(env); resign(env, d, lambda m: [f.pop("size_padded") for f in m["files"]])
with pytest.raises(Reject, match="schema"):
sub(env, d)
# V5: receiver routes a signed-but-malformed manifest to the DLQ (no exception, no poison loop)
def test_receiver_non_int_size_padded_to_dlq(w):
res = send(w); d = os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"])
mp = os.path.join(d, "manifest.json"); m = json.load(open(mp))
for f in m["files"]:
f["size_padded"] = "big"
mb = canonical(m); open(mp, "wb").write(mb); open(mp + ".sig", "w").write(b64(w["sk"].sign(mb)))
assert w["recv"]().run_once()[res["batch_id"]] == "dlq"
def test_non_int_size_padded_reject(env):
d = pub(env); resign(env, d, lambda m: [f.__setitem__("size_padded", "4096") for f in m["files"]])
with pytest.raises(Reject):
sub(env, d)
# V6: expected_head mismatch and malformed lines fail verify (no exception)
def test_ledger_expected_head_and_malformed(tmp_path):
L = Ledger(str(tmp_path / "l.jsonl")); h1 = L.append("a", t=1); L.append("b", t=2)
assert L.verify() and L.verify(expected_head=h1)
assert not L.verify(expected_head="f" * 64)
open(L.path, "a").write("{not json\n")
assert L.verify() is False
# V7: created in the future (beyond skew) is rejected
def test_future_created_rejected(env):
d = pub(env); resign(env, d, lambda m: m.__setitem__("created", NOW + 10_000))
with pytest.raises(Reject, match="future"):
sub(env, d)
# V8: exact-match human gap ack, logged to the ledger
def test_ack_gap_to_must_match_and_is_logged(env):
pub(env); d2 = pub(env); skid, seq = skid_seq(d2)
kw = dict(now=NOW + 5)
with pytest.raises(Held):
ingest(d2, env["ident"], env["registry"], env["sstate"], "c1", ack_gap_to=f"{skid}:{seq + 1}", operator="h", **kw)
with pytest.raises(Held):
ingest(d2, env["ident"], env["registry"], env["sstate"], "c1", ack_gap_to=f"ed25519:other:{seq}", operator="h", **kw)
with pytest.raises(ValueError):
ingest(d2, env["ident"], env["registry"], env["sstate"], "c1", ack_gap_to=f"{skid}:{seq}", **kw)
out = ingest(d2, env["ident"], env["registry"], env["sstate"], "c1", ack_gap_to=f"{skid}:{seq}", operator="h", **kw)
assert out["gap_alarm"] is True
L = Ledger(os.path.join(os.path.dirname(env["sstate"]), "ledger.jsonl"))
e = [x for x in L.entries() if x["event"] == "gap_acked"]
assert e and e[0]["operator"] == "h" and e[0]["skid"] == skid and (e[0]["from"], e[0]["to"]) == (0, seq)
assert L.verify()
def test_cli_has_no_blanket_ack_gap():
src = open(aicep_sub.__file__).read()
assert '"--ack-gap"' not in src and '"--ack-gap-to"' in src
# ---- v2.4 fixes (design review re-review W1-W4)
import subprocess
from aicep_common import LedgerError, MAX_TTL
def test_w1_expires_not_after_created_rejected(env):
d = pub(env); resign(env, d, lambda m: m.__setitem__("expires", m["created"]))
with pytest.raises(Reject, match="expires <= created"):
sub(env, d)
def test_w1_ttl_at_cap_ok_over_cap_rejected(env):
sub(env, pub(env, ttl=MAX_TTL))
with pytest.raises(Reject, match="MAX_TTL"):
sub(env, pub(env, ttl=MAX_TTL + 1))
def test_w3_reanchor_is_human_only_and_logged(tmp_path):
p = str(tmp_path / "l.jsonl"); L = Ledger(p); L.append("e", t=1); L.append("e", t=2)
os.remove(L.anchor_path)
with pytest.raises(LedgerError):
L.append("e", t=3)
with pytest.raises(PermissionError):
L.reanchor("")
L.reanchor("human-x", t=4)
assert L.verify() and L.entries()[-1]["event"] == "reanchored" and L.entries()[-1]["operator"] == "human-x"
L.append("e", t=5); assert L.verify()
def test_w3_reanchor_refuses_broken_links(tmp_path):
p = str(tmp_path / "l.jsonl"); L = Ledger(p); L.append("e", t=1); L.append("e", t=2); L.append("e", t=3)
lines = open(p).read().splitlines(True); open(p, "w").writelines([lines[0], lines[2]])
with pytest.raises(LedgerError):
L.reanchor("human-x")
def test_w3_cli_reanchor(tmp_path):
p = str(tmp_path / "l.jsonl"); L = Ledger(p); L.append("e", t=1); os.remove(L.anchor_path)
sub_py = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "aicep_sub.py")
assert subprocess.run([sys.executable, sub_py, "--reanchor-ledger", p], capture_output=True).returncode != 0
r = subprocess.run([sys.executable, sub_py, "--reanchor-ledger", p, "--operator", "human-x"], capture_output=True)
assert r.returncode == 0 and Ledger(p).verify()
def test_w3_receiver_start_fails_closed_on_truncated_ledger(w):
send(w); send(w); w["recv"]().run_once()
p = w["t"] / "r" / "ledger.jsonl"
lines = open(p).read().splitlines(True); open(p, "w").writelines(lines[:-3])
with pytest.raises(LedgerError):
w["recv"]()
Ledger(str(p)).reanchor("human-x") # human re-anchor still can't hide receipt heads that are gone
with pytest.raises(LedgerError):
w["recv"]()
def test_w4_existing_v1_sender_then_genesis_chain(w):
def put(seq, prev):
b = publish(OBJ, [w["recip"]], w["sk"], "A", "c1", str(w["t"] / "f"), str(w["t"] / "fps.json"),
now=w["clk"].t, seq=seq, ttl=3600, prev_batch_id=prev)
w["bus"].publish(b["batch_dir"], {"compartment": "c1", "batch_id": b["batch_id"], "folder": "agent-a"}); return b["batch_id"]
r = w["recv"]()
b1 = put(1, None); assert r.run_once()[b1][0] == "committed" # legacy v1 history
b2 = put(2, "genesis"); assert r.run_once()[b2][0] == "committed" # outbox with no last_published
b3 = put(3, b2); assert r.run_once()[b3][0] == "committed"
b4 = put(4, "wrong"); assert r.run_once()[b4] == "dlq"
def test_w3_reanchor_blank_operator_refused(tmp_path):
L = Ledger(str(tmp_path / "l.jsonl")); L.append("e", t=1); os.remove(L.anchor_path)
for op in ("", " ", "\t\n", None):
with pytest.raises(PermissionError):
L.reanchor(op)
L.reanchor(" human-x ", t=2)
assert L.entries()[-1]["operator"] == "human-x" and L.verify()
