""" Extra coverage tests for local_deep_research.news.flask_api. Targets uncovered/under-tested branches: - safe_error_message: context='' produces generic message without 'while' - _is_job_owned_by_user: no args attr, empty args, no user_sessions, system job - get_news_feed: error-in-result with 'must be between' yields 400 vs generic 500 - create_subscription: get_json(force=True) raises (invalid JSON body) - vote_on_news: ValueError with 'not found' yields 404 - submit_feedback: ValueError with 'must be' yields 400; generic ValueError yields 400 - get_subscription: null/undefined ID returns 400; valid ID returns data; exception 500 - update_subscription: api returns error with 'not found' -> 404 vs other -> 400 - delete_subscription: success path; not-found path - run_subscription_now: subscription not found; response.ok but non-success status; response not ok - stop_scheduler: not running; no scheduler instance - check_subscriptions_now: no scheduler; scheduler not running - trigger_cleanup: scheduler not running - get_active_users: show_all=False filters to current user only - scheduler_stats: show_all=False scopes; last_activity=None - get_search_history: unauthenticated returns empty - add_search_history: unauthenticated returns 401 - clear_search_history: unauthenticated returns success - get_scheduler_status: APScheduler get_jobs raises exception - update_subscription_folder: not found; refresh_interval recalc with/without last_refresh - check_overdue_subscriptions: inner exception per-subscription """ from unittest.mock import MagicMock, patch import pytest from flask import Flask MODULE = "local_deep_research.news.flask_api" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_app(): """Create a minimal Flask app with only the news_api blueprint.""" app = Flask(__name__) app.config["SECRET_KEY"] = "test-secret" app.config["WTF_CSRF_ENABLED"] = False app.config["TESTING"] = True from local_deep_research.news.flask_api import news_api_bp app.register_blueprint(news_api_bp, url_prefix="/news/api") return app def _auth_session(client, username="testuser"): """Inject an authenticated session.""" with client.session_transaction() as sess: sess["username"] = username def _common_patches(): """Return dict of context-manager patches for auth + db plumbing.""" return { "db_manager": patch( "local_deep_research.web.auth.decorators.db_manager" ), "get_user_id": patch(f"{MODULE}.get_user_id", return_value="testuser"), "get_settings_manager": patch(f"{MODULE}.get_settings_manager"), } @pytest.fixture def app(): """Flask app fixture.""" return _make_app() @pytest.fixture def client(app): """Test client fixture.""" return app.test_client() @pytest.fixture def authed_client(client): """Test client with authenticated session.""" _auth_session(client) return client # --------------------------------------------------------------------------- # safe_error_message # --------------------------------------------------------------------------- class TestSafeErrorMessageExtraCoverage: """Edge cases for safe_error_message.""" def test_generic_error_empty_context(self): """Empty context string produces message without 'while'.""" from local_deep_research.news.flask_api import safe_error_message result = safe_error_message(RuntimeError("boom"), context="") assert result == "An error occurred" assert "while" not in result def test_generic_error_with_context(self): """Non-empty context string includes 'while '.""" from local_deep_research.news.flask_api import safe_error_message result = safe_error_message(RuntimeError("boom"), context="loading") assert "while loading" in result # --------------------------------------------------------------------------- # _is_job_owned_by_user # --------------------------------------------------------------------------- class TestIsJobOwnedByUser: """Branch coverage for _is_job_owned_by_user.""" def test_no_args_attribute(self): """Job without args attribute falls through to scheduler check.""" from local_deep_research.news.flask_api import _is_job_owned_by_user job = MagicMock(spec=[]) # no .args scheduler = MagicMock(spec=[]) # no .user_sessions assert _is_job_owned_by_user(job, "alice", scheduler) is False def test_empty_args_list(self): """Job with empty args list does not match.""" from local_deep_research.news.flask_api import _is_job_owned_by_user job = MagicMock() job.args = [] scheduler = MagicMock(spec=[]) assert _is_job_owned_by_user(job, "alice", scheduler) is False def test_args_first_element_mismatch(self): """Job args[0] does not equal username; scheduler has no user_sessions.""" from local_deep_research.news.flask_api import _is_job_owned_by_user job = MagicMock() job.args = ["bob"] scheduler = MagicMock(spec=[]) assert _is_job_owned_by_user(job, "alice", scheduler) is False def test_fallback_scheduled_jobs_match(self): """Job matched via scheduler.user_sessions scheduled_jobs set.""" from local_deep_research.news.flask_api import _is_job_owned_by_user job = MagicMock() job.args = ["other_user"] job.id = "job-42" scheduler = MagicMock() scheduler.user_sessions = { "alice": {"scheduled_jobs": {"job-42", "job-99"}} } assert _is_job_owned_by_user(job, "alice", scheduler) is True # --------------------------------------------------------------------------- # GET /feed – error-in-result branches # --------------------------------------------------------------------------- class TestGetNewsFeedErrorBranches: """Covers the error-in-result branching in get_news_feed.""" def test_error_with_must_be_between_returns_400(self, authed_client): """When api result error contains 'must be between', status is 400.""" cp = _common_patches() error_result = { "error": "limit must be between 1 and 100", "news_items": [], } with ( cp["db_manager"] as mdb, cp["get_user_id"], cp["get_settings_manager"] as msm, patch(f"{MODULE}.api.get_news_feed", return_value=error_result), ): mdb.is_user_connected.return_value = True sm = MagicMock() sm.get_setting.return_value = 20 msm.return_value = sm resp = authed_client.get("/news/api/feed") assert resp.status_code == 400 def test_error_without_must_be_between_returns_500(self, authed_client): """When api result error does NOT contain 'must be between', status is 500.""" cp = _common_patches() error_result = { "error": "database connection failed", "news_items": [], } with ( cp["db_manager"] as mdb, cp["get_user_id"], cp["get_settings_manager"] as msm, patch(f"{MODULE}.api.get_news_feed", return_value=error_result), ): mdb.is_user_connected.return_value = True sm = MagicMock() sm.get_setting.return_value = 20 msm.return_value = sm resp = authed_client.get("/news/api/feed") assert resp.status_code == 500 # --------------------------------------------------------------------------- # POST /subscribe – invalid JSON # --------------------------------------------------------------------------- class TestCreateSubscriptionInvalidJson: """Covers the get_json(force=True) exception branch.""" def test_invalid_json_body_returns_400(self, authed_client): """Malformed JSON body returns 400.""" cp = _common_patches() with cp["db_manager"] as mdb, cp["get_user_id"]: mdb.is_user_connected.return_value = True resp = authed_client.post( "/news/api/subscribe", data="not-json{{{", content_type="application/json", ) assert resp.status_code == 400 # --------------------------------------------------------------------------- # POST /vote – ValueError branches # --------------------------------------------------------------------------- class TestVoteOnNewsValueErrorBranches: """Covers ValueError handling in vote_on_news.""" def test_value_error_not_found_returns_404(self, authed_client): """ValueError with 'not found' yields 404.""" cp = _common_patches() with ( cp["db_manager"] as mdb, cp["get_user_id"], patch( f"{MODULE}.api.submit_feedback", side_effect=ValueError("Card not found"), ), ): mdb.is_user_connected.return_value = True resp = authed_client.post( "/news/api/vote", json={"card_id": "c1", "vote": "up"}, ) assert resp.status_code == 404 assert "not found" in resp.get_json()["error"].lower() def test_value_error_other_returns_400(self, authed_client): """ValueError without 'not found' yields 400.""" cp = _common_patches() with ( cp["db_manager"] as mdb, cp["get_user_id"], patch( f"{MODULE}.api.submit_feedback", side_effect=ValueError("vote must be up or down"), ), ): mdb.is_user_connected.return_value = True resp = authed_client.post( "/news/api/vote", json={"card_id": "c1", "vote": "invalid"}, ) assert resp.status_code == 400 # --------------------------------------------------------------------------- # POST /feedback/ – ValueError 'must be' branch # --------------------------------------------------------------------------- class TestSubmitFeedbackValueErrors: """Covers ValueError branches in submit_feedback route.""" def test_must_be_error_returns_400(self, authed_client): """ValueError with 'must be' in message yields 400.""" cp = _common_patches() with ( cp["db_manager"] as mdb, cp["get_user_id"], patch( f"{MODULE}.api.submit_feedback", side_effect=ValueError("vote must be 'up' or 'down'"), ), ): mdb.is_user_connected.return_value = True resp = authed_client.post( "/news/api/feedback/card-1", json={"vote": "sideways"}, ) assert resp.status_code == 400 assert "Invalid input value" in resp.get_json()["error"] def test_not_found_error_returns_404(self, authed_client): """ValueError with 'not found' yields 404.""" cp = _common_patches() with ( cp["db_manager"] as mdb, cp["get_user_id"], patch( f"{MODULE}.api.submit_feedback", side_effect=ValueError("Card not found in database"), ), ): mdb.is_user_connected.return_value = True resp = authed_client.post( "/news/api/feedback/card-1", json={"vote": "up"}, ) assert resp.status_code == 404 # --------------------------------------------------------------------------- # PUT /subscriptions/ – update error branches # --------------------------------------------------------------------------- class TestUpdateSubscriptionErrorBranches: """Covers error-in-result branching in update_subscription.""" def test_api_error_not_found_returns_404(self, authed_client): """When api.update_subscription result has 'not found' error -> 404.""" cp = _common_patches() with ( cp["db_manager"] as mdb, cp["get_user_id"], patch( f"{MODULE}.api.update_subscription", return_value={"error": "Subscription not found"}, ), ): mdb.is_user_connected.return_value = True resp = authed_client.put( "/news/api/subscriptions/sub-1", json={"query": "new query"}, ) assert resp.status_code == 404 def test_api_error_other_returns_400(self, authed_client): """When api.update_subscription result has other error -> 400.""" cp = _common_patches() with ( cp["db_manager"] as mdb, cp["get_user_id"], patch( f"{MODULE}.api.update_subscription", return_value={"error": "Invalid field value"}, ), ): mdb.is_user_connected.return_value = True resp = authed_client.put( "/news/api/subscriptions/sub-1", json={"query": "new query"}, ) assert resp.status_code == 400 # --------------------------------------------------------------------------- # POST /scheduler/stop – branches # --------------------------------------------------------------------------- class TestStopSchedulerBranches: """Covers stop_scheduler edge cases.""" def test_scheduler_not_running_returns_200(self, authed_client): """Scheduler exists but is not running.""" cp = _common_patches() mock_scheduler = MagicMock() mock_scheduler.is_running = False with ( cp["db_manager"] as mdb, patch(f"{MODULE}.get_env_setting", return_value=True), ): mdb.is_user_connected.return_value = True with authed_client.application.test_request_context(): authed_client.application.background_job_scheduler = ( mock_scheduler ) resp = authed_client.post("/news/api/scheduler/stop") assert resp.status_code == 200 data = resp.get_json() assert "not running" in data.get("message", "").lower() def test_no_scheduler_instance_returns_404(self, authed_client): """No news_scheduler attribute on app.""" cp = _common_patches() with ( cp["db_manager"] as mdb, patch(f"{MODULE}.get_env_setting", return_value=True), ): mdb.is_user_connected.return_value = True # Ensure no news_scheduler attribute if hasattr(authed_client.application, "background_job_scheduler"): delattr(authed_client.application, "background_job_scheduler") resp = authed_client.post("/news/api/scheduler/stop") assert resp.status_code == 404 data = resp.get_json() assert "no scheduler" in data.get("message", "").lower() # --------------------------------------------------------------------------- # POST /scheduler/check-now – branches # --------------------------------------------------------------------------- class TestCheckSubscriptionsNowBranches: """Covers check_subscriptions_now edge cases.""" def test_no_scheduler_returns_503(self, authed_client): """No scheduler attribute on app returns 503.""" cp = _common_patches() with ( cp["db_manager"] as mdb, patch(f"{MODULE}.get_env_setting", return_value=True), ): mdb.is_user_connected.return_value = True if hasattr(authed_client.application, "background_job_scheduler"): delattr(authed_client.application, "background_job_scheduler") resp = authed_client.post("/news/api/scheduler/check-now") assert resp.status_code == 503 def test_scheduler_not_running_returns_503(self, authed_client): """Scheduler exists but not running returns 503.""" cp = _common_patches() mock_scheduler = MagicMock() mock_scheduler.is_running = False with ( cp["db_manager"] as mdb, patch(f"{MODULE}.get_env_setting", return_value=True), ): mdb.is_user_connected.return_value = True authed_client.application.background_job_scheduler = mock_scheduler resp = authed_client.post("/news/api/scheduler/check-now") assert resp.status_code == 503 # --------------------------------------------------------------------------- # POST /scheduler/cleanup-now – scheduler not running # --------------------------------------------------------------------------- class TestTriggerCleanupBranches: """Covers trigger_cleanup edge cases.""" def test_scheduler_not_running_returns_400(self, authed_client): """Scheduler not running returns 400.""" cp = _common_patches() mock_scheduler = MagicMock() mock_scheduler.is_running = False with ( cp["db_manager"] as mdb, patch(f"{MODULE}.get_env_setting", return_value=True), patch( f"{MODULE}.get_background_job_scheduler" if False else "local_deep_research.scheduler.background.get_background_job_scheduler", return_value=mock_scheduler, ), ): mdb.is_user_connected.return_value = True resp = authed_client.post("/news/api/scheduler/cleanup-now") assert resp.status_code == 400 assert "not running" in resp.get_json()["error"].lower() # --------------------------------------------------------------------------- # GET /scheduler/status – APScheduler get_jobs exception # --------------------------------------------------------------------------- class TestSchedulerStatusApschedulerException: """Covers the exception path when get_jobs raises.""" def test_apscheduler_exception_sets_count_zero(self, authed_client): """When scheduler.scheduler.get_jobs() raises, apscheduler_job_count=0.""" cp = _common_patches() mock_scheduler = MagicMock() mock_scheduler.is_running = True mock_scheduler.config = {} mock_scheduler.user_sessions = {"testuser": {"scheduled_jobs": set()}} mock_scheduler.scheduler.get_jobs.side_effect = RuntimeError( "lock error" ) with ( cp["db_manager"] as mdb, patch( "local_deep_research.scheduler.background.get_background_job_scheduler", return_value=mock_scheduler, ), patch(f"{MODULE}.get_env_setting", return_value=True), ): mdb.is_user_connected.return_value = True resp = authed_client.get("/news/api/scheduler/status") assert resp.status_code == 200 data = resp.get_json() assert data["apscheduler_job_count"] == 0 # --------------------------------------------------------------------------- # GET /scheduler/users – show_all=False filtering # --------------------------------------------------------------------------- class TestGetActiveUsersFiltering: """Covers user filtering when show_all=False.""" def test_only_current_user_returned(self, authed_client): """When show_all=False, only current user is returned.""" cp = _common_patches() mock_scheduler = MagicMock() mock_scheduler.get_user_sessions_summary.return_value = [ {"user_id": "testuser", "jobs": 2}, {"user_id": "otheruser", "jobs": 5}, ] with ( cp["db_manager"] as mdb, patch( "local_deep_research.scheduler.background.get_background_job_scheduler", return_value=mock_scheduler, ), patch(f"{MODULE}.get_env_setting", return_value=False), ): mdb.is_user_connected.return_value = True resp = authed_client.get("/news/api/scheduler/users") assert resp.status_code == 200 data = resp.get_json() assert data["active_users"] == 1 assert all(u["user_id"] == "testuser" for u in data["users"]) # --------------------------------------------------------------------------- # GET /search-history – unauthenticated # --------------------------------------------------------------------------- class TestSearchHistoryUnauthenticated: """Covers unauthenticated paths in search-history endpoints.""" def test_get_history_no_user_returns_empty(self, authed_client): """When current_user returns None, empty history is returned.""" cp = _common_patches() with ( cp["db_manager"] as mdb, patch(f"{MODULE}.get_user_id", return_value=None), patch( "local_deep_research.web.auth.decorators.current_user", return_value=None, ), ): mdb.is_user_connected.return_value = True resp = authed_client.get("/news/api/search-history") assert resp.status_code == 200 data = resp.get_json() assert data["search_history"] == [] def test_add_history_no_user_returns_401(self, authed_client): """When current_user returns None, add returns 401.""" cp = _common_patches() with ( cp["db_manager"] as mdb, patch( "local_deep_research.web.auth.decorators.current_user", return_value=None, ), ): mdb.is_user_connected.return_value = True resp = authed_client.post( "/news/api/search-history", json={"query": "test"}, ) assert resp.status_code == 401 def test_clear_history_no_user_returns_success(self, authed_client): """When current_user returns None, clear returns success.""" cp = _common_patches() with ( cp["db_manager"] as mdb, patch( "local_deep_research.web.auth.decorators.current_user", return_value=None, ), ): mdb.is_user_connected.return_value = True resp = authed_client.delete("/news/api/search-history") assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "success" # --------------------------------------------------------------------------- # GET /scheduler/stats – last_activity None branch # --------------------------------------------------------------------------- class TestSchedulerStatsLastActivityNone: """Covers scheduler_stats when last_activity is None.""" def test_last_activity_none_serialized(self, authed_client): """When last_activity is None, it serializes as null.""" cp = _common_patches() mock_scheduler = MagicMock() mock_scheduler.is_running = True mock_scheduler.user_sessions = { "testuser": { "last_activity": None, "scheduled_jobs": set(), } } mock_scheduler._credential_store.retrieve.return_value = None mock_scheduler.scheduler = None # no APScheduler with ( cp["db_manager"] as mdb, patch( "local_deep_research.scheduler.background.get_background_job_scheduler", return_value=mock_scheduler, ), patch(f"{MODULE}.get_env_setting", return_value=True), ): mdb.is_user_connected.return_value = True resp = authed_client.get("/news/api/scheduler/stats") assert resp.status_code == 200 data = resp.get_json() user_info = data["user_sessions"]["testuser"] assert user_info["last_activity"] is None assert user_info["has_password"] is False # --------------------------------------------------------------------------- # DELETE /subscriptions/ – success and not-found # --------------------------------------------------------------------------- class TestDeleteSubscriptionBranches: """Covers delete_subscription success and not-found paths.""" def test_successful_delete(self, authed_client): """Successful deletion returns success message.""" cp = _common_patches() with ( cp["db_manager"] as mdb, cp["get_user_id"], patch(f"{MODULE}.api.delete_subscription", return_value=True), ): mdb.is_user_connected.return_value = True resp = authed_client.delete("/news/api/subscriptions/sub-1") assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "success" assert "sub-1" in data["message"] def test_not_found_returns_404(self, authed_client): """When delete returns False, 404 is returned.""" cp = _common_patches() with ( cp["db_manager"] as mdb, cp["get_user_id"], patch(f"{MODULE}.api.delete_subscription", return_value=False), ): mdb.is_user_connected.return_value = True resp = authed_client.delete("/news/api/subscriptions/sub-999") assert resp.status_code == 404 # --------------------------------------------------------------------------- # GET /subscriptions/current – error in result # --------------------------------------------------------------------------- class TestGetCurrentUserSubscriptionsErrorInResult: """Covers the error-in-result branch for get_current_user_subscriptions.""" def test_api_error_returns_500(self, authed_client): """When api.get_subscriptions returns error dict, returns 500.""" cp = _common_patches() with ( cp["db_manager"] as mdb, cp["get_user_id"], patch( f"{MODULE}.api.get_subscriptions", return_value={"error": "DB connection lost"}, ), ): mdb.is_user_connected.return_value = True resp = authed_client.get("/news/api/subscriptions/current") assert resp.status_code == 500 assert ( "Failed to retrieve subscriptions" in resp.get_json()["error"] ) # --------------------------------------------------------------------------- # POST /research/ – exception # --------------------------------------------------------------------------- class TestResearchNewsItemException: """Covers exception path in research_news_item.""" def test_exception_returns_500(self, authed_client): """Exception during research returns 500 with safe message.""" cp = _common_patches() with ( cp["db_manager"] as mdb, cp["get_user_id"], patch( f"{MODULE}.api.research_news_item", side_effect=RuntimeError("LLM unavailable"), ), ): mdb.is_user_connected.return_value = True resp = authed_client.post( "/news/api/research/card-1", json={"depth": "detailed"}, ) assert resp.status_code == 500 data = resp.get_json() assert "error" in data # Internal message should not leak assert "LLM unavailable" not in data["error"] # --------------------------------------------------------------------------- # GET /subscription/stats – exception # --------------------------------------------------------------------------- class TestSubscriptionStatsException: """Covers exception path in get_subscription_stats.""" def test_exception_returns_500(self, authed_client): """Exception during stats retrieval returns 500.""" cp = _common_patches() with ( cp["db_manager"] as mdb, cp["get_user_id"], patch( f"{MODULE}.get_user_db_session", side_effect=RuntimeError("DB locked"), ), ): mdb.is_user_connected.return_value = True resp = authed_client.get("/news/api/subscription/stats") assert resp.status_code == 500 assert "error" in resp.get_json()