import random import uuid from datetime import UTC, datetime, timedelta from app.clips_feed import ( Affinity, Candidate, apply_event, cap_weights, compose_page, decay_weights, decode_cursor, encode_cursor, event_kind_for_weights, pick_affinity, pick_explore, score_candidate, split_counts, ) def test_cursor_roundtrip_and_garbage(): pk = uuid.uuid4() c = encode_cursor("abc123", pk, 7) assert decode_cursor(c) == ("abc123", pk, 7) assert decode_cursor("not-a-cursor") is None assert decode_cursor("") is None def test_event_kind_mapping(): assert event_kind_for_weights("impression", 12000) == "dwell" assert event_kind_for_weights("impression", 3000) is None assert event_kind_for_weights("skip", 500) == "skip" assert event_kind_for_weights("skip", 5000) is None assert event_kind_for_weights("like", None) == "like" assert event_kind_for_weights("autoswipe", None) is None def test_apply_event_like_and_not_interested_reason(): aff = Affinity() changed = apply_event(aff, kind="like", dwell_ms=None, reason=None, tags=["big-tits", "amateur"], niches=["just-boobs"], creator="kitsune") assert changed assert aff.tags["big-tits"] == 5.0 and aff.niches["just-boobs"] == 5.0 and aff.creators["kitsune"] == 4.0 apply_event(aff, kind="not_interested", dwell_ms=None, reason="tag:amateur", tags=["amateur"], niches=[], creator=None) # -4 za zdarzenie i -8 za wskazany tag assert aff.tags["amateur"] == 5.0 - 4.0 - 8.0 assert not aff.is_empty() assert aff.top("tags", 1) == ["big-tits"] def test_cap_and_decay(): d = {f"t{i}": float(i) for i in range(100)} capped = cap_weights(d, 10) assert len(capped) == 10 and "t99" in capped and "t0" not in capped decayed = decay_weights({"a": 1.0, "b": 0.2}, factor=0.5, floor=0.2) assert decayed == {"a": 0.5} # b spada pod prog i znika def _cand(creator, likes=0, tags=(), niches=(), days_old=30): return Candidate( id=uuid.uuid4(), creator=creator, likes=likes, published_at=datetime.now(UTC) - timedelta(days=days_old), tags=list(tags), niches=list(niches), ) def test_pick_affinity_respects_creator_cap_and_order(): aff = Affinity(tags={"x": 10.0}) cands = [_cand("a", tags=["x"]) for _ in range(5)] + [_cand("b", tags=["x"], likes=1)] picked = pick_affinity(cands, aff, 4, max_per_creator=2) creators = [c.creator for c in picked] assert creators.count("a") == 2 and "b" in creators and len(picked) == 3 def test_score_prefers_fresh_and_liked(): aff = Affinity() old = _cand("a", likes=0, days_old=100) fresh = _cand("a", likes=0, days_old=1) liked = _cand("a", likes=1000, days_old=100) assert score_candidate(fresh, aff) > score_candidate(old, aff) assert score_candidate(liked, aff) > score_candidate(old, aff) def test_split_and_compose_deterministic(): assert split_counts(10, False) == (0, 10) assert split_counts(10, True) == (8, 2) a = [_cand("a") for _ in range(8)] e = [_cand("e") for _ in range(2)] p1 = compose_page(list(a), list(e), random.Random("s:1")) p2 = compose_page(list(a), list(e), random.Random("s:1")) assert [c.id for c, _ in p1] == [c.id for c, _ in p2] assert sum(1 for _, slot in p1 if slot == "explore") == 2 def test_pick_explore_weighted_but_complete(): cands = [_cand("a", likes=0), _cand("b", likes=10**6)] out = pick_explore(cands, 2, random.Random(1)) assert {c.creator for c in out} == {"a", "b"}