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
710 lines
25 KiB
Python
710 lines
25 KiB
Python
"""
|
|
Comprehensive tests for news/core/storage_manager.py
|
|
|
|
Tests cover:
|
|
- StorageManager initialization
|
|
- User feed retrieval
|
|
- Trending news
|
|
- Interaction recording
|
|
- Card operations
|
|
- Cleanup operations
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, patch
|
|
|
|
|
|
class TestStorageManagerInit:
|
|
"""Tests for StorageManager initialization."""
|
|
|
|
def test_init_creates_instance(self):
|
|
"""Test that StorageManager can be initialized."""
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.get_relevance_service"
|
|
) as mock_relevance:
|
|
mock_relevance.return_value = Mock()
|
|
from local_deep_research.news.core.storage_manager import (
|
|
StorageManager,
|
|
)
|
|
|
|
manager = StorageManager()
|
|
|
|
assert manager._cards is None
|
|
assert manager._ratings is None
|
|
assert manager._preferences is None
|
|
assert manager.relevance_service is not None
|
|
|
|
|
|
class TestInteractionType:
|
|
"""Tests for InteractionType enum."""
|
|
|
|
def test_interaction_types_exist(self):
|
|
"""Test that all interaction types are defined."""
|
|
from local_deep_research.news.core.storage_manager import (
|
|
InteractionType,
|
|
)
|
|
|
|
assert InteractionType.VIEW.value == "view"
|
|
assert InteractionType.VOTE_UP.value == "vote_up"
|
|
assert InteractionType.VOTE_DOWN.value == "vote_down"
|
|
assert InteractionType.RESEARCH.value == "research"
|
|
assert InteractionType.SHARE.value == "share"
|
|
|
|
|
|
class TestStorageManagerProperties:
|
|
"""Tests for StorageManager property accessors."""
|
|
|
|
@pytest.fixture
|
|
def storage_manager(self):
|
|
"""Create a StorageManager instance for testing."""
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.get_relevance_service"
|
|
) as mock_relevance:
|
|
mock_relevance.return_value = Mock()
|
|
from local_deep_research.news.core.storage_manager import (
|
|
StorageManager,
|
|
)
|
|
|
|
return StorageManager()
|
|
|
|
def test_cards_property_with_session(self, storage_manager):
|
|
"""Test cards property when session is available."""
|
|
mock_session = Mock()
|
|
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=mock_session
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLCardStorage"
|
|
) as mock_storage:
|
|
mock_storage.return_value = Mock()
|
|
_ = storage_manager.cards
|
|
mock_storage.assert_called_once_with(mock_session)
|
|
|
|
def test_cards_property_without_session_raises(self, storage_manager):
|
|
"""Test cards property raises when no session available."""
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=None
|
|
):
|
|
with pytest.raises(RuntimeError) as exc_info:
|
|
_ = storage_manager.cards
|
|
|
|
assert "No database session" in str(exc_info.value)
|
|
|
|
def test_ratings_property_with_session(self, storage_manager):
|
|
"""Test ratings property when session is available."""
|
|
mock_session = Mock()
|
|
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=mock_session
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLRatingStorage"
|
|
) as mock_storage:
|
|
mock_storage.return_value = Mock()
|
|
_ = storage_manager.ratings
|
|
mock_storage.assert_called_once_with(mock_session)
|
|
|
|
def test_preferences_property_with_session(self, storage_manager):
|
|
"""Test preferences property when session is available."""
|
|
mock_session = Mock()
|
|
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=mock_session
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLPreferenceStorage"
|
|
) as mock_storage:
|
|
mock_storage.return_value = Mock()
|
|
_ = storage_manager.preferences
|
|
mock_storage.assert_called_once_with(mock_session)
|
|
|
|
|
|
class TestGetUserFeed:
|
|
"""Tests for get_user_feed method."""
|
|
|
|
@pytest.fixture
|
|
def storage_manager(self):
|
|
"""Create a StorageManager instance for testing."""
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.get_relevance_service"
|
|
) as mock_relevance:
|
|
mock_relevance.return_value = Mock()
|
|
from local_deep_research.news.core.storage_manager import (
|
|
StorageManager,
|
|
)
|
|
|
|
return StorageManager()
|
|
|
|
def test_get_user_feed_success(self, storage_manager):
|
|
"""Test successful user feed retrieval."""
|
|
mock_prefs_storage = Mock()
|
|
mock_prefs_storage.get.return_value = {"liked_categories": ["Tech"]}
|
|
|
|
mock_cards_storage = Mock()
|
|
mock_cards_storage.list.return_value = [
|
|
{"id": "card1"},
|
|
{"id": "card2"},
|
|
]
|
|
|
|
mock_card1 = Mock()
|
|
mock_card2 = Mock()
|
|
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=Mock()
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLPreferenceStorage",
|
|
return_value=mock_prefs_storage,
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLCardStorage",
|
|
return_value=mock_cards_storage,
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.load_card.side_effect = [
|
|
mock_card1,
|
|
mock_card2,
|
|
]
|
|
|
|
storage_manager.relevance_service.personalize_feed.return_value = [
|
|
mock_card1,
|
|
mock_card2,
|
|
]
|
|
|
|
result = storage_manager.get_user_feed(
|
|
"user1", limit=10
|
|
)
|
|
|
|
assert len(result) == 2
|
|
|
|
def test_get_user_feed_without_preferences(self, storage_manager):
|
|
"""Test user feed when user has no preferences."""
|
|
mock_prefs_storage = Mock()
|
|
mock_prefs_storage.get.return_value = None
|
|
|
|
mock_cards_storage = Mock()
|
|
mock_cards_storage.list.return_value = [{"id": "card1"}]
|
|
|
|
mock_card = Mock()
|
|
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=Mock()
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLPreferenceStorage",
|
|
return_value=mock_prefs_storage,
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLCardStorage",
|
|
return_value=mock_cards_storage,
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.load_card.return_value = mock_card
|
|
storage_manager.relevance_service.personalize_feed.return_value = [
|
|
mock_card
|
|
]
|
|
|
|
storage_manager.get_user_feed("user1")
|
|
|
|
# Should use user_id filter only
|
|
mock_cards_storage.list.assert_called_once()
|
|
call_args = mock_cards_storage.list.call_args
|
|
assert call_args[1]["filters"]["user_id"] == "user1"
|
|
|
|
def test_get_user_feed_with_card_types(self, storage_manager):
|
|
"""Test user feed filtered by card types."""
|
|
mock_prefs_storage = Mock()
|
|
mock_prefs_storage.get.return_value = None
|
|
|
|
mock_cards_storage = Mock()
|
|
mock_cards_storage.list.return_value = []
|
|
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=Mock()
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLPreferenceStorage",
|
|
return_value=mock_prefs_storage,
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLCardStorage",
|
|
return_value=mock_cards_storage,
|
|
):
|
|
storage_manager.relevance_service.personalize_feed.return_value = []
|
|
|
|
storage_manager.get_user_feed(
|
|
"user1", card_types=["news", "alert"]
|
|
)
|
|
|
|
call_args = mock_cards_storage.list.call_args
|
|
assert call_args[1]["filters"]["card_type"] == [
|
|
"news",
|
|
"alert",
|
|
]
|
|
|
|
def test_get_user_feed_handles_error(self, storage_manager):
|
|
"""Test that errors are handled gracefully."""
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=Mock()
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLPreferenceStorage"
|
|
) as mock_prefs:
|
|
mock_prefs.return_value.get.side_effect = Exception("DB Error")
|
|
|
|
result = storage_manager.get_user_feed("user1")
|
|
|
|
assert result == []
|
|
|
|
|
|
class TestGetTrendingNews:
|
|
"""Tests for get_trending_news method."""
|
|
|
|
@pytest.fixture
|
|
def storage_manager(self):
|
|
"""Create a StorageManager instance for testing."""
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.get_relevance_service"
|
|
) as mock_relevance:
|
|
mock_relevance.return_value = Mock()
|
|
from local_deep_research.news.core.storage_manager import (
|
|
StorageManager,
|
|
)
|
|
|
|
return StorageManager()
|
|
|
|
def test_get_trending_news_success(self, storage_manager):
|
|
"""Test successful trending news retrieval."""
|
|
mock_cards = [Mock(), Mock()]
|
|
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.get_recent_cards.return_value = mock_cards
|
|
storage_manager.relevance_service.filter_trending.return_value = (
|
|
mock_cards
|
|
)
|
|
|
|
result = storage_manager.get_trending_news(hours=24, limit=10)
|
|
|
|
mock_factory.get_recent_cards.assert_called_once_with(
|
|
hours=24, card_types=["news"], limit=20
|
|
)
|
|
storage_manager.relevance_service.filter_trending.assert_called_once()
|
|
assert len(result) == 2
|
|
|
|
def test_get_trending_news_handles_error(self, storage_manager):
|
|
"""Test that errors are handled gracefully."""
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.get_recent_cards.side_effect = Exception("Error")
|
|
|
|
result = storage_manager.get_trending_news()
|
|
|
|
assert result == []
|
|
|
|
|
|
class TestRecordInteraction:
|
|
"""Tests for record_interaction method."""
|
|
|
|
@pytest.fixture
|
|
def storage_manager(self):
|
|
"""Create a StorageManager instance for testing."""
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.get_relevance_service"
|
|
) as mock_relevance:
|
|
mock_relevance.return_value = Mock()
|
|
from local_deep_research.news.core.storage_manager import (
|
|
StorageManager,
|
|
)
|
|
|
|
return StorageManager()
|
|
|
|
def test_record_view_interaction(self, storage_manager):
|
|
"""Test recording a view interaction."""
|
|
from local_deep_research.news.core.storage_manager import (
|
|
InteractionType,
|
|
)
|
|
|
|
mock_card = Mock()
|
|
mock_card.interaction = {"views": 5}
|
|
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.load_card.return_value = mock_card
|
|
mock_factory.update_card.return_value = True
|
|
|
|
result = storage_manager.record_interaction(
|
|
"user1", "card1", InteractionType.VIEW
|
|
)
|
|
|
|
assert result is True
|
|
assert mock_card.interaction["viewed"] is True
|
|
assert mock_card.interaction["views"] == 6
|
|
|
|
def test_record_vote_up_interaction(self, storage_manager):
|
|
"""Test recording a vote up interaction."""
|
|
from local_deep_research.news.core.storage_manager import (
|
|
InteractionType,
|
|
)
|
|
|
|
mock_card = Mock()
|
|
mock_card.interaction = {"votes_up": 10}
|
|
|
|
mock_ratings = Mock()
|
|
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=Mock()
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLRatingStorage",
|
|
return_value=mock_ratings,
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.load_card.return_value = mock_card
|
|
mock_factory.update_card.return_value = True
|
|
|
|
result = storage_manager.record_interaction(
|
|
"user1", "card1", InteractionType.VOTE_UP
|
|
)
|
|
|
|
assert result is True
|
|
assert mock_card.interaction["voted"] == "up"
|
|
assert mock_card.interaction["votes_up"] == 11
|
|
mock_ratings.save.assert_called_once()
|
|
|
|
def test_record_vote_down_interaction(self, storage_manager):
|
|
"""Test recording a vote down interaction."""
|
|
from local_deep_research.news.core.storage_manager import (
|
|
InteractionType,
|
|
)
|
|
|
|
mock_card = Mock()
|
|
mock_card.interaction = {"votes_down": 3}
|
|
|
|
mock_ratings = Mock()
|
|
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=Mock()
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLRatingStorage",
|
|
return_value=mock_ratings,
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.load_card.return_value = mock_card
|
|
mock_factory.update_card.return_value = True
|
|
|
|
result = storage_manager.record_interaction(
|
|
"user1", "card1", InteractionType.VOTE_DOWN
|
|
)
|
|
|
|
assert result is True
|
|
assert mock_card.interaction["voted"] == "down"
|
|
assert mock_card.interaction["votes_down"] == 4
|
|
|
|
def test_record_research_interaction(self, storage_manager):
|
|
"""Test recording a research interaction."""
|
|
from local_deep_research.news.core.storage_manager import (
|
|
InteractionType,
|
|
)
|
|
|
|
mock_card = Mock()
|
|
mock_card.interaction = {"research_count": 2}
|
|
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.load_card.return_value = mock_card
|
|
mock_factory.update_card.return_value = True
|
|
|
|
result = storage_manager.record_interaction(
|
|
"user1", "card1", InteractionType.RESEARCH
|
|
)
|
|
|
|
assert result is True
|
|
assert mock_card.interaction["researched"] is True
|
|
assert mock_card.interaction["research_count"] == 3
|
|
|
|
def test_record_interaction_with_metadata(self, storage_manager):
|
|
"""Test recording interaction with metadata."""
|
|
from local_deep_research.news.core.storage_manager import (
|
|
InteractionType,
|
|
)
|
|
|
|
mock_card = Mock()
|
|
mock_card.interaction = {"views": 0}
|
|
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.load_card.return_value = mock_card
|
|
mock_factory.update_card.return_value = True
|
|
|
|
result = storage_manager.record_interaction(
|
|
"user1",
|
|
"card1",
|
|
InteractionType.VIEW,
|
|
metadata={"source": "feed"},
|
|
)
|
|
|
|
assert result is True
|
|
assert "InteractionType.VIEW_metadata" in mock_card.interaction
|
|
|
|
def test_record_interaction_card_not_found(self, storage_manager):
|
|
"""Test recording interaction when card not found."""
|
|
from local_deep_research.news.core.storage_manager import (
|
|
InteractionType,
|
|
)
|
|
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.load_card.return_value = None
|
|
|
|
result = storage_manager.record_interaction(
|
|
"user1", "nonexistent", InteractionType.VIEW
|
|
)
|
|
|
|
assert result is False
|
|
|
|
def test_record_interaction_handles_error(self, storage_manager):
|
|
"""Test that errors are handled gracefully."""
|
|
from local_deep_research.news.core.storage_manager import (
|
|
InteractionType,
|
|
)
|
|
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.load_card.side_effect = Exception("Error")
|
|
|
|
result = storage_manager.record_interaction(
|
|
"user1", "card1", InteractionType.VIEW
|
|
)
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestCardOperations:
|
|
"""Tests for card-related operations."""
|
|
|
|
@pytest.fixture
|
|
def storage_manager(self):
|
|
"""Create a StorageManager instance for testing."""
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.get_relevance_service"
|
|
) as mock_relevance:
|
|
mock_relevance.return_value = Mock()
|
|
from local_deep_research.news.core.storage_manager import (
|
|
StorageManager,
|
|
)
|
|
|
|
return StorageManager()
|
|
|
|
def test_get_card_success(self, storage_manager):
|
|
"""Test successful card retrieval."""
|
|
mock_card = Mock()
|
|
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.load_card.return_value = mock_card
|
|
|
|
result = storage_manager.get_card("card1")
|
|
|
|
assert result == mock_card
|
|
mock_factory.load_card.assert_called_once_with("card1")
|
|
|
|
def test_get_card_not_found(self, storage_manager):
|
|
"""Test card retrieval when not found."""
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.load_card.return_value = None
|
|
|
|
result = storage_manager.get_card("nonexistent")
|
|
|
|
assert result is None
|
|
|
|
def test_get_card_handles_error(self, storage_manager):
|
|
"""Test that errors are handled gracefully."""
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.load_card.side_effect = Exception("Error")
|
|
|
|
result = storage_manager.get_card("card1")
|
|
|
|
assert result is None
|
|
|
|
def test_update_card_success(self, storage_manager):
|
|
"""Test successful card update."""
|
|
mock_card = Mock()
|
|
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.update_card.return_value = True
|
|
|
|
result = storage_manager.update_card(mock_card)
|
|
|
|
assert result is True
|
|
mock_factory.update_card.assert_called_once_with(mock_card)
|
|
|
|
def test_update_card_handles_error(self, storage_manager):
|
|
"""Test that errors are handled gracefully."""
|
|
mock_card = Mock()
|
|
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.CardFactory"
|
|
) as mock_factory:
|
|
mock_factory.update_card.side_effect = Exception("Error")
|
|
|
|
result = storage_manager.update_card(mock_card)
|
|
|
|
assert result is False
|
|
|
|
def test_get_card_interactions_success(self, storage_manager):
|
|
"""Test successful interaction retrieval."""
|
|
mock_ratings = [
|
|
{"user_id": "user1", "value": 1, "created_at": "2024-01-01"},
|
|
{"user_id": "user2", "value": -1, "created_at": "2024-01-02"},
|
|
]
|
|
mock_storage = Mock()
|
|
mock_storage.list.return_value = mock_ratings
|
|
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=Mock()
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLRatingStorage",
|
|
return_value=mock_storage,
|
|
):
|
|
result = storage_manager.get_card_interactions("card1")
|
|
|
|
assert len(result) == 2
|
|
assert result[0]["interaction_type"] == "vote"
|
|
assert result[0]["interaction_data"]["vote"] == "up"
|
|
assert result[1]["interaction_data"]["vote"] == "down"
|
|
|
|
|
|
class TestCleanupOldData:
|
|
"""Tests for cleanup_old_data method."""
|
|
|
|
@pytest.fixture
|
|
def storage_manager(self):
|
|
"""Create a StorageManager instance for testing."""
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.get_relevance_service"
|
|
) as mock_relevance:
|
|
mock_relevance.return_value = Mock()
|
|
from local_deep_research.news.core.storage_manager import (
|
|
StorageManager,
|
|
)
|
|
|
|
return StorageManager()
|
|
|
|
def test_cleanup_old_data_success(self, storage_manager):
|
|
"""Test successful data cleanup."""
|
|
mock_old_cards = [{"id": "card1"}, {"id": "card2"}]
|
|
mock_old_ratings = [{"id": "rating1"}]
|
|
|
|
mock_card_storage = Mock()
|
|
mock_card_storage.list.return_value = mock_old_cards
|
|
mock_card_storage.delete.return_value = True
|
|
|
|
mock_rating_storage = Mock()
|
|
mock_rating_storage.list.return_value = mock_old_ratings
|
|
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=Mock()
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLCardStorage",
|
|
return_value=mock_card_storage,
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLRatingStorage",
|
|
return_value=mock_rating_storage,
|
|
):
|
|
result = storage_manager.cleanup_old_data(days=30)
|
|
|
|
assert result["cards"] == 2
|
|
assert result["ratings"] == 1
|
|
|
|
def test_cleanup_old_data_handles_error(self, storage_manager):
|
|
"""Test that errors are handled gracefully."""
|
|
with patch.object(
|
|
storage_manager, "_get_current_session", return_value=Mock()
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.SQLCardStorage"
|
|
) as mock_storage:
|
|
mock_storage.return_value.list.side_effect = Exception("Error")
|
|
|
|
result = storage_manager.cleanup_old_data()
|
|
|
|
assert result == {}
|
|
|
|
|
|
class TestGetCurrentSession:
|
|
"""Tests for _get_current_session method."""
|
|
|
|
def test_get_session_from_flask_context(self):
|
|
"""Test getting session from Flask g context (via lazy helper)."""
|
|
mock_session = Mock()
|
|
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.get_relevance_service"
|
|
) as mock_relevance:
|
|
mock_relevance.return_value = Mock()
|
|
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.has_app_context",
|
|
return_value=True,
|
|
):
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_g_db_session",
|
|
return_value=mock_session,
|
|
):
|
|
from local_deep_research.news.core.storage_manager import (
|
|
StorageManager,
|
|
)
|
|
|
|
manager = StorageManager()
|
|
result = manager._get_current_session()
|
|
|
|
assert result == mock_session
|
|
|
|
def test_get_session_no_app_context(self):
|
|
"""Test getting session when no Flask context."""
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.get_relevance_service"
|
|
) as mock_relevance:
|
|
mock_relevance.return_value = Mock()
|
|
|
|
with patch(
|
|
"local_deep_research.news.core.storage_manager.has_app_context",
|
|
return_value=False,
|
|
):
|
|
from local_deep_research.news.core.storage_manager import (
|
|
StorageManager,
|
|
)
|
|
|
|
manager = StorageManager()
|
|
result = manager._get_current_session()
|
|
|
|
assert result is None
|