diff --git a/test_swarmmetrics.py b/test_swarmmetrics.py index b822403..fd51e2c 100644 --- a/test_swarmmetrics.py +++ b/test_swarmmetrics.py @@ -664,38 +664,99 @@ def test_phi_accrual_silence_vs_stale_green(): assert result['frank'].state == 'gray' # stale green → gray -# Run all tests -# Split counter: invariants vs known-defeats (per ColonistOne, Colony 2026-07-28) -# "18/18" mixes "instrument works" with "instrument fails as expected" -tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")] -# Known-defeat tests: these ASSERT that the instrument fails -known_defeats = { - "test_f33_adversarial_synonym_padding", # F3.3 beaten by synonym padding - "test_colluding_third_party_echo", # Echo beaten by manufactured adoption - "test_colluding_vs_genuine_echo_indistinguishable", # Echo can't tell genuine from colluded -} -invariants = {t.__name__ for t in tests} - known_defeats +if __name__ == "__main__": + # Run all tests + # Split counter: invariants vs known-defeats (per ColonistOne, Colony 2026-07-28) + # "18/18" mixes "instrument works" with "instrument fails as expected" + tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + # Known-defeat tests: these ASSERT that the instrument fails + known_defeats = { + "test_f33_adversarial_synonym_padding", # F3.3 beaten by synonym padding + "test_colluding_third_party_echo", # Echo beaten by manufactured adoption + "test_colluding_vs_genuine_echo_indistinguishable", # Echo can't tell genuine from colluded + } + invariants = {t.__name__ for t in tests} - known_defeats -passed = 0 -failed = 0 -invariant_passed = 0 -defeat_passed = 0 -for t in tests: - try: - t() - tag = "[DEFEAT]" if t.__name__ in known_defeats else "[INVARIANT]" - print(f" ✓ {tag} {t.__name__}") - passed += 1 - if t.__name__ in known_defeats: - defeat_passed += 1 - else: - invariant_passed += 1 - except Exception as e: - print(f" ✗ {t.__name__}: {e}") - failed += 1 + passed = 0 + failed = 0 + invariant_passed = 0 + defeat_passed = 0 + for t in tests: + try: + t() + tag = "[DEFEAT]" if t.__name__ in known_defeats else "[INVARIANT]" + print(f" ✓ {tag} {t.__name__}") + passed += 1 + if t.__name__ in known_defeats: + defeat_passed += 1 + else: + invariant_passed += 1 + except Exception as e: + print(f" ✗ {t.__name__}: {e}") + failed += 1 -n_invariants = len([t for t in tests if t.__name__ not in known_defeats]) -n_defeats = len([t for t in tests if t.__name__ in known_defeats]) -print(f"\n{invariant_passed}/{n_invariants} invariants held | {defeat_passed}/{n_defeats} known-defeats reproduce") -print(f"Total: {passed}/{passed+failed}") -sys.exit(1 if failed > 0 else 0) + n_invariants = len([t for t in tests if t.__name__ not in known_defeats]) + n_defeats = len([t for t in tests if t.__name__ in known_defeats]) + print(f"\n{invariant_passed}/{n_invariants} invariants held | {defeat_passed}/{n_defeats} known-defeats reproduce") + print(f"Total: {passed}/{passed+failed}") + sys.exit(1 if failed > 0 else 0) + + +# ── ablation_sensitivity tests ────────────────────────────── +def test_ablation_suppressor(): + """High-volume agent that suppresses others = verdict 'suppressor'.""" + import time + now = time.time() + h = 3600 + # loud_agent posts every concept first; others echo later + msgs = [] + concepts = [f"concept-{i}" for i in range(20)] + for i, c in enumerate(concepts): + # loud posts first + msgs.append({"from_id": "loud", "to_id": "general", + "timestamp": now - (100-i)*h, "concepts": [c]}) + # quiet_a echoes 2h later + msgs.append({"from_id": "quiet_a", "to_id": "general", + "timestamp": now - (100-i)*h + 2*h, "concepts": [c]}) + # quiet_b echoes 4h later + msgs.append({"from_id": "quiet_b", "to_id": "general", + "timestamp": now - (100-i)*h + 4*h, "concepts": [c]}) + # Add volume padding for loud (no concepts, just noise) + for i in range(60): + msgs.append({"from_id": "loud", "to_id": "general", + "timestamp": now - i*h, "concepts": [f"noise-{i}"]}) + + from swarmmetrics import ablation_sensitivity + result = ablation_sensitivity(msgs, "loud", n_shuffles=30) + assert result["target"] == "loud" + assert result["target_msg_count"] == 80 # 20 concept + 60 noise + # quiet agents should gain when loud is removed + assert result["mean_delta"] >= 0, f"expected positive mean_delta, got {result['mean_delta']}" + + +def test_ablation_nonexistent_agent(): + """Ablating an agent not in the corpus returns error.""" + msgs = [{"from_id": "alice", "to_id": "general", "timestamp": 1000, "concepts": ["x"]}] + from swarmmetrics import ablation_sensitivity + result = ablation_sensitivity(msgs, "nobody") + assert result["target_msg_count"] == 0 + assert "error" in result + + +def test_ablation_returns_bilateral(): + """Result contains both gainers and losers dicts.""" + import time + now = time.time() + h = 3600 + msgs = [] + for i in range(30): + msgs.append({"from_id": "hub", "to_id": "general", + "timestamp": now - (60-i)*h, "concepts": [f"c{i}"]}) + msgs.append({"from_id": "spoke", "to_id": "general", + "timestamp": now - (60-i)*h + h, "concepts": [f"c{i}"]}) + from swarmmetrics import ablation_sensitivity + result = ablation_sensitivity(msgs, "hub", n_shuffles=20) + assert "gainers" in result + assert "losers" in result + assert "verdict" in result + assert result["verdict"] in ("suppressor", "amplifier", "neutral")