""" Coverage tests for local_deep_research/web/app.py. Targets the ~50 missing statements by exercising each branch of main() in isolation. All heavy imports inside main() are patched at their canonical module paths. Tests: 1. test_main_creates_app_and_runs — create_app + socket_service.run called 2. test_main_https_warning — use_https=True emits warning, doesn't raise 3. test_main_cleanup_scheduler_start — debug=False starts cleanup scheduler 4. test_main_cleanup_scheduler_failure — exception in scheduler start is swallowed 5. test_main_shutdown_scheduler_atexit — news_scheduler.stop() is called 6. test_main_shutdown_databases_atexit — db_manager.close_all_databases() called 7. test_flush_logs_on_exit — first atexit handler flushes log queue """ import os from unittest.mock import MagicMock, patch import pytest MODULE = "local_deep_research.web.app" _CLEANUP_MOD = "local_deep_research.web.auth.connection_cleanup" _SESSION_MOD = "local_deep_research.web.auth.session_manager" _DB_MOD = "local_deep_research.database.encrypted_db" @pytest.fixture(autouse=True) def _stub_log_queue_daemon(): """Prevent main() from spawning a real log-queue daemon thread during tests.""" with ( patch(f"{MODULE}.start_log_queue_processor"), patch(f"{MODULE}.stop_log_queue_processor"), ): yield def _handler_by_name(captured, name): """Locate an atexit handler by function name (order-independent).""" for fn in captured: if getattr(fn, "__name__", None) == name: return fn raise AssertionError( f"No captured handler named {name!r}; got " f"{[getattr(fn, '__name__', repr(fn)) for fn in captured]}" ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_config(host="127.0.0.1", port=5000, debug=False, use_https=False): return {"host": host, "port": port, "debug": debug, "use_https": use_https} def _run_main(config, extra_env=None, cleanup_side_effect=None): """ Patch everything main() needs, run it, and return the mock objects. Returns a dict with keys: socket_service, app, scheduler, atexit_calls. """ mock_socket_service = MagicMock() mock_app = MagicMock() mock_scheduler = MagicMock() atexit_calls = [] patches = [ patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(mock_app, mock_socket_service) ), patch("atexit.register", side_effect=atexit_calls.append), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", side_effect=cleanup_side_effect, return_value=mock_scheduler, ), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), ] env_patch = patch.dict("os.environ", extra_env or {}, clear=False) for p in patches: p.start() env_patch.start() try: from local_deep_research.web import app as app_module app_module.main() finally: for p in reversed(patches): p.stop() env_patch.stop() return { "socket_service": mock_socket_service, "app": mock_app, "scheduler": mock_scheduler, "atexit_calls": atexit_calls, } # --------------------------------------------------------------------------- # 1. test_main_creates_app_and_runs # --------------------------------------------------------------------------- class TestMainCreatesAppAndRuns: """main() must call create_app() and then socket_service.run().""" def test_create_app_is_called(self): config = _make_config() with ( patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(MagicMock(), MagicMock()) ) as mock_create, patch("atexit.register"), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", return_value=MagicMock(), ), ): from local_deep_research.web import app as app_module app_module.main() mock_create.assert_called_once() def test_socket_service_run_called(self): config = _make_config(host="0.0.0.0", port=8080, debug=False) result = _run_main(config) result["socket_service"].run.assert_called_once_with( host="0.0.0.0", port=8080, debug=False ) def test_socket_service_run_forwards_debug_true(self): config = _make_config(host="localhost", port=5001, debug=True) result = _run_main(config, extra_env={"WERKZEUG_RUN_MAIN": "true"}) result["socket_service"].run.assert_called_once_with( host="localhost", port=5001, debug=True ) def test_main_completes_without_exception(self): # audit: PUNCHLIST reviewed 2026-05 — KEEP (ASSERT_TRUE). config = _make_config() # Should not raise _run_main(config) def test_config_logger_called_with_debug_flag(self): config = _make_config(debug=True) with ( patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger") as mock_cl, patch( f"{MODULE}.create_app", return_value=(MagicMock(), MagicMock()) ), patch("atexit.register"), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", return_value=MagicMock(), ), patch.dict( "os.environ", {"WERKZEUG_RUN_MAIN": "true"}, clear=False ), ): from local_deep_research.web import app as app_module app_module.main() mock_cl.assert_called_once_with("ldr_web", debug=True) # --------------------------------------------------------------------------- # 2. test_main_https_warning # --------------------------------------------------------------------------- class TestMainHttpsWarning: """When use_https=True, main() logs a warning but does not raise.""" def test_https_true_does_not_raise(self): config = _make_config(use_https=True) # Must not raise result = _run_main(config) # Server still runs result["socket_service"].run.assert_called_once() def test_https_true_emits_logger_warning(self): config = _make_config(use_https=True) with ( patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(MagicMock(), MagicMock()) ), patch("atexit.register"), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", return_value=MagicMock(), ), # Intercept logger to verify the HTTPS warning path was reached patch(f"{MODULE}.logger") as mock_logger, ): from local_deep_research.web import app as app_module app_module.main() # logger.info("Starting server with HTTPS ...") and logger.warning(...) should be called assert mock_logger.info.called or mock_logger.warning.called def test_https_false_skips_https_branch(self): config = _make_config(use_https=False) with ( patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(MagicMock(), MagicMock()) ), patch("atexit.register"), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", return_value=MagicMock(), ), patch(f"{MODULE}.logger") as mock_logger, ): from local_deep_research.web import app as app_module app_module.main() # "Starting server with HTTPS" message should NOT be logged https_info_calls = [ c for c in mock_logger.info.call_args_list if "HTTPS" in str(c) ] assert len(https_info_calls) == 0 # --------------------------------------------------------------------------- # 3. test_main_cleanup_scheduler_start # --------------------------------------------------------------------------- class TestMainCleanupSchedulerStart: """Cleanup scheduler is started when debug=False (or WERKZEUG_RUN_MAIN=true).""" def test_scheduler_started_when_debug_false(self): config = _make_config(debug=False) with ( patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(MagicMock(), MagicMock()) ), patch("atexit.register"), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", return_value=MagicMock(), ) as mock_start, ): from local_deep_research.web import app as app_module app_module.main() mock_start.assert_called_once() def test_scheduler_started_when_werkzeug_run_main_set(self): config = _make_config(debug=True) with ( patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(MagicMock(), MagicMock()) ), patch("atexit.register"), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", return_value=MagicMock(), ) as mock_start, patch.dict( "os.environ", {"WERKZEUG_RUN_MAIN": "true"}, clear=False ), ): from local_deep_research.web import app as app_module app_module.main() mock_start.assert_called_once() def test_scheduler_not_started_in_debug_without_werkzeug(self): config = _make_config(debug=True) env_without = { k: v for k, v in os.environ.items() if k != "WERKZEUG_RUN_MAIN" } with ( patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(MagicMock(), MagicMock()) ), patch("atexit.register"), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", return_value=MagicMock(), ) as mock_start, patch.dict("os.environ", env_without, clear=True), ): from local_deep_research.web import app as app_module app_module.main() mock_start.assert_not_called() def test_scheduler_shutdown_lambda_registered_as_atexit(self): """After starting cleanup scheduler, main() registers a lambda atexit handler.""" config = _make_config(debug=False) atexit_calls = [] mock_scheduler = MagicMock() with ( patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(MagicMock(), MagicMock()) ), patch("atexit.register", side_effect=atexit_calls.append), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", return_value=mock_scheduler, ), ): from local_deep_research.web import app as app_module app_module.main() # At least 4 atexit handlers: shutdown_db, shutdown_scheduler, # cleanup lambda, flush_logs, (+ stop_log_queue_processor) assert len(atexit_calls) >= 4 # The cleanup_scheduler.shutdown call is an anonymous lambda — locate # it by calling each handler and finding the one that triggers # mock_scheduler.shutdown. for handler in atexit_calls: if getattr(handler, "__name__", None) == "": handler() break mock_scheduler.shutdown.assert_called_once_with(wait=False) # --------------------------------------------------------------------------- # 4. test_main_cleanup_scheduler_failure # --------------------------------------------------------------------------- class TestMainCleanupSchedulerFailure: """If start_connection_cleanup_scheduler raises, main() must not abort.""" def test_scheduler_exception_is_swallowed(self): config = _make_config(debug=False) result = _run_main( config, cleanup_side_effect=RuntimeError("scheduler boom") ) # Server still runs despite scheduler failure result["socket_service"].run.assert_called_once() def test_scheduler_exception_logs_warning(self): config = _make_config(debug=False) with ( patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(MagicMock(), MagicMock()) ), patch("atexit.register"), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", side_effect=Exception("bad scheduler"), ), patch(f"{MODULE}.logger") as mock_logger, ): from local_deep_research.web import app as app_module app_module.main() mock_logger.warning.assert_called() def test_cleanup_scheduler_none_when_it_fails(self): """When scheduler startup fails, cleanup_scheduler is set to None — no atexit lambda.""" config = _make_config(debug=False) atexit_calls = [] with ( patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(MagicMock(), MagicMock()) ), patch("atexit.register", side_effect=atexit_calls.append), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", side_effect=RuntimeError("fail"), ), ): from local_deep_research.web import app as app_module app_module.main() # When scheduler fails, no cleanup-scheduler lambda is registered. # The other handlers (shutdown_databases, shutdown_scheduler, # flush_logs_on_exit, stop_log_queue_processor) are still registered. handler_names = [getattr(h, "__name__", None) for h in atexit_calls] assert "" not in handler_names assert "shutdown_databases" in handler_names assert "shutdown_scheduler" in handler_names assert "flush_logs_on_exit" in handler_names # --------------------------------------------------------------------------- # 5. test_main_shutdown_scheduler_atexit # --------------------------------------------------------------------------- class TestMainShutdownSchedulerAtexit: """The shutdown_scheduler atexit closure calls app.background_job_scheduler.stop().""" def _run_and_capture(self, mock_app): captured = [] config = _make_config() patches = [ patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch(f"{MODULE}.create_app", return_value=(mock_app, MagicMock())), patch("atexit.register", side_effect=captured.append), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", return_value=MagicMock(), ), ] for p in patches: p.start() try: from local_deep_research.web import app as app_module app_module.main() finally: for p in reversed(patches): p.stop() return captured def test_news_scheduler_stop_called(self): """shutdown_scheduler calls app.background_job_scheduler.stop() when it exists.""" mock_app = MagicMock() mock_background_job_scheduler = MagicMock() mock_app.background_job_scheduler = mock_background_job_scheduler captured = self._run_and_capture(mock_app) # Invoke all atexit handlers to find shutdown_scheduler for handler in captured: try: handler() except Exception: pass mock_background_job_scheduler.stop.assert_called_once() def test_no_error_when_news_scheduler_missing(self): """shutdown_scheduler is a no-op if app has no news_scheduler.""" # spec=[] means no attributes exist on the mock mock_app = MagicMock(spec=[]) captured = self._run_and_capture(mock_app) # The atexit handlers registered by app.main() must not raise even # when the app has no news_scheduler attribute. Collect any errors # explicitly so the assertion fails loudly if a handler does raise. errors = [] for handler in captured: try: handler() except Exception as exc: errors.append(exc) assert errors == [], ( f"atexit handlers raised when app has no news_scheduler: {errors}" ) def test_news_scheduler_stop_exception_is_swallowed(self): """shutdown_scheduler swallows exceptions from news_scheduler.stop().""" mock_app = MagicMock() mock_app.background_job_scheduler = MagicMock() mock_app.background_job_scheduler.stop.side_effect = RuntimeError( "scheduler error" ) captured = self._run_and_capture(mock_app) # Even though stop() raises RuntimeError, the SUT's shutdown_scheduler # closure swallows it so the atexit chain keeps running. Collect any # errors that escape — there should be none. errors = [] for handler in captured: try: handler() except Exception as exc: errors.append(exc) assert errors == [], ( f"atexit handlers leaked exceptions despite swallow: {errors}" ) # And confirm the swallowed call actually happened. mock_app.background_job_scheduler.stop.assert_called_once() # --------------------------------------------------------------------------- # 6. test_main_shutdown_databases_atexit # --------------------------------------------------------------------------- class TestMainShutdownDatabasesAtexit: """The shutdown_databases atexit closure calls db_manager.close_all_databases().""" def _run_and_capture(self): captured = [] config = _make_config() patches = [ patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(MagicMock(), MagicMock()) ), patch("atexit.register", side_effect=captured.append), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", return_value=MagicMock(), ), ] for p in patches: p.start() try: from local_deep_research.web import app as app_module app_module.main() finally: for p in reversed(patches): p.stop() return captured def test_close_all_databases_called(self): """The shutdown_databases atexit handler calls db_manager.close_all_databases().""" captured = self._run_and_capture() mock_db_manager = MagicMock() handler = _handler_by_name(captured, "shutdown_databases") with patch(f"{_DB_MOD}.db_manager", mock_db_manager): handler() mock_db_manager.close_all_databases.assert_called_once() def test_shutdown_databases_swallows_import_error(self): """shutdown_databases does not propagate exceptions.""" # audit: PUNCHLIST reviewed 2026-05 — KEEP (ASSERT_TRUE). captured = self._run_and_capture() handler = _handler_by_name(captured, "shutdown_databases") # Simulate import failure inside the handler with patch(f"{_DB_MOD}.db_manager", side_effect=ImportError("gone")): handler() # must not raise def test_shutdown_databases_swallows_runtime_error(self): """shutdown_databases swallows RuntimeError from close_all_databases.""" # audit: PUNCHLIST reviewed 2026-05 — KEEP (ASSERT_TRUE). captured = self._run_and_capture() handler = _handler_by_name(captured, "shutdown_databases") mock_db = MagicMock() mock_db.close_all_databases.side_effect = RuntimeError("db error") with patch(f"{_DB_MOD}.db_manager", mock_db): handler() # must not raise def test_at_least_three_atexit_handlers_registered(self): """main() always registers at least 3 atexit handlers.""" captured = self._run_and_capture() # shutdown_databases + shutdown_scheduler + flush_logs_on_exit # (+ optional cleanup-scheduler lambda, + stop_log_queue_processor) assert len(captured) >= 3 # --------------------------------------------------------------------------- # 7. test_flush_logs_on_exit # --------------------------------------------------------------------------- class TestFlushLogsOnExit: """The flush_logs_on_exit atexit handler flushes the log queue.""" def _run_and_capture(self, extra_patches=None): captured = [] config = _make_config() patches = [ patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(MagicMock(), MagicMock()) ), patch("atexit.register", side_effect=captured.append), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", return_value=MagicMock(), ), ] + (extra_patches or []) for p in patches: p.start() try: from local_deep_research.web import app as app_module app_module.main() finally: for p in reversed(patches): p.stop() return captured def test_first_handler_is_flush_logs(self): """Calling the flush_logs_on_exit atexit handler flushes the log queue.""" captured = self._run_and_capture() flush_handler = _handler_by_name(captured, "flush_logs_on_exit") # Patch flush_log_queue at the module where flush_logs_on_exit imports it with patch(f"{MODULE}.flush_log_queue") as mock_flush: flush_handler() mock_flush.assert_called_once() def test_flush_handler_swallows_exceptions(self): """flush_logs_on_exit does not propagate exceptions from flush_log_queue.""" captured = self._run_and_capture() flush_handler = _handler_by_name(captured, "flush_logs_on_exit") with patch( f"{MODULE}.flush_log_queue", side_effect=RuntimeError("log err") ): flush_handler() # must not raise def test_flush_handler_uses_flask_app_context(self): """flush_logs_on_exit creates a minimal Flask app context before flushing.""" captured = self._run_and_capture() flush_handler = _handler_by_name(captured, "flush_logs_on_exit") mock_flask_app = MagicMock() mock_flask_app.app_context.return_value.__enter__ = MagicMock( return_value=None ) mock_flask_app.app_context.return_value.__exit__ = MagicMock( return_value=False ) # Flask is imported lazily inside flush_logs_on_exit as `from flask import Flask` # patch it at the flask package level so the local import picks up the mock with ( patch("flask.Flask", return_value=mock_flask_app), patch(f"{MODULE}.flush_log_queue") as mock_flush, ): flush_handler() mock_flask_app.app_context.assert_called_once() mock_flush.assert_called_once() def test_atexit_registered_before_create_app(self): """The flush log atexit handler is registered early (before socket run).""" atexit_order = [] run_order = [] config = _make_config() mock_socket_service = MagicMock() mock_socket_service.run.side_effect = lambda **_: run_order.append( "run" ) def register_side_effect(fn): atexit_order.append(fn) patches = [ patch(f"{MODULE}.load_server_config", return_value=config), patch(f"{MODULE}.config_logger"), patch( f"{MODULE}.create_app", return_value=(MagicMock(), mock_socket_service), ), patch("atexit.register", side_effect=register_side_effect), patch(f"{_SESSION_MOD}.session_manager", MagicMock(), create=True), patch(f"{_DB_MOD}.db_manager", MagicMock(), create=True), patch( f"{_CLEANUP_MOD}.start_connection_cleanup_scheduler", return_value=MagicMock(), ), ] for p in patches: p.start() try: from local_deep_research.web import app as app_module app_module.main() finally: for p in reversed(patches): p.stop() # At least one handler must have been registered assert len(atexit_order) >= 1