""" Comprehensive tests for local_deep_research.news.flask_api endpoints. These tests properly mock auth decorators and the underlying api module functions to exercise actual endpoint logic, not just status codes. """ from unittest.mock import MagicMock, patch import pytest from flask import Flask # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _auth_session(client, username="testuser"): """Set up a valid authenticated session.""" with client.session_transaction() as sess: sess["username"] = username def _auth_patches(): """Return a dict of common patches needed for auth + db plumbing.""" return { "db_manager": patch( "local_deep_research.web.auth.decorators.db_manager" ), "get_user_id": patch( "local_deep_research.news.flask_api.get_user_id", return_value="testuser", ), "get_settings_manager": patch( "local_deep_research.news.flask_api.get_settings_manager", ), } # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def app(): app = Flask(__name__) app.config["SECRET_KEY"] = "test-secret-key" 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 @pytest.fixture def client(app): return app.test_client() @pytest.fixture def authed_client(client): """Return client with auth session pre-configured.""" _auth_session(client) return client # --------------------------------------------------------------------------- # 1. GET /feed (get_news_feed) # --------------------------------------------------------------------------- class TestGetNewsFeed: """Tests for the /feed endpoint.""" def test_returns_news_items(self, authed_client): patches = _auth_patches() mock_result = { "news_items": [{"id": "1", "title": "Test News"}], "total": 1, } with ( patches["db_manager"] as mock_db, patches["get_user_id"], patches["get_settings_manager"] as mock_sm, patch( "local_deep_research.news.flask_api.api.get_news_feed", return_value=mock_result, ) as mock_feed, ): mock_db.is_user_connected.return_value = True mock_sm_instance = MagicMock() mock_sm_instance.get_setting.return_value = 20 mock_sm.return_value = mock_sm_instance resp = authed_client.get("/news/api/feed") assert resp.status_code == 200 data = resp.get_json() assert len(data["news_items"]) == 1 assert data["news_items"][0]["title"] == "Test News" mock_feed.assert_called_once_with( user_id="testuser", limit=20, use_cache=True, focus=None, search_strategy=None, subscription_id=None, ) def test_with_query_params(self, authed_client): patches = _auth_patches() mock_result = {"news_items": [], "total": 0} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patches["get_settings_manager"] as mock_sm, patch( "local_deep_research.news.flask_api.api.get_news_feed", return_value=mock_result, ) as mock_feed, ): mock_db.is_user_connected.return_value = True mock_sm_instance = MagicMock() mock_sm_instance.get_setting.return_value = 20 mock_sm.return_value = mock_sm_instance resp = authed_client.get( "/news/api/feed?limit=5&strategy=latest&subscription_id=sub1&focus=tech" ) assert resp.status_code == 200 mock_feed.assert_called_once_with( user_id="testuser", limit=5, use_cache=True, focus="tech", search_strategy="latest", subscription_id="sub1", ) def test_use_cache_false(self, authed_client): patches = _auth_patches() mock_result = {"news_items": []} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patches["get_settings_manager"] as mock_sm, patch( "local_deep_research.news.flask_api.api.get_news_feed", return_value=mock_result, ) as mock_feed, ): mock_db.is_user_connected.return_value = True mock_sm_instance = MagicMock() mock_sm_instance.get_setting.return_value = 20 mock_sm.return_value = mock_sm_instance resp = authed_client.get("/news/api/feed?use_cache=false") assert resp.status_code == 200 mock_feed.assert_called_once() assert mock_feed.call_args.kwargs["use_cache"] is False def test_api_error_in_result(self, authed_client): patches = _auth_patches() mock_result = { "error": "limit must be between 1 and 100", "news_items": [], } with ( patches["db_manager"] as mock_db, patches["get_user_id"], patches["get_settings_manager"] as mock_sm, patch( "local_deep_research.news.flask_api.api.get_news_feed", return_value=mock_result, ), ): mock_db.is_user_connected.return_value = True mock_sm_instance = MagicMock() mock_sm_instance.get_setting.return_value = 20 mock_sm.return_value = mock_sm_instance resp = authed_client.get("/news/api/feed?limit=-1") assert resp.status_code == 400 data = resp.get_json() assert "error" in data def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patches["get_settings_manager"] as mock_sm, patch( "local_deep_research.news.flask_api.api.get_news_feed", side_effect=RuntimeError("boom"), ), ): mock_db.is_user_connected.return_value = True mock_sm_instance = MagicMock() mock_sm_instance.get_setting.return_value = 20 mock_sm.return_value = mock_sm_instance resp = authed_client.get("/news/api/feed") assert resp.status_code == 500 data = resp.get_json() assert "error" in data assert data["news_items"] == [] def test_unauthenticated_is_blocked(self, client): """Without session username, login_required returns JSON 401 for the /news/api/feed API endpoint.""" resp = client.get("/news/api/feed") assert resp.status_code == 401 assert resp.get_json()["error"] == "Authentication required" # --------------------------------------------------------------------------- # 2. POST /subscribe (create_subscription) # --------------------------------------------------------------------------- class TestCreateSubscription: """Tests for the /subscribe endpoint.""" def test_successful_creation(self, authed_client): patches = _auth_patches() mock_result = {"id": "sub-123", "status": "active", "query": "AI news"} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.create_subscription", return_value=mock_result, ) as mock_create, ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/subscribe", json={"query": "AI news", "subscription_type": "search"}, ) assert resp.status_code == 200 data = resp.get_json() assert data["id"] == "sub-123" mock_create.assert_called_once() call_kwargs = mock_create.call_args.kwargs assert call_kwargs["query"] == "AI news" assert call_kwargs["user_id"] == "testuser" assert call_kwargs["subscription_type"] == "search" def test_missing_query_returns_400(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/subscribe", json={"subscription_type": "search"}, ) assert resp.status_code == 400 data = resp.get_json() assert "query" in data["error"].lower() def test_no_json_body_returns_400(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/subscribe", data="not json", content_type="text/plain", ) assert resp.status_code == 400 def test_with_all_optional_fields(self, authed_client): patches = _auth_patches() mock_result = {"id": "sub-456", "status": "active"} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.create_subscription", return_value=mock_result, ) as mock_create, ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/subscribe", json={ "query": "ML papers", "subscription_type": "topic", "refresh_minutes": 60, "model_provider": "OPENAI", "model": "gpt-4", "search_strategy": "deep", "custom_endpoint": "https://example.com", "name": "ML Feed", "folder_id": "fold-1", "is_active": True, "search_engine": "google", "search_iterations": 3, "questions_per_iteration": 5, }, ) assert resp.status_code == 200 call_kwargs = mock_create.call_args.kwargs assert call_kwargs["model_provider"] == "openai" assert call_kwargs["name"] == "ML Feed" assert call_kwargs["search_iterations"] == 3 def test_value_error_returns_400(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.create_subscription", side_effect=ValueError("bad value"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/subscribe", json={"query": "test"}, ) assert resp.status_code == 400 # --------------------------------------------------------------------------- # 3. POST /vote (vote_on_news) # --------------------------------------------------------------------------- class TestVoteOnNews: """Tests for the /vote endpoint.""" def test_upvote(self, authed_client): patches = _auth_patches() mock_result = {"status": "success", "vote": "up"} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.submit_feedback", return_value=mock_result, ) as mock_vote, ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/vote", json={"card_id": "card-1", "vote": "up"}, ) assert resp.status_code == 200 mock_vote.assert_called_once_with( card_id="card-1", user_id="testuser", vote="up" ) def test_downvote(self, authed_client): patches = _auth_patches() mock_result = {"status": "success", "vote": "down"} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.submit_feedback", return_value=mock_result, ) as mock_vote, ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/vote", json={"card_id": "card-2", "vote": "down"}, ) assert resp.status_code == 200 mock_vote.assert_called_once_with( card_id="card-2", user_id="testuser", vote="down" ) def test_missing_card_id(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/vote", json={"vote": "up"}, ) assert resp.status_code == 400 assert "required" in resp.get_json()["error"].lower() def test_missing_vote(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/vote", json={"card_id": "card-1"}, ) assert resp.status_code == 400 def test_not_found_error(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.submit_feedback", side_effect=ValueError("Card not found"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/vote", json={"card_id": "bad-id", "vote": "up"}, ) assert resp.status_code == 404 # --------------------------------------------------------------------------- # 4. POST /feedback/batch (get_batch_feedback) # --------------------------------------------------------------------------- class TestGetBatchFeedback: """Tests for the /feedback/batch endpoint.""" def test_get_votes_for_cards(self, authed_client): patches = _auth_patches() mock_result = {"votes": {"c1": "up", "c2": "down"}} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.get_votes_for_cards", return_value=mock_result, ) as mock_batch, ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/feedback/batch", json={"card_ids": ["c1", "c2"]}, ) assert resp.status_code == 200 data = resp.get_json() assert data["votes"]["c1"] == "up" mock_batch.assert_called_once_with( card_ids=["c1", "c2"], user_id="testuser" ) def test_empty_card_ids_returns_empty(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/feedback/batch", json={"card_ids": []}, ) assert resp.status_code == 200 assert resp.get_json()["votes"] == {} def test_missing_card_ids_returns_empty(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/feedback/batch", json={}, ) assert resp.status_code == 200 assert resp.get_json()["votes"] == {} def test_value_error_not_found(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.get_votes_for_cards", side_effect=ValueError("User not found"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/feedback/batch", json={"card_ids": ["c1"]}, ) assert resp.status_code == 404 def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.get_votes_for_cards", side_effect=RuntimeError("db error"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/feedback/batch", json={"card_ids": ["c1"]}, ) assert resp.status_code == 500 # --------------------------------------------------------------------------- # 5. POST /feedback/ (submit_feedback) # --------------------------------------------------------------------------- class TestSubmitFeedback: """Tests for the /feedback/ endpoint.""" def test_submit_vote(self, authed_client): patches = _auth_patches() mock_result = {"status": "recorded"} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.submit_feedback", return_value=mock_result, ) as mock_fb, ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/feedback/card-99", json={"vote": "up"}, ) assert resp.status_code == 200 mock_fb.assert_called_once_with( card_id="card-99", user_id="testuser", vote="up" ) def test_missing_vote_returns_400(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/feedback/card-1", json={"something": "else"}, ) assert resp.status_code == 400 assert "vote" in resp.get_json()["error"].lower() def test_not_found_error(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.submit_feedback", side_effect=ValueError("Card not found"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/feedback/card-bad", json={"vote": "up"}, ) assert resp.status_code == 404 def test_invalid_value_error(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.submit_feedback", side_effect=ValueError("Vote must be up or down"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/feedback/card-1", json={"vote": "invalid"}, ) assert resp.status_code == 400 # --------------------------------------------------------------------------- # 6. GET /subscriptions/current (get_current_user_subscriptions) # --------------------------------------------------------------------------- class TestGetCurrentUserSubscriptions: """Tests for the /subscriptions/current endpoint.""" def test_returns_subscriptions(self, authed_client): patches = _auth_patches() mock_result = { "subscriptions": [ {"id": "s1", "query": "AI"}, {"id": "s2", "query": "ML"}, ] } with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.get_subscriptions", return_value=mock_result, ) as mock_subs, ): mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/subscriptions/current") assert resp.status_code == 200 data = resp.get_json() assert len(data["subscriptions"]) == 2 mock_subs.assert_called_once_with("testuser") def test_api_error_returns_500(self, authed_client): patches = _auth_patches() mock_result = {"error": "Database unavailable"} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.get_subscriptions", return_value=mock_result, ), ): mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/subscriptions/current") assert resp.status_code == 500 def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.get_subscriptions", side_effect=RuntimeError("crash"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/subscriptions/current") assert resp.status_code == 500 # --------------------------------------------------------------------------- # 7. GET /subscriptions/ (get_subscription) # --------------------------------------------------------------------------- class TestGetSubscription: """Tests for the /subscriptions/ endpoint.""" def test_returns_subscription(self, authed_client): patches = _auth_patches() mock_result = {"id": "s1", "query": "AI news", "status": "active"} with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.api.get_subscription", return_value=mock_result, ) as mock_get, ): mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/subscriptions/s1") assert resp.status_code == 200 data = resp.get_json() assert data["id"] == "s1" mock_get.assert_called_once_with("s1") def test_not_found(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.api.get_subscription", return_value=None, ), ): mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/subscriptions/nonexistent") assert resp.status_code == 404 def test_null_id_returns_400(self, authed_client): patches = _auth_patches() with patches["db_manager"] as mock_db: mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/subscriptions/null") assert resp.status_code == 400 def test_undefined_id_returns_400(self, authed_client): patches = _auth_patches() with patches["db_manager"] as mock_db: mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/subscriptions/undefined") assert resp.status_code == 400 def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.api.get_subscription", side_effect=RuntimeError("db error"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/subscriptions/s1") assert resp.status_code == 500 # --------------------------------------------------------------------------- # 8. PUT /subscriptions/ (update_subscription) # --------------------------------------------------------------------------- class TestUpdateSubscription: """Tests for the PUT /subscriptions/ endpoint.""" def test_successful_update(self, authed_client): patches = _auth_patches() mock_result = {"id": "s1", "status": "updated"} with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.api.update_subscription", return_value=mock_result, ) as mock_update, ): mock_db.is_user_connected.return_value = True resp = authed_client.put( "/news/api/subscriptions/s1", json={"query": "Updated query", "name": "New Name"}, ) assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "updated" # Verify field mapping: query -> query_or_topic call_args = mock_update.call_args assert call_args[0][0] == "s1" update_data = call_args[0][1] assert update_data["query_or_topic"] == "Updated query" assert update_data["name"] == "New Name" def test_field_mapping(self, authed_client): """Verify all field mappings work correctly.""" patches = _auth_patches() mock_result = {"id": "s1", "status": "updated"} with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.api.update_subscription", return_value=mock_result, ) as mock_update, ): mock_db.is_user_connected.return_value = True resp = authed_client.put( "/news/api/subscriptions/s1", json={ "refresh_minutes": 30, "is_active": False, "search_engine": "bing", }, ) assert resp.status_code == 200 update_data = mock_update.call_args[0][1] assert update_data["refresh_interval_minutes"] == 30 assert update_data["is_active"] is False assert update_data["search_engine"] == "bing" def test_error_in_result_not_found(self, authed_client): patches = _auth_patches() mock_result = {"error": "Subscription not found"} with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.api.update_subscription", return_value=mock_result, ), ): mock_db.is_user_connected.return_value = True resp = authed_client.put( "/news/api/subscriptions/bad-id", json={"name": "test"}, ) assert resp.status_code == 404 def test_error_in_result_generic(self, authed_client): patches = _auth_patches() mock_result = {"error": "Invalid data provided"} with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.api.update_subscription", return_value=mock_result, ), ): mock_db.is_user_connected.return_value = True resp = authed_client.put( "/news/api/subscriptions/s1", json={"name": "x"}, ) assert resp.status_code == 400 def test_no_json_body_returns_400(self, authed_client): patches = _auth_patches() with patches["db_manager"] as mock_db: mock_db.is_user_connected.return_value = True resp = authed_client.put( "/news/api/subscriptions/s1", data="not json", content_type="text/plain", ) assert resp.status_code == 400 # --------------------------------------------------------------------------- # 9. DELETE /subscriptions/ (delete_subscription) # --------------------------------------------------------------------------- class TestDeleteSubscription: """Tests for the DELETE /subscriptions/ endpoint.""" def test_successful_delete(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.api.delete_subscription", return_value=True, ) as mock_del, ): mock_db.is_user_connected.return_value = True resp = authed_client.delete("/news/api/subscriptions/s1") assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "success" assert "s1" in data["message"] mock_del.assert_called_once_with("s1") def test_not_found(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.api.delete_subscription", return_value=False, ), ): mock_db.is_user_connected.return_value = True resp = authed_client.delete("/news/api/subscriptions/bad-id") assert resp.status_code == 404 def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.api.delete_subscription", side_effect=RuntimeError("db error"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.delete("/news/api/subscriptions/s1") assert resp.status_code == 500 # --------------------------------------------------------------------------- # 10. GET /subscriptions//history (get_subscription_history) # --------------------------------------------------------------------------- class TestGetSubscriptionHistory: """Tests for the /subscriptions//history endpoint.""" def test_returns_history(self, authed_client): patches = _auth_patches() mock_result = { "history": [ {"id": "h1", "date": "2026-01-01"}, {"id": "h2", "date": "2026-01-02"}, ] } with ( patches["db_manager"] as mock_db, patches["get_settings_manager"] as mock_sm, patch( "local_deep_research.news.flask_api.api.get_subscription_history", return_value=mock_result, ) as mock_hist, ): mock_db.is_user_connected.return_value = True mock_sm_instance = MagicMock() mock_sm_instance.get_setting.return_value = 20 mock_sm.return_value = mock_sm_instance resp = authed_client.get("/news/api/subscriptions/s1/history") assert resp.status_code == 200 data = resp.get_json() assert len(data["history"]) == 2 mock_hist.assert_called_once_with("s1", 20) def test_custom_limit(self, authed_client): patches = _auth_patches() mock_result = {"history": []} with ( patches["db_manager"] as mock_db, patches["get_settings_manager"] as mock_sm, patch( "local_deep_research.news.flask_api.api.get_subscription_history", return_value=mock_result, ) as mock_hist, ): mock_db.is_user_connected.return_value = True mock_sm_instance = MagicMock() mock_sm_instance.get_setting.return_value = 20 mock_sm.return_value = mock_sm_instance resp = authed_client.get( "/news/api/subscriptions/s1/history?limit=5" ) assert resp.status_code == 200 mock_hist.assert_called_once_with("s1", 5) def test_error_in_result(self, authed_client): patches = _auth_patches() mock_result = {"error": "Database error", "history": []} with ( patches["db_manager"] as mock_db, patches["get_settings_manager"] as mock_sm, patch( "local_deep_research.news.flask_api.api.get_subscription_history", return_value=mock_result, ), ): mock_db.is_user_connected.return_value = True mock_sm_instance = MagicMock() mock_sm_instance.get_setting.return_value = 20 mock_sm.return_value = mock_sm_instance resp = authed_client.get("/news/api/subscriptions/s1/history") assert resp.status_code == 500 # --------------------------------------------------------------------------- # 11. POST /preferences (save_preferences) # --------------------------------------------------------------------------- class TestSavePreferences: """Tests for the /preferences endpoint.""" def test_save_preferences(self, authed_client): patches = _auth_patches() mock_result = {"status": "saved"} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.save_news_preferences", return_value=mock_result, ) as mock_save, ): mock_db.is_user_connected.return_value = True prefs = {"theme": "dark", "language": "en"} resp = authed_client.post( "/news/api/preferences", json={"preferences": prefs}, ) assert resp.status_code == 200 mock_save.assert_called_once_with("testuser", prefs) def test_empty_preferences(self, authed_client): patches = _auth_patches() mock_result = {"status": "saved"} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.save_news_preferences", return_value=mock_result, ) as mock_save, ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/preferences", json={}, ) assert resp.status_code == 200 mock_save.assert_called_once_with("testuser", {}) def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.api.save_news_preferences", side_effect=RuntimeError("fail"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/preferences", json={"preferences": {"a": 1}}, ) assert resp.status_code == 500 # --------------------------------------------------------------------------- # 12. GET /categories (get_categories) # --------------------------------------------------------------------------- class TestGetCategories: """Tests for the /categories endpoint.""" def test_returns_categories(self, authed_client): mock_result = { "categories": [ {"name": "tech", "count": 10}, {"name": "science", "count": 5}, ] } patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.api.get_news_categories", return_value=mock_result, ) as mock_cats, ): mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/categories") assert resp.status_code == 200 data = resp.get_json() assert len(data["categories"]) == 2 mock_cats.assert_called_once() def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.api.get_news_categories", side_effect=RuntimeError("fail"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/categories") assert resp.status_code == 500 # --------------------------------------------------------------------------- # 13. GET /subscription/folders (get_folders) # --------------------------------------------------------------------------- class TestGetFolders: """Tests for the /subscription/folders GET endpoint.""" def test_returns_folders(self, authed_client): patches = _auth_patches() mock_folder = MagicMock() mock_folder.to_dict.return_value = {"id": "f1", "name": "Work"} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.get_user_db_session" ) as mock_sess, patch("local_deep_research.news.flask_api.FolderManager") as MockFM, ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=MagicMock() ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) MockFM.return_value.get_user_folders.return_value = [mock_folder] resp = authed_client.get("/news/api/subscription/folders") assert resp.status_code == 200 data = resp.get_json() assert len(data) == 1 assert data[0]["name"] == "Work" def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.get_user_db_session", side_effect=RuntimeError("db fail"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/subscription/folders") assert resp.status_code == 500 # --------------------------------------------------------------------------- # 14. POST /subscription/folders (create_folder) # --------------------------------------------------------------------------- class TestCreateFolder: """Tests for the POST /subscription/folders endpoint.""" def test_create_folder(self, authed_client): patches = _auth_patches() mock_folder = MagicMock() mock_folder.to_dict.return_value = { "id": "f-new", "name": "Research", } mock_session = MagicMock() mock_query = MagicMock() mock_query.filter_by.return_value.first.return_value = None mock_session.query.return_value = mock_query with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.get_user_db_session" ) as mock_sess, patch("local_deep_research.news.flask_api.FolderManager") as MockFM, ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=mock_session ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) MockFM.return_value.create_folder.return_value = mock_folder resp = authed_client.post( "/news/api/subscription/folders", json={"name": "Research", "description": "Research topics"}, ) assert resp.status_code == 201 data = resp.get_json() assert data["name"] == "Research" def test_missing_name_returns_400(self, authed_client): patches = _auth_patches() with patches["db_manager"] as mock_db: mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/subscription/folders", json={"description": "no name"}, ) assert resp.status_code == 400 def test_duplicate_folder_returns_409(self, authed_client): patches = _auth_patches() mock_session = MagicMock() mock_existing = MagicMock() mock_query = MagicMock() mock_query.filter_by.return_value.first.return_value = mock_existing mock_session.query.return_value = mock_query with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.get_user_db_session" ) as mock_sess, patch("local_deep_research.news.flask_api.FolderManager"), ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=mock_session ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) resp = authed_client.post( "/news/api/subscription/folders", json={"name": "Existing"}, ) assert resp.status_code == 409 # --------------------------------------------------------------------------- # 15. PUT /subscription/folders/ (update_folder) # --------------------------------------------------------------------------- class TestUpdateFolder: """Tests for the PUT /subscription/folders/ endpoint.""" def test_update_folder(self, authed_client): patches = _auth_patches() mock_folder = MagicMock() mock_folder.to_dict.return_value = {"id": "f1", "name": "Updated"} with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.get_user_db_session" ) as mock_sess, patch("local_deep_research.news.flask_api.FolderManager") as MockFM, ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=MagicMock() ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) MockFM.return_value.update_folder.return_value = mock_folder resp = authed_client.put( "/news/api/subscription/folders/f1", json={"name": "Updated"}, ) assert resp.status_code == 200 data = resp.get_json() assert data["name"] == "Updated" def test_folder_not_found(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.get_user_db_session" ) as mock_sess, patch("local_deep_research.news.flask_api.FolderManager") as MockFM, ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=MagicMock() ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) MockFM.return_value.update_folder.return_value = None resp = authed_client.put( "/news/api/subscription/folders/bad-id", json={"name": "X"}, ) assert resp.status_code == 404 # --------------------------------------------------------------------------- # 16. DELETE /subscription/folders/ (delete_folder) # --------------------------------------------------------------------------- class TestDeleteFolder: """Tests for the DELETE /subscription/folders/ endpoint.""" def test_delete_folder(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.get_user_db_session" ) as mock_sess, patch("local_deep_research.news.flask_api.FolderManager") as MockFM, ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=MagicMock() ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) MockFM.return_value.delete_folder.return_value = True resp = authed_client.delete("/news/api/subscription/folders/f1") assert resp.status_code == 200 assert resp.get_json()["status"] == "deleted" def test_delete_folder_not_found(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.get_user_db_session" ) as mock_sess, patch("local_deep_research.news.flask_api.FolderManager") as MockFM, ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=MagicMock() ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) MockFM.return_value.delete_folder.return_value = False resp = authed_client.delete("/news/api/subscription/folders/bad-id") assert resp.status_code == 404 def test_delete_folder_with_move_to(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.get_user_db_session" ) as mock_sess, patch("local_deep_research.news.flask_api.FolderManager") as MockFM, ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=MagicMock() ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) MockFM.return_value.delete_folder.return_value = True resp = authed_client.delete( "/news/api/subscription/folders/f1?move_to=f2" ) assert resp.status_code == 200 MockFM.return_value.delete_folder.assert_called_once_with( "f1", "f2" ) # --------------------------------------------------------------------------- # 17. GET /subscription/subscriptions/organized (get_subscriptions_organized) # --------------------------------------------------------------------------- class TestGetSubscriptionsOrganized: """Tests for the /subscription/subscriptions/organized endpoint.""" def test_returns_organized(self, authed_client): # get_subscriptions_by_folder returns plain dicts in a # {"folders": [{"folder": {...}, "subscriptions": [...]}], # "uncategorized": [...]} shape; the route flattens to # {folder_name: [subscription, ...]}. (Calling .to_dict() on those # dicts previously 500'd the route.) patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.get_user_db_session" ) as mock_sess, patch("local_deep_research.news.flask_api.FolderManager") as MockFM, ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=MagicMock() ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) MockFM.return_value.get_subscriptions_by_folder.return_value = { "folders": [ { "folder": {"id": "f1", "name": "Work"}, "subscriptions": [{"id": "s1", "query_or_topic": "AI"}], }, ], "uncategorized": [], } resp = authed_client.get( "/news/api/subscription/subscriptions/organized" ) assert resp.status_code == 200 data = resp.get_json() assert data["Work"] == [{"id": "s1", "query_or_topic": "AI"}] assert data["uncategorized"] == [] def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.get_user_db_session", side_effect=RuntimeError("db fail"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.get( "/news/api/subscription/subscriptions/organized" ) assert resp.status_code == 500 # --------------------------------------------------------------------------- # 18. GET /subscription/stats (get_subscription_stats) # --------------------------------------------------------------------------- class TestGetSubscriptionStats: """Tests for the /subscription/stats endpoint.""" def test_returns_stats(self, authed_client): patches = _auth_patches() mock_stats = {"total": 5, "active": 3, "paused": 2} with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.get_user_db_session" ) as mock_sess, patch("local_deep_research.news.flask_api.FolderManager") as MockFM, ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=MagicMock() ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) MockFM.return_value.get_subscription_stats.return_value = mock_stats resp = authed_client.get("/news/api/subscription/stats") assert resp.status_code == 200 data = resp.get_json() assert data["total"] == 5 assert data["active"] == 3 def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patches["get_user_id"], patch( "local_deep_research.news.flask_api.get_user_db_session", side_effect=RuntimeError("db fail"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/subscription/stats") assert resp.status_code == 500 # --------------------------------------------------------------------------- # 19. GET /search-history (get_search_history) # --------------------------------------------------------------------------- class TestGetSearchHistory: """Tests for the GET /search-history endpoint.""" def test_returns_history(self, authed_client): patches = _auth_patches() mock_item = MagicMock() mock_item.to_dict.return_value = { "id": 1, "query": "test", "type": "filter", } mock_db_session = MagicMock() mock_db_session.query.return_value.order_by.return_value.limit.return_value.all.return_value = [ mock_item ] with ( patches["db_manager"] as mock_db, patch( "local_deep_research.web.auth.decorators.current_user", return_value="testuser", ), patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_sess, ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=mock_db_session ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) resp = authed_client.get("/news/api/search-history") assert resp.status_code == 200 data = resp.get_json() assert len(data["search_history"]) == 1 assert data["search_history"][0]["query"] == "test" def test_unauthenticated_returns_empty(self, authed_client): """When current_user returns None, should return empty list.""" patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.web.auth.decorators.current_user", return_value=None, ), ): mock_db.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_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.web.auth.decorators.current_user", return_value="testuser", ), patch( "local_deep_research.database.session_context.get_user_db_session", side_effect=RuntimeError("db error"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/search-history") assert resp.status_code == 500 # --------------------------------------------------------------------------- # 20. POST /search-history (add_search_history) # --------------------------------------------------------------------------- class TestAddSearchHistory: """Tests for the POST /search-history endpoint.""" def test_add_search(self, authed_client): patches = _auth_patches() mock_db_session = MagicMock() mock_history_instance = MagicMock() mock_history_instance.id = 42 with ( patches["db_manager"] as mock_db, patch( "local_deep_research.web.auth.decorators.current_user", return_value="testuser", ), patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_sess, patch( "local_deep_research.database.models.UserNewsSearchHistory", return_value=mock_history_instance, ), ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=mock_db_session ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) resp = authed_client.post( "/news/api/search-history", json={ "query": "test search", "type": "filter", "resultCount": 10, }, ) assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "success" mock_db_session.add.assert_called_once() mock_db_session.commit.assert_called_once() def test_missing_query_returns_400(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.web.auth.decorators.current_user", return_value="testuser", ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/search-history", json={"type": "filter"}, ) assert resp.status_code == 400 def test_no_body_returns_400(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.web.auth.decorators.current_user", return_value="testuser", ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/search-history", json={}, ) assert resp.status_code == 400 def test_unauthenticated_returns_401(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.web.auth.decorators.current_user", return_value=None, ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/search-history", json={"query": "test"}, ) assert resp.status_code == 401 # --------------------------------------------------------------------------- # 21. DELETE /search-history (clear_search_history) # --------------------------------------------------------------------------- class TestClearSearchHistory: """Tests for the DELETE /search-history endpoint.""" def test_clear_history(self, authed_client): patches = _auth_patches() mock_db_session = MagicMock() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.web.auth.decorators.current_user", return_value="testuser", ), patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_sess, ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=mock_db_session ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) resp = authed_client.delete("/news/api/search-history") assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "success" mock_db_session.query.return_value.delete.assert_called_once() mock_db_session.commit.assert_called_once() def test_unauthenticated_returns_success(self, authed_client): """When current_user returns None, endpoint returns success (no-op).""" patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.web.auth.decorators.current_user", return_value=None, ), ): mock_db.is_user_connected.return_value = True resp = authed_client.delete("/news/api/search-history") assert resp.status_code == 200 assert resp.get_json()["status"] == "success" def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.web.auth.decorators.current_user", return_value="testuser", ), patch( "local_deep_research.database.session_context.get_user_db_session", side_effect=RuntimeError("db error"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.delete("/news/api/search-history") assert resp.status_code == 500 # --------------------------------------------------------------------------- # 22. POST /check-overdue (check_overdue_subscriptions) # --------------------------------------------------------------------------- class TestCheckOverdueSubscriptions: """Tests for the /check-overdue endpoint.""" def test_no_overdue_subscriptions(self, authed_client): patches = _auth_patches() mock_db_session = MagicMock() mock_db_session.query.return_value.filter.return_value.all.return_value = [] with ( patches["db_manager"] as mock_db, patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_sess, ): mock_db.is_user_connected.return_value = True mock_sess.return_value.__enter__ = MagicMock( return_value=mock_db_session ) mock_sess.return_value.__exit__ = MagicMock(return_value=False) resp = authed_client.post("/news/api/check-overdue") assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "success" assert data["overdue_found"] == 0 assert data["started"] == 0 def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.database.session_context.get_user_db_session", side_effect=RuntimeError("db error"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post("/news/api/check-overdue") assert resp.status_code == 500 # --------------------------------------------------------------------------- # Additional: Scheduler endpoints # --------------------------------------------------------------------------- class TestSchedulerStatus: """Tests for the /scheduler/status endpoint.""" def test_returns_status(self, authed_client): patches = _auth_patches() mock_scheduler = MagicMock() mock_scheduler.is_running = True mock_scheduler.config = {"interval": 60} mock_scheduler.user_sessions = {"testuser": {"scheduled_jobs": {"j1"}}} mock_scheduler.scheduler = None with ( patches["db_manager"] as mock_db, patch( "local_deep_research.scheduler.background.get_background_job_scheduler", return_value=mock_scheduler, ), patch( "local_deep_research.news.flask_api.get_env_setting", return_value=False, ), ): mock_db.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["is_running"] is True assert data["scheduler_available"] is True def test_exception_returns_500(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.scheduler.background.get_background_job_scheduler", side_effect=RuntimeError("fail"), ), ): mock_db.is_user_connected.return_value = True resp = authed_client.get("/news/api/scheduler/status") assert resp.status_code == 500 class TestSchedulerStartStop: """Tests for scheduler start/stop endpoints.""" def test_start_scheduler(self, authed_client): patches = _auth_patches() mock_scheduler = MagicMock() mock_scheduler.is_running = False mock_scheduler.user_sessions = {} with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.get_env_setting", return_value=True, ), patch( "local_deep_research.scheduler.background.get_background_job_scheduler", return_value=mock_scheduler, ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post("/news/api/scheduler/start") assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "success" mock_scheduler.start.assert_called_once() def test_start_already_running(self, authed_client): patches = _auth_patches() mock_scheduler = MagicMock() mock_scheduler.is_running = True with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.get_env_setting", return_value=True, ), patch( "local_deep_research.scheduler.background.get_background_job_scheduler", return_value=mock_scheduler, ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post("/news/api/scheduler/start") assert resp.status_code == 200 assert "already running" in resp.get_json()["message"].lower() def test_start_blocked_by_setting(self, authed_client): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.get_env_setting", return_value=False, ), ): mock_db.is_user_connected.return_value = True resp = authed_client.post("/news/api/scheduler/start") assert resp.status_code == 403 def test_stop_scheduler(self, authed_client, app): patches = _auth_patches() mock_scheduler = MagicMock() mock_scheduler.is_running = True with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.get_env_setting", return_value=True, ), ): mock_db.is_user_connected.return_value = True app.background_job_scheduler = mock_scheduler resp = authed_client.post("/news/api/scheduler/stop") assert resp.status_code == 200 mock_scheduler.stop.assert_called_once() def test_stop_no_scheduler(self, authed_client, app): patches = _auth_patches() with ( patches["db_manager"] as mock_db, patch( "local_deep_research.news.flask_api.get_env_setting", return_value=True, ), ): mock_db.is_user_connected.return_value = True # Ensure no background_job_scheduler attribute if hasattr(app, "background_job_scheduler"): delattr(app, "background_job_scheduler") resp = authed_client.post("/news/api/scheduler/stop") assert resp.status_code == 404 # --------------------------------------------------------------------------- # Edge case: require_json_body decorator # --------------------------------------------------------------------------- class TestRequireJsonBody: """Test that endpoints with @require_json_body reject non-JSON requests.""" def test_vote_without_json(self, authed_client): patches = _auth_patches() with patches["db_manager"] as mock_db: mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/vote", data="plain text", content_type="text/plain", ) assert resp.status_code == 400 def test_feedback_batch_without_json(self, authed_client): patches = _auth_patches() with patches["db_manager"] as mock_db: mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/feedback/batch", data="plain text", content_type="text/plain", ) assert resp.status_code == 400 def test_preferences_without_json(self, authed_client): patches = _auth_patches() with patches["db_manager"] as mock_db: mock_db.is_user_connected.return_value = True resp = authed_client.post( "/news/api/preferences", data="plain text", content_type="text/plain", ) assert resp.status_code == 400