""" Tests for news/web.py - Flask blueprint for news system web routes. Tests cover: - create_news_blueprint: Blueprint creation and route registration - Page routes: news, subscriptions - new_subscription_page: Form rendering with default settings - edit_subscription_page: Subscription loading and error handling - health_check: Database connectivity checks - load_user_settings: User settings loading from database """ import pytest from unittest.mock import MagicMock, patch from flask import Flask class _AlwaysAuthenticated: def __contains__(self, key): return True def __getitem__(self, key): return "testuser" @pytest.fixture(autouse=True) def _bypass_login_required(monkeypatch): # These tests exercise route logic, not auth. PR #3129 added # @login_required to news routes; auth itself is covered in tests/web/auth/. monkeypatch.setattr( "local_deep_research.web.auth.decorators.session", _AlwaysAuthenticated(), ) monkeypatch.setattr( "local_deep_research.web.auth.decorators.db_manager.is_user_connected", lambda *args, **kwargs: True, ) @pytest.fixture def app(): """Create a Flask app for testing.""" app = Flask(__name__) app.config["SECRET_KEY"] = "test-secret-key" app.config["WTF_CSRF_ENABLED"] = False app.config["TESTING"] = True # Register the news blueprint from local_deep_research.news.web import create_news_blueprint bp = create_news_blueprint() app.register_blueprint(bp, url_prefix="/news") return app @pytest.fixture def client(app): """Create a test client.""" return app.test_client() class TestCreateNewsBlueprint: """Tests for create_news_blueprint function.""" def test_blueprint_created_successfully(self): """Test that blueprint is created with correct name.""" from local_deep_research.news.web import create_news_blueprint bp = create_news_blueprint() assert bp.name == "news" def test_blueprint_registers_flask_api(self): """Test that news_api_bp is registered as sub-blueprint.""" from local_deep_research.news.web import create_news_blueprint bp = create_news_blueprint() # Check that the API blueprint is registered # _blueprints is a list of (blueprint, options) tuples registered_names = [b[0].name for b in bp._blueprints] assert "news_api" in registered_names def test_blueprint_has_expected_routes(self, app): """Test that all expected routes are registered.""" expected_routes = [ "/news/", "/news/subscriptions", "/news/subscriptions/new", "/news/subscriptions//edit", "/news/health", ] registered_rules = [rule.rule for rule in app.url_map.iter_rules()] for route in expected_routes: assert route in registered_rules, f"Route {route} not found" class TestNewsPageRoute: """Tests for news_page route.""" def test_news_page_returns_200(self, client): """Test that news page returns 200 status.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" response = client.get("/news/") assert response.status_code == 200 def test_news_page_renders_correct_template(self, client): """Test that correct template is rendered.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" client.get("/news/") mock_render.assert_called_once() args, kwargs = mock_render.call_args assert args[0] == "pages/news.html" def test_news_page_passes_strategies_list(self, client): """Test that strategies are passed to template.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" client.get("/news/") args, kwargs = mock_render.call_args assert "strategies" in kwargs assert len(kwargs["strategies"]) >= 5 strategy_names = [s["name"] for s in kwargs["strategies"]] assert "source-based" in strategy_names assert "focused-iteration" in strategy_names assert "focused-iteration-standard" in strategy_names def test_news_page_passes_strategies_with_source_based(self, client): """Test that source-based strategy is present in strategies.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" client.get("/news/") args, kwargs = mock_render.call_args strategy_names = [s["name"] for s in kwargs["strategies"]] assert "source-based" in strategy_names class TestSubscriptionsPageRoute: """Tests for subscriptions_page route.""" def test_subscriptions_page_returns_200(self, client): """Test that subscriptions page returns 200 status.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" response = client.get("/news/subscriptions") assert response.status_code == 200 def test_subscriptions_page_renders_correct_template(self, client): """Test that correct template is rendered.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" client.get("/news/subscriptions") mock_render.assert_called_once_with("pages/subscriptions.html") class TestNewSubscriptionPageRoute: """Tests for new_subscription_page route.""" def test_new_subscription_page_returns_200(self, client): """Test that new subscription page returns 200 status.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" response = client.get("/news/subscriptions/new") assert response.status_code == 200 def test_new_subscription_page_renders_correct_template(self, client): """Test that correct template is rendered.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" client.get("/news/subscriptions/new") args, kwargs = mock_render.call_args assert args[0] == "pages/news-subscription-form.html" def test_new_subscription_page_anonymous_user_uses_defaults(self, client): """Test that anonymous user doesn't trigger DB lookup.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" # Session doesn't have username, defaults to "anonymous" client.get("/news/subscriptions/new") # No DB session call expected for anonymous user mock_render.assert_called_once() def test_new_subscription_page_passes_null_subscription(self, client): """Test that subscription is None for new subscription.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" client.get("/news/subscriptions/new") args, kwargs = mock_render.call_args assert kwargs["subscription"] is None def test_new_subscription_page_has_default_iterations(self, client): """Test default iterations is 3.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" client.get("/news/subscriptions/new") args, kwargs = mock_render.call_args assert kwargs["default_settings"]["iterations"] == 3 def test_new_subscription_page_has_default_questions_per_iteration( self, client ): """Test default questions_per_iteration is 5.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" client.get("/news/subscriptions/new") args, kwargs = mock_render.call_args assert kwargs["default_settings"]["questions_per_iteration"] == 5 def test_new_subscription_page_has_default_search_engine(self, client): """Test default search_engine is 'searxng'.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" client.get("/news/subscriptions/new") args, kwargs = mock_render.call_args assert kwargs["default_settings"]["search_engine"] == "searxng" def test_new_subscription_page_has_default_model_provider(self, client): """Test default model_provider is 'OLLAMA'.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" client.get("/news/subscriptions/new") args, kwargs = mock_render.call_args assert kwargs["default_settings"]["model_provider"] == "ollama" def test_new_subscription_page_has_default_search_strategy(self, client): """Test default search_strategy is 'source-based'.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" client.get("/news/subscriptions/new") args, kwargs = mock_render.call_args assert ( kwargs["default_settings"]["search_strategy"] == "source-based" ) def test_new_subscription_page_logged_in_user_loads_settings(self, app): """Test that logged-in user triggers DB session.""" with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" # Patch at the source location where it's imported from with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db: mock_session = MagicMock() mock_db.return_value.__enter__ = MagicMock( return_value=mock_session ) mock_db.return_value.__exit__ = MagicMock( return_value=False ) client.get("/news/subscriptions/new") # get_user_db_session is called once for loading settings assert mock_db.call_count == 1 def test_new_subscription_page_logged_in_user_calls_load_user_settings( self, app ): """Test that load_user_settings helper is called for logged-in user.""" with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" # Patch at the source location where it's imported from with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db: mock_session = MagicMock() mock_db.return_value.__enter__ = MagicMock( return_value=mock_session ) mock_db.return_value.__exit__ = MagicMock( return_value=False ) with patch( "local_deep_research.news.web.load_user_settings" ) as mock_load: client.get("/news/subscriptions/new") mock_load.assert_called_once() class TestEditSubscriptionPageRoute: """Tests for edit_subscription_page route.""" def test_edit_subscription_page_returns_200(self, client): """Test that edit page returns 200 with valid subscription.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" with patch("local_deep_research.news.web.api") as mock_api: mock_api.get_subscription.return_value = {"id": "sub-123"} response = client.get("/news/subscriptions/sub-123/edit") assert response.status_code == 200 def test_edit_subscription_page_renders_correct_template(self, client): """Test that correct template is rendered.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" with patch("local_deep_research.news.web.api") as mock_api: mock_api.get_subscription.return_value = {"id": "sub-123"} client.get("/news/subscriptions/sub-123/edit") args, kwargs = mock_render.call_args assert args[0] == "pages/news-subscription-form.html" def test_edit_subscription_page_passes_subscription_data(self, client): """Test that subscription data is passed to template.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" with patch("local_deep_research.news.web.api") as mock_api: subscription_data = { "id": "sub-123", "topic": "AI News", "schedule": "daily", } mock_api.get_subscription.return_value = subscription_data client.get("/news/subscriptions/sub-123/edit") args, kwargs = mock_render.call_args assert kwargs["subscription"] == subscription_data def test_edit_subscription_page_subscription_not_found(self, client): """Test error message when subscription not found.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" with patch("local_deep_research.news.web.api") as mock_api: mock_api.get_subscription.return_value = None client.get("/news/subscriptions/nonexistent/edit") args, kwargs = mock_render.call_args assert kwargs["error"] == "Subscription not found" assert kwargs["subscription"] is None def test_edit_subscription_page_api_exception(self, client): """Test error handling when API raises exception.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" with patch("local_deep_research.news.web.api") as mock_api: mock_api.get_subscription.side_effect = RuntimeError( "DB connection failed" ) client.get("/news/subscriptions/sub-123/edit") args, kwargs = mock_render.call_args assert kwargs["error"] == "Error loading subscription" assert kwargs["subscription"] is None def test_edit_subscription_page_anonymous_user_skips_settings(self, client): """Test that anonymous user doesn't call get_user_db_session.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" with patch("local_deep_research.news.web.api") as mock_api: mock_api.get_subscription.return_value = {"id": "sub-123"} with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db: client.get("/news/subscriptions/sub-123/edit") # get_user_db_session should not be called for anonymous mock_db.assert_not_called() def test_edit_subscription_page_logged_in_user_loads_settings(self, app): """Test that logged-in user triggers settings loading.""" with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" with patch("local_deep_research.news.web.api") as mock_api: mock_api.get_subscription.return_value = {"id": "sub-123"} with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db: mock_session = MagicMock() mock_db.return_value.__enter__ = MagicMock( return_value=mock_session ) mock_db.return_value.__exit__ = MagicMock( return_value=False ) with patch( "local_deep_research.news.web.load_user_settings" ): client.get("/news/subscriptions/sub-123/edit") # get_user_db_session is called for loading settings assert mock_db.call_count == 1 def test_edit_subscription_page_logs_subscription_id(self, client): """Test that subscription ID is logged.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" with patch("local_deep_research.news.web.api") as mock_api: mock_api.get_subscription.return_value = {"id": "sub-123"} with patch( "local_deep_research.news.web.logger" ) as mock_logger: client.get("/news/subscriptions/sub-123/edit") mock_logger.info.assert_called() # Check that sub-123 is mentioned in the log call_args = mock_logger.info.call_args[0][0] assert "sub-123" in call_args def test_edit_subscription_page_passes_default_settings(self, client): """Test that default settings are always passed.""" with patch( "local_deep_research.news.web.render_template" ) as mock_render: mock_render.return_value = "rendered" with patch("local_deep_research.news.web.api") as mock_api: mock_api.get_subscription.return_value = {"id": "sub-123"} client.get("/news/subscriptions/sub-123/edit") args, kwargs = mock_render.call_args assert "default_settings" in kwargs assert kwargs["default_settings"]["iterations"] == 3 class TestHealthCheckRoute: """Tests for health_check route.""" def test_health_check_healthy_response(self, client): """Test healthy response format.""" with patch( "local_deep_research.news.core.storage_manager.StorageManager" ) as MockStorage: mock_storage = MagicMock() MockStorage.return_value = mock_storage response = client.get("/news/health") assert response.status_code == 200 data = response.get_json() assert data["status"] == "healthy" assert data["database"] == "connected" def test_health_check_returns_json(self, client): """Test that response content-type is JSON.""" with patch( "local_deep_research.news.core.storage_manager.StorageManager" ) as MockStorage: mock_storage = MagicMock() MockStorage.return_value = mock_storage response = client.get("/news/health") assert response.content_type == "application/json" def test_health_check_calls_storage_manager(self, client): """Test that StorageManager is instantiated.""" with patch( "local_deep_research.news.core.storage_manager.StorageManager" ) as MockStorage: mock_storage = MagicMock() MockStorage.return_value = mock_storage client.get("/news/health") MockStorage.assert_called_once() def test_health_check_calls_get_user_feed(self, client): """Test that a simple query is executed.""" with patch( "local_deep_research.news.core.storage_manager.StorageManager" ) as MockStorage: mock_storage = MagicMock() MockStorage.return_value = mock_storage client.get("/news/health") mock_storage.get_user_feed.assert_called_once_with( "health_check", limit=1 ) def test_health_check_unhealthy_storage_error(self, client): """Test unhealthy response when StorageManager fails.""" with patch( "local_deep_research.news.core.storage_manager.StorageManager" ) as MockStorage: MockStorage.side_effect = RuntimeError("DB init failed") response = client.get("/news/health") assert response.status_code == 500 data = response.get_json() assert data["status"] == "unhealthy" def test_health_check_unhealthy_query_error(self, client): """Test unhealthy response when query fails.""" with patch( "local_deep_research.news.core.storage_manager.StorageManager" ) as MockStorage: mock_storage = MagicMock() mock_storage.get_user_feed.side_effect = RuntimeError( "Query failed" ) MockStorage.return_value = mock_storage response = client.get("/news/health") assert response.status_code == 500 data = response.get_json() assert data["status"] == "unhealthy" def test_health_check_does_not_expose_internal_errors(self, client): """Test that internal error details are not exposed.""" with patch( "local_deep_research.news.core.storage_manager.StorageManager" ) as MockStorage: MockStorage.side_effect = RuntimeError( "Connection to localhost:5432 failed" ) response = client.get("/news/health") data = response.get_json() # Should not expose connection string or internal details assert "localhost" not in str(data) assert "5432" not in str(data) assert data["error"] == "An internal error has occurred." class TestLoadUserSettings: """Tests for load_user_settings function.""" def test_load_user_settings_no_session_returns_early(self): """Test that no session returns without modification.""" from local_deep_research.news.web import load_user_settings default_settings = {"iterations": 3} load_user_settings(default_settings, db_session=None, username="test") # Settings should be unchanged assert default_settings["iterations"] == 3 def test_load_user_settings_updates_iterations(self): """Test that iterations setting is loaded.""" from local_deep_research.news.web import load_user_settings default_settings = {"iterations": 3} with patch( "local_deep_research.utilities.db_utils.get_settings_manager" ) as mock_get: mock_manager = MagicMock() mock_manager.get_setting.side_effect = lambda key, default: ( 10 if key == "search.iterations" else default ) mock_get.return_value = mock_manager mock_session = MagicMock() load_user_settings(default_settings, mock_session, "testuser") assert default_settings["iterations"] == 10 def test_load_user_settings_updates_questions_per_iteration(self): """Test that questions_per_iteration setting is loaded.""" from local_deep_research.news.web import load_user_settings default_settings = {"questions_per_iteration": 5} with patch( "local_deep_research.utilities.db_utils.get_settings_manager" ) as mock_get: mock_manager = MagicMock() mock_manager.get_setting.side_effect = lambda key, default: ( 8 if key == "search.questions_per_iteration" else default ) mock_get.return_value = mock_manager mock_session = MagicMock() load_user_settings(default_settings, mock_session, "testuser") assert default_settings["questions_per_iteration"] == 8 def test_load_user_settings_updates_search_engine(self): """Test that search_engine setting comes from 'search.tool'.""" from local_deep_research.news.web import load_user_settings default_settings = {"search_engine": "searxng"} with patch( "local_deep_research.utilities.db_utils.get_settings_manager" ) as mock_get: mock_manager = MagicMock() mock_manager.get_setting.side_effect = lambda key, default: ( "google" if key == "search.tool" else default ) mock_get.return_value = mock_manager mock_session = MagicMock() load_user_settings(default_settings, mock_session, "testuser") assert default_settings["search_engine"] == "google" def test_load_user_settings_updates_model_provider(self): """Test that model_provider setting comes from 'llm.provider'.""" from local_deep_research.news.web import load_user_settings default_settings = {"model_provider": "OLLAMA"} with patch( "local_deep_research.utilities.db_utils.get_settings_manager" ) as mock_get: mock_manager = MagicMock() mock_manager.get_setting.side_effect = lambda key, default: ( "OPENAI" if key == "llm.provider" else default ) mock_get.return_value = mock_manager mock_session = MagicMock() load_user_settings(default_settings, mock_session, "testuser") assert default_settings["model_provider"] == "OPENAI" def test_load_user_settings_updates_model(self): """Test that model setting comes from 'llm.model'.""" from local_deep_research.news.web import load_user_settings default_settings = {"model": ""} with patch( "local_deep_research.utilities.db_utils.get_settings_manager" ) as mock_get: mock_manager = MagicMock() mock_manager.get_setting.side_effect = lambda key, default: ( "gpt-4" if key == "llm.model" else default ) mock_get.return_value = mock_manager mock_session = MagicMock() load_user_settings(default_settings, mock_session, "testuser") assert default_settings["model"] == "gpt-4" def test_load_user_settings_updates_search_strategy(self): """Test that search_strategy setting is loaded.""" from local_deep_research.news.web import load_user_settings default_settings = {"search_strategy": "source-based"} with patch( "local_deep_research.utilities.db_utils.get_settings_manager" ) as mock_get: mock_manager = MagicMock() mock_manager.get_setting.side_effect = lambda key, default: ( "topic-based" if key == "search.search_strategy" else default ) mock_get.return_value = mock_manager mock_session = MagicMock() load_user_settings(default_settings, mock_session, "testuser") assert default_settings["search_strategy"] == "topic-based" def test_load_user_settings_updates_custom_endpoint(self): """Test that custom_endpoint comes from 'llm.openai_endpoint.url'.""" from local_deep_research.news.web import load_user_settings default_settings = {"custom_endpoint": ""} with patch( "local_deep_research.utilities.db_utils.get_settings_manager" ) as mock_get: mock_manager = MagicMock() mock_manager.get_setting.side_effect = lambda key, default: ( "https://custom.api.com" if key == "llm.openai_endpoint.url" else default ) mock_get.return_value = mock_manager mock_session = MagicMock() load_user_settings(default_settings, mock_session, "testuser") assert ( default_settings["custom_endpoint"] == "https://custom.api.com" ) def test_load_user_settings_exception_uses_defaults(self): """Test that defaults are preserved when exception occurs.""" from local_deep_research.news.web import load_user_settings default_settings = { "iterations": 3, "questions_per_iteration": 5, "search_engine": "searxng", } with patch( "local_deep_research.utilities.db_utils.get_settings_manager" ) as mock_get: mock_get.side_effect = RuntimeError("Settings service unavailable") mock_session = MagicMock() # Should not raise load_user_settings(default_settings, mock_session, "testuser") # Defaults should be preserved assert default_settings["iterations"] == 3 assert default_settings["questions_per_iteration"] == 5 assert default_settings["search_engine"] == "searxng" def test_load_user_settings_calls_get_settings_manager(self): """Test that get_settings_manager is called with correct args.""" from local_deep_research.news.web import load_user_settings default_settings = {"iterations": 3} with patch( "local_deep_research.utilities.db_utils.get_settings_manager" ) as mock_get: mock_manager = MagicMock() mock_manager.get_setting.return_value = 3 mock_get.return_value = mock_manager mock_session = MagicMock() load_user_settings(default_settings, mock_session, "testuser") mock_get.assert_called_once_with(mock_session, "testuser") def test_load_user_settings_mutates_dict_in_place(self): """Test that dictionary is mutated in place.""" from local_deep_research.news.web import load_user_settings default_settings = {"iterations": 3} original_id = id(default_settings) with patch( "local_deep_research.utilities.db_utils.get_settings_manager" ) as mock_get: mock_manager = MagicMock() mock_manager.get_setting.side_effect = lambda key, default: ( 10 if key == "search.iterations" else default ) mock_get.return_value = mock_manager mock_session = MagicMock() load_user_settings(default_settings, mock_session, "testuser") # Same object should be modified assert id(default_settings) == original_id assert default_settings["iterations"] == 10