"""Tests for SwarmMetrics.""" import time import sys sys.path.insert(0, ".") from swarmmetrics import ( _gini, _half_life_weight, score_reciprocity, score_channels, score_echo, detect_shadows, analyze ) now = time.time() day = 86400 def test_gini_single(): """Single speaker = max inequality.""" assert _gini([10]) == 1.0 def test_gini_equal(): """Equal speakers = 0 inequality.""" g = _gini([10, 10, 10, 10]) assert abs(g) < 0.01, f"Expected ~0, got {g}" def test_gini_dominated(): """One dominant speaker.""" g = _gini([1, 1, 1, 100]) assert g > 0.5, f"Expected >0.5, got {g}" def test_half_life_zero(): """Message at t=now has weight ~1.""" w = _half_life_weight(0) assert abs(w - 1.0) < 0.01 def test_half_life_decay(): """Message at t=tau has weight ~0.5.""" w = _half_life_weight(7 * 86400, tau_days=7.0) assert abs(w - 0.5) < 0.01, f"Expected ~0.5, got {w}" def test_reciprocity_symmetric(): """Equal exchange = high F3.""" msgs = [ {"from_id": "a", "to_id": "b", "timestamp": now - 0.1*day}, {"from_id": "b", "to_id": "a", "timestamp": now - 0.1*day}, ] edges = score_reciprocity(msgs, now=now) pair = ("a", "b") assert pair in edges assert edges[pair].reciprocity_f3 > 0.6, f"Expected >0.6, got {edges[pair].reciprocity_f3}" def test_reciprocity_asymmetric(): """One-way communication = F3 near 0.""" msgs = [ {"from_id": "a", "to_id": "b", "timestamp": now - 0.1*day}, {"from_id": "a", "to_id": "b", "timestamp": now - 0.2*day}, {"from_id": "a", "to_id": "b", "timestamp": now - 0.3*day}, ] edges = score_reciprocity(msgs, now=now) pair = ("a", "b") assert edges[pair].reciprocity_f3 < 0.01 def test_channel_classification(): """Monologue, dialogue, broadcast detection.""" msgs = [ {"from_id": "a", "to_id": "b", "channel": "mono", "timestamp": now}, {"from_id": "a", "to_id": "c", "channel": "mono", "timestamp": now}, {"from_id": "a", "to_id": "b", "channel": "talk", "timestamp": now}, {"from_id": "b", "to_id": "a", "channel": "talk", "timestamp": now}, ] channels = score_channels(msgs) assert channels["mono"].classification == "monologue" assert channels["talk"].classification == "dialogue" def test_echo_coefficient(): """Concept introduced by X, echoed by Y, X gets credit.""" msgs = [ {"from_id": "x", "to_id": "y", "timestamp": now - 5*day, "concepts": ["alpha"]}, {"from_id": "y", "to_id": "z", "timestamp": now - 3*day, "concepts": ["alpha"]}, ] echo = score_echo(msgs) assert echo.get("x", 0) > 0, f"X should have echo credit, got {echo}" def test_shadow_detection(): """Node silent for >7 days is a shadow.""" msgs = [ {"from_id": "ghost", "to_id": "alive", "timestamp": now - 14*day}, {"from_id": "alive", "to_id": "ghost", "timestamp": now - 0.5*day}, ] shadows = detect_shadows(msgs, now=now, silence_days=7.0) assert "ghost" in shadows assert "alive" not in shadows def test_full_analysis(): """Full analyze() returns ScoredGraph with all sections.""" msgs = [ {"from_id": "a", "to_id": "b", "timestamp": now - 1*day, "channel": "ch1", "concepts": ["x"]}, {"from_id": "b", "to_id": "a", "timestamp": now - 0.5*day, "channel": "ch1"}, ] result = analyze(msgs, now=now) assert result.summary["total_messages"] == 2 assert result.summary["total_nodes"] == 2 assert len(result.edges) == 1 assert len(result.channels) == 1 assert len(result.nodes) == 2 def test_f32_zombie_broadcast(): """F3.2 must rank 5-message dialogue above 100-ping broadcast. Regression test for reticuli's arithmetic proof (Colony, 2026-07-27): F3.1 count*log(1+r) lets volume buy back low reciprocity.""" msgs = [] # 5-message dialogue, high reciprocity (3+2) for i in range(3): msgs.append({"from_id": "alice", "to_id": "bob", "timestamp": now - 1*day + i*100}) for i in range(2): msgs.append({"from_id": "bob", "to_id": "alice", "timestamp": now - 0.5*day + i*100}) # 100-ping broadcast, low reciprocity (95+5) for i in range(95): msgs.append({"from_id": "carol", "to_id": "dave", "timestamp": now - 2*day + i*60}) for i in range(5): msgs.append({"from_id": "dave", "to_id": "carol", "timestamp": now - 1.5*day + i*600}) edges = score_reciprocity(msgs, now=now) dialogue = [e for e in edges.values() if "alice" in (e.source, e.target)][0] broadcast = [e for e in edges.values() if "carol" in (e.source, e.target)][0] assert dialogue.reciprocity_f32 > broadcast.reciprocity_f32, \ f"F3.2 zombie fail: dialogue={dialogue.reciprocity_f32:.4f} <= broadcast={broadcast.reciprocity_f32:.4f}" def test_f33_zombie_broadcast_concepts(): """F3.3 unique-concept-count must crush spam even harder than F3.2. 100 pings with 2 concepts vs 5 messages with 5 concepts. Idea from 小風 (ClawdChat): replace count with unique topic count.""" msgs = [] # 5-message dialogue, high reciprocity (3+2), 5 unique concepts concepts_rich = [["graph", "reciprocity"], ["decay", "metrics"], ["shadow"], ["echo", "idf"], ["influence"]] for i in range(3): msgs.append({"from_id": "alice", "to_id": "bob", "timestamp": now - 1*day + i*100, "concepts": concepts_rich[i]}) for i in range(2): msgs.append({"from_id": "bob", "to_id": "alice", "timestamp": now - 0.5*day + i*100, "concepts": concepts_rich[3+i]}) # 100 pings, low reciprocity (95+5), only 2 unique concepts (repeated) for i in range(95): msgs.append({"from_id": "carol", "to_id": "dave", "timestamp": now - 2*day + i*60, "concepts": ["hello", "ping"]}) for i in range(5): msgs.append({"from_id": "dave", "to_id": "carol", "timestamp": now - 1.5*day + i*600, "concepts": ["hello"]}) edges = score_reciprocity(msgs, now=now) dialogue = [e for e in edges.values() if "alice" in (e.source, e.target)][0] broadcast = [e for e in edges.values() if "carol" in (e.source, e.target)][0] # F3.3 must separate them assert dialogue.reciprocity_f33 > broadcast.reciprocity_f33, \ (f"F3.3 zombie fail: dialogue={dialogue.reciprocity_f33:.4f} " f"<= broadcast={broadcast.reciprocity_f33:.4f}") # F3.3 spread should be wider than F3.2 f33_ratio = dialogue.reciprocity_f33 / max(broadcast.reciprocity_f33, 0.001) f32_ratio = dialogue.reciprocity_f32 / max(broadcast.reciprocity_f32, 0.001) assert f33_ratio > f32_ratio, \ f"F3.3 should separate more than F3.2: f33_ratio={f33_ratio:.1f} <= f32_ratio={f32_ratio:.1f}" def test_f33_fallback_no_concepts(): """F3.3 falls back to F3.2 when messages have no concepts.""" msgs = [ {"from_id": "a", "to_id": "b", "timestamp": now - 1*day}, {"from_id": "b", "to_id": "a", "timestamp": now - 0.5*day}, ] edges = score_reciprocity(msgs, now=now) edge = list(edges.values())[0] assert edge.reciprocity_f33 == edge.reciprocity_f32, \ f"F3.3 should equal F3.2 with no concepts: f33={edge.reciprocity_f33}, f32={edge.reciprocity_f32}" def test_f32_idf_echo_combined(): """F3.2 and IDF echo work together in full analyze().""" msgs = [ {"from_id": "a", "to_id": "b", "timestamp": now - 1*day, "channel": "ch", "concepts": ["rare-concept"]}, {"from_id": "b", "to_id": "a", "timestamp": now - 0.9*day, "channel": "ch", "concepts": ["rare-concept"]}, ] result = analyze(msgs, now=now) assert len(result.edges) == 1 edge = list(result.edges.values())[0] assert edge.reciprocity_f32 > 0, "F3.2 should be positive for reciprocal edge" # Run all tests tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")] passed = 0 failed = 0 for t in tests: try: t() print(f" ✓ {t.__name__}") passed += 1 except Exception as e: print(f" ✗ {t.__name__}: {e}") failed += 1 print(f"\n{passed}/{passed+failed} tests passed") sys.exit(1 if failed > 0 else 0)