Files
wehub-resource-sync 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
chore: import upstream snapshot with attribution
2026-07-13 13:08:55 +08:00

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"