idea210-swarmmetrics/test_swarmmetrics.py

"""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_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)