tests/test_contributor_profile.py

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from world.contributor_profile import ContributorProfileStore


def test_alice_jia_tan_pattern():
    store = ContributorProfileStore()

    # 10 small docs PRs, all merged — alice builds trust
    for i in range(10):
        store.observe(
            "alice",
            step=i,
            is_pr=True,
            diff_size=20 + (i % 5),
            files=[f"docs/page_{i}.md", f"docs/section_{i}.md"],
            outcome="merged",
        )

    baseline = store.features_for_observation("alice")
    print(f"alice baseline: {baseline}")
    assert baseline["n_prs_seen"] == 10
    assert baseline["n_prs_merged"] == 10
    assert baseline["trust_score"] > 0.85, f"trust should be high after 10 merges: {baseline['trust_score']}"
    assert baseline["mean_diff_size"] < 30
    baseline_anomaly = baseline["anomaly_score_ewma"]
    baseline_delta = baseline["behavioral_delta"]

    # 1 huge build-system PR — should spike behavioural delta + anomaly
    store.observe(
        "alice",
        step=11,
        is_pr=True,
        diff_size=15000,
        files=["build/configure.ac", "build/Makefile.am", "m4/build_to_host.m4"],
        outcome=None,
    )

    payload = store.features_for_observation("alice")
    print(f"alice payload PR: {payload}")
    assert payload["behavioral_delta"] > baseline_delta * 10 + 100, (
        f"behavioural_delta should spike: baseline={baseline_delta} payload={payload['behavioral_delta']}"
    )
    assert payload["anomaly_score_ewma"] > baseline_anomaly + 1.0, (
        f"anomaly EWMA should jump: baseline={baseline_anomaly} payload={payload['anomaly_score_ewma']}"
    )
    assert payload["subsystem_entropy"] > baseline["subsystem_entropy"], (
        "entropy should rise once a new subsystem is touched"
    )

    # 5 mixed bob PRs — verify trust_score in (0, 1)
    bob_events = [
        (50, ["src/foo.py"], "merged"),
        (200, ["src/bar.py", "src/baz.py"], "merged"),
        (800, ["build/build.py"], "rejected"),
        (75, ["tests/test_x.py"], "merged"),
        (1200, ["src/core.py", "src/api.py"], "flagged"),
    ]
    for i, (size, files, outcome) in enumerate(bob_events):
        store.observe(
            "bob",
            step=20 + i,
            is_pr=True,
            diff_size=size,
            files=files,
            outcome=outcome,
        )

    bob = store.features_for_observation("bob")
    print(f"bob mixed: {bob}")
    assert 0.0 < bob["trust_score"] < 1.0, f"bob trust should be in (0, 1): {bob['trust_score']}"
    assert bob["n_prs_seen"] == 5
    assert bob["n_prs_merged"] == 3
    assert bob["subsystem_entropy"] > 1.0, "bob touched several subsystems → entropy > 1 bit"

    # all_features sanity
    everyone = store.all_features()
    assert set(everyone.keys()) == {"alice", "bob"}
    assert store.get("ghost") is None

    print("smoke summary:")
    print(f"  alice trust={payload['trust_score']:.3f} anomaly={payload['anomaly_score_ewma']:.3f} delta={payload['behavioral_delta']:.1f}")
    print(f"  bob   trust={bob['trust_score']:.3f} anomaly={bob['anomaly_score_ewma']:.3f} entropy={bob['subsystem_entropy']:.3f}")


if __name__ == "__main__":
    test_alice_jia_tan_pattern()
    print("PASS test_alice_jia_tan_pattern")
    print("ALL GREEN")