reference/tests/test_review_r4.py

"""design review v2.5 review: strict-xfail tests for real bugs found in per-agent folders. Flip to pass when fixed."""
import 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__)))
from test_delivery import w, send  # noqa: F401
from test_review_r3 import add_sender

UNKNOWN_CAP = 64      # any fixed bound is fine; must not be O(number of junk folders)


def P(w, *a):
    return os.path.join(w["bus"].root, "c1", *a)


# fixed v2.5.1 (was strict xfail): "X1: unknown_folders_logged grows without bound (one entry + ledger row per junk folder)"
def test_x1_unknown_folder_log_is_bounded(w):
    send(w)
    for i in range(300):
        os.makedirs(P(w, "pub", f"junk-{i}"))
    r = w["recv"](); r.run_once()
    assert len(r._r().get("unknown_folders_logged", [])) <= UNKNOWN_CAP


# fixed v2.5.1 (was strict xfail): "X2: every folder_mismatch emits 2 email alerts (dlq + folder_mismatch), no coalescing"
def test_x2_folder_mismatch_alerts_are_coalesced(w):
    add_sender(w, "agent-beta")
    a = send(w)
    for i in range(20):
        shutil.copytree(P(w, "pub", "agent-a", a["batch_id"]), P(w, "pub", "agent-beta", f"x{i}"))
    r = w["recv"](); r.run_once()
    assert sum(1 for n in r.notices if n.get("email")) <= 2


# fixed v2.5.1 (was strict xfail): "X3: held batch released to its stale _meta folder after a folder migration, stranding the chain"
def test_x3_held_batch_survives_folder_migration(w):
    a1 = send(w); a2 = send(w); b1 = a1["batch_id"]
    shutil.move(P(w, "pub", "agent-a", b1), w["t"] / "stash")       # seq1 late -> seq2 held
    assert w["recv"]().run_once()[a2["batch_id"]] == "held"
    for e in w["reg"].values():                                      # documented migration: rename + move
        if e["folder"] == "agent-a":
            e["folder"] = "agent-a2"
    os.rename(P(w, "pub", "agent-a"), P(w, "pub", "agent-a2"))
    shutil.move(w["t"] / "stash", P(w, "pub", "agent-a2", b1))
    out = w["recv"]().run_once(); out.update(w["recv"]().run_once())
    assert out.get(a2["batch_id"], ("",))[0] == "committed"