7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
5549 lines
184 KiB
Python
5549 lines
184 KiB
Python
"""
|
|
Tests for news/recommender/base_recommender.py
|
|
|
|
Tests cover:
|
|
- BaseRecommender initialization
|
|
- Progress callback handling
|
|
- User preference access
|
|
- Abstract method requirements
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock
|
|
from abc import ABC
|
|
|
|
|
|
class TestBaseRecommenderInit:
|
|
"""Tests for BaseRecommender initialization."""
|
|
|
|
def test_base_recommender_is_abstract(self):
|
|
"""BaseRecommender is an abstract class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
assert issubclass(BaseRecommender, ABC)
|
|
|
|
def test_base_recommender_has_abstract_method(self):
|
|
"""BaseRecommender requires generate_recommendations."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
assert hasattr(BaseRecommender, "generate_recommendations")
|
|
|
|
|
|
class TestConcreteRecommender:
|
|
"""Tests using a concrete implementation of BaseRecommender."""
|
|
|
|
@pytest.fixture
|
|
def mock_preference_manager(self):
|
|
"""Create mock preference manager."""
|
|
mock = Mock()
|
|
mock.get_preferences.return_value = {"topic": "test"}
|
|
return mock
|
|
|
|
@pytest.fixture
|
|
def mock_rating_system(self):
|
|
"""Create mock rating system."""
|
|
mock = Mock()
|
|
mock.get_user_ratings.return_value = []
|
|
return mock
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
return TestRecommender
|
|
|
|
def test_recommender_initialization_with_defaults(
|
|
self, concrete_recommender
|
|
):
|
|
"""Recommender initializes with default None values."""
|
|
recommender = concrete_recommender()
|
|
|
|
assert recommender.preference_manager is None
|
|
assert recommender.rating_system is None
|
|
assert recommender.topic_registry is None
|
|
assert recommender.search_system is None
|
|
assert recommender.progress_callback is None
|
|
|
|
def test_recommender_initialization_with_dependencies(
|
|
self, concrete_recommender, mock_preference_manager, mock_rating_system
|
|
):
|
|
"""Recommender initializes with provided dependencies."""
|
|
recommender = concrete_recommender(
|
|
preference_manager=mock_preference_manager,
|
|
rating_system=mock_rating_system,
|
|
)
|
|
|
|
assert recommender.preference_manager is mock_preference_manager
|
|
assert recommender.rating_system is mock_rating_system
|
|
|
|
def test_strategy_name_is_class_name(self, concrete_recommender):
|
|
"""Strategy name is set to class name."""
|
|
recommender = concrete_recommender()
|
|
|
|
assert recommender.strategy_name == "TestRecommender"
|
|
|
|
|
|
class TestProgressCallback:
|
|
"""Tests for progress callback functionality."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
self._update_progress("Processing", 50, {"step": 1})
|
|
return []
|
|
|
|
return TestRecommender
|
|
|
|
def test_set_progress_callback(self, concrete_recommender):
|
|
"""Progress callback can be set."""
|
|
recommender = concrete_recommender()
|
|
callback = Mock()
|
|
|
|
recommender.set_progress_callback(callback)
|
|
|
|
assert recommender.progress_callback is callback
|
|
|
|
def test_update_progress_calls_callback(self, concrete_recommender):
|
|
"""_update_progress calls the callback when set."""
|
|
recommender = concrete_recommender()
|
|
callback = Mock()
|
|
recommender.set_progress_callback(callback)
|
|
|
|
recommender._update_progress("Test message", 50, {"key": "value"})
|
|
|
|
callback.assert_called_once_with("Test message", 50, {"key": "value"})
|
|
|
|
def test_update_progress_does_nothing_without_callback(
|
|
self, concrete_recommender
|
|
):
|
|
"""_update_progress doesn't fail without callback."""
|
|
recommender = concrete_recommender()
|
|
|
|
# Should not raise
|
|
recommender._update_progress("Test message", 50, {})
|
|
|
|
def test_update_progress_default_metadata(self, concrete_recommender):
|
|
"""_update_progress uses empty dict for default metadata."""
|
|
recommender = concrete_recommender()
|
|
callback = Mock()
|
|
recommender.set_progress_callback(callback)
|
|
|
|
recommender._update_progress("Test message", 50)
|
|
|
|
callback.assert_called_once_with("Test message", 50, {})
|
|
|
|
|
|
class TestUserPreferences:
|
|
"""Tests for user preference handling."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def get_prefs(self, user_id):
|
|
return self._get_user_preferences(user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_get_user_preferences_with_manager(self, concrete_recommender):
|
|
"""_get_user_preferences returns preferences when manager available."""
|
|
mock_manager = Mock()
|
|
mock_manager.get_preferences.return_value = {"topic": "test"}
|
|
|
|
recommender = concrete_recommender(preference_manager=mock_manager)
|
|
prefs = recommender.get_prefs("user123")
|
|
|
|
assert prefs == {"topic": "test"}
|
|
mock_manager.get_preferences.assert_called_once_with("user123")
|
|
|
|
def test_get_user_preferences_without_manager(self, concrete_recommender):
|
|
"""_get_user_preferences returns empty dict without manager."""
|
|
recommender = concrete_recommender()
|
|
prefs = recommender.get_prefs("user123")
|
|
|
|
assert prefs == {}
|
|
|
|
|
|
class TestGenerateRecommendations:
|
|
"""Tests for the generate_recommendations abstract method."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return [{"id": 1, "topic": "test"}]
|
|
|
|
return TestRecommender
|
|
|
|
def test_generate_recommendations_returns_list(self, concrete_recommender):
|
|
"""generate_recommendations returns a list."""
|
|
recommender = concrete_recommender()
|
|
result = recommender.generate_recommendations("user123")
|
|
|
|
assert isinstance(result, list)
|
|
|
|
def test_generate_recommendations_accepts_context(
|
|
self, concrete_recommender
|
|
):
|
|
"""generate_recommendations accepts optional context."""
|
|
recommender = concrete_recommender()
|
|
|
|
# Should not raise
|
|
result = recommender.generate_recommendations(
|
|
"user123", context={"page": "home"}
|
|
)
|
|
|
|
assert isinstance(result, list)
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for _get_user_ratings
|
|
# =============================================================================
|
|
|
|
|
|
class TestGetUserRatings:
|
|
"""Tests for the _get_user_ratings method."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def get_ratings(self, user_id, limit=50):
|
|
return self._get_user_ratings(user_id, limit)
|
|
|
|
return TestRecommender
|
|
|
|
def test_returns_ratings_from_system(self, concrete_recommender):
|
|
"""_get_user_ratings returns ratings from rating_system."""
|
|
mock_rating_system = Mock()
|
|
mock_rating_system.get_recent_ratings.return_value = [
|
|
{"card_id": "card1", "rating": 5},
|
|
{"card_id": "card2", "rating": 3},
|
|
]
|
|
|
|
recommender = concrete_recommender(rating_system=mock_rating_system)
|
|
ratings = recommender.get_ratings("user123")
|
|
|
|
assert len(ratings) == 2
|
|
assert ratings[0]["card_id"] == "card1"
|
|
mock_rating_system.get_recent_ratings.assert_called_once_with(
|
|
"user123", 50
|
|
)
|
|
|
|
def test_respects_limit_parameter(self, concrete_recommender):
|
|
"""_get_user_ratings passes limit to rating_system."""
|
|
mock_rating_system = Mock()
|
|
mock_rating_system.get_recent_ratings.return_value = []
|
|
|
|
recommender = concrete_recommender(rating_system=mock_rating_system)
|
|
recommender.get_ratings("user123", limit=10)
|
|
|
|
mock_rating_system.get_recent_ratings.assert_called_once_with(
|
|
"user123", 10
|
|
)
|
|
|
|
def test_returns_empty_when_no_rating_system(self, concrete_recommender):
|
|
"""_get_user_ratings returns empty list when no rating_system."""
|
|
recommender = concrete_recommender()
|
|
ratings = recommender.get_ratings("user123")
|
|
|
|
assert ratings == []
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for _execute_search
|
|
# =============================================================================
|
|
|
|
|
|
class TestExecuteSearch:
|
|
"""Tests for the _execute_search method."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def do_search(self, query, strategy=None):
|
|
return self._execute_search(query, strategy)
|
|
|
|
return TestRecommender
|
|
|
|
def test_returns_error_when_no_search_system(self, concrete_recommender):
|
|
"""_execute_search returns error dict when no search_system."""
|
|
recommender = concrete_recommender()
|
|
|
|
result = recommender.do_search("test query")
|
|
|
|
assert "error" in result
|
|
assert result["error"] == "Search system not configured"
|
|
|
|
def test_uses_news_aggregation_strategy(self, concrete_recommender):
|
|
"""_execute_search uses news_aggregation as default strategy."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = {"results": []}
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
|
|
# Strategy defaults to "news_aggregation" but analyze_topic is called
|
|
recommender.do_search("test query")
|
|
|
|
mock_search.analyze_topic.assert_called_once_with("test query")
|
|
|
|
def test_calls_analyze_topic(self, concrete_recommender):
|
|
"""_execute_search calls analyze_topic with the query."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = {"findings": ["test"]}
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
result = recommender.do_search("climate change news")
|
|
|
|
mock_search.analyze_topic.assert_called_once_with("climate change news")
|
|
assert result == {"findings": ["test"]}
|
|
|
|
def test_handles_exception_gracefully(self, concrete_recommender):
|
|
"""_execute_search returns error dict on exception."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.side_effect = Exception("Search failed")
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
result = recommender.do_search("test query")
|
|
|
|
assert "error" in result
|
|
assert result["error"] == "Recommendation search failed"
|
|
|
|
def test_logs_search_execution(self, concrete_recommender):
|
|
"""_execute_search logs errors on failure."""
|
|
from unittest.mock import patch
|
|
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.side_effect = Exception("Network error")
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
|
|
with patch(
|
|
"local_deep_research.news.recommender.base_recommender.logger"
|
|
) as mock_logger:
|
|
recommender.do_search("test query")
|
|
|
|
mock_logger.exception.assert_called_once()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for _filter_by_preferences (CRITICAL)
|
|
# =============================================================================
|
|
|
|
|
|
class TestFilterByPreferences:
|
|
"""Tests for the _filter_by_preferences method - core filtering logic."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
@pytest.fixture
|
|
def sample_cards(self):
|
|
"""Create sample NewsCard objects for testing."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
|
|
cards = [
|
|
NewsCard(
|
|
topic="AI in healthcare",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Technology",
|
|
impact_score=8,
|
|
),
|
|
NewsCard(
|
|
topic="Climate change policy",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Environment",
|
|
impact_score=6,
|
|
),
|
|
NewsCard(
|
|
topic="Stock market update",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Finance",
|
|
impact_score=4,
|
|
),
|
|
NewsCard(
|
|
topic="Machine learning breakthrough",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Technology",
|
|
impact_score=9,
|
|
),
|
|
]
|
|
return cards
|
|
|
|
def test_returns_all_cards_with_empty_preferences(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Empty preferences returns all cards unchanged."""
|
|
recommender = concrete_recommender()
|
|
|
|
result = recommender.filter_cards(sample_cards, {})
|
|
|
|
assert len(result) == len(sample_cards)
|
|
|
|
def test_adds_preference_boost_for_liked_categories(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Cards in liked_categories get preference_boost in metadata."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["Technology"]}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
# Technology cards should have boost
|
|
tech_cards = [c for c in result if c.category == "Technology"]
|
|
assert len(tech_cards) == 2
|
|
for card in tech_cards:
|
|
assert card.metadata.get("preference_boost") == 1.2
|
|
|
|
# Non-technology cards should not have boost
|
|
non_tech_cards = [c for c in result if c.category != "Technology"]
|
|
for card in non_tech_cards:
|
|
assert "preference_boost" not in card.metadata
|
|
|
|
def test_filters_low_impact_cards(self, concrete_recommender, sample_cards):
|
|
"""Cards below impact_threshold are filtered out."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": 5}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
# Only cards with impact_score >= 5 should remain
|
|
assert len(result) == 3
|
|
for card in result:
|
|
assert card.impact_score >= 5
|
|
|
|
def test_removes_cards_with_disliked_topics(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Cards matching disliked_topics are removed."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["stock", "market"]}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
# "Stock market update" should be filtered out
|
|
assert len(result) == 3
|
|
topics = [c.topic for c in result]
|
|
assert "Stock market update" not in topics
|
|
|
|
def test_topic_matching_converts_topic_to_lowercase(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Disliked topics must be lowercase since topic.lower() is used."""
|
|
recommender = concrete_recommender()
|
|
# Disliked topics must be lowercase to match (topic is lowercased)
|
|
preferences = {"disliked_topics": ["ai"]}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
# "AI in healthcare" should be filtered out (topic is lowercased)
|
|
topics = [c.topic for c in result]
|
|
assert "AI in healthcare" not in topics
|
|
|
|
def test_handles_multiple_filter_criteria(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Multiple filter criteria are applied together."""
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": ["Technology"],
|
|
"impact_threshold": 7,
|
|
"disliked_topics": ["stock"],
|
|
}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
# Should filter by impact >= 7 AND not contain "stock"
|
|
# Remaining: AI in healthcare (8), Machine learning breakthrough (9)
|
|
assert len(result) == 2
|
|
for card in result:
|
|
assert card.impact_score >= 7
|
|
|
|
def test_handles_empty_cards_list(self, concrete_recommender):
|
|
"""Empty cards list returns empty list."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": 5}
|
|
|
|
result = recommender.filter_cards([], preferences)
|
|
|
|
assert result == []
|
|
|
|
def test_handles_none_values_in_safe_fields(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""None values in liked_categories and disliked_topics are handled safely."""
|
|
recommender = concrete_recommender()
|
|
# Note: None in impact_threshold will raise TypeError (comparison with None)
|
|
# Only liked_categories and disliked_topics handle None gracefully
|
|
preferences = {
|
|
"liked_categories": None,
|
|
"disliked_topics": None,
|
|
}
|
|
|
|
# Should not raise
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
# All cards should be returned (no filtering applied for None values)
|
|
assert len(result) == len(sample_cards)
|
|
|
|
def test_preserves_original_card_order(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Filtering preserves the original order of cards."""
|
|
recommender = concrete_recommender()
|
|
# Filter out one card
|
|
preferences = {"disliked_topics": ["stock"]}
|
|
|
|
original_order = [
|
|
c.topic for c in sample_cards if "stock" not in c.topic.lower()
|
|
]
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
result_order = [c.topic for c in result]
|
|
|
|
assert result_order == original_order
|
|
|
|
def test_filters_partial_topic_matches(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Disliked topics filter on substring matches."""
|
|
recommender = concrete_recommender()
|
|
# "machine" should match "Machine learning breakthrough"
|
|
preferences = {"disliked_topics": ["machine"]}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
topics = [c.topic for c in result]
|
|
assert "Machine learning breakthrough" not in topics
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for _sort_by_relevance
|
|
# =============================================================================
|
|
|
|
|
|
class TestSortByRelevance:
|
|
"""Tests for the _sort_by_relevance method."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
@pytest.fixture
|
|
def sample_cards(self):
|
|
"""Create sample NewsCard objects with varying scores."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
|
|
cards = [
|
|
NewsCard(
|
|
topic="Low impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=3,
|
|
),
|
|
NewsCard(
|
|
topic="High impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=9,
|
|
),
|
|
NewsCard(
|
|
topic="Medium impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=6,
|
|
),
|
|
]
|
|
return cards
|
|
|
|
def test_sorts_by_impact_score_descending(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Cards are sorted by impact_score in descending order."""
|
|
recommender = concrete_recommender()
|
|
|
|
result = recommender.sort_cards(sample_cards, "user123")
|
|
|
|
# Should be ordered: 9, 6, 3
|
|
assert result[0].impact_score == 9
|
|
assert result[1].impact_score == 6
|
|
assert result[2].impact_score == 3
|
|
|
|
def test_applies_preference_boost(self, concrete_recommender, sample_cards):
|
|
"""Preference boost affects sorting order."""
|
|
recommender = concrete_recommender()
|
|
|
|
# Give the low impact card a boost
|
|
sample_cards[0].metadata["preference_boost"] = 5.0 # Low impact (3)
|
|
|
|
result = recommender.sort_cards(sample_cards, "user123")
|
|
|
|
# Low impact card (3 * 5.0 = 15) should now be first
|
|
# Score calculation: (impact/10) * boost
|
|
# Low: (3/10) * 5.0 = 1.5
|
|
# High: (9/10) * 1.0 = 0.9
|
|
# Medium: (6/10) * 1.0 = 0.6
|
|
assert result[0].topic == "Low impact"
|
|
|
|
def test_handles_equal_scores_stably(self, concrete_recommender):
|
|
"""Cards with equal scores maintain stable sort."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="First", source=source, user_id="user1", impact_score=5
|
|
),
|
|
NewsCard(
|
|
topic="Second", source=source, user_id="user1", impact_score=5
|
|
),
|
|
NewsCard(
|
|
topic="Third", source=source, user_id="user1", impact_score=5
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards(cards, "user123")
|
|
|
|
# Python's sort is stable - equal elements maintain relative order
|
|
assert len(result) == 3
|
|
|
|
def test_handles_empty_list(self, concrete_recommender):
|
|
"""Empty list returns empty list."""
|
|
recommender = concrete_recommender()
|
|
|
|
result = recommender.sort_cards([], "user123")
|
|
|
|
assert result == []
|
|
|
|
def test_score_calculation(self, concrete_recommender):
|
|
"""Score is calculated as (impact/10) * boost."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
|
|
# Card with impact 8 and boost 1.5
|
|
# Score = (8/10) * 1.5 = 1.2
|
|
card_boosted = NewsCard(
|
|
topic="Boosted", source=source, user_id="user1", impact_score=8
|
|
)
|
|
card_boosted.metadata["preference_boost"] = 1.5
|
|
|
|
# Card with impact 10 and no boost (default 1.0)
|
|
# Score = (10/10) * 1.0 = 1.0
|
|
card_high = NewsCard(
|
|
topic="High", source=source, user_id="user1", impact_score=10
|
|
)
|
|
|
|
cards = [card_high, card_boosted]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards(cards, "user123")
|
|
|
|
# Boosted card (1.2) should rank higher than high impact (1.0)
|
|
assert result[0].topic == "Boosted"
|
|
assert result[1].topic == "High"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for get_strategy_info
|
|
# =============================================================================
|
|
|
|
|
|
class TestGetStrategyInfo:
|
|
"""Tests for the get_strategy_info method."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class MyCustomRecommender(BaseRecommender):
|
|
"""Custom recommender for testing info display."""
|
|
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
return MyCustomRecommender
|
|
|
|
def test_returns_correct_structure(self, concrete_recommender):
|
|
"""get_strategy_info returns dict with expected keys."""
|
|
recommender = concrete_recommender()
|
|
|
|
info = recommender.get_strategy_info()
|
|
|
|
assert "name" in info
|
|
assert "has_preference_manager" in info
|
|
assert "has_rating_system" in info
|
|
assert "has_search_system" in info
|
|
assert "description" in info
|
|
|
|
def test_boolean_flags_reflect_dependencies(self, concrete_recommender):
|
|
"""has_* flags accurately reflect dependency availability."""
|
|
# Without dependencies
|
|
recommender_empty = concrete_recommender()
|
|
info_empty = recommender_empty.get_strategy_info()
|
|
|
|
assert info_empty["has_preference_manager"] is False
|
|
assert info_empty["has_rating_system"] is False
|
|
assert info_empty["has_search_system"] is False
|
|
|
|
# With dependencies
|
|
recommender_full = concrete_recommender(
|
|
preference_manager=Mock(),
|
|
rating_system=Mock(),
|
|
search_system=Mock(),
|
|
)
|
|
info_full = recommender_full.get_strategy_info()
|
|
|
|
assert info_full["has_preference_manager"] is True
|
|
assert info_full["has_rating_system"] is True
|
|
assert info_full["has_search_system"] is True
|
|
|
|
def test_uses_class_name_as_strategy(self, concrete_recommender):
|
|
"""Strategy name is derived from class name."""
|
|
recommender = concrete_recommender()
|
|
|
|
info = recommender.get_strategy_info()
|
|
|
|
assert info["name"] == "MyCustomRecommender"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for topic_registry attribute
|
|
# =============================================================================
|
|
|
|
|
|
class TestTopicRegistry:
|
|
"""Tests for topic_registry attribute handling."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
return TestRecommender
|
|
|
|
def test_topic_registry_is_none_by_default(self, concrete_recommender):
|
|
"""topic_registry is None when not provided."""
|
|
recommender = concrete_recommender()
|
|
|
|
assert recommender.topic_registry is None
|
|
|
|
def test_topic_registry_can_be_set(self, concrete_recommender):
|
|
"""topic_registry can be set during initialization."""
|
|
mock_registry = Mock()
|
|
mock_registry.get_topics.return_value = ["AI", "Climate"]
|
|
|
|
recommender = concrete_recommender(topic_registry=mock_registry)
|
|
|
|
assert recommender.topic_registry is mock_registry
|
|
assert recommender.topic_registry.get_topics() == ["AI", "Climate"]
|
|
|
|
|
|
# =============================================================================
|
|
# Additional tests for _execute_search
|
|
# =============================================================================
|
|
|
|
|
|
class TestExecuteSearchAdditional:
|
|
"""Additional tests for _execute_search edge cases."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def do_search(self, query, strategy=None):
|
|
return self._execute_search(query, strategy)
|
|
|
|
return TestRecommender
|
|
|
|
def test_ignores_strategy_parameter(self, concrete_recommender):
|
|
"""_execute_search calls analyze_topic regardless of strategy parameter."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = {"results": []}
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
|
|
# Strategy parameter is set but analyze_topic is still used
|
|
recommender.do_search("test query", strategy="custom_strategy")
|
|
|
|
mock_search.analyze_topic.assert_called_once_with("test query")
|
|
|
|
def test_handles_none_result_from_search(self, concrete_recommender):
|
|
"""_execute_search handles None return from analyze_topic."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = None
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
result = recommender.do_search("test query")
|
|
|
|
assert result is None
|
|
|
|
def test_handles_empty_dict_result(self, concrete_recommender):
|
|
"""_execute_search handles empty dict return."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = {}
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
result = recommender.do_search("test query")
|
|
|
|
assert result == {}
|
|
|
|
def test_handles_various_exception_types(self, concrete_recommender):
|
|
"""_execute_search handles various exception types."""
|
|
mock_search = Mock()
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
|
|
# Test with different exception types
|
|
for exc_type in [ValueError, RuntimeError, KeyError, TimeoutError]:
|
|
mock_search.analyze_topic.side_effect = exc_type("Error")
|
|
|
|
result = recommender.do_search("test query")
|
|
|
|
assert "error" in result
|
|
assert result["error"] == "Recommendation search failed"
|
|
|
|
def test_logs_warning_when_no_search_system(self, concrete_recommender):
|
|
"""_execute_search logs warning when search system is not available."""
|
|
from unittest.mock import patch
|
|
|
|
recommender = concrete_recommender()
|
|
|
|
with patch(
|
|
"local_deep_research.news.recommender.base_recommender.logger"
|
|
) as mock_logger:
|
|
recommender.do_search("test query")
|
|
|
|
mock_logger.warning.assert_called_once()
|
|
|
|
|
|
# =============================================================================
|
|
# Additional tests for _filter_by_preferences
|
|
# =============================================================================
|
|
|
|
|
|
class TestFilterByPreferencesAdditional:
|
|
"""Additional tests for _filter_by_preferences edge cases."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
@pytest.fixture
|
|
def sample_cards(self):
|
|
"""Create sample NewsCard objects."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
|
|
return [
|
|
NewsCard(
|
|
topic="Python programming tips",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Programming",
|
|
impact_score=7,
|
|
),
|
|
NewsCard(
|
|
topic="JavaScript frameworks",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Programming",
|
|
impact_score=5,
|
|
),
|
|
NewsCard(
|
|
topic="Data science trends",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Data Science",
|
|
impact_score=8,
|
|
),
|
|
]
|
|
|
|
def test_only_impact_threshold_filter(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Test filtering with only impact_threshold preference."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": 6}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
# Should keep cards with score >= 6
|
|
assert len(result) == 2
|
|
topics = [c.topic for c in result]
|
|
assert "JavaScript frameworks" not in topics # score 5
|
|
|
|
def test_only_liked_categories_filter(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Test filtering with only liked_categories preference."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["Data Science"]}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
# All cards returned, but Data Science cards get boost
|
|
assert len(result) == 3
|
|
|
|
data_science_cards = [c for c in result if c.category == "Data Science"]
|
|
programming_cards = [c for c in result if c.category == "Programming"]
|
|
|
|
assert data_science_cards[0].metadata.get("preference_boost") == 1.2
|
|
for card in programming_cards:
|
|
assert "preference_boost" not in card.metadata
|
|
|
|
def test_only_disliked_topics_filter(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Test filtering with only disliked_topics preference."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["javascript"]}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
assert len(result) == 2
|
|
topics = [c.topic for c in result]
|
|
assert "JavaScript frameworks" not in topics
|
|
|
|
def test_disliked_topics_with_multiple_keywords(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Test disliked_topics with multiple keywords."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["python", "javascript"]}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
assert len(result) == 1
|
|
assert result[0].topic == "Data science trends"
|
|
|
|
def test_empty_liked_categories_list(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Empty liked_categories list applies no boost."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": []}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
# All cards returned, none with boost
|
|
assert len(result) == 3
|
|
for card in result:
|
|
assert "preference_boost" not in card.metadata
|
|
|
|
def test_empty_disliked_topics_list(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Empty disliked_topics list filters nothing."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": []}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
assert len(result) == 3
|
|
|
|
def test_all_cards_filtered_out(self, concrete_recommender, sample_cards):
|
|
"""All cards can be filtered out if threshold too high."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": 100}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
def test_boost_does_not_modify_impact_score(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Preference boost goes to metadata, not impact_score."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["Programming"]}
|
|
|
|
original_scores = [c.impact_score for c in sample_cards]
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
result_scores = [c.impact_score for c in result]
|
|
assert original_scores == result_scores
|
|
|
|
def test_unrecognized_preference_keys_ignored(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Unrecognized preference keys are ignored."""
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"unknown_key": "value",
|
|
"another_unknown": [1, 2, 3],
|
|
"impact_threshold": 6,
|
|
}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
# Only impact_threshold applied
|
|
assert len(result) == 2
|
|
|
|
|
|
# =============================================================================
|
|
# Additional tests for _sort_by_relevance
|
|
# =============================================================================
|
|
|
|
|
|
class TestSortByRelevanceAdditional:
|
|
"""Additional tests for _sort_by_relevance edge cases."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_single_card_returns_same_card(self, concrete_recommender):
|
|
"""Single card list returns that card."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Single", source=source, user_id="user1", impact_score=5
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards([card], "user123")
|
|
|
|
assert len(result) == 1
|
|
assert result[0].topic == "Single"
|
|
|
|
def test_sort_with_zero_impact_scores(self, concrete_recommender):
|
|
"""Cards with zero impact scores are sorted correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Zero", source=source, user_id="user1", impact_score=0
|
|
),
|
|
NewsCard(
|
|
topic="Five", source=source, user_id="user1", impact_score=5
|
|
),
|
|
NewsCard(
|
|
topic="Zero2", source=source, user_id="user1", impact_score=0
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards(cards, "user123")
|
|
|
|
assert result[0].topic == "Five"
|
|
assert result[0].impact_score == 5
|
|
|
|
def test_sort_with_negative_boost(self, concrete_recommender):
|
|
"""Negative boost reduces effective score."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card_boosted = NewsCard(
|
|
topic="Negative Boost",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=10,
|
|
)
|
|
card_boosted.metadata["preference_boost"] = 0.1 # Very low boost
|
|
|
|
card_normal = NewsCard(
|
|
topic="Normal", source=source, user_id="user1", impact_score=5
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards([card_boosted, card_normal], "user123")
|
|
|
|
# Normal (5/10 * 1.0 = 0.5) > Negative Boost (10/10 * 0.1 = 0.1)
|
|
assert result[0].topic == "Normal"
|
|
|
|
def test_sort_with_very_large_boost(self, concrete_recommender):
|
|
"""Very large boost values work correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card_huge_boost = NewsCard(
|
|
topic="Huge Boost", source=source, user_id="user1", impact_score=1
|
|
)
|
|
card_huge_boost.metadata["preference_boost"] = 100.0
|
|
|
|
card_high_score = NewsCard(
|
|
topic="High Score", source=source, user_id="user1", impact_score=10
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards(
|
|
[card_high_score, card_huge_boost], "user123"
|
|
)
|
|
|
|
# Huge Boost (1/10 * 100 = 10) > High Score (10/10 * 1 = 1)
|
|
assert result[0].topic == "Huge Boost"
|
|
|
|
def test_sort_preserves_card_objects(self, concrete_recommender):
|
|
"""Sorting returns the same card objects, not copies."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
original_card = NewsCard(
|
|
topic="Original", source=source, user_id="user1", impact_score=5
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards([original_card], "user123")
|
|
|
|
assert result[0] is original_card
|
|
|
|
|
|
# =============================================================================
|
|
# Additional progress callback tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestProgressCallbackAdditional:
|
|
"""Additional tests for progress callback functionality."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
self._update_progress("Starting", 0)
|
|
self._update_progress("Processing", 50, {"step": "middle"})
|
|
self._update_progress("Complete", 100, {"step": "done"})
|
|
return []
|
|
|
|
return TestRecommender
|
|
|
|
def test_callback_receives_all_updates(self, concrete_recommender):
|
|
"""Callback receives all progress updates in order."""
|
|
recommender = concrete_recommender()
|
|
calls = []
|
|
|
|
def track_callback(message, progress, metadata):
|
|
calls.append((message, progress, metadata))
|
|
|
|
recommender.set_progress_callback(track_callback)
|
|
recommender.generate_recommendations("user123")
|
|
|
|
assert len(calls) == 3
|
|
assert calls[0] == ("Starting", 0, {})
|
|
assert calls[1] == ("Processing", 50, {"step": "middle"})
|
|
assert calls[2] == ("Complete", 100, {"step": "done"})
|
|
|
|
def test_callback_can_be_replaced(self, concrete_recommender):
|
|
"""Progress callback can be replaced."""
|
|
recommender = concrete_recommender()
|
|
|
|
first_callback = Mock()
|
|
second_callback = Mock()
|
|
|
|
recommender.set_progress_callback(first_callback)
|
|
recommender.set_progress_callback(second_callback)
|
|
|
|
recommender._update_progress("Test", 50)
|
|
|
|
first_callback.assert_not_called()
|
|
second_callback.assert_called_once()
|
|
|
|
def test_callback_can_be_removed(self, concrete_recommender):
|
|
"""Progress callback can be set to None to remove it."""
|
|
recommender = concrete_recommender()
|
|
callback = Mock()
|
|
|
|
recommender.set_progress_callback(callback)
|
|
recommender.set_progress_callback(None)
|
|
|
|
# Should not raise
|
|
recommender._update_progress("Test", 50)
|
|
callback.assert_not_called()
|
|
|
|
def test_update_progress_with_none_percent(self, concrete_recommender):
|
|
"""_update_progress works with None progress percent."""
|
|
recommender = concrete_recommender()
|
|
callback = Mock()
|
|
recommender.set_progress_callback(callback)
|
|
|
|
recommender._update_progress("Status update", None, {"key": "value"})
|
|
|
|
callback.assert_called_once_with(
|
|
"Status update", None, {"key": "value"}
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for get_strategy_info additional coverage
|
|
# =============================================================================
|
|
|
|
|
|
class TestGetStrategyInfoAdditional:
|
|
"""Additional tests for get_strategy_info."""
|
|
|
|
def test_description_from_docstring(self):
|
|
"""get_strategy_info uses class docstring for description."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class DocumentedRecommender(BaseRecommender):
|
|
"""This is a custom recommendation strategy for testing."""
|
|
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
recommender = DocumentedRecommender()
|
|
info = recommender.get_strategy_info()
|
|
|
|
assert (
|
|
info["description"]
|
|
== "This is a custom recommendation strategy for testing."
|
|
)
|
|
|
|
def test_description_fallback_when_no_docstring(self):
|
|
"""get_strategy_info uses fallback when no docstring."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class UndocumentedRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
# Remove docstring
|
|
UndocumentedRecommender.__doc__ = None
|
|
|
|
recommender = UndocumentedRecommender()
|
|
info = recommender.get_strategy_info()
|
|
|
|
assert info["description"] == "No description available"
|
|
|
|
|
|
# =============================================================================
|
|
# Integration tests for recommender
|
|
# =============================================================================
|
|
|
|
|
|
class TestRecommenderIntegration:
|
|
"""Integration tests for recommender workflow."""
|
|
|
|
@pytest.fixture
|
|
def full_recommender(self):
|
|
"""Create a recommender with all dependencies."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class FullRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
# Get preferences
|
|
prefs = self._get_user_preferences(user_id)
|
|
|
|
# Get ratings (exercises the method, result not used in this test)
|
|
self._get_user_ratings(user_id, limit=10)
|
|
|
|
# Create some cards
|
|
from local_deep_research.news.core.base_card import (
|
|
NewsCard,
|
|
CardSource,
|
|
)
|
|
|
|
source = CardSource(type="recommendation")
|
|
cards = [
|
|
NewsCard(
|
|
topic="AI News",
|
|
source=source,
|
|
user_id=user_id,
|
|
category="Technology",
|
|
impact_score=8,
|
|
),
|
|
NewsCard(
|
|
topic="Sports Update",
|
|
source=source,
|
|
user_id=user_id,
|
|
category="Sports",
|
|
impact_score=6,
|
|
),
|
|
NewsCard(
|
|
topic="Weather Report",
|
|
source=source,
|
|
user_id=user_id,
|
|
category="Weather",
|
|
impact_score=4,
|
|
),
|
|
]
|
|
|
|
# Filter and sort
|
|
filtered = self._filter_by_preferences(cards, prefs)
|
|
sorted_cards = self._sort_by_relevance(filtered, user_id)
|
|
|
|
return sorted_cards
|
|
|
|
return FullRecommender
|
|
|
|
def test_full_workflow_without_preferences(self, full_recommender):
|
|
"""Test complete workflow without user preferences."""
|
|
mock_pref_manager = Mock()
|
|
mock_pref_manager.get_preferences.return_value = {}
|
|
|
|
mock_rating_system = Mock()
|
|
mock_rating_system.get_recent_ratings.return_value = []
|
|
|
|
recommender = full_recommender(
|
|
preference_manager=mock_pref_manager,
|
|
rating_system=mock_rating_system,
|
|
)
|
|
|
|
result = recommender.generate_recommendations("user123")
|
|
|
|
# All cards returned, sorted by impact score
|
|
assert len(result) == 3
|
|
assert result[0].impact_score == 8 # AI News
|
|
assert result[1].impact_score == 6 # Sports
|
|
assert result[2].impact_score == 4 # Weather
|
|
|
|
def test_full_workflow_with_category_boost(self, full_recommender):
|
|
"""Test workflow with category boost preference."""
|
|
mock_pref_manager = Mock()
|
|
mock_pref_manager.get_preferences.return_value = {
|
|
"liked_categories": ["Sports"]
|
|
}
|
|
|
|
recommender = full_recommender(preference_manager=mock_pref_manager)
|
|
result = recommender.generate_recommendations("user123")
|
|
|
|
# Sports gets 1.2x boost: 6/10 * 1.2 = 0.72
|
|
# AI: 8/10 * 1.0 = 0.8
|
|
# Sports should still be second, but has boost
|
|
assert len(result) == 3
|
|
assert result[0].topic == "AI News" # Still highest
|
|
sports_card = [c for c in result if c.topic == "Sports Update"][0]
|
|
assert sports_card.metadata.get("preference_boost") == 1.2
|
|
|
|
def test_full_workflow_with_impact_filter(self, full_recommender):
|
|
"""Test workflow with impact threshold filter."""
|
|
mock_pref_manager = Mock()
|
|
mock_pref_manager.get_preferences.return_value = {"impact_threshold": 5}
|
|
|
|
recommender = full_recommender(preference_manager=mock_pref_manager)
|
|
result = recommender.generate_recommendations("user123")
|
|
|
|
# Weather (4) filtered out
|
|
assert len(result) == 2
|
|
topics = [c.topic for c in result]
|
|
assert "Weather Report" not in topics
|
|
|
|
def test_full_workflow_with_disliked_topics(self, full_recommender):
|
|
"""Test workflow with disliked topics filter."""
|
|
mock_pref_manager = Mock()
|
|
mock_pref_manager.get_preferences.return_value = {
|
|
"disliked_topics": ["sports"]
|
|
}
|
|
|
|
recommender = full_recommender(preference_manager=mock_pref_manager)
|
|
result = recommender.generate_recommendations("user123")
|
|
|
|
# Sports filtered out
|
|
assert len(result) == 2
|
|
topics = [c.topic for c in result]
|
|
assert "Sports Update" not in topics
|
|
|
|
def test_full_workflow_with_all_filters(self, full_recommender):
|
|
"""Test workflow with all filter types combined."""
|
|
mock_pref_manager = Mock()
|
|
mock_pref_manager.get_preferences.return_value = {
|
|
"liked_categories": ["Technology"],
|
|
"impact_threshold": 5,
|
|
"disliked_topics": ["weather"],
|
|
}
|
|
|
|
recommender = full_recommender(preference_manager=mock_pref_manager)
|
|
result = recommender.generate_recommendations("user123")
|
|
|
|
# Weather filtered by disliked topics
|
|
# Low impact items filtered by threshold
|
|
# Technology boosted
|
|
assert len(result) == 2
|
|
|
|
# AI News should be first (boosted and high impact)
|
|
assert result[0].topic == "AI News"
|
|
assert result[0].metadata.get("preference_boost") == 1.2
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for edge cases with NewsCard creation
|
|
# =============================================================================
|
|
|
|
|
|
class TestNewsCardEdgeCases:
|
|
"""Tests for edge cases when working with NewsCard objects."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_filter_with_none_category(self, concrete_recommender):
|
|
"""Filter handles cards with None category."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="No category",
|
|
source=source,
|
|
user_id="user1",
|
|
category=None,
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["Technology"]}
|
|
|
|
# Should not raise
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 1
|
|
assert "preference_boost" not in result[0].metadata
|
|
|
|
def test_sort_handles_cards_with_existing_metadata(
|
|
self, concrete_recommender
|
|
):
|
|
"""Sort handles cards that already have metadata."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="With metadata",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
card.metadata["existing_key"] = "existing_value"
|
|
card.metadata["preference_boost"] = 2.0
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards([card], "user123")
|
|
|
|
# Original metadata preserved
|
|
assert result[0].metadata["existing_key"] == "existing_value"
|
|
assert result[0].metadata["preference_boost"] == 2.0
|
|
|
|
|
|
# =============================================================================
|
|
# Parameterized tests for impact threshold
|
|
# =============================================================================
|
|
|
|
|
|
class TestImpactThresholdParameterized:
|
|
"""Parameterized tests for impact threshold filtering."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
@pytest.fixture
|
|
def cards_with_range_of_scores(self):
|
|
"""Create cards with impact scores from 1 to 10."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
return [
|
|
NewsCard(
|
|
topic=f"Score {i}",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=i,
|
|
)
|
|
for i in range(1, 11)
|
|
]
|
|
|
|
@pytest.mark.parametrize(
|
|
"threshold,expected_count",
|
|
[
|
|
(1, 10), # All cards
|
|
(5, 6), # Scores 5-10
|
|
(7, 4), # Scores 7-10
|
|
(10, 1), # Only score 10
|
|
(11, 0), # None pass
|
|
],
|
|
)
|
|
def test_threshold_filters_correctly(
|
|
self,
|
|
concrete_recommender,
|
|
cards_with_range_of_scores,
|
|
threshold,
|
|
expected_count,
|
|
):
|
|
"""Verify threshold filters out cards below it."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": threshold}
|
|
|
|
result = recommender.filter_cards(
|
|
cards_with_range_of_scores, preferences
|
|
)
|
|
|
|
assert len(result) == expected_count
|
|
for card in result:
|
|
assert card.impact_score >= threshold
|
|
|
|
|
|
# =============================================================================
|
|
# Parameterized tests for preference boost
|
|
# =============================================================================
|
|
|
|
|
|
class TestPreferenceBoostParameterized:
|
|
"""Parameterized tests for preference boost behavior."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
@pytest.mark.parametrize(
|
|
"impact,boost,expected_score",
|
|
[
|
|
(10, 1.0, 1.0), # 10/10 * 1.0 = 1.0
|
|
(10, 2.0, 2.0), # 10/10 * 2.0 = 2.0
|
|
(5, 1.0, 0.5), # 5/10 * 1.0 = 0.5
|
|
(5, 2.0, 1.0), # 5/10 * 2.0 = 1.0
|
|
(8, 1.5, 1.2), # 8/10 * 1.5 = 1.2
|
|
(0, 1.0, 0.0), # 0/10 * 1.0 = 0.0
|
|
(10, 0.0, 0.0), # 10/10 * 0.0 = 0.0
|
|
],
|
|
)
|
|
def test_score_calculation_formula(
|
|
self, concrete_recommender, impact, boost, expected_score
|
|
):
|
|
"""Verify score = (impact/10) * boost formula."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test", source=source, user_id="user1", impact_score=impact
|
|
)
|
|
card.metadata["preference_boost"] = boost
|
|
|
|
recommender = concrete_recommender()
|
|
|
|
# Sort with single card to verify it processes correctly
|
|
result = recommender.sort_cards([card], "user123")
|
|
|
|
# Verify by checking the card is returned (score calculation happened)
|
|
assert len(result) == 1
|
|
|
|
|
|
# =============================================================================
|
|
# Parameterized tests for disliked topics matching
|
|
# =============================================================================
|
|
|
|
|
|
class TestDislikedTopicsParameterized:
|
|
"""Parameterized tests for disliked topics matching."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
@pytest.mark.parametrize(
|
|
"card_topic,disliked_topics,should_be_filtered",
|
|
[
|
|
("AI in Healthcare", ["ai"], True), # Lowercase match
|
|
(
|
|
"AI in Healthcare",
|
|
["AI"],
|
|
False,
|
|
), # Case-sensitive (topic lowercased)
|
|
("Machine Learning News", ["machine"], True), # Partial match
|
|
("Deep Learning", ["learning"], True), # Word match
|
|
("Python Tips", ["java"], False), # No match
|
|
("JavaScript Guide", ["script"], True), # Substring match
|
|
("Data Science", ["data", "science"], True), # Multiple matches
|
|
("Weather Report", ["tech", "sports"], False), # No matches
|
|
],
|
|
)
|
|
def test_topic_matching_behavior(
|
|
self,
|
|
concrete_recommender,
|
|
card_topic,
|
|
disliked_topics,
|
|
should_be_filtered,
|
|
):
|
|
"""Verify topic matching with various patterns."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic=card_topic, source=source, user_id="user1", impact_score=5
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": disliked_topics}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
if should_be_filtered:
|
|
assert len(result) == 0, f"Expected {card_topic} to be filtered"
|
|
else:
|
|
assert len(result) == 1, f"Expected {card_topic} to NOT be filtered"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for sorting stability
|
|
# =============================================================================
|
|
|
|
|
|
class TestSortingStability:
|
|
"""Tests for sorting stability and correctness."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_sort_many_cards(self, concrete_recommender):
|
|
"""Sort handles many cards correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic=f"Card {i}",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=i % 10 + 1, # Scores 1-10 repeating
|
|
)
|
|
for i in range(100)
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards(cards, "user123")
|
|
|
|
assert len(result) == 100
|
|
|
|
# Verify descending order
|
|
for i in range(len(result) - 1):
|
|
assert result[i].impact_score >= result[i + 1].impact_score
|
|
|
|
def test_sort_with_identical_cards(self, concrete_recommender):
|
|
"""Sort handles many identical cards."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Same",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
for _ in range(10)
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards(cards, "user123")
|
|
|
|
assert len(result) == 10
|
|
|
|
def test_sort_with_mixed_boosts(self, concrete_recommender):
|
|
"""Sort correctly handles mixed boost values."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
|
|
# Create cards with various impact scores and boosts
|
|
card1 = NewsCard(
|
|
topic="Low Impact High Boost",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=2,
|
|
)
|
|
card1.metadata["preference_boost"] = 5.0 # Score: 2/10 * 5 = 1.0
|
|
|
|
card2 = NewsCard(
|
|
topic="High Impact No Boost",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=9,
|
|
) # Score: 9/10 * 1 = 0.9
|
|
|
|
card3 = NewsCard(
|
|
topic="Medium Both",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=6,
|
|
)
|
|
card3.metadata["preference_boost"] = 1.5 # Score: 6/10 * 1.5 = 0.9
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards([card2, card1, card3], "user123")
|
|
|
|
# card1 should be first (highest score: 1.0)
|
|
assert result[0].topic == "Low Impact High Boost"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for method chaining scenarios
|
|
# =============================================================================
|
|
|
|
|
|
class TestMethodChaining:
|
|
"""Tests for realistic method chaining scenarios."""
|
|
|
|
@pytest.fixture
|
|
def full_recommender(self):
|
|
"""Create recommender with all operations exposed."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class ChainableRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_and_sort(self, cards, preferences, user_id):
|
|
"""Chain filter and sort operations."""
|
|
filtered = self._filter_by_preferences(cards, preferences)
|
|
sorted_cards = self._sort_by_relevance(filtered, user_id)
|
|
return sorted_cards
|
|
|
|
return ChainableRecommender
|
|
|
|
def test_filter_then_sort(self, full_recommender):
|
|
"""Filter and sort chain works correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Sports News",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Sports",
|
|
impact_score=9,
|
|
),
|
|
NewsCard(
|
|
topic="Tech News",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Technology",
|
|
impact_score=7,
|
|
),
|
|
NewsCard(
|
|
topic="Low Impact Tech",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Technology",
|
|
impact_score=3,
|
|
),
|
|
]
|
|
|
|
recommender = full_recommender()
|
|
preferences = {
|
|
"liked_categories": ["Technology"],
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
result = recommender.filter_and_sort(cards, preferences, "user123")
|
|
|
|
# Low Impact Tech filtered out (score 3 < 5)
|
|
assert len(result) == 2
|
|
|
|
# Tech News should be first (boosted and above threshold)
|
|
# Score: 7/10 * 1.2 = 0.84
|
|
# Sports: 9/10 * 1.0 = 0.9
|
|
# But wait - Sports has higher base score without boost
|
|
# Let's verify the order
|
|
topics = [c.topic for c in result]
|
|
assert "Low Impact Tech" not in topics
|
|
|
|
def test_empty_after_all_filtered(self, full_recommender):
|
|
"""Empty result after all cards filtered is sorted correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Low Score",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=1,
|
|
),
|
|
]
|
|
|
|
recommender = full_recommender()
|
|
preferences = {"impact_threshold": 10}
|
|
|
|
result = recommender.filter_and_sort(cards, preferences, "user123")
|
|
|
|
assert result == []
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for dependency injection patterns
|
|
# =============================================================================
|
|
|
|
|
|
class TestDependencyInjection:
|
|
"""Tests for various dependency injection scenarios."""
|
|
|
|
def test_all_dependencies_none(self):
|
|
"""Recommender works with all dependencies as None."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class MinimalRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
# Exercise all methods to verify they don't crash
|
|
self._get_user_preferences(user_id)
|
|
self._get_user_ratings(user_id)
|
|
self._execute_search("test")
|
|
return []
|
|
|
|
recommender = MinimalRecommender()
|
|
|
|
# All should work without crashing
|
|
result = recommender.generate_recommendations("user123")
|
|
|
|
assert result == []
|
|
|
|
def test_partial_dependencies(self):
|
|
"""Recommender works with some dependencies set."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
mock_pref_manager = Mock()
|
|
mock_pref_manager.get_preferences.return_value = {"key": "value"}
|
|
|
|
class PartialRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
prefs = self._get_user_preferences(user_id)
|
|
ratings = self._get_user_ratings(user_id) # No rating system
|
|
return [prefs, ratings]
|
|
|
|
recommender = PartialRecommender(preference_manager=mock_pref_manager)
|
|
result = recommender.generate_recommendations("user123")
|
|
|
|
assert result[0] == {"key": "value"} # From preference manager
|
|
assert result[1] == [] # Empty from missing rating system
|
|
|
|
def test_dependencies_can_be_mocked(self):
|
|
"""All dependencies can be replaced with mocks."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
mock_pref = Mock()
|
|
mock_pref.get_preferences.return_value = {"pref": True}
|
|
|
|
mock_rating = Mock()
|
|
mock_rating.get_recent_ratings.return_value = [{"rating": 5}]
|
|
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = {"results": ["item"]}
|
|
|
|
mock_registry = Mock()
|
|
mock_registry.get_topics.return_value = ["topic1"]
|
|
|
|
class FullMockedRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
recommender = FullMockedRecommender(
|
|
preference_manager=mock_pref,
|
|
rating_system=mock_rating,
|
|
search_system=mock_search,
|
|
topic_registry=mock_registry,
|
|
)
|
|
|
|
info = recommender.get_strategy_info()
|
|
|
|
assert info["has_preference_manager"] is True
|
|
assert info["has_rating_system"] is True
|
|
assert info["has_search_system"] is True
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for card attribute edge cases
|
|
# =============================================================================
|
|
|
|
|
|
class TestCardAttributeEdgeCases:
|
|
"""Tests for handling cards with unusual attribute values."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_filter_with_empty_topic(self, concrete_recommender):
|
|
"""Filter handles cards with empty topic string."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["test"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
# Empty topic should not match any disliked topic
|
|
assert len(result) == 1
|
|
|
|
def test_filter_with_whitespace_topic(self, concrete_recommender):
|
|
"""Filter handles cards with whitespace-only topic."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic=" ",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["test"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_sort_with_maximum_impact_score(self, concrete_recommender):
|
|
"""Sort handles maximum impact score correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Max Score",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=100, # Very high score
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards([card], "user123")
|
|
|
|
assert len(result) == 1
|
|
assert result[0].impact_score == 100
|
|
|
|
def test_filter_with_special_chars_in_category(self, concrete_recommender):
|
|
"""Filter handles categories with special characters."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Tech & Science",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["Tech & Science"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 1
|
|
assert result[0].metadata.get("preference_boost") == 1.2
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for preference edge cases
|
|
# =============================================================================
|
|
|
|
|
|
class TestPreferenceEdgeCases:
|
|
"""Tests for unusual preference configurations."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
@pytest.fixture
|
|
def sample_cards(self):
|
|
"""Create sample cards for testing."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
return [
|
|
NewsCard(
|
|
topic="Test Card",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Technology",
|
|
impact_score=5,
|
|
),
|
|
]
|
|
|
|
def test_empty_string_in_disliked_topics(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Empty string in disliked_topics matches everything."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": [""]}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
# Empty string is in every topic.lower()
|
|
assert len(result) == 0
|
|
|
|
def test_very_long_disliked_topic(self, concrete_recommender, sample_cards):
|
|
"""Very long disliked topic that won't match anything."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["a" * 1000]}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
assert len(result) == 1 # No match
|
|
|
|
def test_disliked_topic_with_special_regex_chars(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Disliked topics with regex special chars work as literals."""
|
|
recommender = concrete_recommender()
|
|
# These are regex special chars but should be treated as literals
|
|
preferences = {"disliked_topics": ["[test]", ".*", "^$"]}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
# None of these regex patterns should match literal topic text
|
|
assert len(result) == 1
|
|
|
|
def test_negative_impact_threshold(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Negative impact threshold allows all cards."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": -10}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
assert len(result) == 1 # All cards pass
|
|
|
|
def test_very_high_impact_threshold(
|
|
self, concrete_recommender, sample_cards
|
|
):
|
|
"""Very high impact threshold filters all cards."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": 1000}
|
|
|
|
result = recommender.filter_cards(sample_cards, preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
def test_multiple_matching_categories(self, concrete_recommender):
|
|
"""Card matching multiple liked categories gets boost once."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Technology",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
# Category appears multiple times in list
|
|
preferences = {
|
|
"liked_categories": ["Technology", "Technology", "Science"]
|
|
}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
# Should only get 1.2 boost, not compounded
|
|
assert result[0].metadata.get("preference_boost") == 1.2
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for rating system integration
|
|
# =============================================================================
|
|
|
|
|
|
class TestRatingSystemIntegration:
|
|
"""Tests for rating system integration."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def get_ratings(self, user_id, limit=50):
|
|
return self._get_user_ratings(user_id, limit)
|
|
|
|
return TestRecommender
|
|
|
|
def test_rating_system_called_with_correct_args(self, concrete_recommender):
|
|
"""Rating system is called with correct user_id and limit."""
|
|
mock_rating_system = Mock()
|
|
mock_rating_system.get_recent_ratings.return_value = []
|
|
|
|
recommender = concrete_recommender(rating_system=mock_rating_system)
|
|
recommender.get_ratings("user456", limit=25)
|
|
|
|
mock_rating_system.get_recent_ratings.assert_called_once_with(
|
|
"user456", 25
|
|
)
|
|
|
|
def test_rating_system_returns_complex_data(self, concrete_recommender):
|
|
"""Rating system can return complex rating data."""
|
|
mock_rating_system = Mock()
|
|
mock_rating_system.get_recent_ratings.return_value = [
|
|
{"card_id": "c1", "rating": 5, "timestamp": "2024-01-15T12:00:00"},
|
|
{"card_id": "c2", "rating": 3, "feedback": "good article"},
|
|
{
|
|
"card_id": "c3",
|
|
"rating": 1,
|
|
"disliked": True,
|
|
"topics": ["spam"],
|
|
},
|
|
]
|
|
|
|
recommender = concrete_recommender(rating_system=mock_rating_system)
|
|
result = recommender.get_ratings("user123")
|
|
|
|
assert len(result) == 3
|
|
assert result[0]["rating"] == 5
|
|
assert result[2]["disliked"] is True
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for search system edge cases
|
|
# =============================================================================
|
|
|
|
|
|
class TestSearchSystemEdgeCases:
|
|
"""Tests for search system edge cases."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def search(self, query, strategy=None):
|
|
return self._execute_search(query, strategy)
|
|
|
|
return TestRecommender
|
|
|
|
def test_search_with_empty_query(self, concrete_recommender):
|
|
"""Search handles empty query string."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = {"results": []}
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
result = recommender.search("")
|
|
|
|
mock_search.analyze_topic.assert_called_once_with("")
|
|
assert result == {"results": []}
|
|
|
|
def test_search_with_unicode_query(self, concrete_recommender):
|
|
"""Search handles unicode query string."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = {"results": ["日本語"]}
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
result = recommender.search("日本語ニュース")
|
|
|
|
mock_search.analyze_topic.assert_called_once_with("日本語ニュース")
|
|
assert "日本語" in result["results"]
|
|
|
|
def test_search_returns_large_result(self, concrete_recommender):
|
|
"""Search handles large result set."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = {
|
|
"results": [f"item_{i}" for i in range(1000)]
|
|
}
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
result = recommender.search("test")
|
|
|
|
assert len(result["results"]) == 1000
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for complex filter combinations
|
|
# =============================================================================
|
|
|
|
|
|
class TestComplexFilterCombinations:
|
|
"""Tests for complex combinations of filters."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
@pytest.fixture
|
|
def diverse_cards(self):
|
|
"""Create diverse set of cards for testing."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
return [
|
|
NewsCard(
|
|
topic="Breaking: AI Revolution",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Technology",
|
|
impact_score=9,
|
|
),
|
|
NewsCard(
|
|
topic="Sports Update: Football Finals",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Sports",
|
|
impact_score=7,
|
|
),
|
|
NewsCard(
|
|
topic="Weather: Storm Warning",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Weather",
|
|
impact_score=4,
|
|
),
|
|
NewsCard(
|
|
topic="Finance: Stock Market Crash",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Finance",
|
|
impact_score=8,
|
|
),
|
|
NewsCard(
|
|
topic="Tech News: Python Update",
|
|
source=source,
|
|
user_id="user1",
|
|
category="Technology",
|
|
impact_score=5,
|
|
),
|
|
]
|
|
|
|
def test_filter_boost_then_threshold(
|
|
self, concrete_recommender, diverse_cards
|
|
):
|
|
"""Boost is applied before threshold filtering."""
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": ["Technology"],
|
|
"impact_threshold": 6,
|
|
}
|
|
|
|
result = recommender.filter_cards(diverse_cards, preferences)
|
|
|
|
# Weather (4) and Python Update (5) filtered by threshold
|
|
assert len(result) == 3
|
|
# AI Revolution should have boost
|
|
ai_card = [c for c in result if "AI" in c.topic][0]
|
|
assert ai_card.metadata.get("preference_boost") == 1.2
|
|
|
|
def test_filter_disliked_then_threshold(
|
|
self, concrete_recommender, diverse_cards
|
|
):
|
|
"""Disliked topics and threshold both apply."""
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"disliked_topics": ["stock", "crash"],
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
result = recommender.filter_cards(diverse_cards, preferences)
|
|
|
|
# Finance filtered by disliked, Weather filtered by threshold
|
|
assert len(result) == 3
|
|
topics = [c.topic for c in result]
|
|
assert not any("Stock" in t or "Crash" in t for t in topics)
|
|
|
|
def test_all_filters_combined(self, concrete_recommender, diverse_cards):
|
|
"""All filter types applied together."""
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": ["Technology"],
|
|
"disliked_topics": ["weather", "storm"],
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
result = recommender.filter_cards(diverse_cards, preferences)
|
|
|
|
# Weather filtered (disliked + threshold)
|
|
# Remaining: AI, Sports, Finance, Python
|
|
# But Python also filtered by threshold (5 >= 5, so it passes)
|
|
assert len(result) == 4
|
|
|
|
# Verify AI and Python have boosts
|
|
tech_cards = [c for c in result if c.category == "Technology"]
|
|
for card in tech_cards:
|
|
assert card.metadata.get("preference_boost") == 1.2
|
|
|
|
def test_filter_then_sort_integration(
|
|
self, concrete_recommender, diverse_cards
|
|
):
|
|
"""Filter and sort work together correctly."""
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": ["Finance"],
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
filtered = recommender.filter_cards(diverse_cards, preferences)
|
|
sorted_cards = recommender.sort_cards(filtered, "user123")
|
|
|
|
# Verify sorted by score (impact/10 * boost)
|
|
# AI: 9/10 * 1.0 = 0.9
|
|
# Sports: 7/10 * 1.0 = 0.7
|
|
# Finance: 8/10 * 1.2 = 0.96 (boosted)
|
|
# Python: 5/10 * 1.0 = 0.5
|
|
assert (
|
|
sorted_cards[0].category == "Finance"
|
|
) # Highest score due to boost
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for strategy name inheritance
|
|
# =============================================================================
|
|
|
|
|
|
class TestStrategyNameInheritance:
|
|
"""Tests for strategy name behavior across inheritance."""
|
|
|
|
def test_strategy_name_from_subclass(self):
|
|
"""Strategy name is derived from the actual subclass."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class MySpecialRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
recommender = MySpecialRecommender()
|
|
assert recommender.strategy_name == "MySpecialRecommender"
|
|
|
|
def test_strategy_name_in_nested_inheritance(self):
|
|
"""Strategy name works with nested inheritance."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class IntermediateRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
class FinalRecommender(IntermediateRecommender):
|
|
pass
|
|
|
|
recommender = FinalRecommender()
|
|
assert recommender.strategy_name == "FinalRecommender"
|
|
|
|
def test_strategy_info_includes_correct_name(self):
|
|
"""get_strategy_info returns the correct strategy name."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class AnalyticsRecommender(BaseRecommender):
|
|
"""Recommender using analytics data."""
|
|
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
recommender = AnalyticsRecommender()
|
|
info = recommender.get_strategy_info()
|
|
|
|
assert info["name"] == "AnalyticsRecommender"
|
|
assert "analytics" in info["description"].lower()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for context parameter handling
|
|
# =============================================================================
|
|
|
|
|
|
class TestContextParameterHandling:
|
|
"""Tests for context parameter in generate_recommendations."""
|
|
|
|
def test_context_is_passed_to_implementation(self):
|
|
"""Context parameter is available in implementation."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class ContextAwareRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
# Return context for testing
|
|
return [{"received_context": context}]
|
|
|
|
recommender = ContextAwareRecommender()
|
|
context = {"page": "home", "device": "mobile", "time": "morning"}
|
|
|
|
result = recommender.generate_recommendations(
|
|
"user123", context=context
|
|
)
|
|
|
|
assert result[0]["received_context"] == context
|
|
|
|
def test_context_defaults_to_none(self):
|
|
"""Context defaults to None when not provided."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class ContextCheckRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return [{"context_was_none": context is None}]
|
|
|
|
recommender = ContextCheckRecommender()
|
|
result = recommender.generate_recommendations("user123")
|
|
|
|
assert result[0]["context_was_none"] is True
|
|
|
|
def test_context_with_complex_data(self):
|
|
"""Context can contain complex nested data."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class NestedContextRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return [{"context": context}]
|
|
|
|
recommender = NestedContextRecommender()
|
|
context = {
|
|
"user_history": [{"page": "news", "duration": 120}],
|
|
"preferences": {"theme": "dark"},
|
|
"session": {"id": "abc123", "start": "2024-01-15T12:00:00"},
|
|
}
|
|
|
|
result = recommender.generate_recommendations(
|
|
"user123", context=context
|
|
)
|
|
|
|
assert result[0]["context"]["user_history"][0]["duration"] == 120
|
|
assert result[0]["context"]["preferences"]["theme"] == "dark"
|
|
|
|
|
|
# =============================================================================
|
|
# Stress tests for recommender
|
|
# =============================================================================
|
|
|
|
|
|
class TestRecommenderStress:
|
|
"""Stress tests for recommender with large data sets."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_filter_1000_cards(self, concrete_recommender):
|
|
"""Filter handles 1000 cards efficiently."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic=f"News Item {i}",
|
|
source=source,
|
|
user_id="user1",
|
|
category=f"Category{i % 10}",
|
|
impact_score=i % 10 + 1,
|
|
)
|
|
for i in range(1000)
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": 5}
|
|
|
|
result = recommender.filter_cards(cards, preferences)
|
|
|
|
# Cards with impact 5-10 should pass (600 cards)
|
|
assert len(result) == 600
|
|
|
|
def test_sort_1000_cards(self, concrete_recommender):
|
|
"""Sort handles 1000 cards efficiently."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic=f"News Item {i}",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=i % 100,
|
|
)
|
|
for i in range(1000)
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards(cards, "user123")
|
|
|
|
assert len(result) == 1000
|
|
# Verify descending order
|
|
for i in range(len(result) - 1):
|
|
assert result[i].impact_score >= result[i + 1].impact_score
|
|
|
|
def test_filter_and_sort_large_dataset(self, concrete_recommender):
|
|
"""Combined filter and sort on large dataset."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
categories = ["Tech", "Sports", "Finance", "Health", "Entertainment"]
|
|
cards = [
|
|
NewsCard(
|
|
topic=f"Article about {categories[i % 5]} topic {i}",
|
|
source=source,
|
|
user_id="user1",
|
|
category=categories[i % 5],
|
|
impact_score=(i % 10) + 1,
|
|
)
|
|
for i in range(500)
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": ["Tech", "Finance"],
|
|
"impact_threshold": 3,
|
|
}
|
|
|
|
filtered = recommender.filter_cards(cards, preferences)
|
|
sorted_cards = recommender.sort_cards(filtered, "user123")
|
|
|
|
# Verify filtering worked
|
|
for card in sorted_cards:
|
|
assert card.impact_score >= 3
|
|
|
|
# Verify sorting worked
|
|
for i in range(len(sorted_cards) - 1):
|
|
score_i = (
|
|
sorted_cards[i].impact_score
|
|
/ 10.0
|
|
* sorted_cards[i].metadata.get("preference_boost", 1.0)
|
|
)
|
|
score_next = (
|
|
sorted_cards[i + 1].impact_score
|
|
/ 10.0
|
|
* sorted_cards[i + 1].metadata.get("preference_boost", 1.0)
|
|
)
|
|
assert score_i >= score_next
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for abstract method enforcement in recommender
|
|
# =============================================================================
|
|
|
|
|
|
class TestRecommenderAbstractEnforcement:
|
|
"""Tests for abstract method enforcement in BaseRecommender."""
|
|
|
|
def test_cannot_instantiate_base_class(self):
|
|
"""BaseRecommender cannot be instantiated directly."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
with pytest.raises(TypeError) as exc_info:
|
|
BaseRecommender()
|
|
|
|
assert "abstract" in str(exc_info.value).lower()
|
|
|
|
def test_must_implement_generate_recommendations(self):
|
|
"""Subclass must implement generate_recommendations."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class IncompleteRecommender(BaseRecommender):
|
|
pass
|
|
|
|
with pytest.raises(TypeError):
|
|
IncompleteRecommender()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for data validation in recommender
|
|
# =============================================================================
|
|
|
|
|
|
class TestRecommenderDataValidation:
|
|
"""Tests for data validation in recommender operations."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_filter_returns_list(self, concrete_recommender):
|
|
"""Filter always returns a list."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
|
|
result = recommender.filter_cards(cards, {})
|
|
assert isinstance(result, list)
|
|
|
|
result = recommender.filter_cards([], {})
|
|
assert isinstance(result, list)
|
|
|
|
def test_sort_returns_list(self, concrete_recommender):
|
|
"""Sort always returns a list."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
|
|
result = recommender.sort_cards(cards, "user123")
|
|
assert isinstance(result, list)
|
|
|
|
result = recommender.sort_cards([], "user123")
|
|
assert isinstance(result, list)
|
|
|
|
def test_get_strategy_info_returns_dict(self, concrete_recommender):
|
|
"""get_strategy_info always returns a dict."""
|
|
recommender = concrete_recommender()
|
|
|
|
result = recommender.get_strategy_info()
|
|
|
|
assert isinstance(result, dict)
|
|
|
|
def test_get_strategy_info_has_required_keys(self, concrete_recommender):
|
|
"""get_strategy_info contains all required keys."""
|
|
recommender = concrete_recommender()
|
|
|
|
result = recommender.get_strategy_info()
|
|
|
|
required_keys = [
|
|
"name",
|
|
"has_preference_manager",
|
|
"has_rating_system",
|
|
"has_search_system",
|
|
"description",
|
|
]
|
|
|
|
for key in required_keys:
|
|
assert key in result, f"Missing required key: {key}"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for multiple users
|
|
# =============================================================================
|
|
|
|
|
|
class TestMultipleUsers:
|
|
"""Tests for handling multiple users."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
prefs = self._get_user_preferences(user_id)
|
|
return [{"user_id": user_id, "prefs": prefs}]
|
|
|
|
return TestRecommender
|
|
|
|
def test_different_users_get_different_preferences(
|
|
self, concrete_recommender
|
|
):
|
|
"""Different users receive their own preferences."""
|
|
mock_pref_manager = Mock()
|
|
mock_pref_manager.get_preferences.side_effect = lambda uid: {
|
|
"user_id": uid,
|
|
"setting": f"value_for_{uid}",
|
|
}
|
|
|
|
recommender = concrete_recommender(preference_manager=mock_pref_manager)
|
|
|
|
result1 = recommender.generate_recommendations("user1")
|
|
result2 = recommender.generate_recommendations("user2")
|
|
|
|
assert result1[0]["prefs"]["user_id"] == "user1"
|
|
assert result2[0]["prefs"]["user_id"] == "user2"
|
|
assert result1[0]["prefs"]["setting"] == "value_for_user1"
|
|
assert result2[0]["prefs"]["setting"] == "value_for_user2"
|
|
|
|
def test_user_id_passed_to_sort(self, concrete_recommender):
|
|
"""User ID is passed to sort method."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TrackingRecommender(BaseRecommender):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.sort_user_ids = []
|
|
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
self.sort_user_ids.append(user_id)
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Test", source=source, user_id="user1", impact_score=5
|
|
)
|
|
]
|
|
|
|
recommender = TrackingRecommender()
|
|
recommender.sort_cards(cards, "user_abc")
|
|
recommender.sort_cards(cards, "user_xyz")
|
|
|
|
assert "user_abc" in recommender.sort_user_ids
|
|
assert "user_xyz" in recommender.sort_user_ids
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for callback error handling
|
|
# =============================================================================
|
|
|
|
|
|
class TestCallbackErrorHandling:
|
|
"""Tests for handling errors in callbacks."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
self._update_progress("Step 1", 50)
|
|
return []
|
|
|
|
return TestRecommender
|
|
|
|
def test_callback_exception_propagates(self, concrete_recommender):
|
|
"""Exception in callback propagates to caller."""
|
|
recommender = concrete_recommender()
|
|
|
|
def failing_callback(msg, pct, meta):
|
|
raise ValueError("Callback failed!")
|
|
|
|
recommender.set_progress_callback(failing_callback)
|
|
|
|
with pytest.raises(ValueError) as exc_info:
|
|
recommender.generate_recommendations("user123")
|
|
|
|
assert "Callback failed!" in str(exc_info.value)
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for category boost behavior
|
|
# =============================================================================
|
|
|
|
|
|
class TestCategoryBoostBehavior:
|
|
"""Tests for category boost application in filtering."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class with filter method."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
def test_boost_is_exactly_1_2(self, concrete_recommender):
|
|
"""Preference boost for liked categories is exactly 1.2."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="tech",
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["tech"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert result[0].metadata.get("preference_boost") == 1.2
|
|
|
|
def test_no_boost_for_non_liked_category(self, concrete_recommender):
|
|
"""Cards not in liked categories get no boost."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="sports",
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["tech"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert result[0].metadata.get("preference_boost") is None
|
|
|
|
def test_multiple_liked_categories(self, concrete_recommender):
|
|
"""Multiple liked categories all receive boost."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Tech news",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="tech",
|
|
),
|
|
NewsCard(
|
|
topic="Science news",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="science",
|
|
),
|
|
NewsCard(
|
|
topic="Sports news",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="sports",
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["tech", "science"]}
|
|
|
|
result = recommender.filter_cards(cards, preferences)
|
|
|
|
assert result[0].metadata.get("preference_boost") == 1.2 # tech
|
|
assert result[1].metadata.get("preference_boost") == 1.2 # science
|
|
assert result[2].metadata.get("preference_boost") is None # sports
|
|
|
|
def test_empty_liked_categories_list(self, concrete_recommender):
|
|
"""Empty liked_categories list doesn't apply boost."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="tech",
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": []}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
# Empty list is falsy, so no boost should be applied
|
|
assert result[0].metadata.get("preference_boost") is None
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for impact threshold filtering
|
|
# =============================================================================
|
|
|
|
|
|
class TestImpactThresholdFiltering:
|
|
"""Tests for impact threshold filtering behavior."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class with filter method."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
def test_filters_below_threshold(self, concrete_recommender):
|
|
"""Cards below threshold are filtered out."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Low impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=3,
|
|
),
|
|
NewsCard(
|
|
topic="High impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=7,
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": 5}
|
|
|
|
result = recommender.filter_cards(cards, preferences)
|
|
|
|
assert len(result) == 1
|
|
assert result[0].topic == "High impact"
|
|
|
|
def test_threshold_is_inclusive(self, concrete_recommender):
|
|
"""Cards exactly at threshold are kept."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="At threshold",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": 5}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_threshold_zero_keeps_all(self, concrete_recommender):
|
|
"""Threshold of zero keeps all cards."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Zero impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=0,
|
|
),
|
|
NewsCard(
|
|
topic="Some impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=3,
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": 0}
|
|
|
|
result = recommender.filter_cards(cards, preferences)
|
|
|
|
assert len(result) == 2
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for disliked topics filtering
|
|
# =============================================================================
|
|
|
|
|
|
class TestDislikedTopicsFiltering:
|
|
"""Tests for disliked topics filtering behavior."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class with filter method."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
def test_exact_match_filtered(self, concrete_recommender):
|
|
"""Exact topic match is filtered."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="politics",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["politics"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
def test_substring_match_filtered(self, concrete_recommender):
|
|
"""Substring match is filtered."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="US politics today",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["politics"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
def test_case_insensitive_matching(self, concrete_recommender):
|
|
"""Topic matching is case insensitive."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="POLITICS NEWS",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["politics"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
def test_multiple_disliked_topics(self, concrete_recommender):
|
|
"""Multiple disliked topics are all filtered."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Politics update",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
),
|
|
NewsCard(
|
|
topic="Celebrity gossip",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
),
|
|
NewsCard(
|
|
topic="Tech news",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["politics", "celebrity"]}
|
|
|
|
result = recommender.filter_cards(cards, preferences)
|
|
|
|
assert len(result) == 1
|
|
assert result[0].topic == "Tech news"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for sort score calculation
|
|
# =============================================================================
|
|
|
|
|
|
class TestSortScoreCalculation:
|
|
"""Tests for the score calculation in sorting."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class with sort method."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_higher_impact_ranks_first(self, concrete_recommender):
|
|
"""Cards with higher impact score rank first."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Low impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=3,
|
|
),
|
|
NewsCard(
|
|
topic="High impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=9,
|
|
),
|
|
NewsCard(
|
|
topic="Medium impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=6,
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards(cards, "user123")
|
|
|
|
assert result[0].topic == "High impact"
|
|
assert result[1].topic == "Medium impact"
|
|
assert result[2].topic == "Low impact"
|
|
|
|
def test_boost_affects_ranking(self, concrete_recommender):
|
|
"""Preference boost affects final ranking."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
|
|
# Card with lower impact but boost
|
|
card1 = NewsCard(
|
|
topic="Boosted card",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
card1.metadata["preference_boost"] = 2.0 # Strong boost
|
|
|
|
# Card with higher impact but no boost
|
|
card2 = NewsCard(
|
|
topic="Normal card",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=8,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards([card1, card2], "user123")
|
|
|
|
# 5/10 * 2.0 = 1.0 vs 8/10 * 1.0 = 0.8
|
|
assert result[0].topic == "Boosted card"
|
|
assert result[1].topic == "Normal card"
|
|
|
|
def test_score_formula(self, concrete_recommender):
|
|
"""Score formula is (impact_score / 10) * boost."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
|
|
# Create cards with known scores
|
|
# Card A: 10/10 * 1.0 = 1.0
|
|
card_a = NewsCard(
|
|
topic="A",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=10,
|
|
)
|
|
|
|
# Card B: 5/10 * 1.5 = 0.75
|
|
card_b = NewsCard(
|
|
topic="B",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
card_b.metadata["preference_boost"] = 1.5
|
|
|
|
# Card C: 6/10 * 1.2 = 0.72
|
|
card_c = NewsCard(
|
|
topic="C",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=6,
|
|
)
|
|
card_c.metadata["preference_boost"] = 1.2
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards([card_b, card_c, card_a], "user123")
|
|
|
|
# Expected order: A (1.0), B (0.75), C (0.72)
|
|
assert result[0].topic == "A"
|
|
assert result[1].topic == "B"
|
|
assert result[2].topic == "C"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for search system integration
|
|
# =============================================================================
|
|
|
|
|
|
class TestSearchSystemIntegration:
|
|
"""Tests for search system integration."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def execute_search(self, query, strategy=None):
|
|
return self._execute_search(query, strategy)
|
|
|
|
return TestRecommender
|
|
|
|
def test_uses_news_aggregation_by_default(self, concrete_recommender):
|
|
"""Search uses news_aggregation strategy by default."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = {"results": []}
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
recommender.execute_search("test query")
|
|
|
|
mock_search.analyze_topic.assert_called_once_with("test query")
|
|
|
|
def test_handles_search_exception(self, concrete_recommender):
|
|
"""Search exception returns error dict."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.side_effect = Exception("Search failed")
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
result = recommender.execute_search("test query")
|
|
|
|
assert "error" in result
|
|
|
|
def test_no_search_system_returns_error(self, concrete_recommender):
|
|
"""No search system returns error dict."""
|
|
recommender = concrete_recommender(search_system=None)
|
|
result = recommender.execute_search("test query")
|
|
|
|
assert "error" in result
|
|
assert "not configured" in result["error"]
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for preference manager integration
|
|
# =============================================================================
|
|
|
|
|
|
class TestPreferenceManagerIntegration:
|
|
"""Tests for preference manager integration."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def get_prefs(self, user_id):
|
|
return self._get_user_preferences(user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_calls_preference_manager_with_user_id(self, concrete_recommender):
|
|
"""Preference manager is called with correct user ID."""
|
|
mock_pref = Mock()
|
|
mock_pref.get_preferences.return_value = {"setting": "value"}
|
|
|
|
recommender = concrete_recommender(preference_manager=mock_pref)
|
|
recommender.get_prefs("user_abc")
|
|
|
|
mock_pref.get_preferences.assert_called_once_with("user_abc")
|
|
|
|
def test_returns_preferences_from_manager(self, concrete_recommender):
|
|
"""Returns preferences from manager."""
|
|
mock_pref = Mock()
|
|
mock_pref.get_preferences.return_value = {
|
|
"liked_categories": ["tech"],
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
recommender = concrete_recommender(preference_manager=mock_pref)
|
|
result = recommender.get_prefs("user123")
|
|
|
|
assert result["liked_categories"] == ["tech"]
|
|
assert result["impact_threshold"] == 5
|
|
|
|
def test_no_preference_manager_returns_empty_dict(
|
|
self, concrete_recommender
|
|
):
|
|
"""No preference manager returns empty dict."""
|
|
recommender = concrete_recommender(preference_manager=None)
|
|
result = recommender.get_prefs("user123")
|
|
|
|
assert result == {}
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for filter and sort pipeline
|
|
# =============================================================================
|
|
|
|
|
|
class TestFilterSortPipeline:
|
|
"""Tests for the complete filter and sort pipeline."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def process_cards(self, cards, user_id, preferences):
|
|
filtered = self._filter_by_preferences(cards, preferences)
|
|
sorted_cards = self._sort_by_relevance(filtered, user_id)
|
|
return sorted_cards
|
|
|
|
return TestRecommender
|
|
|
|
def test_complete_pipeline(self, concrete_recommender):
|
|
"""Complete filter and sort pipeline works correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Politics low",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=8,
|
|
),
|
|
NewsCard(
|
|
topic="Tech high",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=7,
|
|
category="tech",
|
|
),
|
|
NewsCard(
|
|
topic="Science medium",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=6,
|
|
),
|
|
NewsCard(
|
|
topic="Low impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=2,
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": ["tech"],
|
|
"disliked_topics": ["politics"],
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
result = recommender.process_cards(cards, "user123", preferences)
|
|
|
|
# Should filter out: "Politics low" (disliked), "Low impact" (below threshold)
|
|
# Should boost: "Tech high" (liked category)
|
|
assert len(result) == 2
|
|
|
|
# Tech high with boost: 7/10 * 1.2 = 0.84
|
|
# Science medium: 6/10 * 1.0 = 0.6
|
|
assert result[0].topic == "Tech high"
|
|
assert result[1].topic == "Science medium"
|
|
|
|
def test_pipeline_with_empty_input(self, concrete_recommender):
|
|
"""Pipeline handles empty input."""
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["tech"]}
|
|
|
|
result = recommender.process_cards([], "user123", preferences)
|
|
|
|
assert result == []
|
|
|
|
def test_pipeline_with_empty_preferences(self, concrete_recommender):
|
|
"""Pipeline handles empty preferences."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Card 1",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
),
|
|
NewsCard(
|
|
topic="Card 2",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=3,
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.process_cards(cards, "user123", {})
|
|
|
|
# No filtering, just sorting by impact
|
|
assert len(result) == 2
|
|
assert result[0].topic == "Card 1"
|
|
assert result[1].topic == "Card 2"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for rating system limit parameter
|
|
# =============================================================================
|
|
|
|
|
|
class TestRatingSystemLimitParameter:
|
|
"""Tests for the limit parameter in _get_user_ratings."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def get_ratings(self, user_id, limit=50):
|
|
return self._get_user_ratings(user_id, limit)
|
|
|
|
return TestRecommender
|
|
|
|
def test_default_limit_is_50(self, concrete_recommender):
|
|
"""Default limit is 50."""
|
|
mock_rating = Mock()
|
|
mock_rating.get_recent_ratings.return_value = []
|
|
|
|
recommender = concrete_recommender(rating_system=mock_rating)
|
|
recommender.get_ratings("user123")
|
|
|
|
mock_rating.get_recent_ratings.assert_called_once_with("user123", 50)
|
|
|
|
def test_custom_limit_is_passed(self, concrete_recommender):
|
|
"""Custom limit is passed to rating system."""
|
|
mock_rating = Mock()
|
|
mock_rating.get_recent_ratings.return_value = []
|
|
|
|
recommender = concrete_recommender(rating_system=mock_rating)
|
|
recommender.get_ratings("user123", limit=100)
|
|
|
|
mock_rating.get_recent_ratings.assert_called_once_with("user123", 100)
|
|
|
|
def test_small_limit(self, concrete_recommender):
|
|
"""Small limit of 1 works correctly."""
|
|
mock_rating = Mock()
|
|
mock_rating.get_recent_ratings.return_value = [{"id": 1, "rating": 5}]
|
|
|
|
recommender = concrete_recommender(rating_system=mock_rating)
|
|
result = recommender.get_ratings("user123", limit=1)
|
|
|
|
assert len(result) == 1
|
|
mock_rating.get_recent_ratings.assert_called_once_with("user123", 1)
|
|
|
|
def test_zero_limit(self, concrete_recommender):
|
|
"""Zero limit works correctly."""
|
|
mock_rating = Mock()
|
|
mock_rating.get_recent_ratings.return_value = []
|
|
|
|
recommender = concrete_recommender(rating_system=mock_rating)
|
|
result = recommender.get_ratings("user123", limit=0)
|
|
|
|
assert result == []
|
|
mock_rating.get_recent_ratings.assert_called_once_with("user123", 0)
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for topic registry dependency
|
|
# =============================================================================
|
|
|
|
|
|
class TestTopicRegistryDependency:
|
|
"""Tests for topic registry dependency handling."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
return TestRecommender
|
|
|
|
def test_topic_registry_stored(self, concrete_recommender):
|
|
"""Topic registry is stored correctly."""
|
|
mock_registry = Mock()
|
|
|
|
recommender = concrete_recommender(topic_registry=mock_registry)
|
|
|
|
assert recommender.topic_registry is mock_registry
|
|
|
|
def test_topic_registry_none_by_default(self, concrete_recommender):
|
|
"""Topic registry is None by default."""
|
|
recommender = concrete_recommender()
|
|
|
|
assert recommender.topic_registry is None
|
|
|
|
def test_topic_registry_in_strategy_info(self, concrete_recommender):
|
|
"""Topic registry presence is reflected in strategy info."""
|
|
# Note: get_strategy_info doesn't include topic_registry currently
|
|
# but has_search_system is there. This tests the pattern.
|
|
mock_registry = Mock()
|
|
|
|
recommender = concrete_recommender(topic_registry=mock_registry)
|
|
info = recommender.get_strategy_info()
|
|
|
|
# Verify the strategy info structure
|
|
assert "has_search_system" in info
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for metadata handling in cards
|
|
# =============================================================================
|
|
|
|
|
|
class TestMetadataHandlingInCards:
|
|
"""Tests for metadata handling when filtering cards."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class with filter method."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
def test_existing_metadata_preserved(self, concrete_recommender):
|
|
"""Existing metadata is preserved when adding boost."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="tech",
|
|
)
|
|
card.metadata["existing_key"] = "existing_value"
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["tech"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert result[0].metadata["existing_key"] == "existing_value"
|
|
assert result[0].metadata["preference_boost"] == 1.2
|
|
|
|
def test_metadata_not_mutated_for_non_liked(self, concrete_recommender):
|
|
"""Metadata is not modified for non-liked categories."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="sports",
|
|
)
|
|
card.metadata["original"] = "value"
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["tech"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert result[0].metadata == {"original": "value"}
|
|
|
|
def test_boost_overwrites_existing_boost(self, concrete_recommender):
|
|
"""Preference boost overwrites any existing boost."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="tech",
|
|
)
|
|
card.metadata["preference_boost"] = 0.5 # Old boost
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["tech"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert result[0].metadata["preference_boost"] == 1.2
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for multiple preference combinations
|
|
# =============================================================================
|
|
|
|
|
|
class TestMultiplePreferenceCombinations:
|
|
"""Tests for complex preference combinations."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
def test_boost_then_dislike_filter(self, concrete_recommender):
|
|
"""Boosted card can still be filtered by disliked topics."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Tech Politics",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="tech",
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": ["tech"],
|
|
"disliked_topics": ["politics"],
|
|
}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
# Should be filtered out despite being in liked category
|
|
assert len(result) == 0
|
|
|
|
def test_boost_then_threshold_filter(self, concrete_recommender):
|
|
"""Boosted card can still be filtered by impact threshold."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Tech News",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=3,
|
|
category="tech",
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": ["tech"],
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
# Boost is applied but card is still filtered by threshold
|
|
assert len(result) == 0
|
|
|
|
def test_all_preferences_combined_pass(self, concrete_recommender):
|
|
"""Card passes all combined preference filters."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="AI Innovation",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=8,
|
|
category="tech",
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": ["tech"],
|
|
"disliked_topics": ["politics", "sports"],
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 1
|
|
assert result[0].metadata.get("preference_boost") == 1.2
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for docstring in strategy info
|
|
# =============================================================================
|
|
|
|
|
|
class TestDocstringInStrategyInfo:
|
|
"""Tests for docstring handling in get_strategy_info."""
|
|
|
|
def test_class_docstring_used(self):
|
|
"""Class docstring is used as description."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class DocumentedRecommender(BaseRecommender):
|
|
"""This is a custom recommender for testing purposes."""
|
|
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
recommender = DocumentedRecommender()
|
|
info = recommender.get_strategy_info()
|
|
|
|
assert "custom recommender" in info["description"].lower()
|
|
|
|
def test_no_docstring_returns_default(self):
|
|
"""No docstring returns default message."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class UndocumentedRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
# Remove docstring explicitly
|
|
UndocumentedRecommender.__doc__ = None
|
|
|
|
recommender = UndocumentedRecommender()
|
|
info = recommender.get_strategy_info()
|
|
|
|
assert info["description"] == "No description available"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for user_id in methods
|
|
# =============================================================================
|
|
|
|
|
|
class TestUserIdInMethods:
|
|
"""Tests for user_id parameter handling in methods."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def get_prefs(self, user_id):
|
|
return self._get_user_preferences(user_id)
|
|
|
|
def get_ratings(self, user_id, limit=50):
|
|
return self._get_user_ratings(user_id, limit)
|
|
|
|
return TestRecommender
|
|
|
|
def test_uuid_user_id_in_preferences(self, concrete_recommender):
|
|
"""UUID user_id is passed correctly to preferences."""
|
|
mock_pref = Mock()
|
|
mock_pref.get_preferences.return_value = {}
|
|
|
|
recommender = concrete_recommender(preference_manager=mock_pref)
|
|
user_id = "550e8400-e29b-41d4-a716-446655440000"
|
|
|
|
recommender.get_prefs(user_id)
|
|
|
|
mock_pref.get_preferences.assert_called_once_with(user_id)
|
|
|
|
def test_email_user_id_in_ratings(self, concrete_recommender):
|
|
"""Email user_id is passed correctly to ratings."""
|
|
mock_rating = Mock()
|
|
mock_rating.get_recent_ratings.return_value = []
|
|
|
|
recommender = concrete_recommender(rating_system=mock_rating)
|
|
user_id = "user@example.com"
|
|
|
|
recommender.get_ratings(user_id)
|
|
|
|
mock_rating.get_recent_ratings.assert_called_once_with(user_id, 50)
|
|
|
|
def test_empty_user_id(self, concrete_recommender):
|
|
"""Empty user_id is handled without error."""
|
|
mock_pref = Mock()
|
|
mock_pref.get_preferences.return_value = {}
|
|
|
|
recommender = concrete_recommender(preference_manager=mock_pref)
|
|
|
|
# Should not raise
|
|
recommender.get_prefs("")
|
|
|
|
mock_pref.get_preferences.assert_called_once_with("")
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for card ordering preservation
|
|
# =============================================================================
|
|
|
|
|
|
class TestCardOrderingPreservation:
|
|
"""Tests for order preservation in filtering."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
def test_order_preserved_with_no_filters(self, concrete_recommender):
|
|
"""Order is preserved when no filters apply."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic=f"Topic {i}",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
for i in range(10)
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.filter_cards(cards, {})
|
|
|
|
# Order should be preserved
|
|
for i, card in enumerate(result):
|
|
assert card.topic == f"Topic {i}"
|
|
|
|
def test_order_preserved_with_boost(self, concrete_recommender):
|
|
"""Order is preserved when applying category boost."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic=f"Topic {i}",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="tech",
|
|
)
|
|
for i in range(5)
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["tech"]}
|
|
|
|
result = recommender.filter_cards(cards, preferences)
|
|
|
|
# Order should be preserved
|
|
for i, card in enumerate(result):
|
|
assert card.topic == f"Topic {i}"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for impact score edge values
|
|
# =============================================================================
|
|
|
|
|
|
class TestImpactScoreEdgeValues:
|
|
"""Tests for edge values in impact score handling."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_zero_impact_score(self, concrete_recommender):
|
|
"""Zero impact score is handled correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Zero impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=0,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards([card], "user123")
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_negative_impact_score_in_sort(self, concrete_recommender):
|
|
"""Negative impact score is handled in sorting."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Negative",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=-5,
|
|
),
|
|
NewsCard(
|
|
topic="Positive",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards(cards, "user123")
|
|
|
|
# Positive should come first
|
|
assert result[0].topic == "Positive"
|
|
assert result[1].topic == "Negative"
|
|
|
|
def test_very_high_impact_score(self, concrete_recommender):
|
|
"""Very high impact score is handled correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="High impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=1000000,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards([card], "user123")
|
|
|
|
assert len(result) == 1
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for float impact thresholds
|
|
# =============================================================================
|
|
|
|
|
|
class TestFloatImpactThresholds:
|
|
"""Tests for float impact threshold values."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
def test_float_threshold(self, concrete_recommender):
|
|
"""Float threshold works correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Below",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
),
|
|
NewsCard(
|
|
topic="Above",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=6,
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": 5.5}
|
|
|
|
result = recommender.filter_cards(cards, preferences)
|
|
|
|
assert len(result) == 1
|
|
assert result[0].topic == "Above"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for search result variations
|
|
# =============================================================================
|
|
|
|
|
|
class TestSearchResultVariations:
|
|
"""Tests for handling various search result formats."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def execute_search(self, query, strategy=None):
|
|
return self._execute_search(query, strategy)
|
|
|
|
return TestRecommender
|
|
|
|
def test_search_returns_dict(self, concrete_recommender):
|
|
"""Search returns dict results correctly."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = {
|
|
"results": ["r1", "r2"],
|
|
"count": 2,
|
|
}
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
result = recommender.execute_search("test")
|
|
|
|
assert result["results"] == ["r1", "r2"]
|
|
assert result["count"] == 2
|
|
|
|
def test_search_returns_empty_dict(self, concrete_recommender):
|
|
"""Search returns empty dict correctly."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = {}
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
result = recommender.execute_search("test")
|
|
|
|
assert result == {}
|
|
|
|
def test_search_returns_nested_results(self, concrete_recommender):
|
|
"""Search returns nested results correctly."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = {
|
|
"data": {
|
|
"articles": [{"title": "A"}, {"title": "B"}],
|
|
"metadata": {"total": 100},
|
|
}
|
|
}
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
result = recommender.execute_search("test")
|
|
|
|
assert result["data"]["articles"][0]["title"] == "A"
|
|
assert result["data"]["metadata"]["total"] == 100
|
|
|
|
def test_search_with_none_result(self, concrete_recommender):
|
|
"""Search handles None result from analyze_topic."""
|
|
mock_search = Mock()
|
|
mock_search.analyze_topic.return_value = None
|
|
|
|
recommender = concrete_recommender(search_system=mock_search)
|
|
result = recommender.execute_search("test")
|
|
|
|
assert result is None
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for progress callback variations
|
|
# =============================================================================
|
|
|
|
|
|
class TestProgressCallbackVariations:
|
|
"""Tests for various progress callback scenarios."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def send_progress(self, msg, pct, meta=None):
|
|
self._update_progress(msg, pct, meta)
|
|
|
|
return TestRecommender
|
|
|
|
def test_callback_with_none_percent(self, concrete_recommender):
|
|
"""Callback handles None percent value."""
|
|
recommender = concrete_recommender()
|
|
callback = Mock()
|
|
recommender.set_progress_callback(callback)
|
|
|
|
recommender.send_progress("Message", None, {"key": "value"})
|
|
|
|
callback.assert_called_once_with("Message", None, {"key": "value"})
|
|
|
|
def test_callback_with_zero_percent(self, concrete_recommender):
|
|
"""Callback handles 0 percent value."""
|
|
recommender = concrete_recommender()
|
|
callback = Mock()
|
|
recommender.set_progress_callback(callback)
|
|
|
|
recommender.send_progress("Starting", 0, {})
|
|
|
|
callback.assert_called_once_with("Starting", 0, {})
|
|
|
|
def test_callback_with_100_percent(self, concrete_recommender):
|
|
"""Callback handles 100 percent value."""
|
|
recommender = concrete_recommender()
|
|
callback = Mock()
|
|
recommender.set_progress_callback(callback)
|
|
|
|
recommender.send_progress("Complete", 100, {})
|
|
|
|
callback.assert_called_once_with("Complete", 100, {})
|
|
|
|
def test_callback_called_multiple_times(self, concrete_recommender):
|
|
"""Callback can be called multiple times."""
|
|
recommender = concrete_recommender()
|
|
callback = Mock()
|
|
recommender.set_progress_callback(callback)
|
|
|
|
recommender.send_progress("Step 1", 25)
|
|
recommender.send_progress("Step 2", 50)
|
|
recommender.send_progress("Step 3", 75)
|
|
recommender.send_progress("Done", 100)
|
|
|
|
assert callback.call_count == 4
|
|
|
|
def test_callback_can_be_changed(self, concrete_recommender):
|
|
"""Callback can be replaced with a new one."""
|
|
recommender = concrete_recommender()
|
|
|
|
callback1 = Mock()
|
|
callback2 = Mock()
|
|
|
|
recommender.set_progress_callback(callback1)
|
|
recommender.send_progress("First", 50)
|
|
|
|
recommender.set_progress_callback(callback2)
|
|
recommender.send_progress("Second", 75)
|
|
|
|
callback1.assert_called_once()
|
|
callback2.assert_called_once()
|
|
|
|
def test_callback_can_be_cleared(self, concrete_recommender):
|
|
"""Callback can be cleared by setting to None."""
|
|
recommender = concrete_recommender()
|
|
|
|
callback = Mock()
|
|
recommender.set_progress_callback(callback)
|
|
recommender.send_progress("First", 50)
|
|
|
|
recommender.set_progress_callback(None)
|
|
recommender.send_progress("Second", 75) # Should not raise
|
|
|
|
callback.assert_called_once()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for category matching behavior
|
|
# =============================================================================
|
|
|
|
|
|
class TestCategoryMatchingBehavior:
|
|
"""Tests for category matching in filtering."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
def test_category_exact_match_required(self, concrete_recommender):
|
|
"""Category matching requires exact match (not substring)."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="technology",
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["tech"]} # Partial match
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
# Should NOT boost because "tech" != "technology"
|
|
assert result[0].metadata.get("preference_boost") is None
|
|
|
|
def test_category_case_sensitive(self, concrete_recommender):
|
|
"""Category matching is case-sensitive."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="Tech",
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["tech"]} # Different case
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
# Should NOT boost because "Tech" != "tech"
|
|
assert result[0].metadata.get("preference_boost") is None
|
|
|
|
def test_category_none_handling(self, concrete_recommender):
|
|
"""Cards with None category are not boosted."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category=None,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["tech"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert result[0].metadata.get("preference_boost") is None
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for disliked topic edge cases
|
|
# =============================================================================
|
|
|
|
|
|
class TestDislikedTopicEdgeCases:
|
|
"""Tests for edge cases in disliked topic filtering."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
def test_disliked_topic_at_start(self, concrete_recommender):
|
|
"""Disliked topic at start of card topic is filtered."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Politics: Today's News",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["politics"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
def test_disliked_topic_at_end(self, concrete_recommender):
|
|
"""Disliked topic at end of card topic is filtered."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Breaking News in Politics",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["politics"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
def test_disliked_topic_in_middle(self, concrete_recommender):
|
|
"""Disliked topic in middle of card topic is filtered."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Today's Politics Update",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["politics"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
def test_disliked_topic_with_punctuation(self, concrete_recommender):
|
|
"""Disliked topic adjacent to punctuation is filtered."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Politics!",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["politics"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for multiple recommender instances
|
|
# =============================================================================
|
|
|
|
|
|
class TestMultipleRecommenderInstances:
|
|
"""Tests for multiple recommender instances."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
return TestRecommender
|
|
|
|
def test_instances_are_independent(self, concrete_recommender):
|
|
"""Multiple instances have independent state."""
|
|
mock_pref1 = Mock()
|
|
mock_pref1.get_preferences.return_value = {"user": "1"}
|
|
|
|
mock_pref2 = Mock()
|
|
mock_pref2.get_preferences.return_value = {"user": "2"}
|
|
|
|
r1 = concrete_recommender(preference_manager=mock_pref1)
|
|
r2 = concrete_recommender(preference_manager=mock_pref2)
|
|
|
|
prefs1 = r1._get_user_preferences("user")
|
|
prefs2 = r2._get_user_preferences("user")
|
|
|
|
assert prefs1["user"] == "1"
|
|
assert prefs2["user"] == "2"
|
|
|
|
def test_callback_is_instance_specific(self, concrete_recommender):
|
|
"""Callbacks are instance-specific."""
|
|
r1 = concrete_recommender()
|
|
r2 = concrete_recommender()
|
|
|
|
callback1 = Mock()
|
|
callback2 = Mock()
|
|
|
|
r1.set_progress_callback(callback1)
|
|
r2.set_progress_callback(callback2)
|
|
|
|
r1._update_progress("R1", 50)
|
|
r2._update_progress("R2", 75)
|
|
|
|
callback1.assert_called_once_with("R1", 50, {})
|
|
callback2.assert_called_once_with("R2", 75, {})
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for recommendation generation contract
|
|
# =============================================================================
|
|
|
|
|
|
class TestRecommendationGenerationContract:
|
|
"""Tests for the generate_recommendations contract."""
|
|
|
|
def test_must_accept_user_id(self):
|
|
"""generate_recommendations must accept user_id parameter."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class ValidRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return [{"user": user_id}]
|
|
|
|
recommender = ValidRecommender()
|
|
result = recommender.generate_recommendations("user123")
|
|
|
|
assert result[0]["user"] == "user123"
|
|
|
|
def test_context_is_optional(self):
|
|
"""generate_recommendations context parameter is optional."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class ValidRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return [{"context": context}]
|
|
|
|
recommender = ValidRecommender()
|
|
|
|
# Without context
|
|
result1 = recommender.generate_recommendations("user123")
|
|
assert result1[0]["context"] is None
|
|
|
|
# With context
|
|
result2 = recommender.generate_recommendations(
|
|
"user123", {"key": "value"}
|
|
)
|
|
assert result2[0]["context"]["key"] == "value"
|
|
|
|
def test_can_return_empty_list(self):
|
|
"""generate_recommendations can return empty list."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class EmptyRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
recommender = EmptyRecommender()
|
|
result = recommender.generate_recommendations("user123")
|
|
|
|
assert result == []
|
|
|
|
def test_can_return_news_cards(self):
|
|
"""generate_recommendations can return NewsCard objects."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
class CardRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
source = CardSource(type="recommendation")
|
|
return [
|
|
NewsCard(
|
|
topic="Recommended",
|
|
source=source,
|
|
user_id=user_id,
|
|
impact_score=8,
|
|
)
|
|
]
|
|
|
|
recommender = CardRecommender()
|
|
result = recommender.generate_recommendations("user123")
|
|
|
|
assert len(result) == 1
|
|
assert result[0].topic == "Recommended"
|
|
assert result[0].user_id == "user123"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for boost value preservation
|
|
# =============================================================================
|
|
|
|
|
|
class TestBoostValuePreservation:
|
|
"""Tests for boost value handling through the pipeline."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_boost_preserved_through_threshold_filter(
|
|
self, concrete_recommender
|
|
):
|
|
"""Boost is preserved even when threshold filter removes other cards."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Low",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=3,
|
|
category="tech",
|
|
),
|
|
NewsCard(
|
|
topic="High",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=8,
|
|
category="tech",
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": ["tech"],
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
result = recommender.filter_cards(cards, preferences)
|
|
|
|
# Only "High" remains, and it should have the boost
|
|
assert len(result) == 1
|
|
assert result[0].metadata.get("preference_boost") == 1.2
|
|
|
|
def test_boost_affects_sort_order(self, concrete_recommender):
|
|
"""Boost affects the final sort order."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="No boost high impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=9,
|
|
category="sports",
|
|
),
|
|
NewsCard(
|
|
topic="Boosted lower impact",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=8,
|
|
category="tech",
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
|
|
# Apply boost first
|
|
preferences = {"liked_categories": ["tech"]}
|
|
filtered = recommender.filter_cards(cards, preferences)
|
|
|
|
# Then sort
|
|
sorted_cards = recommender.sort_cards(filtered, "user123")
|
|
|
|
# 8/10 * 1.2 = 0.96 > 9/10 * 1.0 = 0.9
|
|
assert sorted_cards[0].topic == "Boosted lower impact"
|
|
assert sorted_cards[1].topic == "No boost high impact"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for special characters in card data
|
|
# =============================================================================
|
|
|
|
|
|
class TestSpecialCharactersInCards:
|
|
"""Tests for handling special characters in card data."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_topic_with_unicode(self, concrete_recommender):
|
|
"""Card with unicode topic is handled correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="日本語ニュース 🎉",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.filter_cards([card], {})
|
|
|
|
assert len(result) == 1
|
|
assert result[0].topic == "日本語ニュース 🎉"
|
|
|
|
def test_topic_with_newlines(self, concrete_recommender):
|
|
"""Card with newlines in topic is handled correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Line 1\nLine 2\nLine 3",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards([card], "user123")
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_disliked_topic_with_unicode(self, concrete_recommender):
|
|
"""Disliked topic matching works with unicode."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="日本語ニュース",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["日本語"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
def test_category_with_special_chars(self, concrete_recommender):
|
|
"""Category with special characters is handled correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
category="tech & science",
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"liked_categories": ["tech & science"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert result[0].metadata.get("preference_boost") == 1.2
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for empty and whitespace handling
|
|
# =============================================================================
|
|
|
|
|
|
class TestEmptyAndWhitespaceHandling:
|
|
"""Tests for handling empty strings and whitespace."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_empty_topic_in_filter(self, concrete_recommender):
|
|
"""Card with empty topic passes through filter."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.filter_cards([card], {})
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_whitespace_only_topic(self, concrete_recommender):
|
|
"""Card with whitespace-only topic is handled."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic=" ",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards([card], "user123")
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_empty_disliked_topics_list(self, concrete_recommender):
|
|
"""Empty disliked_topics list doesn't filter anything."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Any topic",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": []}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_whitespace_in_disliked_topic(self, concrete_recommender):
|
|
"""Whitespace in disliked topic is matched literally."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="topic with spaces",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["with spaces"]}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for impact score at boundaries
|
|
# =============================================================================
|
|
|
|
|
|
class TestImpactScoreBoundaries:
|
|
"""Tests for impact score boundary conditions."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_impact_score_zero_in_sort(self, concrete_recommender):
|
|
"""Zero impact score results in zero score."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Zero",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=0,
|
|
),
|
|
NewsCard(
|
|
topic="Positive",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=1,
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards(cards, "user123")
|
|
|
|
assert result[0].topic == "Positive"
|
|
assert result[1].topic == "Zero"
|
|
|
|
def test_impact_score_exactly_at_threshold(self, concrete_recommender):
|
|
"""Impact score exactly at threshold passes filter."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="At threshold",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": 5}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_impact_score_just_below_threshold(self, concrete_recommender):
|
|
"""Impact score just below threshold is filtered."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Just below",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=4.99,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"impact_threshold": 5}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
def test_impact_score_decimal_precision(self, concrete_recommender):
|
|
"""Impact score handles decimal precision correctly."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Score 5.001",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5.001,
|
|
),
|
|
NewsCard(
|
|
topic="Score 5.002",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5.002,
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards(cards, "user123")
|
|
|
|
# Higher score should come first
|
|
assert result[0].topic == "Score 5.002"
|
|
assert result[1].topic == "Score 5.001"
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for preference key variations
|
|
# =============================================================================
|
|
|
|
|
|
class TestPreferenceKeyVariations:
|
|
"""Tests for various preference key scenarios."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
return TestRecommender
|
|
|
|
def test_unknown_preference_keys_ignored(self, concrete_recommender):
|
|
"""Unknown preference keys are ignored."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"unknown_key": "value",
|
|
"another_unknown": [1, 2, 3],
|
|
}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_none_preference_values(self, concrete_recommender):
|
|
"""None values in preferences are handled."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
card = NewsCard(
|
|
topic="Test",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": None,
|
|
"disliked_topics": None,
|
|
}
|
|
|
|
result = recommender.filter_cards([card], preferences)
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_mixed_preference_types(self, concrete_recommender):
|
|
"""Mixed preference types work together."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic="Tech News High",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=8,
|
|
category="tech",
|
|
),
|
|
NewsCard(
|
|
topic="Sports Low",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=3,
|
|
category="sports",
|
|
),
|
|
NewsCard(
|
|
topic="Politics High",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=9,
|
|
category="politics",
|
|
),
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": ["tech"],
|
|
"disliked_topics": ["politics"],
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
result = recommender.filter_cards(cards, preferences)
|
|
|
|
# Only "Tech News High" should remain
|
|
assert len(result) == 1
|
|
assert result[0].topic == "Tech News High"
|
|
assert result[0].metadata.get("preference_boost") == 1.2
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for large datasets
|
|
# =============================================================================
|
|
|
|
|
|
class TestLargeDatasets:
|
|
"""Tests for handling large numbers of cards."""
|
|
|
|
@pytest.fixture
|
|
def concrete_recommender(self):
|
|
"""Create a concrete recommender class."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
def filter_cards(self, cards, preferences):
|
|
return self._filter_by_preferences(cards, preferences)
|
|
|
|
def sort_cards(self, cards, user_id):
|
|
return self._sort_by_relevance(cards, user_id)
|
|
|
|
return TestRecommender
|
|
|
|
def test_filter_10000_cards(self, concrete_recommender):
|
|
"""Filter handles 10000 cards."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic=f"Topic {i}",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=i % 10,
|
|
category="tech" if i % 2 == 0 else "sports",
|
|
)
|
|
for i in range(10000)
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {
|
|
"liked_categories": ["tech"],
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
result = recommender.filter_cards(cards, preferences)
|
|
|
|
# Should filter correctly
|
|
assert all(card.impact_score >= 5 for card in result)
|
|
|
|
def test_sort_10000_cards(self, concrete_recommender):
|
|
"""Sort handles 10000 cards."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic=f"Topic {i}",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=i % 100,
|
|
)
|
|
for i in range(10000)
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
result = recommender.sort_cards(cards, "user123")
|
|
|
|
# Should be sorted in descending order
|
|
for i in range(len(result) - 1):
|
|
assert result[i].impact_score >= result[i + 1].impact_score
|
|
|
|
def test_filter_all_cards_removed(self, concrete_recommender):
|
|
"""Filter can remove all cards."""
|
|
from local_deep_research.news.core.base_card import NewsCard, CardSource
|
|
|
|
source = CardSource(type="test")
|
|
cards = [
|
|
NewsCard(
|
|
topic=f"Politics {i}",
|
|
source=source,
|
|
user_id="user1",
|
|
impact_score=5,
|
|
)
|
|
for i in range(100)
|
|
]
|
|
|
|
recommender = concrete_recommender()
|
|
preferences = {"disliked_topics": ["politics"]}
|
|
|
|
result = recommender.filter_cards(cards, preferences)
|
|
|
|
assert len(result) == 0
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for strategy info completeness
|
|
# =============================================================================
|
|
|
|
|
|
class TestStrategyInfoCompleteness:
|
|
"""Tests for get_strategy_info completeness."""
|
|
|
|
def test_strategy_info_with_all_dependencies(self):
|
|
"""Strategy info reflects all dependencies."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
"""A test recommender for strategy info."""
|
|
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
mock_pref = Mock()
|
|
mock_rating = Mock()
|
|
mock_search = Mock()
|
|
|
|
recommender = TestRecommender(
|
|
preference_manager=mock_pref,
|
|
rating_system=mock_rating,
|
|
search_system=mock_search,
|
|
)
|
|
|
|
info = recommender.get_strategy_info()
|
|
|
|
assert info["has_preference_manager"] is True
|
|
assert info["has_rating_system"] is True
|
|
assert info["has_search_system"] is True
|
|
|
|
def test_strategy_info_with_no_dependencies(self):
|
|
"""Strategy info reflects no dependencies."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class TestRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
recommender = TestRecommender()
|
|
|
|
info = recommender.get_strategy_info()
|
|
|
|
assert info["has_preference_manager"] is False
|
|
assert info["has_rating_system"] is False
|
|
assert info["has_search_system"] is False
|
|
|
|
def test_strategy_info_name_matches_class(self):
|
|
"""Strategy info name matches class name."""
|
|
from local_deep_research.news.recommender.base_recommender import (
|
|
BaseRecommender,
|
|
)
|
|
|
|
class MyCustomRecommender(BaseRecommender):
|
|
def generate_recommendations(self, user_id, context=None):
|
|
return []
|
|
|
|
recommender = MyCustomRecommender()
|
|
info = recommender.get_strategy_info()
|
|
|
|
assert info["name"] == "MyCustomRecommender"
|