hybrid_engine/tests/test_engine.py

"""Smoke tests on synthetic data (no dataset needed). Run: python -m hybrid_engine.tests.test_engine"""
import numpy as np
from hybrid_engine import HybridRecommender

rng = np.random.default_rng(0)
def vec():
    v = rng.normal(size=1024); return v / np.linalg.norm(v)
def post(i, age):
    return dict(post_id=f"p{i}", community=["r/MachineLearning", "r/robotics", "r/IOT"][i % 3], post_type="text", score=10 + i, comments=i, age_hours=age, title=f"post {i}",
                body="body", tags={"machine-learning": 0.9} if i % 2 else {"robotics": 0.8}, embedding=vec(), author=f"a{i % 5}", impressions=i % 4)
def event(i, t, engaged):
    return dict(post_id=f"h{i}", ts_ms=t, visible_ms=4000, focal_ms=4000 if engaged else 200, hover_ms=500, clicked=engaged, engaged=engaged, community="r/robotics", author="a1", embedding=vec())

def test_smoke():
    eng = HybridRecommender(); now = 1_800_000_000_000.0
    cands = [post(i, age) for i, age in enumerate([1, 5, 30, 100, 400, 2, 50, 800, 12, 200])]
    new = eng.score({"interests": ["machine-learning", "robotics"], "history": []}, cands, now)
    assert all(d["warm_weight"] == 0.0 and d["warm_probability"] is None for d in new), "a brand-new user must get pure v2b"
    assert all(np.isfinite(d["score"]) for d in new)
    weights = []
    for n_eng in (1, 5, 20, 100):
        hist = [event(i, now - 10_000_000 + i * 1000, i < n_eng) for i in range(max(n_eng, 10))]
        res = eng.score({"interests": ["machine-learning"], "history": hist}, cands, now); weights.append(res[0]["warm_weight"])
        assert res[0]["engaged_history"] == n_eng
    assert weights == sorted(weights) and weights[-1] <= eng.cfg["w_max"] + 1e-9 and weights[0] > 0, f"warm weight must grow with history and stay <= w_max: {weights}"
    future = [event(0, now + 5_000, True)]                                  # an interaction after the request must not count
    assert eng.score({"interests": [], "history": future}, cands[:3], now)[0]["engaged_history"] == 0
    fm = {d["post_id"]: d["freshness_multiplier"] for d in new}; ages = {c["post_id"]: c["age_hours"] for c in cands}
    s = sorted(fm, key=lambda p: ages[p]); assert all(fm[s[i]] >= fm[s[i + 1]] for i in range(len(s) - 1)), "older posts must never get a larger freshness multiplier"
    assert all((d["push"] > 0) == (ages[d["post_id"]] <= eng.cfg["push_max_age_hours"]) for d in new), "push only for fresh posts"
    many = [post(i, 5 + i) for i in range(30)]                                # 15 'machine-learning' posts (odd i), 15 'robotics' posts
    feed = eng.rank({"interests": ["machine-learning"], "history": []}, many, now)
    k = eng.cfg["topic_first_slots"]
    assert all(d["topic_match"] for d in feed[:k]), "the first topic_first_slots positions must be topic matches when enough exist"
    plain = HybridRecommender(topic_first_slots=0).rank({"interests": ["machine-learning"], "history": []}, many, now)
    assert [d["score"] for d in plain] == sorted([d["score"] for d in plain], reverse=True), "with the rule off, the order is plain score order"
    few = [post(i, 5) for i in (0, 1, 2, 4, 6)]                               # only one topic match: the rule must fall back gracefully
    assert len(eng.rank({"interests": ["machine-learning"], "history": []}, few, now)) == 5
    top = eng.recommend({"interests": ["robotics"], "history": []}, cands, now, k=3); assert len(top) == 3
    print("all smoke tests passed | warm weight at 1/5/20/100 engaged posts:", [round(w, 3) for w in weights])

def test_no_community_or_flair_in_models():
    """Engine 2.0: community and flair are not model inputs, so renaming them changes no score for a new user
    (with history, community only feeds the user's own per-community rates)."""
    eng = HybridRecommender(); now = 1_800_000_000_000.0
    for which in ("cold", "warm"):
        assert not {"subreddit", "flair"} & set(eng.meta[which]["token_inputs"]), eng.meta[which]["token_inputs"]
    assert "no community" in eng.meta["pca"].get("embedding_text", ""), "content vectors must come from title + body (+ caption) only"
    cands = [post(i, 5 + i) for i in range(12)]
    renamed = [{**c, "community": f"somewhere-{i}", "flair": "Anything"} for i, c in enumerate(cands)]
    u = {"interests": ["machine-learning"], "history": []}
    assert [d["score"] for d in eng.score(u, cands, now)] == [d["score"] for d in eng.score(u, renamed, now)]


if __name__ == "__main__":
    test_smoke()
    test_no_community_or_flair_in_models()
    print("community / flair guard passed")