236 lines
9.4 KiB
Python
236 lines
9.4 KiB
Python
"""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"
|
|
|
|
|
|
def test_f33_gibberish_high_score():
|
|
"""F3.3 CANNOT detect quality — only filter spam.
|
|
High-reciprocity, high-diversity gibberish scores well.
|
|
This is not a bug — it's the boundary of what statistics can measure."""
|
|
msgs = []
|
|
# Two agents exchanging diverse gibberish with high reciprocity
|
|
gibberish_concepts = [
|
|
["xkcd", "flurp"], ["zibzab", "quux"], ["bloop", "snarg"],
|
|
["wibble", "grunk"], ["spuzz", "flarb"]
|
|
]
|
|
for i in range(5):
|
|
msgs.append({"from_id": "alice", "to_id": "bob",
|
|
"timestamp": now - 1*day + i*100,
|
|
"concepts": gibberish_concepts[i]})
|
|
for i in range(4):
|
|
msgs.append({"from_id": "bob", "to_id": "alice",
|
|
"timestamp": now - 0.5*day + i*100,
|
|
"concepts": [f"nonsense_{i}", f"drivel_{i}"]})
|
|
|
|
edges = score_reciprocity(msgs, now=now)
|
|
edge = list(edges.values())[0]
|
|
|
|
# Gibberish scores well — high diversity + high reciprocity
|
|
assert edge.reciprocity_f33 > 0.5, \
|
|
f"Gibberish should score high on F3.3: {edge.reciprocity_f33:.4f}"
|
|
# This is the proof: F3.3 measures not-spam, not quality.
|
|
# 18 unique concepts, reciprocity ~0.8 → high score.
|
|
# The formula chain is an anti-spam filter, not a quality metric.
|
|
|
|
|
|
# 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)
|