245 lines
8.4 KiB
Python
245 lines
8.4 KiB
Python
"""
|
|
Tests for peer module.
|
|
"""
|
|
import pytest
|
|
from datetime import datetime, timedelta
|
|
|
|
from src.peer.models import (
|
|
PeerRepr, Tier,
|
|
ExplicitPreference, DeducedPreference, InducedPreference
|
|
)
|
|
from src.peer.decay import DecayCalculator, DecayResult, DECAY_RATE, MIN_SCORE
|
|
from src.peer.inference import ExplicitInference, DeductiveInference, InductiveInference, InferenceContext
|
|
|
|
|
|
class TestPeerRepr:
|
|
def test_create_peer_repr(self):
|
|
"""Test basic PeerRepr creation."""
|
|
peer = PeerRepr(peer_id="user-1", tier=Tier.ACTIVE)
|
|
assert peer.peer_id == "user-1"
|
|
assert peer.tier == Tier.ACTIVE
|
|
assert len(peer.explicit_prefs) == 0
|
|
|
|
def test_add_explicit_preference(self):
|
|
"""Test adding explicit preferences."""
|
|
peer = PeerRepr(peer_id="user-1")
|
|
peer.add_explicit("item-a", 0.9, "rating")
|
|
|
|
assert "item-a" in peer.explicit_prefs
|
|
assert peer.get_explicit_score("item-a") == 0.9
|
|
|
|
def test_add_deductive_preference(self):
|
|
"""Test adding deduced preferences."""
|
|
peer = PeerRepr(peer_id="user-1")
|
|
peer.add_deductive("item-b", 0.7, "transitive", evidence=["item-a"])
|
|
|
|
assert "item-b" in peer.deduced_prefs
|
|
assert peer.get_deductive_score("item-b") == 0.7
|
|
|
|
def test_add_inductive_preference(self):
|
|
"""Test adding induced preferences."""
|
|
peer = PeerRepr(peer_id="user-1")
|
|
peer.add_inductive("item-c", 0.8, "cluster", supporting_items=["item-a", "item-b"])
|
|
|
|
assert "item-c" in peer.induced_prefs
|
|
assert peer.get_inductive_score("item-c") == 0.8
|
|
|
|
def test_max_scores(self):
|
|
"""Test max score calculations."""
|
|
peer = PeerRepr(peer_id="user-1")
|
|
peer.add_explicit("item-a", 0.9, "rating")
|
|
peer.add_explicit("item-b", 0.7, "rating")
|
|
peer.add_deductive("item-c", 0.8, "transitive")
|
|
|
|
assert peer.max_explicit == 0.9
|
|
assert peer.max_deductive == 0.8
|
|
|
|
def test_final_confidence_calculation(self):
|
|
"""Test weighted confidence aggregation."""
|
|
peer = PeerRepr(peer_id="user-1")
|
|
peer.add_explicit("item-a", 1.0, "rating")
|
|
peer.add_deductive("item-b", 1.0, "transitive")
|
|
peer.add_inductive("item-c", 1.0, "cluster")
|
|
|
|
# Formula: 0.35*max_deductive + 0.35*max_inductive + 0.30*max_explicit
|
|
expected = 0.35 * 1.0 + 0.35 * 1.0 + 0.30 * 1.0
|
|
assert peer.final_confidence == expected
|
|
|
|
def test_all_item_ids(self):
|
|
"""Test collecting all item IDs."""
|
|
peer = PeerRepr(peer_id="user-1")
|
|
peer.add_explicit("item-a", 0.9, "rating")
|
|
peer.add_deductive("item-b", 0.8, "transitive")
|
|
peer.add_inductive("item-c", 0.7, "cluster")
|
|
|
|
items = peer.all_item_ids
|
|
assert len(items) == 3
|
|
assert "item-a" in items
|
|
assert "item-b" in items
|
|
assert "item-c" in items
|
|
|
|
def test_tier_assignment(self):
|
|
"""Test tier property."""
|
|
peer = PeerRepr(peer_id="user-1", tier=Tier.CORE)
|
|
assert peer.tier == Tier.CORE
|
|
|
|
|
|
class TestDecayCalculator:
|
|
def test_normal_decay(self):
|
|
"""Test normal decay rate."""
|
|
calc = DecayCalculator()
|
|
|
|
# After 10 days: score = max(0.1, 1.0 - 10 * 0.015) = max(0.1, 0.85) = 0.85
|
|
score = calc.calculate(1.0, 10.0, Tier.ACTIVE)
|
|
assert score == 0.85
|
|
|
|
def test_core_no_decay(self):
|
|
"""Test core tier never decays."""
|
|
calc = DecayCalculator()
|
|
|
|
score = calc.calculate(0.9, 100.0, Tier.CORE)
|
|
assert score == 0.9 # No decay
|
|
|
|
def test_minimum_score_floor(self):
|
|
"""Test that score doesn't go below minimum."""
|
|
calc = DecayCalculator()
|
|
|
|
# After 100 days: score = max(0.1, 1.0 - 100 * 0.015) = max(0.1, -0.5) = 0.1
|
|
score = calc.calculate(1.0, 100.0, Tier.ACTIVE)
|
|
assert score == 0.1
|
|
|
|
def test_passive_faster_decay(self):
|
|
"""Test passive tier decays faster."""
|
|
calc = DecayCalculator()
|
|
|
|
# Passive decays 50% faster
|
|
# After 10 days: 1.0 - 10 * 0.015 * 1.5 = 0.775
|
|
score_passive = calc.calculate(1.0, 10.0, Tier.PASSIVE)
|
|
score_active = calc.calculate(1.0, 10.0, Tier.ACTIVE)
|
|
|
|
assert score_passive < score_active
|
|
|
|
def test_deprecated_faster_decay(self):
|
|
"""Test deprecated tier decays fastest."""
|
|
calc = DecayCalculator()
|
|
|
|
# Deprecated decays 100% faster
|
|
score_deprecated = calc.calculate(1.0, 10.0, Tier.DEPRECATED)
|
|
score_active = calc.calculate(1.0, 10.0, Tier.ACTIVE)
|
|
|
|
assert score_deprecated < score_active
|
|
|
|
def test_decay_result_details(self):
|
|
"""Test DecayResult contains all details."""
|
|
calc = DecayCalculator()
|
|
|
|
result = calc.calculate_result(1.0, 10.0, Tier.ACTIVE)
|
|
|
|
assert isinstance(result, DecayResult)
|
|
assert result.original_score == 1.0
|
|
assert result.days_elapsed == 10.0
|
|
assert result.decay_applied == pytest.approx(0.15)
|
|
|
|
|
|
class TestExplicitInference:
|
|
def test_extract_like_action(self):
|
|
"""Test extracting from like action."""
|
|
inferrer = ExplicitInference()
|
|
context = InferenceContext(peer=PeerRepr(peer_id="user-1"))
|
|
|
|
pref = inferrer.extract(context, "item-1", "like", 1.0)
|
|
|
|
assert pref.item_id == "item-1"
|
|
assert pref.score == 1.0
|
|
assert pref.source == "like"
|
|
|
|
def test_extract_rating_normalization(self):
|
|
"""Test rating normalization to 0-1."""
|
|
inferrer = ExplicitInference()
|
|
context = InferenceContext(peer=PeerRepr(peer_id="user-1"))
|
|
|
|
# 5-star rating -> 5/5 = 1.0
|
|
pref = inferrer.extract(context, "item-1", "rating_5", 4.0)
|
|
assert pref.score == 0.8
|
|
|
|
def test_extract_percentage(self):
|
|
"""Test percentage action."""
|
|
inferrer = ExplicitInference()
|
|
context = InferenceContext(peer=PeerRepr(peer_id="user-1"))
|
|
|
|
pref = inferrer.extract(context, "item-1", "percentage", 75.0)
|
|
assert pref.score == 0.75
|
|
|
|
|
|
class TestDeductiveInference:
|
|
def test_infer_deductive_preference(self):
|
|
"""Test deductive inference."""
|
|
inferrer = DeductiveInference()
|
|
context = InferenceContext(peer=PeerRepr(peer_id="user-1"))
|
|
|
|
# Setup: User likes item-a
|
|
peer = context.peer
|
|
peer.add_explicit("item-a", 0.9, "rating")
|
|
|
|
pref = inferrer.infer(
|
|
context, "item-b", "item-a", 0.9,
|
|
inference_type="transitive", confidence=0.8
|
|
)
|
|
|
|
assert pref.item_id == "item-b"
|
|
assert pref.inference_type == "transitive"
|
|
assert 0 < pref.score <= 0.9 # transitive has 0.9 strength
|
|
|
|
|
|
class TestInductiveInference:
|
|
def test_discover_from_cluster(self):
|
|
"""Test discovering preference from cluster."""
|
|
inferrer = InductiveInference()
|
|
context = InferenceContext(peer=PeerRepr(peer_id="user-1"))
|
|
|
|
# Setup: User likes item-a
|
|
peer = context.peer
|
|
peer.add_explicit("item-a", 0.9, "rating")
|
|
|
|
cluster_membership = {
|
|
"cluster-1": ["item-a", "item-b", "item-c"]
|
|
}
|
|
|
|
pref = inferrer.discover_from_clusters(
|
|
context, "item-b", ["item-a"], cluster_membership, 0.8
|
|
)
|
|
|
|
assert pref.item_id == "item-b"
|
|
assert pref.pattern_type == "cluster"
|
|
|
|
def test_discover(self):
|
|
"""Test basic discover method."""
|
|
inferrer = InductiveInference()
|
|
context = InferenceContext(peer=PeerRepr(peer_id="user-1"))
|
|
|
|
pref = inferrer.discover(
|
|
context, "item-x", "cluster", ["item-a", "item-b"], 0.8, 0.75
|
|
)
|
|
|
|
assert pref.item_id == "item-x"
|
|
assert pref.pattern_type == "cluster"
|
|
assert 0 < pref.score <= 0.8
|
|
|
|
|
|
class TestTierEnum:
|
|
def test_tier_values(self):
|
|
"""Test all tier values exist."""
|
|
assert Tier.CORE.value == "core"
|
|
assert Tier.ACTIVE.value == "active"
|
|
assert Tier.PASSIVE.value == "passive"
|
|
assert Tier.DEPRECATED.value == "deprecated"
|
|
|
|
def test_tier_assignment_affects_decay(self):
|
|
"""Test different tiers produce different decay results."""
|
|
calc = DecayCalculator()
|
|
|
|
core = calc.calculate(1.0, 10.0, Tier.CORE)
|
|
active = calc.calculate(1.0, 10.0, Tier.ACTIVE)
|
|
|
|
# Core should be higher (no decay)
|
|
assert core > active |