idea189-replicaweave-safe-m.../src/idea189_replicaweave_safe_m.../test_cartridge.py

73 lines
2.7 KiB
Python

"""TestCartridge spec and a few built-in deterministic cartridges.
Each cartridge is a small function that, given a delta and a seed, returns a TestReport.
TestReport is a dict with keys: passed (bool), score (0-100), logs (list), counterexample_seed (optional int)
"""
from dataclasses import dataclass
from typing import Callable, Dict, Any, List
import random
@dataclass
class TestReport:
passed: bool
score: int
logs: List[str]
counterexample_seed: int = None
@dataclass
class TestCartridge:
name: str
version: str
run: Callable[[Dict[str, Any], int], TestReport]
def _deterministic_outcome(seed: int, fail_rate: float = 0.2):
r = random.Random(seed)
return r.random() >= fail_rate
def network_partition_cartridge(delta, seed: int) -> TestReport:
# deterministic check: failure occurs when random < 0.15
ok = _deterministic_outcome(seed, fail_rate=0.15)
logs = [f"network_partition(seed={seed}) -> {'ok' if ok else 'fail'}"]
return TestReport(passed=ok, score=(90 if ok else 20), logs=logs, counterexample_seed=(None if ok else seed))
def resource_exhaustion_cartridge(delta, seed: int) -> TestReport:
# estimate resource impact from delta size
size = len(delta.get("ops", []))
r = random.Random(seed + size)
ok = r.random() >= 0.25
logs = [f"resource_exhaustion(seed={seed},ops={size}) -> {'ok' if ok else 'fail'}"]
score = max(0, 100 - size * 5) if ok else 10
return TestReport(passed=ok, score=score, logs=logs, counterexample_seed=(None if ok else seed))
def equivocation_cartridge(delta, seed: int) -> TestReport:
# simple heuristic: fail if two ops from different actors target same path
ops = delta.get("ops", [])
targets = {}
for op in ops:
t = op.get("target")
targets.setdefault(t, set()).add(op.get("actor"))
equiv = any(len(s) > 1 for s in targets.values())
logs = [f"equivocation check -> {'equivocated' if equiv else 'clean'}"]
return TestReport(passed=not equiv, score=(50 if not equiv else 0), logs=logs, counterexample_seed=(seed if equiv else None))
def actuator_mock_cartridge(delta, seed: int) -> TestReport:
# simulate actuator invocation risk; use seed to decide
ok = _deterministic_outcome(seed, fail_rate=0.1)
logs = [f"actuator_mock(seed={seed}) -> {'ok' if ok else 'danger'}"]
return TestReport(passed=ok, score=(95 if ok else 5), logs=logs, counterexample_seed=(None if ok else seed))
built_in_cartridges = [
TestCartridge("network_partition", "0.1", network_partition_cartridge),
TestCartridge("resource_exhaustion", "0.1", resource_exhaustion_cartridge),
TestCartridge("equivocation", "0.1", equivocation_cartridge),
TestCartridge("actuator_mock", "0.1", actuator_mock_cartridge),
]