diff --git a/prod/clean_arch/dita_v2/build_native_artifact.py b/prod/clean_arch/dita_v2/build_native_artifact.py new file mode 100644 index 0000000..fe1667f --- /dev/null +++ b/prod/clean_arch/dita_v2/build_native_artifact.py @@ -0,0 +1,36 @@ +"""Build and verify the DITAv2 Rust artifact outside the live runner. + +Usage: + python -m prod.clean_arch.dita_v2.build_native_artifact + python -m prod.clean_arch.dita_v2.build_native_artifact --rollback +""" + +from __future__ import annotations + +import argparse +from pathlib import Path + +from .native_artifact import build_verified_artifact, rollback_artifact + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--target-dir", + type=Path, + default=Path("/root/.cargo/dita_v2_target"), + ) + parser.add_argument("--rollback", action="store_true") + args = parser.parse_args() + crate_dir = Path(__file__).resolve().with_name("_rust_kernel") + result = ( + rollback_artifact(args.target_dir, crate_dir) + if args.rollback + else build_verified_artifact(crate_dir, args.target_dir) + ) + print(result) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/prod/clean_arch/dita_v2/conftest.py b/prod/clean_arch/dita_v2/conftest.py new file mode 100644 index 0000000..b0578f8 --- /dev/null +++ b/prod/clean_arch/dita_v2/conftest.py @@ -0,0 +1,10 @@ +"""DITAv2 pytest collection policy. + +The files below are source generators or frozen backups, not executable tests. +Both ``_gen_test.py`` paths match pytest's default ``*_test.py`` pattern and +perform file generation at import time, so collecting them corrupts the test +boundary and can fail before the real suite loads. +""" + +collect_ignore = ["_gen_test.py"] +collect_ignore_glob = ["_backup_20260530/*"] diff --git a/prod/clean_arch/dita_v2/test_account_region_inmem.py b/prod/clean_arch/dita_v2/test_account_region_inmem.py new file mode 100644 index 0000000..5284aa0 --- /dev/null +++ b/prod/clean_arch/dita_v2/test_account_region_inmem.py @@ -0,0 +1,1381 @@ +"""Phase-0 tests: InMemoryZincPlane account region — deep coverage. + +Test TYPES (not just count): + 1. Property-based: hypothesis-driven fuzz over random snapshots + 2. Combinatorial: param grid over fields, states, sequences + 3. Concurrency: race detectors, starvation, multi-waiter order + 4. Behavioral: invariants that must hold for ANY sequence + 5. Mutational: known bug injection → test must RED + 6. Contract: TypeScript-style protocol satisfaction + 7. State-machine: exhaustive small-state-space enumeration + 8. Serialization: payload→snapshot→dict→payload→snapshot identity + 9. Stress: long chains, GC pressure, deterministic replay + 10. Leak: no reference escape across publish/read boundary + +Every mutational test is annotated: # MUTATION: → +""" + +from __future__ import annotations + +import gc +import math +import random +import sys +import threading +import time +import copy +import itertools + +sys.path.insert(0, "/mnt/dolphinng5_predict") + +from prod.clean_arch.dita_v2.contracts import AccountStateSnapshot +from prod.clean_arch.dita_v2.zinc_plane import InMemoryZincPlane +from prod.clean_arch.dita_v2.real_zinc_plane import _account_from_payload + +_SEED = 42 +random.seed(_SEED) + +# ══════════════════════════════════════════════════════════════════════════════ +# 1. PROPERTY-BASED — field-level generators + invariants +# ══════════════════════════════════════════════════════════════════════════════ + +def _random_snapshot(rng: random.Random = random) -> AccountStateSnapshot: + """Generate a random valid AccountStateSnapshot.""" + wb = rng.uniform(-1e6, 1e7) + am = rng.uniform(0, max(0, wb)) if wb > 0 else rng.uniform(-1e6, 1e6) + um = rng.uniform(0, max(0, wb - am)) if (wb - am) > 0 else rng.uniform(-1e6, 1e6) + return AccountStateSnapshot( + wallet_balance=wb, + available_margin=am, + used_margin=um, + event_seq=rng.randint(0, 2**31 - 1), + mono_ns=rng.randint(0, 2**63 - 1), + e_live=rng.choice([True, False]), + reconcile_ok=rng.choice([True, False]), + ) + + +class TestPropertyBased: + """Invariants that hold across many random snapshots.""" + + def test_roundtrip_property(self) -> None: + """For ANY snapshot, publish→read fields survive.""" + plane = InMemoryZincPlane() + for i in range(2000): + snap = _random_snapshot() + plane.publish_account(snap) + read = plane.read_account() + assert read.wallet_balance == snap.wallet_balance, f"iter {i}: wb {read.wallet_balance} != {snap.wallet_balance}" + assert read.available_margin == snap.available_margin + assert read.used_margin == snap.used_margin + assert read.event_seq == snap.event_seq + assert read.mono_ns == snap.mono_ns + assert read.e_live is snap.e_live + assert read.reconcile_ok is snap.reconcile_ok + + def test_payload_identity_property(self) -> None: + """For ANY snapshot, _account_from_payload(as_dict()) == snap.""" + for i in range(2000): + snap = _random_snapshot() + decoded = _account_from_payload(snap.as_dict()) + assert decoded == snap, f"iter {i}: {decoded} != {snap}" + + def test_as_dict_types_property(self) -> None: + """For ANY snapshot, as_dict() returns correct types.""" + for i in range(2000): + snap = _random_snapshot() + d = snap.as_dict() + assert isinstance(d["wallet_balance"], float), type(d["wallet_balance"]) + assert isinstance(d["available_margin"], float) + assert isinstance(d["used_margin"], float) + assert isinstance(d["event_seq"], int) + assert isinstance(d["mono_ns"], int) + assert isinstance(d["e_live"], bool) + assert isinstance(d["reconcile_ok"], bool) + + def test_default_is_resilient_property(self) -> None: + """Fresh plane survives 2000 random operations without state corruption.""" + plane = InMemoryZincPlane() + for i in range(2000): + snap = _random_snapshot() + plane.publish_account(snap) + _ = plane.read_account() + _ = plane.wait_on_account(timeout_ms=0) + # After all ops, a new publish works + plane.publish_account(_random_snapshot()) + assert plane.read_account() is not None + + def test_seq_monotonic_property(self) -> None: + """Internal seq always increases, never wraps negative.""" + plane = InMemoryZincPlane() + for i in range(2000): + prev_seq = plane._account_seq + plane.publish_account(_random_snapshot()) + assert plane._account_seq > prev_seq, f"iter {i}: seq did not increase" + + def test_underscore_fields_match_publish(self) -> None: + """After N random publishes, field values match the LAST published.""" + plane = InMemoryZincPlane() + last = None + for i in range(500): + last = _random_snapshot() + plane.publish_account(last) + read = plane.read_account() + assert read == last, f"after 500 publishes, last snap doesn't match" + + def test_random_poison_always_defaults(self) -> None: + """Random garbage in payload → default snapshot, never exception.""" + rng = random.Random(999) + for i in range(2000): + # generate a dict with random key/value types + n_fields = rng.randint(0, 20) + payload = {} + for _ in range(n_fields): + k = rng.choice(["wallet_balance", "available_margin", "used_margin", + "event_seq", "mono_ns", "e_live", "reconcile_ok", + "foo", "bar", "nested", "_internal"]) + t = rng.choice([None, "string", 42, 3.14, True, [1,2], {"a":1}, b"bytes"]) + payload[k] = t + result = _account_from_payload(payload) + assert isinstance(result, AccountStateSnapshot) + + +class TestPropertyBasedBoundary: + """Boundary values around float/int limits.""" + + def test_fp_ulp_preserved(self) -> None: + """Last-ulp precision preserved — no rounding on transport.""" + plane = InMemoryZincPlane() + snap = AccountStateSnapshot( + wallet_balance=0.1 + 0.2, # famously not 0.3 in binary + available_margin=1.0 / 3.0, + used_margin=math.pi, + event_seq=0, mono_ns=0, e_live=True, reconcile_ok=False, + ) + plane.publish_account(snap) + read = plane.read_account() + assert read.wallet_balance == 0.1 + 0.2 + assert read.available_margin == 1.0 / 3.0 + assert read.used_margin == math.pi + + def test_float_min_positive(self) -> None: + plane = InMemoryZincPlane() + tiny = AccountStateSnapshot( + wallet_balance=math.ulp(0.0), # smallest representable positive float + available_margin=0.0, used_margin=0.0, + event_seq=0, mono_ns=0, e_live=False, reconcile_ok=False, + ) + plane.publish_account(tiny) + read = plane.read_account() + assert read.wallet_balance == math.ulp(0.0) + assert read.wallet_balance > 0.0 + + def test_large_int_exact(self) -> None: + """Large ints in event_seq/mono_ns survive exactly.""" + plane = InMemoryZincPlane() + snap = AccountStateSnapshot( + wallet_balance=0.0, available_margin=0.0, used_margin=0.0, + event_seq=2**31 - 1, mono_ns=2**63 - 1, # max int32 / int64 + e_live=True, reconcile_ok=True, + ) + plane.publish_account(snap) + read = plane.read_account() + assert read.event_seq == 2**31 - 1 + assert read.mono_ns == 2**63 - 1 + + def test_float_edge_values(self) -> None: + """Special floats survive.""" + plane = InMemoryZincPlane() + for val in [0.0, -0.0, 1e-300, 1e300, 1.0, -1.0, 1.7976931348623157e308]: + plane.publish_account(AccountStateSnapshot( + wallet_balance=val, available_margin=val, used_margin=val, + event_seq=0, mono_ns=0, e_live=True, reconcile_ok=False, + )) + read = plane.read_account() + if math.isnan(val): + assert math.isnan(read.wallet_balance) + else: + assert read.wallet_balance == val, f"val={val} got={read.wallet_balance}" + assert read.available_margin == val + assert read.used_margin == val + + +# ══════════════════════════════════════════════════════════════════════════════ +# 2. COMBINATORIAL — param grid over fields, states, sequences +# ══════════════════════════════════════════════════════════════════════════════ + +# All possible values for each field (equiv classes) +_WB_VALS = [0.0, 100.0, -100.0, 1e10, 0.000001, math.inf, float("nan"), -0.0] +_AM_VALS = [0.0, 50.0, -50.0, 1e10, 0.0] +_UM_VALS = [0.0, 30.0, -30.0, 1e10, 0.0] +_ES_VALS = [0, 1, 2**31 - 1, -1] +_MN_VALS = [0, 1000, 2**63 - 1, -1] +_BOOL_VALS = [True, False] + + +class TestCombinatorialFieldValues: + """Cartesian product of all equivalence-class field values.""" + + def _test_grid(self, wb, am, um, es, mn, el, ro) -> None: + """Single grid-point test: publish, read, assert identity.""" + if (isinstance(wb, float) and (math.isinf(wb) or math.isnan(wb))): + return # skip inf/nan for margin constraints + if isinstance(am, float) and (math.isinf(am) or math.isnan(am)): + return + if isinstance(um, float) and (math.isinf(um) or math.isnan(um)): + return + snap = AccountStateSnapshot( + wallet_balance=wb, available_margin=am, used_margin=um, + event_seq=es, mono_ns=mn, e_live=el, reconcile_ok=ro, + ) + plane = InMemoryZincPlane() + plane.publish_account(snap) + read = plane.read_account() + assert abs(read.wallet_balance - wb) < 1e-12 or (math.isnan(wb) and math.isnan(read.wallet_balance)), f"wb: {wb} vs {read.wallet_balance}" + assert abs(read.available_margin - am) < 1e-12 or (math.isnan(am) and math.isnan(read.available_margin)) + assert abs(read.used_margin - um) < 1e-12 or (math.isnan(um) and math.isnan(read.used_margin)) + assert read.event_seq == es + assert read.mono_ns == mn + assert read.e_live is el + assert read.reconcile_ok is ro + + def test_all_combinations(self) -> None: + """Test every combination of equivalence-class values.""" + # 8 * 5 * 5 * 4 * 4 * 2 * 2 = 12,800 tests + cart = list(itertools.product(_WB_VALS, _AM_VALS, _UM_VALS, _ES_VALS, _MN_VALS, _BOOL_VALS, _BOOL_VALS)) + count = len(cart) + for idx, (wb, am, um, es, mn, el, ro) in enumerate(cart): + # skip impossible combos: used > available > wallet + if isinstance(wb, (int, float)) and isinstance(am, (int, float)) and isinstance(um, (int, float)): + if not math.isnan(wb) and not math.isinf(wb) and not math.isnan(am) and not math.isinf(am) and not math.isnan(um) and not math.isinf(um): + if abs(um) > abs(am) + 1e-9 and abs(am) > abs(wb) + 1e-9: + continue + self._test_grid(wb, am, um, es, mn, el, ro) + assert count > 12000, f"expected >=12000 combos, got {count}" + + +class TestCombinatorialFieldOrder: + """Field ordering in as_dict doesn't affect identity.""" + + def test_as_dict_field_order_irrelevant(self) -> None: + snap = _random_snapshot() + d1 = snap.as_dict() + d2 = {k: d1[k] for k in reversed(list(d1.keys()))} + assert _account_from_payload(d2) == snap + + def test_as_dict_extra_keys_ignored(self) -> None: + snap = _random_snapshot() + d = snap.as_dict() + d["_extra"] = 42 + d["nested"] = {"a": 1} + assert _account_from_payload(d) == snap + + def test_as_dict_partial_ok(self) -> None: + """Partial dict — missing fields get defaults.""" + d = {"wallet_balance": 100.0} + result = _account_from_payload(d) + assert result.wallet_balance == 100.0 + assert result.available_margin == 0.0 + assert result.e_live is False + + +# ══════════════════════════════════════════════════════════════════════════════ +# 3. CONCURRENCY — race detectors, starvation, waiter ordering +# ══════════════════════════════════════════════════════════════════════════════ + +class TestConcurrencyStress: + """Heavy concurrency patterns.""" + + def test_100_threads_publish(self) -> None: + """100 threads, 100 publishes each = 10,000 concurrent operations.""" + plane = InMemoryZincPlane() + errors: list[Exception] = [] + lock = threading.Lock() + + def worker(n: int) -> None: + try: + for i in range(100): + plane.publish_account(AccountStateSnapshot( + wallet_balance=float(n * 1000 + i), + event_seq=n * 1000 + i, + mono_ns=n * 1000 + i, + e_live=True, reconcile_ok=True, + )) + except Exception as e: + with lock: + errors.append(e) + + threads = [threading.Thread(target=worker, args=(i,), daemon=True) for i in range(100)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=10) + assert len(errors) == 0, f"{len(errors)} errors: {errors[:5]}" + + def test_50_readers_50_writers(self) -> None: + """50 reader threads + 50 writer threads running simultaneously.""" + plane = InMemoryZincPlane() + errors: list[Exception] = [] + lock = threading.Lock() + + def writer() -> None: + try: + for i in range(50): + plane.publish_account(_random_snapshot()) + except Exception as e: + with lock: + errors.append(e) + + def reader() -> None: + try: + for _ in range(50): + snap = plane.read_account() + # verify internal consistency: used + available <= wallet + epsilon + # (not guaranteed due to race, but read should never crash) + _ = snap.as_dict() + except Exception as e: + with lock: + errors.append(e) + + threads = [threading.Thread(target=writer, daemon=True) for _ in range(50)] + threads += [threading.Thread(target=reader, daemon=True) for _ in range(50)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=10) + assert len(errors) == 0, f"{len(errors)} errors: {errors[:5]}" + + def test_wait_notify_chain_10_stages(self) -> None: + """10-stage chain: stage N publishes, stage N+1 waits for it.""" + plane = InMemoryZincPlane() + gates = [threading.Event() for _ in range(11)] + errors: list[str] = [] + error_lock = threading.Lock() + + def stage(n: int) -> None: + if not gates[n].wait(timeout=2.0): + with error_lock: + errors.append(f"stage {n}: predecessor gate timed out") + return + ok = plane.wait_on_account(timeout_ms=1000) + if not ok: + with error_lock: + errors.append(f"stage {n}: account wait timed out") + return + snap = plane.read_account() + if snap.event_seq != n: + with error_lock: + errors.append(f"stage {n}: observed event_seq={snap.event_seq}") + return + plane.publish_account( + AccountStateSnapshot( + wallet_balance=float(n + 1), + event_seq=n + 1, + mono_ns=n + 1, + e_live=True, + reconcile_ok=True, + ) + ) + if n < 10: + gates[n + 1].set() + + threads = [ + threading.Thread(target=stage, args=(i,), daemon=True) + for i in range(1, 11) + ] + for thread in threads: + thread.start() + # Seed stage 1 only after every worker is ready. Later stages remain + # behind explicit predecessor gates, so a broadcast cannot reorder them. + plane.publish_account(AccountStateSnapshot( + wallet_balance=1.0, event_seq=1, mono_ns=1, + e_live=True, reconcile_ok=True, + )) + gates[1].set() + for thread in threads: + thread.join(timeout=3.0) + + assert not [thread for thread in threads if thread.is_alive()], "chain left live workers" + assert errors == [] + assert plane.read_account().event_seq == 11 + + def test_signal_is_consumed_once_and_releases_all_waiters(self) -> None: + """One sequence change has one logical consumer; peers time out cleanly.""" + plane = InMemoryZincPlane() + observed = [False] * 20 + ready = threading.Barrier(21) + + def waiter(i: int) -> None: + ready.wait(timeout=2.0) + observed[i] = plane.wait_on_account(timeout_ms=250) + + threads = [threading.Thread(target=waiter, args=(i,), daemon=True) for i in range(20)] + for t in threads: + t.start() + ready.wait(timeout=2.0) + plane.notify_account() + for t in threads: + t.join(timeout=1.0) + assert not [thread for thread in threads if thread.is_alive()] + assert sum(observed) == 1, f"one sequence change was consumed {sum(observed)} times" + + def test_publish_during_wait_reentry(self) -> None: + """Wait inside a wait (nested) doesn't deadlock.""" + plane = InMemoryZincPlane() + inner_ok = [False] + + def inner_waiter() -> None: + # wait on a new InMemoryZincPlane while holding no lock + p2 = InMemoryZincPlane() + p2.publish_account(_random_snapshot()) + inner = p2.wait_on_account(timeout_ms=500) + inner_ok[0] = inner + + t = threading.Thread(target=inner_waiter, daemon=True) + t.start() + t.join(timeout=3) + assert inner_ok[0] is True + + def test_alternating_publish_read_no_stall(self) -> None: + """100k alternating publish/read operations.""" + plane = InMemoryZincPlane() + for i in range(50000): + if i % 2 == 0: + plane.publish_account(AccountStateSnapshot( + wallet_balance=float(i), event_seq=i, mono_ns=i, + e_live=True, reconcile_ok=(i % 3 == 0), + )) + else: + read = plane.read_account() + # Just check it doesn't crash — value may be stale, that's fine + _ = read.as_dict() + # After cycle completed, last publish is readable + final = plane.read_account() + assert final.event_seq >= 49998, f"seq={final.event_seq}" + + +# ══════════════════════════════════════════════════════════════════════════════ +# 4. BEHAVIORAL INVARIANTS — must hold for ANY sequence +# ══════════════════════════════════════════════════════════════════════════════ + +class TestBehavioralInvariants: + """Properties that must hold regardless of operation history.""" + + def test_read_never_returns_none(self) -> None: + plane = InMemoryZincPlane() + for _ in range(500): + snap = plane.read_account() + assert snap is not None + assert isinstance(snap, AccountStateSnapshot) + + def test_read_never_raises(self) -> None: + plane = InMemoryZincPlane() + # Interleave with other operations + for i in range(500): + if i % 3 == 0: + plane.publish_account(AccountStateSnapshot(event_seq=i)) + elif i % 3 == 1: + plane.notify_account() + # read must never raise + _ = plane.read_account() + + def test_publish_is_idempotent(self) -> None: + """Publishing the same snapshot twice results in same state.""" + plane = InMemoryZincPlane() + snap = _random_snapshot() + plane.publish_account(snap) + plane.publish_account(snap) # same snap again + assert plane.read_account() == snap + + def test_wait_on_zero_timeout_returns_immediately(self) -> None: + """wait_on_account with timeout_ms=0 checks once, no blocking.""" + plane = InMemoryZincPlane() + t0 = time.monotonic() + ok = plane.wait_on_account(timeout_ms=0) + elapsed = time.monotonic() - t0 + assert elapsed < 0.5, f"blocked for {elapsed}s" + assert ok is False # no event yet + + def test_wait_with_negative_timeout(self) -> None: + """Negative timeout_ms is treated as no timeout (infinite wait).""" + plane = InMemoryZincPlane() + + def delayed_publish() -> None: + time.sleep(0.02) + plane.publish_account(_random_snapshot()) + + publisher = threading.Thread(target=delayed_publish, daemon=True) + publisher.start() + t0 = time.monotonic() + ok = plane.wait_on_account(timeout_ms=-1) + elapsed = time.monotonic() - t0 + publisher.join(timeout=1.0) + assert ok is True + assert 0.01 <= elapsed < 1.0 + + def test_mono_ns_is_not_clobbered(self) -> None: + """mono_ns from publish survives — it's user-defined, not auto-overwritten.""" + plane = InMemoryZincPlane() + for ns in [0, 100, 1_000_000_000, 2**63 - 1]: + plane.publish_account(AccountStateSnapshot( + wallet_balance=0.0, available_margin=0.0, used_margin=0.0, + event_seq=0, mono_ns=ns, e_live=False, reconcile_ok=False, + )) + assert plane.read_account().mono_ns == ns + + def test_two_planes_independent(self) -> None: + """No shared state between two InMemoryZincPlanes.""" + p1 = InMemoryZincPlane() + p2 = InMemoryZincPlane() + for i in range(500): + s1 = _random_snapshot() + s2 = _random_snapshot() + p1.publish_account(s1) + p2.publish_account(s2) + assert p1.read_account() == s1 + assert p2.read_account() == s2 + + def test_wait_does_not_consume_event(self) -> None: + """wait_on_account doesn't consume the event — read still sees latest.""" + plane = InMemoryZincPlane() + plane.publish_account(AccountStateSnapshot( + wallet_balance=42.0, event_seq=1, mono_ns=1, e_live=True, reconcile_ok=True, + )) + assert plane.wait_on_account(timeout_ms=500) is True + # After wait, read still returns the same snapshot + assert plane.read_account().wallet_balance == 42.0 + assert plane.read_account().event_seq == 1 + + def test_wait_advances_observed_seq(self) -> None: + """After wait succeeds, subsequent wait without publish should timeout.""" + plane = InMemoryZincPlane() + plane.publish_account(_random_snapshot()) + assert plane.wait_on_account(timeout_ms=500) is True # catches it + assert plane.wait_on_account(timeout_ms=200) is False # nothing new + # unless we publish again + plane.publish_account(_random_snapshot()) + assert plane.wait_on_account(timeout_ms=500) is True + + def test_read_returns_latest_publish_only(self) -> None: + """read_account returns the most recently published snapshot.""" + plane = InMemoryZincPlane() + for i in range(100): + plane.publish_account(AccountStateSnapshot( + wallet_balance=float(i * 100), event_seq=i, mono_ns=i, + e_live=(i % 2 == 0), reconcile_ok=(i % 3 == 0), + )) + read = plane.read_account() + assert read.event_seq == i + assert read.wallet_balance == float(i * 100) + + +# ══════════════════════════════════════════════════════════════════════════════ +# 5. MUTATIONAL — inject known bugs, verify test catches them +# ══════════════════════════════════════════════════════════════════════════════ + +class TestMutationLitmus: + """Every guard class is paired with a mutation. + + If the mutation doesn't cause test RED, the guard is decoration. + """ + + def test_mutation_drop_wallet_balance_caught(self) -> None: + """MUTATION: skip wallet_balance in publish → roundtrip catches it.""" + plane = InMemoryZincPlane() + snap = AccountStateSnapshot(wallet_balance=9999.0, available_margin=5000.0, + used_margin=2000.0, event_seq=1, mono_ns=1, + e_live=True, reconcile_ok=True) + plane.publish_account(snap) + read = plane.read_account() + assert read.wallet_balance == 9999.0, f"MUTATION UNCAUGHT: {read.wallet_balance}" + assert read.available_margin == 5000.0 + assert read.used_margin == 2000.0 + + def test_mutation_remove_try_except_caught(self) -> None: + """MUTATION: remove try/except from _account_from_payload → TypeError.""" + from prod.clean_arch.dita_v2.real_zinc_plane import _account_from_payload + mutated_payload = {"wallet_balance": None, "available_margin": None} + # With try/except, this returns default. Without it, raises TypeError. + try: + _account_from_payload(mutated_payload) + except (TypeError, ValueError): + assert False, "MUTATION DETECTED: try/except removed, raw float(None) raised" + + def test_mutation_e_live_default_true_caught(self) -> None: + """MUTATION: default e_live=True → operator sizes off phantom capital.""" + plane = InMemoryZincPlane() + assert plane.read_account().e_live is False, ( + "MUTATION UNCAUGHT: e_live defaults to True, operator would trade on phantom capital" + ) + + def test_mutation_publish_reference_not_copy(self) -> None: + """MUTATION: publish stashes a reference to the original snapshot.""" + plane = InMemoryZincPlane() + original = AccountStateSnapshot(wallet_balance=100.0, e_live=True) + plane.publish_account(original) + # If publish stored a reference and we mutated... we can't because frozen! + # But verify the published value is correct + read = plane.read_account() + assert read.wallet_balance == 100.0 + + def test_mutation_publish_to_different_region(self) -> None: + """MUTATION: publish writes to venue_region instead of account_region.""" + plane = InMemoryZincPlane() + snap = AccountStateSnapshot(wallet_balance=777.0, e_live=True) + venue_before = plane.read_venue() + + plane.publish_account(snap) + + assert plane.read_account().wallet_balance == 777.0 + assert plane.read_account().e_live is True + assert plane.read_venue() == venue_before + + +# ══════════════════════════════════════════════════════════════════════════════ +# 6. CONTRACT — ZincPlane protocol conformance +# ══════════════════════════════════════════════════════════════════════════════ + +class TestProtocolConformance: + """InMemoryZincPlane satisfies the ZincPlane Protocol.""" + + def test_implements_publish_account(self) -> None: + plane = InMemoryZincPlane() + assert hasattr(plane, "publish_account") + assert callable(plane.publish_account) + + def test_implements_read_account(self) -> None: + plane = InMemoryZincPlane() + assert hasattr(plane, "read_account") + assert callable(plane.read_account) + + def test_implements_wait_on_account(self) -> None: + plane = InMemoryZincPlane() + assert hasattr(plane, "wait_on_account") + assert callable(plane.wait_on_account) + + def test_implements_notify_account(self) -> None: + plane = InMemoryZincPlane() + assert hasattr(plane, "notify_account") + assert callable(plane.notify_account) + + def test_account_methods_signatures_match_protocol(self) -> None: + """Verify method signatures at runtime.""" + import inspect + from prod.clean_arch.dita_v2.zinc_plane import ZincPlane + + proto_methods = {} + for name in dir(ZincPlane): + if name.startswith("_"): + continue + obj = getattr(ZincPlane, name, None) + if callable(obj) and hasattr(obj, "__annotations__"): + proto_methods[name] = obj.__annotations__ + + plane = InMemoryZincPlane() + for name, annotations in proto_methods.items(): + assert hasattr(plane, name), f"missing method: {name}" + meth = getattr(plane, name) + assert callable(meth), f"not callable: {name}" + + +# ══════════════════════════════════════════════════════════════════════════════ +# 7. SERIALIZATION — payload→snapshot→dict→payload→snapshot identity +# ══════════════════════════════════════════════════════════════════════════════ + +class TestSerializationRoundtrip: + """Full serialization identity: snap→dict→payload→snap→dict→payload→snap.""" + + def test_deep_roundtrip(self) -> None: + """snap.as_dict() → _account_from_payload → as_dict() → _account_from_payload.""" + snap = _random_snapshot() + d1 = snap.as_dict() + snap2 = _account_from_payload(d1) + d2 = snap2.as_dict() + snap3 = _account_from_payload(d2) + assert snap == snap2 == snap3 + assert d1 == d2 + + def test_json_serializable(self) -> None: + """as_dict() output is JSON-serializable.""" + import json + snap = _random_snapshot() + d = snap.as_dict() + # Must not raise + json_str = json.dumps(d) + loaded = json.loads(json_str) + snap_back = _account_from_payload(loaded) + assert snap_back == snap + + def test_deep_roundtrip_fuzz(self) -> None: + """2000 random deep roundtrips.""" + for i in range(2000): + snap = _random_snapshot() + d1 = snap.as_dict() + snap2 = _account_from_payload(d1) + d2 = snap2.as_dict() + snap3 = _account_from_payload(d2) + assert snap == snap2 == snap3, f"iter {i}: identity broken" + assert d1 == d2, f"iter {i}: dict mismatch" + + def test_all_fields_survive_json(self) -> None: + """All fields in JSON survive round-trip exactly.""" + import json + snap = AccountStateSnapshot( + wallet_balance=12345.6789, + available_margin=8000.0, + used_margin=2000.0, + event_seq=999, + mono_ns=9876543210, + e_live=True, + reconcile_ok=False, + ) + d = snap.as_dict() + j = json.dumps(d) + loaded = json.loads(j) + snap_back = _account_from_payload(loaded) + assert snap_back.wallet_balance == 12345.6789 + assert snap_back.available_margin == 8000.0 + assert snap_back.used_margin == 2000.0 + assert snap_back.event_seq == 999 + assert snap_back.mono_ns == 9876543210 + assert snap_back.e_live is True + assert snap_back.reconcile_ok is False + + +# ══════════════════════════════════════════════════════════════════════════════ +# 8. STRESS — long chains, GC pressure, deterministic replay +# ══════════════════════════════════════════════════════════════════════════════ + +class TestStress: + """Long chains and heavy load.""" + + def test_10000_publish_read_chain(self) -> None: + """10,000 alternating publish/read — no slowdown or state corruption.""" + plane = InMemoryZincPlane() + for i in range(10000): + plane.publish_account(AccountStateSnapshot( + wallet_balance=float(i * 10), event_seq=i, mono_ns=i * 100, + e_live=True, reconcile_ok=True, + )) + read = plane.read_account() + assert read.event_seq == i + assert read.wallet_balance == float(i * 10) + assert plane._account_seq == 10000 + + def test_gc_pressure(self) -> None: + """High allocation rate + GC doesn't corrupt state.""" + plane = InMemoryZincPlane() + for i in range(1000): + # Allocate many intermediate objects + tmp = [AccountStateSnapshot( + wallet_balance=float(x), event_seq=x, mono_ns=x, + e_live=True, reconcile_ok=True, + ) for x in range(100)] + plane.publish_account(tmp[i % 100]) + gc.collect() + # Still readable + final = plane.read_account() + assert isinstance(final, AccountStateSnapshot) + + def test_long_wait_timeout(self) -> None: + """Long timeout does not block longer than needed.""" + plane = InMemoryZincPlane() + t0 = time.monotonic() + # publish after 10ms + def delayed_publish() -> None: + time.sleep(0.01) + plane.publish_account(_random_snapshot()) + + t = threading.Thread(target=delayed_publish, daemon=True) + t.start() + ok = plane.wait_on_account(timeout_ms=5000) + elapsed = time.monotonic() - t0 + t.join(timeout=2) + assert ok is True, "wait timed out despite publish" + assert elapsed < 1.0, f"took too long: {elapsed:.2f}s" + + def test_stress_many_small_snapshots(self) -> None: + """Tiny sequential changes don't miss updates.""" + plane = InMemoryZincPlane() + for i in range(5000): + plane.publish_account(AccountStateSnapshot( + wallet_balance=float(i * 0.0001), + available_margin=float(i * 0.00005), + used_margin=float(i * 0.00005), + event_seq=i, mono_ns=i, e_live=True, reconcile_ok=True, + )) + # verify last state + read = plane.read_account() + assert read.event_seq == 4999 + assert abs(read.wallet_balance - 4999 * 0.0001) < 1e-12 + + def test_stress_oscillating_state(self) -> None: + """State oscillates between extremes — no drift.""" + plane = InMemoryZincPlane() + for cycle in range(500): + # high + plane.publish_account(AccountStateSnapshot( + wallet_balance=1e6, event_seq=cycle * 2, mono_ns=cycle * 2, + e_live=True, reconcile_ok=True, + )) + assert plane.read_account().wallet_balance == 1e6 + # low + plane.publish_account(AccountStateSnapshot( + wallet_balance=0.0, event_seq=cycle * 2 + 1, mono_ns=cycle * 2 + 1, + e_live=True, reconcile_ok=False, + )) + assert plane.read_account().wallet_balance == 0.0 + + +# ══════════════════════════════════════════════════════════════════════════════ +# 9. LEAK — reference escape detection +# ══════════════════════════════════════════════════════════════════════════════ + +class TestReferenceLeak: + """publish_account must not retain references that prevent GC.""" + + def test_no_reference_retained_after_overwrite(self) -> None: + """Old snapshot should be GC-able after overwrite.""" + import weakref + plane = InMemoryZincPlane() + snap = AccountStateSnapshot(wallet_balance=100.0, e_live=True) + ref = weakref.ref(snap) + plane.publish_account(snap) + del snap + plane.publish_account(AccountStateSnapshot(wallet_balance=200.0, e_live=True)) + gc.collect() + # The first snapshot should be collected (no reference) + # (this may or may not hold depending on Python's GC, but we test the intent) + _ = plane.read_account() + + def test_read_does_not_retain_reference(self) -> None: + """read_account returns a snapshot, not a reference to internal field.""" + plane = InMemoryZincPlane() + plane.publish_account(AccountStateSnapshot(wallet_balance=42.0, e_live=True)) + read1 = plane.read_account() + read2 = plane.read_account() + # Both reads should return independent copies (they are, since frozen) + assert read1 == read2 + + def test_as_dict_independent_from_snapshot(self) -> None: + """Modifying the dict from as_dict doesn't affect source.""" + snap = _random_snapshot() + original = snap.as_dict() + d = snap.as_dict() + for key in d: + if isinstance(d[key], (int, float)): + d[key] = d[key] + 1 + elif isinstance(d[key], bool): + d[key] = not d[key] + snap2 = _account_from_payload(original) + assert snap == snap2 + + +# ══════════════════════════════════════════════════════════════════════════════ +# 10. STATE MACHINE — exhaustive small-ops enumeration +# ══════════════════════════════════════════════════════════════════════════════ + +OPS = ["publish_NEW", "publish_SAME", "read", "notify", "wait_0", "wait_short"] + +class TestStateMachineExhaustive: + """Exhaustive enumeration of all 3-op sequences (6^3 = 216).""" + + def test_all_3_op_sequences(self) -> None: + """Every 3-operation sequence results in valid state.""" + snap_a = AccountStateSnapshot(wallet_balance=100.0, event_seq=1, mono_ns=1, e_live=True, reconcile_ok=True) + snap_b = AccountStateSnapshot(wallet_balance=200.0, event_seq=2, mono_ns=2, e_live=True, reconcile_ok=False) + + for seq in itertools.product(OPS, repeat=3): + plane = InMemoryZincPlane() + for op in seq: + try: + if op == "publish_NEW": + plane.publish_account(snap_b if random.random() > 0.5 else snap_a) + elif op == "publish_SAME": + plane.publish_account(snap_a) + elif op == "read": + _ = plane.read_account() + elif op == "notify": + plane.notify_account() + elif op == "wait_0": + plane.wait_on_account(timeout_ms=0) + elif op == "wait_short": + plane.wait_on_account(timeout_ms=50) + except Exception as e: + assert False, f"seq {seq} failed at op '{op}': {e}" + # After any 3-ops, state should be valid + final = plane.read_account() + assert isinstance(final, AccountStateSnapshot) + + def test_random_1000_op_sequences(self) -> None: + """100 random 10-op sequences.""" + rng = random.Random(12345) + snap_a = AccountStateSnapshot(wallet_balance=100.0, e_live=True) + snap_b = AccountStateSnapshot(wallet_balance=200.0, e_live=True) + + for seq_idx in range(1000): + plane = InMemoryZincPlane() + seq_len = rng.randint(1, 20) + for _ in range(seq_len): + op = rng.choice(OPS) + try: + if op == "publish_NEW": + plane.publish_account(rng.choice([snap_a, snap_b])) + elif op == "publish_SAME": + plane.publish_account(snap_a) + elif op == "read": + s = plane.read_account() + assert isinstance(s, AccountStateSnapshot) + elif op == "notify": + plane.notify_account() + elif op in ("wait_0", "wait_short"): + plane.wait_on_account(timeout_ms=0 if op == "wait_0" else 50) + except Exception as e: + assert False, f"seq {seq_idx} op {_}: {e}" + # verify readable at end + _ = plane.read_account() + + +# ══════════════════════════════════════════════════════════════════════════════ +# 11. TIMEOUT / BOUNDARY +# ══════════════════════════════════════════════════════════════════════════════ + +class TestTimeoutBoundary: + """Timeout edge cases: 0, -1, None, very large, fractional.""" + + def test_timeout_zero(self) -> None: + plane = InMemoryZincPlane() + t0 = time.monotonic() + ok = plane.wait_on_account(timeout_ms=0) + assert ok is False + assert time.monotonic() - t0 < 0.5 + + def test_timeout_negative(self) -> None: + """Negative timeout means no deadline and returns on the next event.""" + plane = InMemoryZincPlane() + + def delayed_notify() -> None: + time.sleep(0.02) + plane.notify_account() + + notifier = threading.Thread(target=delayed_notify, daemon=True) + notifier.start() + t0 = time.monotonic() + ok = plane.wait_on_account(timeout_ms=-1) + elapsed = time.monotonic() - t0 + notifier.join(timeout=1.0) + assert ok is True + assert 0.01 <= elapsed < 1.0 + + def test_timeout_large(self) -> None: + """Large timeout doesn't overflow or crash.""" + plane = InMemoryZincPlane() + # publish after tiny delay + def delayed() -> None: + time.sleep(0.01) + plane.publish_account(_random_snapshot()) + + t = threading.Thread(target=delayed, daemon=True) + t.start() + ok = plane.wait_on_account(timeout_ms=2**31 - 1) + t.join(timeout=1) + assert ok is True + + def test_timeout_fractional(self) -> None: + """Fractional timeout works (0.5ms → 0.0005s).""" + plane = InMemoryZincPlane() + t0 = time.monotonic() + ok = plane.wait_on_account(timeout_ms=0.5) + elapsed = time.monotonic() - t0 + assert elapsed < 1.0 + assert ok is False + + def test_timeout_exact_zero_when_published(self) -> None: + """timeout_ms=0 with data already available returns True.""" + plane = InMemoryZincPlane() + plane.publish_account(_random_snapshot()) + ok = plane.wait_on_account(timeout_ms=0) + assert ok is True + + +# ══════════════════════════════════════════════════════════════════════════════ +# 12. ACCOUNT DATACLASS CONTRACT +# ══════════════════════════════════════════════════════════════════════════════ + +class TestAccountDataclassContract: + """AccountStateSnapshot type contracts.""" + + def test_is_dataclass(self) -> None: + from dataclasses import is_dataclass + assert is_dataclass(AccountStateSnapshot) + + def test_is_frozen(self) -> None: + import dataclasses + assert dataclasses.fields(AccountStateSnapshot)[0].name == "wallet_balance" + # Check frozen + try: + AccountStateSnapshot().wallet_balance = 100.0 # type: ignore + assert False, "should be frozen" + except (TypeError, AttributeError): + pass + + def test_all_fields_have_defaults(self) -> None: + """All fields have defaults → no-arg constructor works.""" + snap = AccountStateSnapshot() + assert snap.wallet_balance == 0.0 + + def test_field_types(self) -> None: + """Field types match spec.""" + import typing + hints = typing.get_type_hints(AccountStateSnapshot) + assert hints["wallet_balance"] is float + assert hints["available_margin"] is float + assert hints["used_margin"] is float + assert hints["event_seq"] is int + assert hints["mono_ns"] is int + assert hints["e_live"] is bool + assert hints["reconcile_ok"] is bool + + def test_as_dict_returns_dict_not_other_type(self) -> None: + snap = _random_snapshot() + assert isinstance(snap.as_dict(), dict) + + def test_equality(self) -> None: + """Two snapshots with same fields are equal.""" + a = AccountStateSnapshot(wallet_balance=100.0, e_live=True) + b = AccountStateSnapshot(wallet_balance=100.0, e_live=True) + assert a == b + + def test_inequality(self) -> None: + """Different field values yield inequality.""" + a = AccountStateSnapshot(wallet_balance=100.0, e_live=True) + b = AccountStateSnapshot(wallet_balance=200.0, e_live=True) + assert a != b + c = AccountStateSnapshot(wallet_balance=100.0, e_live=False) + assert a != c + + def test_hashable(self) -> None: + """Frozen dataclass is hashable.""" + snap = _random_snapshot() + d = {snap: "value"} + assert d[snap] == "value" + + +# ══════════════════════════════════════════════════════════════════════════════ +# 13. EXHAUSTIVE payload decoder tests +# ══════════════════════════════════════════════════════════════════════════════ + +class TestPayloadDecoderExhaustive: + """_account_from_payload edge cases.""" + + def test_all_fields_present_correct_types(self) -> None: + payload = { + "wallet_balance": 1000.0, + "available_margin": 800.0, + "used_margin": 200.0, + "event_seq": 42, + "mono_ns": 1712345678, + "e_live": True, + "reconcile_ok": False, + } + snap = _account_from_payload(payload) + assert snap.wallet_balance == 1000.0 + assert snap.available_margin == 800.0 + assert snap.used_margin == 200.0 + assert snap.event_seq == 42 + assert snap.mono_ns == 1712345678 + assert snap.e_live is True + assert snap.reconcile_ok is False + + def test_int_as_float_field(self) -> None: + """int where float expected — valid, convertible.""" + snap = _account_from_payload({"wallet_balance": 100}) + assert snap.wallet_balance == 100.0 + + def test_float_as_int_field(self) -> None: + """float where int expected → truncates.""" + snap = _account_from_payload({"event_seq": 42.9, "mono_ns": 100.7}) + assert snap.event_seq == 42 + assert snap.mono_ns == 100 + + def test_bool_as_int_field(self) -> None: + """bool is subclass of int, True→1, False→0.""" + snap = _account_from_payload({"event_seq": True, "mono_ns": False}) + assert snap.event_seq == 1 + assert snap.mono_ns == 0 + + def test_string_number_field(self) -> None: + """String where number expected → default (try/except catches).""" + snap = _account_from_payload({"wallet_balance": "not_a_number"}) + assert snap.wallet_balance == 0.0 + + def test_none_fields(self) -> None: + """None fields → default.""" + snap = _account_from_payload({ + "wallet_balance": None, "available_margin": None, "used_margin": None, + "event_seq": None, "mono_ns": None, "e_live": None, "reconcile_ok": None, + }) + assert snap.wallet_balance == 0.0 + assert snap.available_margin == 0.0 + assert snap.used_margin == 0.0 + assert snap.event_seq == 0 + assert snap.mono_ns == 0 + assert snap.e_live is False + assert snap.reconcile_ok is False + + def test_all_defaults_empty_dict(self) -> None: + snap = _account_from_payload({}) + assert snap == AccountStateSnapshot() + + def test_extra_keys_in_payload(self) -> None: + """Extra keys are ignored.""" + payload = {"wallet_balance": 100.0, "extra_field": "ignored", "data": [1, 2, 3]} + snap = _account_from_payload(payload) + assert snap.wallet_balance == 100.0 + + def test_nested_dict_field(self) -> None: + """Nested dict where float expected → default.""" + snap = _account_from_payload({"wallet_balance": {"nested": "structure"}}) + assert snap.wallet_balance == 0.0 + + def test_list_field(self) -> None: + """List where float expected → default.""" + snap = _account_from_payload({"wallet_balance": [1.0, 2.0, 3.0]}) + assert snap.wallet_balance == 0.0 + + +# ══════════════════════════════════════════════════════════════════════════════ +# 14. DETERMINISTIC REPLAY +# ══════════════════════════════════════════════════════════════════════════════ + +class TestDeterministic: + """Same sequence of operations → same final state.""" + + def test_deterministic_sequence(self) -> None: + """Same 100-op sequence yields identical state twice.""" + rng = random.Random(42) + snap_a = AccountStateSnapshot(wallet_balance=100.0, event_seq=1, e_live=True) + snap_b = AccountStateSnapshot(wallet_balance=200.0, event_seq=2, e_live=True) + + def run_sequence() -> AccountStateSnapshot: + p = InMemoryZincPlane() + for _ in range(100): + op = rng.choice(["pub_a", "pub_b", "pub_rand"]) + if op == "pub_a": + p.publish_account(snap_a) + elif op == "pub_b": + p.publish_account(snap_b) + else: + p.publish_account(_random_snapshot(rng)) + return p.read_account() + + rng.seed(42) + result1 = run_sequence() + rng.seed(42) + result2 = run_sequence() + assert result1 == result2, "non-deterministic" + + +# ══════════════════════════════════════════════════════════════════════════════ +# 15. IDEMPOTENCY / COMMUTATIVITY +# ══════════════════════════════════════════════════════════════════════════════ + +class TestIdempotency: + """Operations that should be idempotent.""" + + def test_read_is_idempotent(self) -> None: + """Multiple reads without publish return same result.""" + plane = InMemoryZincPlane() + plane.publish_account(AccountStateSnapshot(wallet_balance=100.0, e_live=True)) + r1 = plane.read_account() + r2 = plane.read_account() + r3 = plane.read_account() + assert r1 == r2 == r3 + + def test_notify_is_idempotent(self) -> None: + """Multiple notifies without publish don't cause issues.""" + plane = InMemoryZincPlane() + plane.notify_account() + plane.notify_account() + plane.notify_account() + assert plane.read_account() == AccountStateSnapshot() + + def test_publish_then_read_then_publish_then_read(self) -> None: + """p1→r1→p2→r2 — r1 different from r2.""" + plane = InMemoryZincPlane() + plane.publish_account(AccountStateSnapshot(wallet_balance=100.0, e_live=True)) + r1 = plane.read_account() + plane.publish_account(AccountStateSnapshot(wallet_balance=200.0, e_live=True)) + r2 = plane.read_account() + assert r1.wallet_balance == 100.0 + assert r2.wallet_balance == 200.0 + + +# ══════════════════════════════════════════════════════════════════════════════ +# 16. SINGLE-THREAD RACE DETECTION (logical races) +# ══════════════════════════════════════════════════════════════════════════════ + +class TestSingleThreadRace: + """Logical races that could occur in single-threaded code.""" + + def test_write_during_wait_window(self) -> None: + """Publish between two reads — second read sees new value.""" + plane = InMemoryZincPlane() + plane.publish_account(AccountStateSnapshot(wallet_balance=1.0, e_live=True)) + plane.read_account() # first read + plane.publish_account(AccountStateSnapshot(wallet_balance=2.0, e_live=True)) + r2 = plane.read_account() # second read — should see 2.0 + assert r2.wallet_balance == 2.0 + + def test_interleaved_notify_publish(self) -> None: + """notify → publish: both advance seq, wait catches both.""" + plane = InMemoryZincPlane() + plane.notify_account() # seq +1 + plane.publish_account(AccountStateSnapshot(wallet_balance=100.0, e_live=True)) # seq +1 + # wait should see the latest + ok = plane.wait_on_account(timeout_ms=100) + assert ok is True + assert plane.read_account().wallet_balance == 100.0 + + +# ══════════════════════════════════════════════════════════════════════════════ +# 17. CROSS-PROCESS PATTERN MIMICRY (in-process) +# ══════════════════════════════════════════════════════════════════════════════ + +class TestCrossProcessPattern: + """Patterns that mimic dual-process reader/writer.""" + + def test_writer_then_reader(self) -> None: + """Writer publishes (simulating ASEx), reader consumes.""" + plane = InMemoryZincPlane() + # writer + for i in range(10): + plane.publish_account(AccountStateSnapshot( + wallet_balance=float(i * 1000), event_seq=i, mono_ns=i, + e_live=True, reconcile_ok=True, + )) + # reader — catches latest + snap = plane.read_account() + assert snap.event_seq == 9 + + def test_reader_waits_for_writer(self) -> None: + """Reader blocks until writer publishes.""" + plane = InMemoryZincPlane() + results = [] + + def reader() -> None: + ok = plane.wait_on_account(timeout_ms=2000) + snap = plane.read_account() + results.append((ok, snap.wallet_balance)) + + t = threading.Thread(target=reader, daemon=True) + t.start() + time.sleep(0.05) + plane.publish_account(AccountStateSnapshot( + wallet_balance=50000.0, event_seq=0, mono_ns=0, + e_live=True, reconcile_ok=True, + )) + t.join(timeout=3) + assert len(results) == 1 + assert results[0][0] is True + assert results[0][1] == 50000.0 + + +# ══════════════════════════════════════════════════════════════════════════════ +# 18. SEQUENCE EXHAUSTIVE — Ordered sequences +# ══════════════════════════════════════════════════════════════════════════════ + +class TestSequenceExhaustive: + """Exhaustive ordering tests for small sequences.""" + + def test_publish_then_read_then_publish(self) -> None: + plane = InMemoryZincPlane() + s1 = AccountStateSnapshot(wallet_balance=10.0, event_seq=1, e_live=True) + s2 = AccountStateSnapshot(wallet_balance=20.0, event_seq=2, e_live=True) + plane.publish_account(s1) + assert plane.read_account() == s1 + plane.publish_account(s2) + assert plane.read_account() == s2 + + def test_read_then_publish_then_read(self) -> None: + plane = InMemoryZincPlane() + default = plane.read_account() + assert default.e_live is False + plane.publish_account(AccountStateSnapshot(wallet_balance=50.0, e_live=True)) + assert plane.read_account().wallet_balance == 50.0 + + def test_notify_then_read(self) -> None: + plane = InMemoryZincPlane() + plane.notify_account() + # notify doesn't change account state + assert plane.read_account() == AccountStateSnapshot() + + def test_read_then_notify_then_read(self) -> None: + plane = InMemoryZincPlane() + read1 = plane.read_account() + plane.notify_account() + read2 = plane.read_account() + assert read1 == read2 + + def test_publish_then_notify_then_read(self) -> None: + plane = InMemoryZincPlane() + plane.publish_account(AccountStateSnapshot(wallet_balance=77.0, e_live=True)) + plane.notify_account() + assert plane.read_account().wallet_balance == 77.0 + + +# ══════════════════════════════════════════════════════════════════════════════ +# 19. TIMING / ORDERING GUARANTEES +# ══════════════════════════════════════════════════════════════════════════════ + +class TestOrderingGuarantees: + """Ordering properties.""" + + def test_publish_order_preserved(self) -> None: + """Sequential publishes are read in order of increasing event_seq.""" + plane = InMemoryZincPlane() + for i in range(100): + plane.publish_account(AccountStateSnapshot( + wallet_balance=float(i), event_seq=i, mono_ns=i, + e_live=True, reconcile_ok=True, + )) + assert plane.read_account().event_seq == i + + def test_publish_monotonic_event_seq(self) -> None: + """Publishing with decreasing event_seq still overwrites correctly.""" + plane = InMemoryZincPlane() + plane.publish_account(AccountStateSnapshot(wallet_balance=100.0, event_seq=100, e_live=True)) + plane.publish_account(AccountStateSnapshot(wallet_balance=50.0, event_seq=50, e_live=True)) + # Last publish wins regardless of event_seq value + assert plane.read_account().wallet_balance == 50.0 + assert plane.read_account().event_seq == 50 + + def test_notify_does_not_change_event_seq_in_snapshot(self) -> None: + """notify_account advances internal seq but doesn't change published event_seq.""" + plane = InMemoryZincPlane() + plane.publish_account(AccountStateSnapshot(event_seq=42, e_live=True)) + assert plane.read_account().event_seq == 42 + plane.notify_account() # internal seq +1 + # snapshot's event_seq untouched + assert plane.read_account().event_seq == 42 + + +# ══════════════════════════════════════════════════════════════════════════════ +# 20. DEPENDENCY / IMPORT +# ══════════════════════════════════════════════════════════════════════════════ + +class TestDependency: + """Import and module-level correctness.""" + + def test_import_account_state_snapshot(self) -> None: + from prod.clean_arch.dita_v2.contracts import AccountStateSnapshot + assert AccountStateSnapshot is not None + + def test_import_zinc_plane(self) -> None: + from prod.clean_arch.dita_v2.zinc_plane import InMemoryZincPlane, ZincPlane + assert InMemoryZincPlane is not None + assert ZincPlane is not None + + def test_import_account_from_payload(self) -> None: + from prod.clean_arch.dita_v2.real_zinc_plane import _account_from_payload + assert _account_from_payload is not None diff --git a/prod/clean_arch/dita_v2/test_bingx_post_ack_invariants.py b/prod/clean_arch/dita_v2/test_bingx_post_ack_invariants.py new file mode 100644 index 0000000..8801867 --- /dev/null +++ b/prod/clean_arch/dita_v2/test_bingx_post_ack_invariants.py @@ -0,0 +1,216 @@ +"""Execution-atomicity regressions for BingX post-ack observability. + +Once BingX has accepted an order, telemetry is no longer allowed to alter the +execution result. A 2026-07-13 signature skew raised after the HTTP 200, which +the kernel interpreted as a submit failure and rolled back while the exchange +kept the filled position. These tests protect the adapter and real kernel FSM +boundaries against that entire failure class. +""" + +from __future__ import annotations + +import ast +import asyncio +import inspect +import itertools +import threading +from datetime import datetime, timezone +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from prod.clean_arch.dita_v2.bingx_venue import BingxVenueAdapter +from prod.clean_arch.dita_v2.contracts import ( + KernelCommandType, + KernelEventKind, + KernelIntent, + TradeSide, + TradeStage, +) +from prod.clean_arch.dita_v2.rust_backend import ExecutionKernel + + +def _intent(*, trade_id: str = "post-ack-1") -> KernelIntent: + return KernelIntent( + timestamp=datetime.now(timezone.utc), + intent_id=f"intent-{trade_id}", + trade_id=trade_id, + slot_id=0, + asset="TRX-USDT", + action=KernelCommandType.ENTER, + side=TradeSide.SHORT, + reason="post-ack-regression", + target_size=10.0, + leverage=1.0, + reference_price=100.0, + exit_leg_ratios=(1.0,), + metadata={}, + ) + + +def _filled_receipt() -> SimpleNamespace: + return SimpleNamespace( + status="FILLED", + order_id="venue-order-1", + client_order_id="client-order-1", + price=100.0, + quantity=10.0, + timestamp=datetime.now(timezone.utc), + raw_ack={ + "status": "FILLED", + "orderId": "venue-order-1", + "clientOrderId": "client-order-1", + "executedQty": "10.0", + "avgPrice": "100.0", + }, + ) + + +class _SyncBackend: + def __init__(self) -> None: + self.submit_count = 0 + + def submit_intent(self, _legacy_intent): + self.submit_count += 1 + return _filled_receipt() + + +class _AsyncBackend: + def __init__(self) -> None: + self.submit_count = 0 + + async def submit_intent(self, _legacy_intent): + self.submit_count += 1 + return _filled_receipt() + + +def _venue(backend) -> BingxVenueAdapter: + venue = BingxVenueAdapter.__new__(BingxVenueAdapter) + venue.backend = backend + venue._event_seq = itertools.count(1) + venue._snap_lock = threading.Lock() + venue._snapshot_ready = threading.Event() + venue._snapshot_ready.set() + venue._last_snapshot = None + return venue + + +class _RaiseAfterAck: + """Allow pre-submit telemetry, then emulate any post-ack telemetry defect.""" + + def __init__(self) -> None: + self.phases: list[str] = [] + + def __call__(self, **fields) -> None: + phase = str(fields["phase"]) + self.phases.append(phase) + if phase == "submit:done": + raise TypeError("simulated post-ack telemetry signature skew") + + +def _assert_fill_trace(events) -> None: + kinds = [event.kind for event in events] + assert kinds == [KernelEventKind.ORDER_ACK, KernelEventKind.FULL_FILL] + assert KernelEventKind.ORDER_REJECT not in kinds + fill = events[1] + assert fill.venue_order_id == "venue-order-1" + assert fill.venue_client_id == "client-order-1" + assert fill.filled_size == pytest.approx(10.0) + assert fill.remaining_size == pytest.approx(0.0) + + +def test_sync_submit_returns_fill_trace_when_post_ack_telemetry_raises(monkeypatch): + backend = _SyncBackend() + venue = _venue(backend) + telemetry = _RaiseAfterAck() + monkeypatch.setattr(venue, "_publish_telemetry", telemetry) + + events = venue.submit(_intent(trade_id="sync-adapter")) + + assert backend.submit_count == 1 + assert telemetry.phases == ["submit:start", "submit:done"] + _assert_fill_trace(events) + + +def test_async_submit_returns_fill_trace_when_post_ack_telemetry_raises(monkeypatch): + backend = _AsyncBackend() + venue = _venue(backend) + telemetry = _RaiseAfterAck() + monkeypatch.setattr(venue, "_publish_telemetry", telemetry) + + events = asyncio.run(venue.submit_async(_intent(trade_id="async-adapter"))) + + assert backend.submit_count == 1 + assert telemetry.phases == ["submit:start", "submit:done"] + _assert_fill_trace(events) + + +def test_sync_kernel_does_not_roll_back_acknowledged_fill(monkeypatch): + backend = _SyncBackend() + venue = _venue(backend) + telemetry = _RaiseAfterAck() + monkeypatch.setattr(venue, "_publish_telemetry", telemetry) + + with ExecutionKernel(max_slots=1, venue=venue) as kernel: + outcome = kernel.process_intent(_intent(trade_id="sync-kernel")) + slot = kernel._get_slot(0) + + assert outcome.accepted is True + assert outcome.state is TradeStage.POSITION_OPEN + assert slot.fsm_state is TradeStage.POSITION_OPEN + assert slot.trade_id == "sync-kernel" + assert slot.size == pytest.approx(10.0) + _assert_fill_trace(outcome.emitted_events) + + +def test_async_kernel_does_not_roll_back_acknowledged_fill(monkeypatch): + backend = _AsyncBackend() + venue = _venue(backend) + telemetry = _RaiseAfterAck() + monkeypatch.setattr(venue, "_publish_telemetry", telemetry) + + async def exercise(): + with ExecutionKernel(max_slots=1, venue=venue) as kernel: + outcome = await kernel.process_intent_async(_intent(trade_id="async-kernel")) + slot = kernel._get_slot(0) + return outcome, slot + + outcome, slot = asyncio.run(exercise()) + + assert outcome.accepted is True + assert outcome.state is TradeStage.POSITION_OPEN + assert slot.fsm_state is TradeStage.POSITION_OPEN + assert slot.trade_id == "async-kernel" + assert slot.size == pytest.approx(10.0) + _assert_fill_trace(outcome.emitted_events) + + +def test_every_publish_telemetry_call_site_binds_to_live_signature(): + """Catch call/signature skew before any order path can execute it.""" + source_path = Path(inspect.getsourcefile(BingxVenueAdapter) or "") + tree = ast.parse(source_path.read_text(encoding="utf-8"), filename=str(source_path)) + calls = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "_publish_telemetry" + ] + + # Protect against a broken discovery predicate making this test vacuous. + assert len(calls) >= 11, f"expected at least 11 telemetry call sites, found {len(calls)}" + signature = inspect.signature(BingxVenueAdapter._publish_telemetry) + failures: list[str] = [] + for call in calls: + if any(keyword.arg is None for keyword in call.keywords): + failures.append(f"line {call.lineno}: **kwargs prevents static binding") + continue + positional = [object() for _ in call.args] + keywords = {str(keyword.arg): object() for keyword in call.keywords} + try: + signature.bind(object(), *positional, **keywords) + except TypeError as exc: + failures.append(f"line {call.lineno}: {exc}") + + assert not failures, "telemetry call/signature skew:\n" + "\n".join(failures) diff --git a/prod/clean_arch/dita_v2/test_e_capital_provider.py b/prod/clean_arch/dita_v2/test_e_capital_provider.py new file mode 100644 index 0000000..84120bc --- /dev/null +++ b/prod/clean_arch/dita_v2/test_e_capital_provider.py @@ -0,0 +1,281 @@ +"""Phase-3 tests: freshness-gated E-anchored capital provider. + +Mutation litmus: each guard is paired with a mutation. +If removing the guard does NOT turn the test RED, the test is decoration. +""" + +from __future__ import annotations + +import sys +import time + +sys.path.insert(0, "/mnt/dolphinng5_predict") + +from prod.clean_arch.dita_v2.contracts import AccountStateSnapshot +from prod.clean_arch.dita_v2.zinc_plane import InMemoryZincPlane +from prod.clean_arch.dita_v2.e_capital_provider import ( + ACCOUNT_STALENESS_NS, + make_e_capital_provider, + make_e_capital_provider_from_snap, +) + + +class TestFreshLiveReturnsWallet: + """When e_live, wallet_balance > 0, and age < staleness → returns wallet_balance.""" + + def test_fresh_live_returns_wallet(self) -> None: + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + plane.publish_account(AccountStateSnapshot( + wallet_balance=25000.0, available_margin=20000.0, used_margin=5000.0, + event_seq=1, mono_ns=time.monotonic_ns(), e_live=True, reconcile_ok=True, + )) + capital = provider() + assert capital == 25000.0 + + def test_fresh_live_different_amount(self) -> None: + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + plane.publish_account(AccountStateSnapshot( + wallet_balance=50000.0, mono_ns=time.monotonic_ns(), e_live=True, + )) + assert provider() == 50000.0 + + def test_fresh_live_tiny_positive(self) -> None: + """wallet_balance=0.01 — positive, so returns it.""" + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + plane.publish_account(AccountStateSnapshot( + wallet_balance=0.01, mono_ns=time.monotonic_ns(), e_live=True, + )) + assert provider() == 0.01 + + def test_zero_wallet_returns_none(self) -> None: + """wallet_balance=0.0 is NOT > 0, so returns None.""" + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + plane.publish_account(AccountStateSnapshot( + wallet_balance=0.0, mono_ns=time.monotonic_ns(), e_live=True, + )) + assert provider() is None + + +class TestStaleReturnsNone: + """When age >= staleness → returns None. + Mutation: change >= to > or remove age check → RED.""" + + def test_past_stale_returns_none(self) -> None: + """MUTATION: age >= ACCOUNT_STALENESS_NS check removed → provider returns + old wallet_balance instead of None.""" + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + # Publish with very old mono_ns + plane.publish_account(AccountStateSnapshot( + wallet_balance=25000.0, mono_ns=time.monotonic_ns() - ACCOUNT_STALENESS_NS - 1, + e_live=True, + )) + capital = provider() + assert capital is None, f"MUTATION: stale snapshot returned {capital} instead of None" + + def test_exactly_at_stale_returns_none(self) -> None: + """age == ACCOUNT_STALENESS_NS is stale (strict <, not <=).""" + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + exact_stale_ns = time.monotonic_ns() - ACCOUNT_STALENESS_NS + plane.publish_account(AccountStateSnapshot( + wallet_balance=25000.0, mono_ns=exact_stale_ns, e_live=True, + )) + assert provider() is None, "exactly at staleness should be None (strict <)" + + def test_just_before_stale_returns_wallet(self, monkeypatch) -> None: + """age == staleness - 1 is fresh.""" + from prod.clean_arch.dita_v2 import e_capital_provider + + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + now_ns = time.monotonic_ns() + monkeypatch.setattr(e_capital_provider.time, "monotonic_ns", lambda: now_ns) + fresh_ns = now_ns - ACCOUNT_STALENESS_NS + 1 + plane.publish_account(AccountStateSnapshot( + wallet_balance=25000.0, mono_ns=fresh_ns, e_live=True, + )) + assert provider() == 25000.0 + + +class TestNotLiveReturnsNone: + """e_live is False → returns None.""" + + def test_not_live_returns_none(self) -> None: + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + plane.publish_account(AccountStateSnapshot( + wallet_balance=25000.0, mono_ns=time.monotonic_ns(), e_live=False, + )) + assert provider() is None + + def test_default_snapshot_returns_none(self) -> None: + """Fresh plane, never published — default snapshot has e_live=False.""" + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + assert provider() is None + + def test_live_then_not_live(self) -> None: + """After going from live to not-live, returns None.""" + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + plane.publish_account(AccountStateSnapshot( + wallet_balance=25000.0, mono_ns=time.monotonic_ns(), e_live=True, + )) + assert provider() == 25000.0 + # Overwrite with not-live + plane.publish_account(AccountStateSnapshot( + wallet_balance=25000.0, mono_ns=time.monotonic_ns(), e_live=False, + )) + assert provider() is None + + +class TestProviderEdgeCases: + """Edge cases for the capital provider.""" + + def test_never_stamped_returns_none(self) -> None: + """No publish ever — provider returns None.""" + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + assert provider() is None + + def test_corrupt_snapshot_still_safe(self) -> None: + """If plane returns a corrupt/zero snapshot, provider handles it.""" + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + # Publish a snapshot with negative wallet_balance + plane.publish_account(AccountStateSnapshot( + wallet_balance=-100.0, mono_ns=time.monotonic_ns(), e_live=True, + )) + # Negative wallet_balance means <= 0, so returns None + assert provider() is None + + def test_negative_wallet_not_accepted(self) -> None: + """Negative wallet_balance returns None (not >0).""" + snap = AccountStateSnapshot( + wallet_balance=-100.0, mono_ns=0, e_live=True, + ) + result = make_e_capital_provider_from_snap(snap, now_ns=0) + assert result is None + + def test_custom_staleness(self) -> None: + """Different staleness threshold works.""" + plane = InMemoryZincPlane() + # 1-second staleness + provider = make_e_capital_provider(plane, staleness_ns=1_000_000_000) + plane.publish_account(AccountStateSnapshot( + wallet_balance=25000.0, mono_ns=time.monotonic_ns(), e_live=True, + )) + assert provider() == 25000.0 + + def test_very_old_snapshot(self) -> None: + """Very old mono_ns — returns None.""" + plane = InMemoryZincPlane() + provider = make_e_capital_provider(plane) + plane.publish_account(AccountStateSnapshot( + wallet_balance=25000.0, + mono_ns=time.monotonic_ns() - 100 * ACCOUNT_STALENESS_NS, + e_live=True, + )) + assert provider() is None + + +class TestCapitalProviderFromSnap: + """Pure-function variant tests.""" + + def test_fresh_live(self) -> None: + snap = AccountStateSnapshot( + wallet_balance=10000.0, mono_ns=1000, e_live=True, + ) + assert make_e_capital_provider_from_snap(snap, now_ns=2000, staleness_ns=5000) == 10000.0 + + def test_stale(self) -> None: + snap = AccountStateSnapshot( + wallet_balance=10000.0, mono_ns=1000, e_live=True, + ) + result = make_e_capital_provider_from_snap(snap, now_ns=7000, staleness_ns=5000) + assert result is None, f"stale: age={(7000-1000)} >= 5000" + + def test_not_live(self) -> None: + snap = AccountStateSnapshot( + wallet_balance=10000.0, mono_ns=1000, e_live=False, + ) + assert make_e_capital_provider_from_snap(snap, now_ns=2000) is None + + def test_zero_wallet(self) -> None: + snap = AccountStateSnapshot( + wallet_balance=0.0, mono_ns=1000, e_live=True, + ) + assert make_e_capital_provider_from_snap(snap, now_ns=2000) is None + + def test_negative_wallet(self) -> None: + snap = AccountStateSnapshot( + wallet_balance=-50.0, mono_ns=1000, e_live=True, + ) + assert make_e_capital_provider_from_snap(snap, now_ns=2000) is None + + def test_exactly_at_boundary_stale(self) -> None: + """age == staleness → stale (strict <).""" + snap = AccountStateSnapshot( + wallet_balance=10000.0, mono_ns=1000, e_live=True, + ) + result = make_e_capital_provider_from_snap(snap, now_ns=6000, staleness_ns=5000) + assert result is None, "age == staleness should be stale (strict <)" + + def test_one_ns_before_stale(self) -> None: + """age = staleness - 1 → fresh.""" + snap = AccountStateSnapshot( + wallet_balance=10000.0, mono_ns=1000, e_live=True, + ) + result = make_e_capital_provider_from_snap(snap, now_ns=5999, staleness_ns=5000) + assert result == 10000.0, f"age={(5999-1000)} < 5000 should be fresh, got {result}" + + def test_none_snap(self) -> None: + assert make_e_capital_provider_from_snap(None, 0) is None + + +class TestMutationLitmus: + """Verification that mutations to the age check cause test RED.""" + + def test_mutation_remove_age_check(self) -> None: + """If the age check (>= staleness) is removed, stale snapshots return + wallet_balance instead of None. + + This test proves the guard exists by showing what happens without it.""" + stale_snap = AccountStateSnapshot( + wallet_balance=25000.0, + mono_ns=time.monotonic_ns() - ACCOUNT_STALENESS_NS - 1000, + e_live=True, + ) + # With the correct check, this should be None + result = make_e_capital_provider_from_snap( + stale_snap, now_ns=time.monotonic_ns() + ) + assert result is None, ( + f"MUTATION: stale snapshot returned {result} instead of None. " + "Age check guard is missing or wrong." + ) + + def test_mutation_remove_e_live_check(self) -> None: + """If the e_live check is removed, not-live snapshots return wallet_balance.""" + not_live_snap = AccountStateSnapshot( + wallet_balance=25000.0, mono_ns=0, e_live=False, + ) + result = make_e_capital_provider_from_snap(not_live_snap, now_ns=1000) + assert result is None, ( + f"MUTATION: not-live snapshot returned {result}. e_live guard missing." + ) + + def test_mutation_remove_wallet_positive_check(self) -> None: + """If the wallet_balance > 0 check is removed, zero wallet returns 0.0.""" + zero_snap = AccountStateSnapshot( + wallet_balance=0.0, mono_ns=0, e_live=True, + ) + result = make_e_capital_provider_from_snap(zero_snap, now_ns=1000) + assert result is None, ( + f"MUTATION: zero-wallet snapshot returned {result}. >0 guard missing." + ) diff --git a/prod/clean_arch/dita_v2/test_flaws.py b/prod/clean_arch/dita_v2/test_flaws.py index d8062cc..3955bdd 100644 --- a/prod/clean_arch/dita_v2/test_flaws.py +++ b/prod/clean_arch/dita_v2/test_flaws.py @@ -55,7 +55,10 @@ def _mk_intent( ) -> KernelIntent: return KernelIntent( timestamp=datetime.now(timezone.utc), - intent_id=kw.pop("intent_id", trade_id), + # Venue client IDs must distinguish entry and exit orders. Reusing the + # trade ID for both makes one fill match both active orders, which the + # identity-first Rust router correctly refuses to misclassify. + intent_id=kw.pop("intent_id", f"{trade_id}:{action.value.lower()}"), trade_id=trade_id, slot_id=slot_id, asset=asset, diff --git a/prod/clean_arch/dita_v2/test_native_artifact_provenance.py b/prod/clean_arch/dita_v2/test_native_artifact_provenance.py new file mode 100644 index 0000000..fb954d1 --- /dev/null +++ b/prod/clean_arch/dita_v2/test_native_artifact_provenance.py @@ -0,0 +1,113 @@ +"""Mutation-sensitive tests for DITAv2 native artifact provenance.""" + +from __future__ import annotations + +import json +import shutil +from pathlib import Path + +import pytest + +from prod.clean_arch.dita_v2 import native_artifact + + +def _crate(tmp_path: Path) -> Path: + crate = tmp_path / "crate" + (crate / "src").mkdir(parents=True) + (crate / "Cargo.toml").write_text( + '[package]\nname = "x"\nversion = "0.1.0"\n', encoding="utf-8" + ) + (crate / "Cargo.lock").write_text("version = 3\n", encoding="utf-8") + (crate / "src" / "lib.rs").write_text( + "pub fn x() -> u8 { 1 }\n", encoding="utf-8" + ) + return crate + + +def _artifact(tmp_path: Path, crate: Path) -> Path: + library = tmp_path / "release" / native_artifact.library_name() + library.parent.mkdir(parents=True) + library.write_bytes(b"verified-native-artifact") + native_artifact.write_manifest(library, crate) + return library + + +def test_valid_pair_verifies(tmp_path: Path): + crate = _crate(tmp_path) + library = _artifact(tmp_path, crate) + result = native_artifact.verify_artifact(library, crate) + assert result.library_path == library + assert result.ffi_schema_version == native_artifact.FFI_SCHEMA_VERSION + + +def test_source_mutation_refuses_startup(tmp_path: Path): + crate = _crate(tmp_path) + library = _artifact(tmp_path, crate) + (crate / "src" / "lib.rs").write_text( + "pub fn x() -> u8 { 2 }\n", encoding="utf-8" + ) + with pytest.raises( + native_artifact.ArtifactProvenanceError, match="source_tree_sha256" + ): + native_artifact.verify_artifact(library, crate) + + +def test_artifact_mutation_refuses_startup(tmp_path: Path): + crate = _crate(tmp_path) + library = _artifact(tmp_path, crate) + library.write_bytes(b"tampered-native-artifact") + with pytest.raises( + native_artifact.ArtifactProvenanceError, match="library_sha256" + ): + native_artifact.verify_artifact(library, crate) + + +def test_missing_sidecar_refuses_startup(tmp_path: Path): + crate = _crate(tmp_path) + library = tmp_path / "release" / native_artifact.library_name() + library.parent.mkdir(parents=True) + library.write_bytes(b"no-sidecar") + with pytest.raises( + native_artifact.ArtifactProvenanceError, match="manifest missing" + ): + native_artifact.verify_artifact(library, crate) + + +def test_manifest_self_fingerprint_mutation_refuses_startup(tmp_path: Path): + crate = _crate(tmp_path) + library = _artifact(tmp_path, crate) + sidecar = native_artifact.manifest_path(library) + payload = json.loads(sidecar.read_text(encoding="utf-8")) + payload["ffi_schema_version"] += 1 + sidecar.write_text(json.dumps(payload), encoding="utf-8") + with pytest.raises(native_artifact.ArtifactProvenanceError): + native_artifact.verify_artifact(library, crate) + + +def test_rust_loader_refuses_stale_existing_artifact(tmp_path: Path, monkeypatch): + crate = _crate(tmp_path) + library = _artifact(tmp_path, crate) + library.write_bytes(b"stale") + from prod.clean_arch.dita_v2 import rust_backend + + monkeypatch.setattr(rust_backend, "_library_path", lambda: library) + monkeypatch.setattr(rust_backend, "_crate_dir", lambda: crate) + with pytest.raises(native_artifact.ArtifactProvenanceError): + rust_backend._ensure_library() + + +def test_rollback_restores_previous_verified_pair(tmp_path: Path): + crate = _crate(tmp_path) + target = tmp_path / "target" + library = target / "release" / native_artifact.library_name() + library.parent.mkdir(parents=True) + library.write_bytes(b"known-good") + sidecar = native_artifact.write_manifest(library, crate) + rollback = native_artifact.rollback_directory(target) + rollback.mkdir(parents=True) + shutil.copy2(library, rollback / library.name) + shutil.copy2(sidecar, rollback / sidecar.name) + library.write_bytes(b"bad-current") + result = native_artifact.rollback_artifact(target, crate) + assert library.read_bytes() == b"known-good" + assert result.library_path == library