""" Tests for news/api.py Tests cover: - get_news_feed() function - submit_feedback() function - get_votes_for_cards() function - _format_time_ago() function - get_subscription() function - get_subscriptions() function - create_subscription() function - delete_subscription() function - update_subscription() function """ import pytest from unittest.mock import Mock, patch, MagicMock from datetime import datetime, timezone, timedelta import json class TestGetNewsFeed: """Tests for get_news_feed function.""" def test_get_news_feed_invalid_limit_raises_exception(self): """Test that invalid limit raises InvalidLimitException.""" from local_deep_research.news.api import get_news_feed from local_deep_research.news.exceptions import InvalidLimitException with pytest.raises(InvalidLimitException): get_news_feed(user_id="test", limit=0) def test_get_news_feed_negative_limit_raises_exception(self): """Test that negative limit raises InvalidLimitException.""" from local_deep_research.news.api import get_news_feed from local_deep_research.news.exceptions import InvalidLimitException with pytest.raises(InvalidLimitException): get_news_feed(user_id="test", limit=-1) def test_get_news_feed_returns_empty_when_no_items(self): """Test that empty result is returned when no news items.""" from local_deep_research.news.api import get_news_feed with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [] mock_session.return_value = mock_db result = get_news_feed(user_id="test", limit=10) assert "news_items" in result assert len(result["news_items"]) == 0 def test_get_news_feed_filters_by_subscription_id(self): """Test that subscription_id filter is applied.""" from local_deep_research.news.api import get_news_feed with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_query = mock_db.query.return_value.filter.return_value mock_query.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [] mock_session.return_value = mock_db get_news_feed(user_id="test", limit=10, subscription_id="sub-123") # Verify filter was applied (query was built) assert mock_db.query.called def test_get_news_feed_handles_json_parsing_errors(self): """Test handling of malformed JSON in research_meta.""" from local_deep_research.news.api import get_news_feed with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) # Create mock result with invalid JSON mock_result = MagicMock() mock_result.id = "test-id" mock_result.query = "news about test" mock_result.title = None mock_result.created_at = datetime.now(timezone.utc) mock_result.completed_at = None mock_result.duration_seconds = 10 mock_result.report_path = None mock_result.report_content = None mock_result.research_meta = "invalid json{" mock_result.status = "completed" mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [ mock_result ] mock_session.return_value = mock_db result = get_news_feed(user_id="test", limit=10) # Should handle error gracefully assert "news_items" in result def test_get_news_feed_extracts_links_from_content(self): """Test that links are extracted from report content.""" from local_deep_research.news.api import get_news_feed with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) # Create mock result with links in content mock_result = MagicMock() mock_result.id = "test-id" mock_result.query = "breaking news today" mock_result.title = "Test News" mock_result.created_at = datetime.now(timezone.utc) mock_result.completed_at = datetime.now(timezone.utc) mock_result.duration_seconds = 10 mock_result.report_path = None mock_result.report_content = ( "Source Title\nURL: https://example.com/article\nMore content" ) mock_result.research_meta = json.dumps( {"is_news_search": True, "generated_headline": "Test Headline"} ) mock_result.status = "completed" mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [ mock_result ] mock_session.return_value = mock_db result = get_news_feed(user_id="test", limit=10) assert "news_items" in result def test_get_news_feed_includes_generated_at_timestamp(self): """Test that generated_at timestamp is included.""" from local_deep_research.news.api import get_news_feed with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [] mock_session.return_value = mock_db result = get_news_feed(user_id="test", limit=10) assert "generated_at" in result def test_get_news_feed_respects_focus_parameter(self): """Test that focus parameter is passed through.""" from local_deep_research.news.api import get_news_feed with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [] mock_session.return_value = mock_db result = get_news_feed(user_id="test", limit=10, focus="technology") assert result.get("focus") == "technology" def test_get_news_feed_handles_database_error(self): """Test handling of database access errors.""" from local_deep_research.news.api import get_news_feed from local_deep_research.news.exceptions import DatabaseAccessException with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_session.side_effect = Exception("Database connection failed") with pytest.raises((DatabaseAccessException, Exception)): get_news_feed(user_id="test", limit=10) def test_get_news_feed_skips_items_without_headline(self): """Test that items without headlines are skipped.""" from local_deep_research.news.api import get_news_feed with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) # Create mock result without headline mock_result = MagicMock() mock_result.id = "test-id" mock_result.query = "breaking news test" mock_result.title = None # No title mock_result.created_at = datetime.now(timezone.utc) mock_result.completed_at = datetime.now(timezone.utc) mock_result.duration_seconds = 10 mock_result.report_path = None mock_result.report_content = "Some content" mock_result.research_meta = json.dumps( { "is_news_search": True, # No generated_headline } ) mock_result.status = "completed" mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [ mock_result ] mock_session.return_value = mock_db result = get_news_feed(user_id="test", limit=10) assert "news_items" in result def test_get_news_feed_skips_in_progress_items(self): """Test that in_progress items are skipped.""" from local_deep_research.news.api import get_news_feed with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [] mock_session.return_value = mock_db result = get_news_feed(user_id="test", limit=10) # Verify query filters by completed status assert "news_items" in result class TestSubmitFeedback: """Tests for submit_feedback function.""" def test_submit_feedback_invalid_vote_raises_error(self): """Test that invalid vote type raises ValueError.""" from local_deep_research.news.api import submit_feedback with patch("flask.has_request_context") as mock_ctx: mock_ctx.return_value = False with pytest.raises(ValueError) as exc_info: submit_feedback("card-1", "testuser", "invalid") assert "Invalid vote type" in str(exc_info.value) def test_submit_feedback_creates_new_rating(self): """Test that new rating is created when none exists.""" from local_deep_research.news.api import submit_feedback with patch("flask.has_request_context") as mock_ctx: mock_ctx.return_value = False with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_db.query.return_value.filter_by.return_value.first.return_value = None mock_db.query.return_value.filter_by.return_value.count.return_value = 0 mock_session.return_value = mock_db result = submit_feedback("card-1", "testuser", "up") assert result["success"] is True assert mock_db.add.called def test_submit_feedback_updates_existing_rating(self): """Test that existing rating is updated.""" from local_deep_research.news.api import submit_feedback with patch("flask.has_request_context") as mock_ctx: mock_ctx.return_value = False with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) existing_rating = MagicMock() existing_rating.rating_value = "down" mock_db.query.return_value.filter_by.return_value.first.return_value = existing_rating mock_db.query.return_value.filter_by.return_value.count.return_value = 1 mock_session.return_value = mock_db result = submit_feedback("card-1", "testuser", "up") assert existing_rating.rating_value == "up" assert result["success"] is True def test_submit_feedback_returns_vote_counts(self): """Test that vote counts are returned.""" from local_deep_research.news.api import submit_feedback with patch("flask.has_request_context") as mock_ctx: mock_ctx.return_value = False with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_db.query.return_value.filter_by.return_value.first.return_value = None mock_db.query.return_value.filter_by.return_value.count.return_value = 5 mock_session.return_value = mock_db result = submit_feedback("card-1", "testuser", "up") assert "upvotes" in result assert "downvotes" in result def test_submit_feedback_no_username_raises_error(self): """Test that missing username raises ValueError.""" from local_deep_research.news.api import submit_feedback with patch("flask.has_request_context") as mock_ctx: mock_ctx.return_value = False with pytest.raises(ValueError): submit_feedback("card-1", None, "up") class TestGetVotesForCards: """Tests for get_votes_for_cards function.""" def test_get_votes_for_cards_returns_votes(self): """Test that votes are returned for each card.""" from local_deep_research.news.api import get_votes_for_cards with patch("flask.has_request_context") as mock_ctx: mock_ctx.return_value = False with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_db.query.return_value.filter_by.return_value.first.return_value = None mock_db.query.return_value.filter_by.return_value.count.return_value = 0 mock_session.return_value = mock_db result = get_votes_for_cards( ["card-1", "card-2"], user_id="testuser" ) assert result["success"] is True assert "votes" in result assert "card-1" in result["votes"] assert "card-2" in result["votes"] def test_get_votes_for_cards_includes_user_vote(self): """Test that user's vote is included.""" from local_deep_research.news.api import get_votes_for_cards with patch("flask.has_request_context") as mock_ctx: mock_ctx.return_value = False with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) user_vote = MagicMock() user_vote.rating_value = "up" mock_db.query.return_value.filter_by.return_value.first.return_value = user_vote mock_db.query.return_value.filter_by.return_value.count.return_value = 1 mock_session.return_value = mock_db result = get_votes_for_cards(["card-1"], user_id="testuser") assert result["votes"]["card-1"]["user_vote"] == "up" def test_get_votes_for_cards_no_username_raises_error(self): """Test that missing username raises error.""" from local_deep_research.news.api import get_votes_for_cards with patch("flask.has_request_context") as mock_ctx: mock_ctx.return_value = False with pytest.raises(ValueError): get_votes_for_cards(["card-1"], user_id=None) def test_get_votes_for_cards_empty_list(self): """Test with empty card list.""" from local_deep_research.news.api import get_votes_for_cards with patch("flask.has_request_context") as mock_ctx: mock_ctx.return_value = False with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_session.return_value = mock_db result = get_votes_for_cards([], user_id="testuser") assert result["success"] is True assert result["votes"] == {} class TestFormatTimeAgo: """Tests for _format_time_ago function.""" def test_format_time_ago_just_now(self): """Test that recent timestamps return 'Just now'.""" from local_deep_research.news.api import _format_time_ago now = datetime.now(timezone.utc) result = _format_time_ago(now.isoformat()) assert result == "Just now" def test_format_time_ago_minutes(self): """Test formatting for minutes ago.""" from local_deep_research.news.api import _format_time_ago past = datetime.now(timezone.utc) - timedelta(minutes=30) result = _format_time_ago(past.isoformat()) assert "minute" in result def test_format_time_ago_hours(self): """Test formatting for hours ago.""" from local_deep_research.news.api import _format_time_ago past = datetime.now(timezone.utc) - timedelta(hours=5) result = _format_time_ago(past.isoformat()) assert "hour" in result def test_format_time_ago_days(self): """Test formatting for days ago.""" from local_deep_research.news.api import _format_time_ago past = datetime.now(timezone.utc) - timedelta(days=3) result = _format_time_ago(past.isoformat()) assert "day" in result def test_format_time_ago_invalid_timestamp(self): """Invalid timestamps raise (caller logs + skips the row).""" from local_deep_research.news.api import _format_time_ago with pytest.raises(ValueError): _format_time_ago("not a valid timestamp") def test_format_time_ago_timezone_handling(self): """Naive timestamps are interpreted as UTC by the formatter. Build the naive string from UTC explicitly — using datetime.now() (local) made this test fail on any non-UTC machine, since the formatter then computed diff = (utc_now - local_now_minus_2h), which collapses to ~0 on a UTC+2 box and returns "Just now". """ from local_deep_research.news.api import _format_time_ago naive_utc = ( datetime.now(timezone.utc) - timedelta(hours=2, seconds=1) ).replace(tzinfo=None) result = _format_time_ago(naive_utc.isoformat()) # The function may return "2 hours ago" or "120 minutes ago" assert "hour" in result or "minute" in result class TestGetSubscription: """Tests for get_subscription function.""" def test_get_subscription_not_found_raises_exception(self): """Test that missing subscription raises SubscriptionNotFoundException.""" from local_deep_research.news.api import get_subscription from local_deep_research.news.exceptions import ( SubscriptionNotFoundException, ) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_db.query.return_value.filter_by.return_value.first.return_value = None mock_session.return_value = mock_db with pytest.raises(SubscriptionNotFoundException): get_subscription("nonexistent-id") def test_get_subscription_returns_formatted_data(self): """Test that subscription data is properly formatted.""" from local_deep_research.news.api import get_subscription with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_sub = MagicMock() mock_sub.id = "sub-123" mock_sub.name = "Test Sub" mock_sub.query_or_topic = "test query" mock_sub.subscription_type = "search" mock_sub.refresh_interval_minutes = 240 mock_sub.status = "active" mock_sub.folder_id = None mock_sub.model_provider = "openai" mock_sub.model = "gpt-4" mock_sub.search_strategy = "default" mock_sub.custom_endpoint = None mock_sub.search_engine = "searxng" mock_sub.search_iterations = 3 mock_sub.questions_per_iteration = 5 mock_sub.created_at = datetime.now(timezone.utc) mock_sub.updated_at = datetime.now(timezone.utc) mock_db.query.return_value.filter_by.return_value.first.return_value = mock_sub mock_session.return_value = mock_db result = get_subscription("sub-123") assert result["id"] == "sub-123" assert result["name"] == "Test Sub" assert result["is_active"] is True class TestCreateSubscription: """Tests for create_subscription function.""" def test_create_subscription_success(self): """Test successful subscription creation.""" from local_deep_research.news.api import create_subscription with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_session.return_value = mock_db with patch( "local_deep_research.news.api._notify_scheduler_about_subscription_change" ): result = create_subscription( user_id="testuser", query="test query", subscription_type="search", refresh_minutes=120, ) assert result["status"] == "success" assert "subscription_id" in result assert mock_db.add.called assert mock_db.commit.called def test_create_subscription_uses_default_refresh_minutes(self): """Test that default refresh minutes is used when not provided.""" from local_deep_research.news.api import create_subscription with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_session.return_value = mock_db with patch( "local_deep_research.news.api._notify_scheduler_about_subscription_change" ): with patch( "local_deep_research.utilities.db_utils.get_settings_manager" ) as mock_settings: mock_mgr = MagicMock() mock_mgr.get_setting.return_value = 240 mock_settings.return_value = mock_mgr result = create_subscription( user_id="testuser", query="test query", ) assert result["status"] == "success" class TestDeleteSubscription: """Tests for delete_subscription function.""" def test_delete_subscription_not_found_raises_exception(self): """Test that missing subscription raises exception.""" from local_deep_research.news.api import delete_subscription from local_deep_research.news.exceptions import ( SubscriptionNotFoundException, ) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_db.query.return_value.filter_by.return_value.first.return_value = None mock_session.return_value = mock_db with pytest.raises(SubscriptionNotFoundException): delete_subscription("nonexistent-id") def test_delete_subscription_success(self): """Test successful subscription deletion.""" from local_deep_research.news.api import delete_subscription with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_sub = MagicMock() mock_db.query.return_value.filter_by.return_value.first.return_value = mock_sub mock_session.return_value = mock_db with patch( "local_deep_research.news.api._notify_scheduler_about_subscription_change" ): result = delete_subscription("sub-123") assert result["status"] == "success" assert mock_db.delete.called assert mock_db.commit.called class TestUpdateSubscription: """Tests for update_subscription function.""" def test_update_subscription_not_found_raises_exception(self): """Test that missing subscription raises exception.""" from local_deep_research.news.api import update_subscription from local_deep_research.news.exceptions import ( SubscriptionNotFoundException, ) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_db.query.return_value.filter_by.return_value.first.return_value = None mock_session.return_value = mock_db with pytest.raises(SubscriptionNotFoundException): update_subscription("nonexistent-id", {"name": "New Name"}) def test_update_subscription_updates_fields(self): """Test that subscription fields are updated.""" from local_deep_research.news.api import update_subscription with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_sub = MagicMock() mock_sub.id = "sub-123" mock_sub.name = "Old Name" mock_sub.query_or_topic = "old query" mock_sub.subscription_type = "search" mock_sub.refresh_interval_minutes = 240 mock_sub.status = "active" mock_sub.folder_id = None mock_sub.model_provider = None mock_sub.model = None mock_sub.search_strategy = None mock_sub.custom_endpoint = None mock_sub.search_engine = None mock_sub.search_iterations = 3 mock_sub.questions_per_iteration = 5 mock_db.query.return_value.filter_by.return_value.first.return_value = mock_sub mock_session.return_value = mock_db with patch( "local_deep_research.news.api._notify_scheduler_about_subscription_change" ): result = update_subscription( "sub-123", {"name": "New Name", "is_active": False} ) assert mock_sub.name == "New Name" assert mock_sub.status == "paused" assert result["status"] == "success" class TestGetSubscriptions: """Tests for get_subscriptions function.""" def test_get_subscriptions_returns_list(self): """Test that subscription list is returned.""" from local_deep_research.news.api import get_subscriptions with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_db.query.return_value.all.return_value = [] mock_session.return_value = mock_db result = get_subscriptions(user_id="testuser") assert "subscriptions" in result assert "total" in result def test_get_subscriptions_counts_runs(self): """Test that total runs are counted.""" from local_deep_research.news.api import get_subscriptions with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_session: mock_db = MagicMock() mock_db.__enter__ = Mock(return_value=mock_db) mock_db.__exit__ = Mock(return_value=False) mock_sub = MagicMock() mock_sub.id = "sub-123" mock_sub.query_or_topic = "test" mock_sub.subscription_type = "search" mock_sub.refresh_interval_minutes = 240 mock_sub.status = "active" mock_sub.created_at = datetime.now(timezone.utc) mock_sub.next_refresh = datetime.now(timezone.utc) mock_sub.last_refresh = None mock_sub.name = "Test" mock_sub.folder_id = None mock_sub.model_provider = "openai" mock_sub.model = "gpt-4o" mock_sub.search_strategy = "focused-iteration" mock_sub.search_engine = "searxng" mock_sub.custom_endpoint = None mock_sub.search_iterations = 3 mock_sub.questions_per_iteration = 5 mock_db.query.return_value.all.return_value = [mock_sub] mock_db.query.return_value.filter.return_value.scalar.return_value = 5 mock_session.return_value = mock_db result = get_subscriptions(user_id="testuser") assert len(result["subscriptions"]) == 1 # The subscription's saved model/search config must be included # (it was previously dropped, so callers silently got None). sub = result["subscriptions"][0] assert sub["model_provider"] == "openai" assert sub["model"] == "gpt-4o" assert sub["search_strategy"] == "focused-iteration" assert sub["search_engine"] == "searxng" assert sub["status"] == "active"