diff --git a/tests/test_coverage_remaining_gaps.py b/tests/test_coverage_remaining_gaps.py new file mode 100644 index 00000000..16c564f9 --- /dev/null +++ b/tests/test_coverage_remaining_gaps.py @@ -0,0 +1,768 @@ +""" +Targeted tests to fill remaining coverage gaps in: +- app/views/base.py (lines 36-41) +- app/views/general.py (lines 64-66, 111, 133-134, 138) +- app/views/status.py (lines 53-54, 59-60, 68-69) +- app/utils/config_validator/providers.py (lines 68-78, 86-100, 243) +- app/utils/settings_sync.py (lines 71-72, 95-98) +- app/api/logs.py (additional branches) +""" + +from unittest.mock import MagicMock, Mock, mock_open, patch + +import pytest + +# =========================================================================== +# app/views/base.py – kwargs context path (lines 36-41) +# =========================================================================== + + +@pytest.mark.unit +class TestViewsBase: + """Tests for template_response_with_version wrapper in app/views/base.py.""" + + def test_kwargs_context_version_injection(self): + """Test version injection when context is passed as a keyword argument.""" + from app.views.base import template_response_with_version + + with patch("app.views.base.original_template_response") as mock_orig: + mock_orig.return_value = "response" + req = MagicMock() + # Make the request NOT have a csrf_token on its state + del req.state.csrf_token # ensure AttributeError on hasattr + + result = template_response_with_version( + "template.html", + context={"request": req, "title": "Test"}, + ) + + mock_orig.assert_called_once() + _, kwargs = mock_orig.call_args + assert "version" in kwargs["context"] + + def test_kwargs_context_csrf_token_injection(self): + """Test CSRF token injection when context has request with csrf_token.""" + from app.views.base import template_response_with_version + + with patch("app.views.base.original_template_response") as mock_orig: + mock_orig.return_value = "response" + req = MagicMock() + req.state.csrf_token = "test-csrf-token" + + template_response_with_version( + "template.html", + context={"request": req}, + ) + + _, kwargs = mock_orig.call_args + assert kwargs["context"].get("csrf_token") == "test-csrf-token" + + def test_positional_context_with_csrf_token(self): + """Test CSRF token injection via positional args.""" + from app.views.base import template_response_with_version + + with patch("app.views.base.original_template_response") as mock_orig: + mock_orig.return_value = "response" + req = MagicMock() + req.state.csrf_token = "my-csrf" + + context = {"request": req} + template_response_with_version("template.html", context) + + args, _ = mock_orig.call_args + assert args[1].get("csrf_token") == "my-csrf" + + def test_kwargs_context_no_request(self): + """Test kwargs context path when request is not in context.""" + from app.views.base import template_response_with_version + + with patch("app.views.base.original_template_response") as mock_orig: + mock_orig.return_value = "response" + + template_response_with_version( + "template.html", + context={"title": "No request"}, + ) + + mock_orig.assert_called_once() + _, kwargs = mock_orig.call_args + assert "version" in kwargs["context"] + + def test_no_args_no_context(self): + """Test with no args and no context (edge case).""" + from app.views.base import template_response_with_version + + with patch("app.views.base.original_template_response") as mock_orig: + mock_orig.return_value = "response" + template_response_with_version("template.html") + + mock_orig.assert_called_once() + + +# =========================================================================== +# app/views/general.py – error branches +# =========================================================================== + + +@pytest.mark.unit +class TestGeneralViewsAdditional: + """Additional tests for general view error branches.""" + + @patch("app.views.general.get_provider_status") + @patch("app.views.general.validate_storage_configs") + @patch("app.views.general.templates") + @pytest.mark.asyncio + async def test_db_error_logged_but_page_still_renders(self, mock_templates, mock_storage, mock_providers): + """Test that a DB error is logged but the page still renders (lines 64-66).""" + from app.views.general import serve_index + + mock_providers.return_value = {} + mock_storage.return_value = {} + mock_templates.TemplateResponse = MagicMock(return_value="response") + + with patch("app.utils.settings_service.get_setting_from_db", return_value=None): + with patch("app.utils.setup_wizard.is_setup_required", return_value=False): + mock_request = MagicMock() + mock_request.query_params.get = MagicMock(return_value=None) + mock_db = MagicMock() + # Make DB query fail to trigger the except block (lines 64-66) + mock_db.query.side_effect = Exception("DB failure") + + result = await serve_index(mock_request, mock_db) + + # Page should still render with 0 processed_files + mock_templates.TemplateResponse.assert_called_once() + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + assert context["stats"]["processed_files"] == 0 + + @patch("app.views.general.templates") + @pytest.mark.asyncio + async def test_favicon_not_found_raises_404(self, mock_templates): + """Test that missing favicon raises 404 HTTPException (line 111).""" + from fastapi import HTTPException + + from app.views.general import favicon + + with patch("pathlib.Path.exists", return_value=False): + with pytest.raises(HTTPException) as exc_info: + favicon() + + assert exc_info.value.status_code == 404 + + @patch("app.views.general.templates") + @pytest.mark.asyncio + async def test_license_fallback_when_no_file_found(self, mock_templates): + """Test that license page uses embedded text when no file found (lines 133-134, 138).""" + from app.views.general import serve_license + + mock_templates.TemplateResponse = MagicMock(return_value="response") + mock_request = MagicMock() + + # Patch open to raise FileNotFoundError for all paths + with patch("builtins.open", side_effect=FileNotFoundError("No such file")): + await serve_license(mock_request) + + # Template should be called with embedded license text + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + assert "Apache License" in context["license_text"] + + def test_license_page_integration_always_renders(self, client): + """Test license page renders (with real or embedded text).""" + response = client.get("/license") + assert response.status_code == 200 + + +# =========================================================================== +# app/views/status.py – Docker exception branches +# =========================================================================== + + +@pytest.mark.unit +class TestStatusViewsAdditional: + """Additional tests for status view exception branches.""" + + @patch("app.views.status.get_provider_status") + @patch("app.views.status.templates") + @patch("app.views.status.settings") + @patch("app.views.status.os.path.exists") + @patch("builtins.open", new_callable=mock_open, read_data="12:docker:/abc123") + @pytest.mark.asyncio + async def test_git_sha_exception_in_docker_env( + self, mock_file, mock_exists, mock_settings, mock_templates, mock_providers + ): + """Test Docker env: git_sha[:7] raises exception → lines 53-54.""" + from app.views.status import status_dashboard + + mock_exists.return_value = True # Docker env exists + mock_providers.return_value = {} + mock_settings.version = "1.0.0" + mock_settings.build_date = "2024-01-01" + # A bool is not subscriptable, so git_sha[:7] raises TypeError → lines 53-54 + mock_settings.git_sha = True + mock_settings.notification_urls = [] + + mock_request = Mock() + + await status_dashboard(mock_request) + + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + assert context["container_info"]["is_docker"] is True + assert context["container_info"]["git_sha"] == "Unknown" + + @patch("app.views.status.get_provider_status") + @patch("app.views.status.templates") + @patch("app.views.status.settings") + @patch("app.views.status.os.path.exists") + @patch("builtins.open", new_callable=mock_open, read_data="12:docker:/abc123") + @pytest.mark.asyncio + async def test_runtime_info_exception_in_docker_env( + self, mock_file, mock_exists, mock_settings, mock_templates, mock_providers + ): + """Test Docker env: settings.runtime_info raises exception → lines 59-60.""" + from unittest.mock import PropertyMock + + from app.views.status import status_dashboard + + mock_exists.return_value = True + mock_providers.return_value = {} + mock_settings.version = "1.0.0" + mock_settings.build_date = "2024-01-01" + mock_settings.git_sha = "abc1234567" + mock_settings.notification_urls = [] + # Use spec to restrict runtime_info to raise AttributeError + # since it's not a validated Settings attribute + type(mock_settings).runtime_info = PropertyMock(side_effect=AttributeError("runtime_info not set")) + + mock_request = Mock() + + await status_dashboard(mock_request) + + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + assert context["container_info"]["is_docker"] is True + + @patch("app.views.status.get_provider_status") + @patch("app.views.status.templates") + @patch("app.views.status.settings") + @patch("app.views.status.os.path.exists") + @pytest.mark.asyncio + async def test_git_sha_exception_in_non_docker_env( + self, mock_exists, mock_settings, mock_templates, mock_providers + ): + """Test non-Docker env: git_sha[:7] raises exception → lines 68-69.""" + from app.views.status import status_dashboard + + mock_exists.return_value = False # Not in Docker + mock_providers.return_value = {} + mock_settings.version = "1.0.0" + mock_settings.build_date = "2024-01-01" + # A bool is not subscriptable, so git_sha[:7] raises TypeError → lines 68-69 + mock_settings.git_sha = True + mock_settings.notification_urls = [] + + mock_request = Mock() + + await status_dashboard(mock_request) + + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + assert context["container_info"]["is_docker"] is False + assert context["container_info"]["git_sha"] == "Unknown" + + +# =========================================================================== +# app/utils/config_validator/providers.py – alternative AI providers +# =========================================================================== + + +@pytest.mark.unit +class TestProvidersAlternativeAI: + """Tests for alternative AI provider branches in get_provider_status.""" + + def test_anthropic_configured(self): + """Test provider status with anthropic AI provider (lines 68-69, 86-87).""" + from app.utils.config_validator.providers import get_provider_status + + with patch("app.utils.config_validator.providers.settings") as mock_settings: + mock_settings.ai_provider = "anthropic" + mock_settings.ai_model = "claude-3-5-sonnet" + mock_settings.anthropic_api_key = "sk-ant-key" + mock_settings.auth_enabled = False + mock_settings.notification_urls = [] + mock_settings.azure_ai_key = None + mock_settings.azure_endpoint = None + mock_settings.azure_region = "eastus" + _set_minimal_provider_settings(mock_settings) + + result = get_provider_status() + + ai_provider = result["AI Provider"] + assert ai_provider["configured"] is True + assert ai_provider["details"]["provider"] == "anthropic" + assert "api_key" in ai_provider["details"] + + def test_gemini_configured(self): + """Test provider status with gemini AI provider (lines 70-71, 88-89).""" + from app.utils.config_validator.providers import get_provider_status + + with patch("app.utils.config_validator.providers.settings") as mock_settings: + mock_settings.ai_provider = "gemini" + mock_settings.ai_model = "gemini-pro" + mock_settings.gemini_api_key = "gemini-key-123" + mock_settings.auth_enabled = False + mock_settings.notification_urls = [] + mock_settings.azure_ai_key = None + mock_settings.azure_endpoint = None + mock_settings.azure_region = "eastus" + _set_minimal_provider_settings(mock_settings) + + result = get_provider_status() + + ai_provider = result["AI Provider"] + assert ai_provider["configured"] is True + assert ai_provider["details"]["provider"] == "gemini" + assert "api_key" in ai_provider["details"] + + def test_ollama_configured(self): + """Test provider status with ollama AI provider (lines 72-73, 90-91).""" + from app.utils.config_validator.providers import get_provider_status + + with patch("app.utils.config_validator.providers.settings") as mock_settings: + mock_settings.ai_provider = "ollama" + mock_settings.ai_model = "llama2" + mock_settings.ollama_base_url = "http://localhost:11434" + mock_settings.auth_enabled = False + mock_settings.notification_urls = [] + mock_settings.azure_ai_key = None + mock_settings.azure_endpoint = None + mock_settings.azure_region = "eastus" + _set_minimal_provider_settings(mock_settings) + + result = get_provider_status() + + ai_provider = result["AI Provider"] + assert ai_provider["configured"] is True + assert ai_provider["details"]["provider"] == "ollama" + assert "base_url" in ai_provider["details"] + + def test_openrouter_configured(self): + """Test provider status with openrouter AI provider (lines 74-75, 92-94).""" + from app.utils.config_validator.providers import get_provider_status + + with patch("app.utils.config_validator.providers.settings") as mock_settings: + mock_settings.ai_provider = "openrouter" + mock_settings.ai_model = "openai/gpt-4" + mock_settings.openrouter_api_key = "or-key-123" + mock_settings.openrouter_base_url = "https://openrouter.ai/api/v1" + mock_settings.auth_enabled = False + mock_settings.notification_urls = [] + mock_settings.azure_ai_key = None + mock_settings.azure_endpoint = None + mock_settings.azure_region = "eastus" + _set_minimal_provider_settings(mock_settings) + + result = get_provider_status() + + ai_provider = result["AI Provider"] + assert ai_provider["configured"] is True + assert ai_provider["details"]["provider"] == "openrouter" + assert "api_key" in ai_provider["details"] + assert "base_url" in ai_provider["details"] + + def test_portkey_configured_with_virtual_key(self): + """Test provider status with portkey AI provider including virtual key (lines 76-77, 95-100).""" + from app.utils.config_validator.providers import get_provider_status + + with patch("app.utils.config_validator.providers.settings") as mock_settings: + mock_settings.ai_provider = "portkey" + mock_settings.ai_model = "claude-3" + mock_settings.portkey_api_key = "pk-key-123" + mock_settings.portkey_virtual_key = "vk-123" + mock_settings.portkey_config = None + mock_settings.auth_enabled = False + mock_settings.notification_urls = [] + mock_settings.azure_ai_key = None + mock_settings.azure_endpoint = None + mock_settings.azure_region = "eastus" + _set_minimal_provider_settings(mock_settings) + + result = get_provider_status() + + ai_provider = result["AI Provider"] + assert ai_provider["configured"] is True + assert ai_provider["details"]["provider"] == "portkey" + assert "api_key" in ai_provider["details"] + assert "virtual_key" in ai_provider["details"] + + def test_portkey_configured_with_config(self): + """Test provider status with portkey AI provider including config.""" + from app.utils.config_validator.providers import get_provider_status + + with patch("app.utils.config_validator.providers.settings") as mock_settings: + mock_settings.ai_provider = "portkey" + mock_settings.ai_model = "claude-3" + mock_settings.portkey_api_key = "pk-key-123" + mock_settings.portkey_virtual_key = None + mock_settings.portkey_config = '{"strategy": "loadbalance"}' + mock_settings.auth_enabled = False + mock_settings.notification_urls = [] + mock_settings.azure_ai_key = None + mock_settings.azure_endpoint = None + mock_settings.azure_region = "eastus" + _set_minimal_provider_settings(mock_settings) + + result = get_provider_status() + + ai_provider = result["AI Provider"] + assert "config" in ai_provider["details"] + + def test_unknown_ai_provider_not_configured(self): + """Test that unknown provider returns configured=False (line 78 – return False).""" + from app.utils.config_validator.providers import get_provider_status + + with patch("app.utils.config_validator.providers.settings") as mock_settings: + mock_settings.ai_provider = "unknown_provider" + mock_settings.ai_model = None + mock_settings.auth_enabled = False + mock_settings.notification_urls = [] + mock_settings.azure_ai_key = None + mock_settings.azure_endpoint = None + mock_settings.azure_region = "eastus" + _set_minimal_provider_settings(mock_settings) + + result = get_provider_status() + + ai_provider = result["AI Provider"] + assert ai_provider["configured"] is False + + def test_nextcloud_url_with_remote_php(self): + """Test that nextcloud base URL is extracted correctly (line 243).""" + from app.utils.config_validator.providers import get_provider_status + + with patch("app.utils.config_validator.providers.settings") as mock_settings: + mock_settings.ai_provider = "openai" + mock_settings.ai_model = "gpt-4o-mini" + mock_settings.openai_api_key = "sk-key" + mock_settings.openai_base_url = "https://api.openai.com/v1" + mock_settings.auth_enabled = False + mock_settings.notification_urls = [] + mock_settings.azure_ai_key = None + mock_settings.azure_endpoint = None + mock_settings.azure_region = "eastus" + # NextCloud URL with /remote.php should be split + mock_settings.nextcloud_upload_url = "https://cloud.example.com/remote.php/dav/files/user" + mock_settings.nextcloud_username = "user" + mock_settings.nextcloud_password = "pass" + mock_settings.nextcloud_folder = "/Docs" + _set_minimal_provider_settings(mock_settings) + + result = get_provider_status() + + nextcloud = result["NextCloud"] + assert nextcloud["details"]["base_url"] == "https://cloud.example.com" + + +def _set_minimal_provider_settings(mock_settings): + """Helper to set the minimal settings attributes needed by get_provider_status.""" + defaults = { + "authentik_client_id": None, + "authentik_client_secret": None, + "authentik_config_url": None, + "admin_username": None, + "oauth_provider_name": None, + "session_secret": "secret", + "notify_on_task_failure": True, + "notify_on_credential_failure": True, + "notify_on_startup": True, + "notify_on_shutdown": False, + "dropbox_app_key": None, + "dropbox_app_secret": None, + "dropbox_refresh_token": None, + "dropbox_folder": "/", + "email_host": None, + "email_default_recipient": None, + "email_port": 587, + "email_username": None, + "email_password": None, + "email_use_tls": True, + "email_sender": None, + "ftp_host": None, + "ftp_username": None, + "ftp_password": None, + "ftp_port": 21, + "ftp_folder": "/", + "ftp_use_tls": True, + "ftp_allow_plaintext": False, + "google_drive_client_id": None, + "google_drive_client_secret": None, + "google_drive_refresh_token": None, + "google_drive_credentials_json": None, + "google_drive_use_oauth": False, + "google_drive_folder_id": None, + "google_drive_delegate_to": None, + "nextcloud_upload_url": None, + "nextcloud_username": None, + "nextcloud_password": None, + "nextcloud_folder": "/", + "onedrive_client_id": None, + "onedrive_client_secret": None, + "onedrive_refresh_token": None, + "onedrive_tenant_id": None, + "onedrive_folder_path": "/", + "paperless_url": None, + "paperless_api_token": None, + "paperless_correspondent_name": None, + "paperless_document_type_name": None, + "s3_bucket_name": None, + "s3_region": None, + "aws_access_key_id": None, + "aws_secret_access_key": None, + "s3_folder_prefix": None, + "sftp_host": None, + "sftp_username": None, + "sftp_password": None, + "sftp_port": 22, + "sftp_folder": "/", + "sftp_private_key": None, + "webdav_url": None, + "webdav_username": None, + "webdav_password": None, + "webdav_folder": "/", + "imap_host": None, + "imap_username": None, + "imap_password": None, + "imap_folder": "INBOX", + "gotenberg_url": "http://localhost:3000", + "rclone_remote": None, + "rclone_base_path": None, + "uptime_kuma_push_url": None, + # AI provider specifics (set None to avoid AttributeError) + "anthropic_api_key": None, + "gemini_api_key": None, + "ollama_base_url": None, + "openrouter_api_key": None, + "openrouter_base_url": None, + "portkey_api_key": None, + "portkey_virtual_key": None, + "portkey_config": None, + "litellm": None, + "openai_api_key": None, + "openai_base_url": "https://api.openai.com/v1", + } + for attr, val in defaults.items(): + if not hasattr(mock_settings, attr) or getattr(mock_settings, attr, "NOTSET") == "NOTSET": + setattr(mock_settings, attr, val) + + +# =========================================================================== +# app/utils/settings_sync.py – reload failure and signal handler +# =========================================================================== + + +@pytest.mark.unit +class TestSettingsSyncAdditional: + """Additional tests for settings_sync covering reload failure branch.""" + + def test_reload_failure_is_logged_not_raised(self): + """Test that a reload failure is logged, not raised (lines 71-72).""" + from app.utils.settings_sync import notify_settings_updated + + with patch("app.utils.settings_sync.redis") as mock_redis_module: + mock_redis_module.from_url.return_value = MagicMock() # Redis OK + with patch("app.utils.config_loader.reload_settings_from_db", side_effect=Exception("reload failed")): + # Should not raise despite reload failure + notify_settings_updated() + + def test_signal_handler_reloads_on_version_change(self): + """Test the task_prerun signal handler reloads settings when version changes (lines 95-98).""" + from app.utils.settings_sync import register_settings_reload_signal + + handler_fn = None + + # Capture the handler; the decorator pattern means connect(weak=False) + # returns a decorator, which is then applied to _reload_if_stale + def capture_connect(fn=None, weak=None, **kwargs): + nonlocal handler_fn + if fn is not None: + handler_fn = fn + return fn + + def decorator(func): + nonlocal handler_fn + handler_fn = func + return func + + return decorator + + with patch("app.utils.settings_sync.task_prerun") as mock_signal: + mock_signal.connect = capture_connect + register_settings_reload_signal() + + assert handler_fn is not None + + # Simulate handler being called with a new version + mock_redis = MagicMock() + mock_redis.get.return_value = b"1234567890.0" # new version + + with patch("app.utils.settings_sync.redis") as mock_redis_mod: + mock_redis_mod.from_url.return_value = mock_redis + with patch("app.utils.config_loader.reload_settings_from_db") as mock_reload: + with patch("app.utils.settings_sync._last_seen_version", ""): + handler_fn(sender=None) # call the signal handler + mock_reload.assert_called_once() + + def test_signal_handler_skips_reload_when_version_unchanged(self): + """Test that handler skips reload when version is same as last seen.""" + import app.utils.settings_sync as sync_module + from app.utils.settings_sync import register_settings_reload_signal + + handler_fn = None + + def capture_connect(fn=None, weak=None, **kwargs): + nonlocal handler_fn + if fn is not None: + handler_fn = fn + return fn + + def decorator(func): + nonlocal handler_fn + handler_fn = func + return func + + return decorator + + with patch("app.utils.settings_sync.task_prerun") as mock_signal: + mock_signal.connect = capture_connect + register_settings_reload_signal() + + assert handler_fn is not None + + version = "999.0" + mock_redis = MagicMock() + mock_redis.get.return_value = version.encode() + + with patch("app.utils.settings_sync.redis") as mock_redis_mod: + mock_redis_mod.from_url.return_value = mock_redis + with patch("app.utils.config_loader.reload_settings_from_db") as mock_reload: + # Set last seen to same version + sync_module._last_seen_version = version + handler_fn(sender=None) + mock_reload.assert_not_called() + + def test_signal_handler_handles_redis_failure(self): + """Test that signal handler skips gracefully on Redis failure.""" + from app.utils.settings_sync import register_settings_reload_signal + + handler_fn = None + + def capture_connect(fn=None, weak=None, **kwargs): + nonlocal handler_fn + if fn is not None: + handler_fn = fn + return fn + + def decorator(func): + nonlocal handler_fn + handler_fn = func + return func + + return decorator + + with patch("app.utils.settings_sync.task_prerun") as mock_signal: + mock_signal.connect = capture_connect + register_settings_reload_signal() + + assert handler_fn is not None + + with patch("app.utils.settings_sync.redis") as mock_redis_mod: + mock_redis_mod.from_url.side_effect = Exception("Redis down") + # Should not raise + handler_fn(sender=None) + + +# =========================================================================== +# app/api/logs.py – additional branches +# =========================================================================== + + +@pytest.mark.integration +class TestLogsApiAdditional: + """Additional tests for logs API to improve branch coverage.""" + + def test_list_logs_filter_by_file_id(self, client, db_session): + """Test filtering logs by file_id.""" + from app.models import FileRecord, ProcessingLog + + file_record = FileRecord( + filehash="hash123", + original_filename="test.pdf", + local_filename="/tmp/test.pdf", + file_size=1024, + mime_type="application/pdf", + ) + db_session.add(file_record) + db_session.commit() + + log = ProcessingLog( + file_id=file_record.id, + task_id="aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + step_name="process", + status="success", + message="Done", + ) + db_session.add(log) + db_session.commit() + + response = client.get(f"/api/logs?file_id={file_record.id}") + assert response.status_code == 200 + data = response.json() + assert all(entry["file_id"] == file_record.id for entry in data) + + def test_list_logs_with_null_timestamp(self, client, db_session): + """Test log with null timestamp serializes to None.""" + from app.models import ProcessingLog + + log = ProcessingLog( + task_id="bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", + step_name="step", + status="success", + message="msg", + ) + log.timestamp = None + db_session.add(log) + db_session.commit() + + response = client.get("/api/logs") + assert response.status_code == 200 + + def test_get_file_logs_returns_200(self, client, db_session): + """Test get file logs endpoint returns data.""" + from app.models import FileRecord, ProcessingLog + + file_record = FileRecord( + filehash="hash456", + original_filename="test2.pdf", + local_filename="/tmp/test2.pdf", + file_size=512, + mime_type="application/pdf", + ) + db_session.add(file_record) + db_session.commit() + + log = ProcessingLog( + file_id=file_record.id, + task_id="cccccccc-cccc-4ccc-8ccc-cccccccccccc", + step_name="step", + status="success", + message="msg", + ) + db_session.add(log) + db_session.commit() + + response = client.get(f"/api/logs/file/{file_record.id}") + assert response.status_code == 200 + data = response.json() + assert len(data["logs"]) >= 1 diff --git a/tests/test_views_filemanager.py b/tests/test_views_filemanager.py new file mode 100644 index 00000000..78f837e3 --- /dev/null +++ b/tests/test_views_filemanager.py @@ -0,0 +1,755 @@ +""" +Tests for app/views/filemanager.py module. + +Tests all helper functions and route handlers for the admin file manager. +""" + +import base64 +import json +import os +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from itsdangerous import TimestampSigner + +from app.models import FileRecord + +_SESSION_SECRET = "test_secret_key_for_testing_must_be_at_least_32_characters_long" + + +def _make_admin_session_cookie() -> str: + """Create a properly signed admin session cookie for tests.""" + session_data = {"user": {"id": "admin", "is_admin": True}} + signer = TimestampSigner(_SESSION_SECRET) + data = base64.b64encode(json.dumps(session_data).encode()).decode("utf-8") + return signer.sign(data).decode("utf-8") + + +@pytest.mark.unit +class TestFormatSize: + """Tests for _format_size helper function.""" + + def test_formats_bytes(self): + """Test formatting bytes.""" + from app.views.filemanager import _format_size + + assert _format_size(512) == "512.0 B" + + def test_formats_kilobytes(self): + """Test formatting kilobytes.""" + from app.views.filemanager import _format_size + + assert _format_size(1024) == "1.0 KB" + + def test_formats_megabytes(self): + """Test formatting megabytes.""" + from app.views.filemanager import _format_size + + result = _format_size(1024 * 1024) + assert result == "1.0 MB" + + def test_formats_gigabytes(self): + """Test formatting gigabytes.""" + from app.views.filemanager import _format_size + + result = _format_size(1024 * 1024 * 1024) + assert result == "1.0 GB" + + def test_formats_terabytes(self): + """Test formatting terabytes.""" + from app.views.filemanager import _format_size + + result = _format_size(1024 * 1024 * 1024 * 1024) + assert result == "1.0 TB" + + def test_formats_zero_bytes(self): + """Test formatting zero bytes.""" + from app.views.filemanager import _format_size + + assert _format_size(0) == "0.0 B" + + def test_formats_partial_kilobytes(self): + """Test formatting partial kilobytes.""" + from app.views.filemanager import _format_size + + result = _format_size(1500) + assert "KB" in result + + +@pytest.mark.unit +class TestSafePath: + """Tests for _safe_path helper function.""" + + def test_valid_empty_rel_path(self, tmp_path): + """Test valid empty relative path returns workdir.""" + from app.views.filemanager import _safe_path + + result = _safe_path(str(tmp_path), "") + assert result == tmp_path.resolve() + + def test_valid_nested_path(self, tmp_path): + """Test valid nested relative path.""" + from app.views.filemanager import _safe_path + + subdir = tmp_path / "subdir" + subdir.mkdir() + result = _safe_path(str(tmp_path), "subdir") + assert result == subdir.resolve() + + def test_traversal_attempt_raises(self, tmp_path): + """Test that path traversal raises ValueError.""" + from app.views.filemanager import _safe_path + + with pytest.raises(ValueError, match="Path traversal detected"): + _safe_path(str(tmp_path), "../../etc/passwd") + + def test_traversal_with_dots_raises(self, tmp_path): + """Test that .. in path raises ValueError when escaping workdir.""" + from app.views.filemanager import _safe_path + + with pytest.raises(ValueError, match="Path traversal detected"): + _safe_path(str(tmp_path), "../outside") + + def test_valid_deep_nested_path(self, tmp_path): + """Test valid deeply nested path.""" + from app.views.filemanager import _safe_path + + deep = tmp_path / "a" / "b" / "c" + deep.mkdir(parents=True) + result = _safe_path(str(tmp_path), "a/b/c") + assert result == deep.resolve() + + +@pytest.mark.unit +class TestFileIcon: + """Tests for _file_icon helper function.""" + + def test_directory_icon(self): + """Test directory returns folder icon.""" + from app.views.filemanager import _file_icon + + result = _file_icon("", True) + assert "fa-folder" in result + + def test_image_icon(self): + """Test image MIME type returns image icon.""" + from app.views.filemanager import _file_icon + + result = _file_icon("image/jpeg", False) + assert "fa-file-image" in result + + def test_pdf_icon(self): + """Test PDF MIME type returns PDF icon.""" + from app.views.filemanager import _file_icon + + result = _file_icon("application/pdf", False) + assert "fa-file-pdf" in result + + def test_text_icon(self): + """Test text MIME type returns text icon.""" + from app.views.filemanager import _file_icon + + result = _file_icon("text/plain", False) + assert "fa-file-alt" in result + + def test_json_icon(self): + """Test JSON MIME type returns code icon.""" + from app.views.filemanager import _file_icon + + result = _file_icon("application/json", False) + assert "fa-file-code" in result + + def test_default_icon(self): + """Test unknown MIME type returns default file icon.""" + from app.views.filemanager import _file_icon + + result = _file_icon("application/octet-stream", False) + assert "fa-file" in result + + def test_html_text_icon(self): + """Test HTML (text/) MIME type returns text icon.""" + from app.views.filemanager import _file_icon + + result = _file_icon("text/html", False) + assert "fa-file-alt" in result + + +@pytest.mark.unit +class TestDbPathSet: + """Tests for _db_path_set helper function.""" + + def test_empty_database(self, db_session): + """Test with empty database returns empty set.""" + from app.views.filemanager import _db_path_set + + result = _db_path_set(db_session) + assert isinstance(result, set) + assert len(result) == 0 + + def test_with_file_records(self, db_session, tmp_path): + """Test with file records returns their paths.""" + from app.views.filemanager import _db_path_set + + test_file = tmp_path / "test.pdf" + test_file.touch() + + record = FileRecord( + filehash="abc123", + original_filename="test.pdf", + local_filename=str(test_file), + file_size=1024, + mime_type="application/pdf", + ) + db_session.add(record) + db_session.commit() + + result = _db_path_set(db_session) + assert str(test_file.resolve()) in result + + def test_with_null_paths(self, db_session): + """Test that null paths are skipped (uses empty strings instead of None).""" + from app.views.filemanager import _db_path_set + + # Use a real file path but with empty original and processed paths + record = FileRecord( + filehash="def456", + original_filename="test.pdf", + local_filename="/tmp/test_def456.pdf", + original_file_path=None, + processed_file_path=None, + file_size=0, + mime_type="application/pdf", + ) + db_session.add(record) + db_session.commit() + + result = _db_path_set(db_session) + # Only local_filename (non-null) should be in the set + assert len(result) == 1 + + +@pytest.mark.unit +class TestScanDir: + """Tests for _scan_dir helper function.""" + + def test_empty_directory(self, tmp_path): + """Test scanning empty directory.""" + from app.views.filemanager import _scan_dir + + result = _scan_dir(tmp_path, tmp_path, set()) + assert result == [] + + def test_directory_with_file(self, tmp_path): + """Test scanning directory with a file.""" + from app.views.filemanager import _scan_dir + + test_file = tmp_path / "test.pdf" + test_file.write_text("content") + + result = _scan_dir(tmp_path, tmp_path, set()) + assert len(result) == 1 + assert result[0]["name"] == "test.pdf" + assert result[0]["is_dir"] is False + assert result[0]["db_status"] == "orphan" + + def test_file_in_db(self, tmp_path): + """Test that file in DB set gets in_db status.""" + from app.views.filemanager import _scan_dir + + test_file = tmp_path / "tracked.pdf" + test_file.write_text("content") + + db_paths = {str(test_file.resolve())} + result = _scan_dir(tmp_path, tmp_path, db_paths) + assert len(result) == 1 + assert result[0]["db_status"] == "in_db" + + def test_directory_entry(self, tmp_path): + """Test that subdirectory has empty db_status.""" + from app.views.filemanager import _scan_dir + + subdir = tmp_path / "subdir" + subdir.mkdir() + + result = _scan_dir(tmp_path, tmp_path, set()) + dirs = [e for e in result if e["is_dir"]] + assert len(dirs) == 1 + assert dirs[0]["db_status"] == "" + assert dirs[0]["size"] == "" + + def test_directories_sorted_first(self, tmp_path): + """Test that directories come before files.""" + from app.views.filemanager import _scan_dir + + (tmp_path / "zfile.txt").write_text("content") + (tmp_path / "adir").mkdir() + + result = _scan_dir(tmp_path, tmp_path, set()) + assert result[0]["is_dir"] is True + assert result[1]["is_dir"] is False + + def test_includes_mime_type_and_icon(self, tmp_path): + """Test that mime type and icon are set.""" + from app.views.filemanager import _scan_dir + + (tmp_path / "doc.pdf").write_text("pdf content") + + result = _scan_dir(tmp_path, tmp_path, set()) + assert result[0]["mime_type"] == "application/pdf" + assert "fa-file-pdf" in result[0]["icon"] + + +@pytest.mark.unit +class TestWalkAllFiles: + """Tests for _walk_all_files helper function.""" + + def test_empty_directory(self, tmp_path): + """Test walking empty directory.""" + from app.views.filemanager import _walk_all_files + + result = _walk_all_files(tmp_path, set()) + assert result == [] + + def test_with_files(self, tmp_path): + """Test walking directory with files.""" + from app.views.filemanager import _walk_all_files + + (tmp_path / "file1.txt").write_text("content") + (tmp_path / "file2.pdf").write_text("content") + + result = _walk_all_files(tmp_path, set()) + assert len(result) == 2 + names = {e["name"] for e in result} + assert "file1.txt" in names + assert "file2.pdf" in names + + def test_with_subdirectories(self, tmp_path): + """Test that subdirectory entries are skipped (only files).""" + from app.views.filemanager import _walk_all_files + + subdir = tmp_path / "subdir" + subdir.mkdir() + (subdir / "nested.pdf").write_text("content") + (tmp_path / "root.pdf").write_text("content") + + result = _walk_all_files(tmp_path, set()) + # Only files, not directories + assert all(not e["is_dir"] for e in result) + assert len(result) == 2 + + def test_db_status_in_db(self, tmp_path): + """Test that tracked files get in_db status.""" + from app.views.filemanager import _walk_all_files + + test_file = tmp_path / "tracked.pdf" + test_file.write_text("content") + + db_paths = {str(test_file.resolve())} + result = _walk_all_files(tmp_path, db_paths) + assert result[0]["db_status"] == "in_db" + + def test_db_status_orphan(self, tmp_path): + """Test that untracked files get orphan status.""" + from app.views.filemanager import _walk_all_files + + (tmp_path / "orphan.pdf").write_text("content") + + result = _walk_all_files(tmp_path, set()) + assert result[0]["db_status"] == "orphan" + + +@pytest.mark.unit +class TestDbRecords: + """Tests for _db_records helper function.""" + + def test_empty_database(self, db_session, tmp_path): + """Test with empty database.""" + from app.views.filemanager import _db_records + + result = _db_records(db_session, tmp_path) + assert result == [] + + def test_with_file_record(self, db_session, tmp_path): + """Test with file record in database.""" + from app.views.filemanager import _db_records + + test_file = tmp_path / "test.pdf" + test_file.write_text("content") + + record = FileRecord( + filehash="abc123", + original_filename="test.pdf", + local_filename=str(test_file), + file_size=1024, + mime_type="application/pdf", + ) + db_session.add(record) + db_session.commit() + + result = _db_records(db_session, tmp_path) + assert len(result) == 1 + assert result[0]["original_filename"] == "test.pdf" + assert result[0]["health"] == "ok" + + def test_missing_file_marked_as_missing(self, db_session, tmp_path): + """Test that files not on disk are marked as missing.""" + from app.views.filemanager import _db_records + + record = FileRecord( + filehash="def456", + original_filename="missing.pdf", + local_filename="/nonexistent/path/missing.pdf", + file_size=1024, + mime_type="application/pdf", + ) + db_session.add(record) + db_session.commit() + + result = _db_records(db_session, tmp_path) + assert len(result) == 1 + assert result[0]["health"] == "missing" + + def test_null_file_size_shows_dash(self, db_session, tmp_path): + """Test that zero file size and null mime_type show dashes.""" + from app.views.filemanager import _db_records + + # file_size=0 is falsy → shows "—"; mime_type=None also shows "—" + record = FileRecord( + filehash="ghi789", + original_filename="test.pdf", + local_filename="/tmp/test_ghi789.pdf", + file_size=0, + mime_type=None, + ) + db_session.add(record) + db_session.commit() + + result = _db_records(db_session, tmp_path) + assert result[0]["file_size"] == "—" + assert result[0]["mime_type"] == "—" + + def test_file_outside_workdir(self, db_session, tmp_path): + """Test handling of file path outside workdir.""" + from app.views.filemanager import _db_records + + record = FileRecord( + filehash="jkl012", + original_filename="outside.pdf", + local_filename="/completely/different/path/outside.pdf", + file_size=512, + mime_type="application/pdf", + ) + db_session.add(record) + db_session.commit() + + result = _db_records(db_session, tmp_path) + assert len(result) == 1 + # Path outside workdir should use full path as rel + assert result[0]["local"]["rel"] is not None + assert result[0]["local"]["exists"] is False + + +@pytest.mark.integration +class TestFilemanagerRoute: + """Integration tests for filemanager route.""" + + def test_redirects_non_admin(self, client): + """Test that non-admin users are redirected.""" + response = client.get("/admin/files", follow_redirects=False) + # Without admin session, should redirect to home + assert response.status_code in (200, 302, 303) + + def test_filesystem_view_with_admin_session(self, client): + """Test filesystem view with admin session cookie.""" + client.cookies.set("session", _make_admin_session_cookie()) + response = client.get("/admin/files?view=filesystem", follow_redirects=False) + assert response.status_code == 200 + + def test_database_view_with_admin_session(self, client): + """Test database view with admin session.""" + client.cookies.set("session", _make_admin_session_cookie()) + response = client.get("/admin/files?view=database", follow_redirects=False) + assert response.status_code == 200 + + def test_reconcile_view_with_admin_session(self, client): + """Test reconcile view with admin session.""" + client.cookies.set("session", _make_admin_session_cookie()) + response = client.get("/admin/files?view=reconcile", follow_redirects=False) + assert response.status_code == 200 + + def test_path_traversal_blocked(self, client): + """Test that path traversal is blocked.""" + client.cookies.set("session", _make_admin_session_cookie()) + response = client.get("/admin/files?view=filesystem&path=../../etc", follow_redirects=False) + # Should still return 200 (blocked silently, falls back to workdir root) + assert response.status_code == 200 + + def test_nonexistent_path_falls_back_to_root(self, client): + """Test that nonexistent path falls back to workdir root.""" + client.cookies.set("session", _make_admin_session_cookie()) + response = client.get("/admin/files?path=nonexistent_dir_xyz", follow_redirects=False) + assert response.status_code == 200 + + def test_with_breadcrumbs(self, client, tmp_path): + """Test that breadcrumbs are generated for nested paths.""" + client.cookies.set("session", _make_admin_session_cookie()) + + # Create a subdirectory in the actual workdir + workdir = os.environ.get("WORKDIR", "/tmp") + subdir = Path(workdir) / "testsubdir" + subdir.mkdir(exist_ok=True) + try: + response = client.get("/admin/files?path=testsubdir", follow_redirects=False) + assert response.status_code == 200 + finally: + subdir.rmdir() + + +@pytest.mark.integration +class TestFilemanagerDownloadRoute: + """Integration tests for filemanager download route.""" + + def test_redirects_non_admin(self, client): + """Test that non-admin users are redirected.""" + response = client.get("/admin/files/download?path=test.pdf", follow_redirects=False) + assert response.status_code in (200, 302, 303) + + def test_download_invalid_path_returns_400(self, client): + """Test that invalid (traversal) path returns 400.""" + client.cookies.set("session", _make_admin_session_cookie()) + response = client.get("/admin/files/download?path=../../etc/passwd", follow_redirects=False) + assert response.status_code == 400 + + def test_download_nonexistent_file_returns_404(self, client): + """Test that nonexistent file returns 404.""" + client.cookies.set("session", _make_admin_session_cookie()) + response = client.get("/admin/files/download?path=nonexistent_file_xyz.pdf", follow_redirects=False) + assert response.status_code == 404 + + def test_download_existing_file(self, client, tmp_path): + """Test downloading an existing file.""" + client.cookies.set("session", _make_admin_session_cookie()) + + # Create a file in the workdir + workdir = os.environ.get("WORKDIR", "/tmp") + test_file = Path(workdir) / "test_download_xyz.txt" + test_file.write_text("test content") + try: + response = client.get("/admin/files/download?path=test_download_xyz.txt", follow_redirects=False) + assert response.status_code == 200 + finally: + test_file.unlink(missing_ok=True) + + +@pytest.mark.unit +class TestFilemanagerRouteUnit: + """Unit tests for filemanager route handler.""" + + @patch("app.views.filemanager.templates") + @patch("app.views.filemanager.settings") + @pytest.mark.asyncio + async def test_filesystem_view_calls_scan_dir(self, mock_settings, mock_templates): + """Test filesystem view calls _scan_dir.""" + from app.views.filemanager import filemanager + + mock_settings.workdir = "/tmp" + mock_templates.TemplateResponse = MagicMock() + + mock_request = MagicMock() + mock_request.query_params.get = MagicMock( + side_effect=lambda key, default=None: "filesystem" if key == "view" else "" + ) + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + + mock_db = MagicMock() + mock_db.query.return_value.order_by.return_value.all.return_value = [] + mock_db.query.return_value.count.return_value = 0 + mock_db.query.return_value.all.return_value = [] + mock_db.query.return_value.filter.return_value.all.return_value = [] + + await filemanager(mock_request, mock_db) + mock_templates.TemplateResponse.assert_called_once() + + @patch("app.views.filemanager.templates") + @patch("app.views.filemanager.settings") + @pytest.mark.asyncio + async def test_database_view_calls_db_records(self, mock_settings, mock_templates): + """Test database view calls _db_records.""" + from app.views.filemanager import filemanager + + mock_settings.workdir = "/tmp" + mock_templates.TemplateResponse = MagicMock() + + mock_request = MagicMock() + mock_request.query_params.get = MagicMock( + side_effect=lambda key, default=None: "database" if key == "view" else "" + ) + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + + mock_db = MagicMock() + mock_db.query.return_value.order_by.return_value.all.return_value = [] + mock_db.query.return_value.count.return_value = 0 + mock_db.query.return_value.all.return_value = [] + mock_db.query.return_value.filter.return_value.all.return_value = [] + + await filemanager(mock_request, mock_db) + mock_templates.TemplateResponse.assert_called_once() + + @patch("app.views.filemanager.templates") + @patch("app.views.filemanager.settings") + @pytest.mark.asyncio + async def test_reconcile_view(self, mock_settings, mock_templates): + """Test reconcile view builds orphan and ghost lists.""" + from app.views.filemanager import filemanager + + mock_settings.workdir = "/tmp" + mock_templates.TemplateResponse = MagicMock() + + mock_request = MagicMock() + mock_request.query_params.get = MagicMock( + side_effect=lambda key, default=None: "reconcile" if key == "view" else "" + ) + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + + mock_db = MagicMock() + mock_db.query.return_value.order_by.return_value.all.return_value = [] + mock_db.query.return_value.count.return_value = 0 + mock_db.query.return_value.all.return_value = [] + + await filemanager(mock_request, mock_db) + mock_templates.TemplateResponse.assert_called_once() + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + assert "orphan_files" in context + assert "ghost_records" in context + + @patch("app.views.filemanager.templates") + @patch("app.views.filemanager.settings") + @pytest.mark.asyncio + async def test_path_traversal_falls_back_to_root(self, mock_settings, mock_templates): + """Test that path traversal attempt falls back to workdir root.""" + from app.views.filemanager import filemanager + + mock_settings.workdir = "/tmp" + mock_templates.TemplateResponse = MagicMock() + + mock_request = MagicMock() + mock_request.query_params.get = MagicMock( + side_effect=lambda key, default=None: "filesystem" if key == "view" else "../../etc" + ) + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + + mock_db = MagicMock() + mock_db.query.return_value.order_by.return_value.all.return_value = [] + mock_db.query.return_value.count.return_value = 0 + mock_db.query.return_value.all.return_value = [] + + # Should not raise, falls back to workdir root + await filemanager(mock_request, mock_db) + mock_templates.TemplateResponse.assert_called_once() + + @patch("app.views.filemanager.templates") + @patch("app.views.filemanager.settings") + @pytest.mark.asyncio + async def test_breadcrumbs_generated_for_nested_path(self, mock_settings, mock_templates, tmp_path): + """Test that breadcrumbs are generated for nested paths.""" + from app.views.filemanager import filemanager + + # Create nested dir for a valid path + nested = tmp_path / "level1" / "level2" + nested.mkdir(parents=True) + + mock_settings.workdir = str(tmp_path) + mock_templates.TemplateResponse = MagicMock() + + mock_request = MagicMock() + mock_request.query_params.get = MagicMock( + side_effect=lambda key, default=None: "filesystem" if key == "view" else "level1/level2" + ) + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + + mock_db = MagicMock() + mock_db.query.return_value.order_by.return_value.all.return_value = [] + mock_db.query.return_value.count.return_value = 0 + mock_db.query.return_value.all.return_value = [] + + await filemanager(mock_request, mock_db) + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + assert len(context["breadcrumbs"]) > 0 + + +@pytest.mark.unit +class TestFilemanagerDownloadUnit: + """Unit tests for filemanager_download route handler.""" + + @pytest.mark.asyncio + async def test_download_path_traversal_raises_400(self, tmp_path): + """Test that path traversal in download raises 400.""" + from fastapi import HTTPException + + from app.views.filemanager import filemanager_download + + with patch("app.views.filemanager.settings") as mock_settings: + mock_settings.workdir = str(tmp_path) + mock_request = MagicMock() + mock_request.query_params.get = MagicMock(return_value="../../etc/passwd") + + with pytest.raises(HTTPException) as exc_info: + await filemanager_download(mock_request) + assert exc_info.value.status_code == 400 + + @pytest.mark.asyncio + async def test_download_nonexistent_file_raises_404(self, tmp_path): + """Test that nonexistent file raises 404.""" + from fastapi import HTTPException + + from app.views.filemanager import filemanager_download + + with patch("app.views.filemanager.settings") as mock_settings: + mock_settings.workdir = str(tmp_path) + mock_request = MagicMock() + mock_request.query_params.get = MagicMock(return_value="nonexistent.pdf") + + with pytest.raises(HTTPException) as exc_info: + await filemanager_download(mock_request) + assert exc_info.value.status_code == 404 + + @pytest.mark.asyncio + async def test_download_directory_raises_404(self, tmp_path): + """Test that trying to download a directory raises 404.""" + from fastapi import HTTPException + + from app.views.filemanager import filemanager_download + + subdir = tmp_path / "subdir" + subdir.mkdir() + + with patch("app.views.filemanager.settings") as mock_settings: + mock_settings.workdir = str(tmp_path) + mock_request = MagicMock() + mock_request.query_params.get = MagicMock(return_value="subdir") + + with pytest.raises(HTTPException) as exc_info: + await filemanager_download(mock_request) + assert exc_info.value.status_code == 404 + + @pytest.mark.asyncio + async def test_download_valid_file_returns_file_response(self, tmp_path): + """Test that valid file returns FileResponse.""" + from fastapi.responses import FileResponse + + from app.views.filemanager import filemanager_download + + test_file = tmp_path / "download_me.pdf" + test_file.write_bytes(b"%PDF-1.4 content") + + with patch("app.views.filemanager.settings") as mock_settings: + mock_settings.workdir = str(tmp_path) + mock_request = MagicMock() + mock_request.query_params.get = MagicMock(return_value="download_me.pdf") + + result = await filemanager_download(mock_request) + assert isinstance(result, FileResponse) + assert result.filename == "download_me.pdf" diff --git a/tests/test_views_settings.py b/tests/test_views_settings.py index 5b8b7865..9f950cdd 100644 --- a/tests/test_views_settings.py +++ b/tests/test_views_settings.py @@ -1,11 +1,24 @@ """Tests for app/views/settings.py module.""" +import base64 +import json from unittest.mock import MagicMock, Mock, patch import pytest +from itsdangerous import TimestampSigner from app.views.settings import require_admin_access +_SESSION_SECRET = "test_secret_key_for_testing_must_be_at_least_32_characters_long" + + +def _make_admin_session_cookie() -> str: + """Create a properly signed session cookie with admin user for tests.""" + session_data = {"user": {"id": "admin", "is_admin": True}} + signer = TimestampSigner(_SESSION_SECRET) + data = base64.b64encode(json.dumps(session_data).encode()).decode("utf-8") + return signer.sign(data).decode("utf-8") + @pytest.mark.unit class TestRequireAdminAccess: @@ -198,3 +211,311 @@ class TestSettingsPageLogic: with pytest.raises(HTTPException) as exc_info: await settings_page(mock_request, mock_db) assert exc_info.value.status_code == 500 + + @patch("app.views.settings.get_all_settings_from_db") + @patch("app.views.settings.get_settings_by_category") + @patch("app.views.settings.get_setting_metadata") + @patch("app.views.settings.mask_sensitive_value") + @patch("app.views.settings.templates") + @patch("app.views.settings.settings") + @patch("app.views.settings.os.environ", {"WORKDIR": "/tmp"}) + @pytest.mark.asyncio + async def test_env_source_when_key_in_environ( + self, mock_settings, mock_templates, mock_mask, mock_metadata, mock_categories, mock_db_settings + ): + """Test that env source is used when key is in os.environ.""" + from app.views.settings import settings_page + + mock_db_settings.return_value = {} + mock_categories.return_value = {"General": ["workdir"]} + mock_metadata.return_value = {"type": "str", "sensitive": False} + mock_settings.workdir = "/tmp" + mock_settings.version = "1.0.0" + + mock_request = Mock() + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + mock_db = Mock() + + await settings_page(mock_request, mock_db) + mock_templates.TemplateResponse.assert_called_once() + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + settings_data = context["settings_data"] + workdir_entry = settings_data["General"][0] + assert workdir_entry["source"] == "environment" + + @patch("app.views.settings.get_all_settings_from_db") + @patch("app.views.settings.get_settings_by_category") + @patch("app.views.settings.get_setting_metadata") + @patch("app.views.settings.templates") + @patch("app.views.settings.settings") + @patch("app.views.settings.os.environ", {}) + @pytest.mark.asyncio + async def test_default_source_when_key_not_in_environ( + self, mock_settings, mock_templates, mock_metadata, mock_categories, mock_db_settings + ): + """Test that default source is used when key is not in environ or DB.""" + from app.views.settings import settings_page + + mock_db_settings.return_value = {} + mock_categories.return_value = {"General": ["workdir"]} + mock_metadata.return_value = {"type": "str", "sensitive": False} + mock_settings.workdir = "/app/workdir" + mock_settings.version = "1.0.0" + + mock_request = Mock() + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + mock_db = Mock() + + await settings_page(mock_request, mock_db) + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + settings_data = context["settings_data"] + workdir_entry = settings_data["General"][0] + assert workdir_entry["source"] == "default" + + @patch("app.views.settings.get_all_settings_from_db") + @patch("app.views.settings.get_settings_by_category") + @patch("app.views.settings.get_setting_metadata") + @patch("app.views.settings.mask_sensitive_value") + @patch("app.views.settings.templates") + @patch("app.views.settings.settings") + @pytest.mark.asyncio + async def test_sensitive_values_masked_in_settings_page( + self, mock_settings, mock_templates, mock_mask, mock_metadata, mock_categories, mock_db_settings + ): + """Test that sensitive values are masked in settings page.""" + from app.views.settings import settings_page + + mock_db_settings.return_value = {"openai_api_key": "sk-real-secret"} + mock_categories.return_value = {"AI": ["openai_api_key"]} + mock_metadata.return_value = {"type": "str", "sensitive": True} + mock_mask.return_value = "sk-****" + mock_settings.version = "1.0.0" + + mock_request = Mock() + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + mock_db = Mock() + + await settings_page(mock_request, mock_db) + mock_mask.assert_called() + + +@pytest.mark.unit +class TestCredentialsPage: + """Tests for credentials_page endpoint.""" + + @patch("app.views.settings.get_all_settings_from_db") + @patch("app.views.settings.SETTING_METADATA") + @patch("app.views.settings.templates") + @patch("app.views.settings.settings") + @pytest.mark.asyncio + async def test_credentials_page_success(self, mock_settings, mock_templates, mock_metadata, mock_db_settings): + """Test credentials page renders successfully.""" + from app.views.settings import credentials_page + + mock_db_settings.return_value = {} + mock_metadata.items.return_value = [ + ("openai_api_key", {"sensitive": True, "category": "AI", "description": "OpenAI key"}), + ("workdir", {"sensitive": False, "category": "General", "description": "Work dir"}), + ] + mock_settings.openai_api_key = None + mock_settings.version = "1.0.0" + + mock_request = Mock() + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + mock_db = Mock() + + await credentials_page(mock_request, mock_db) + mock_templates.TemplateResponse.assert_called_once() + call_args = mock_templates.TemplateResponse.call_args + assert call_args[0][0] == "credentials.html" + + @patch("app.views.settings.get_all_settings_from_db") + @patch("app.views.settings.SETTING_METADATA") + @patch("app.views.settings.templates") + @patch("app.views.settings.settings") + @pytest.mark.asyncio + async def test_credentials_page_source_db(self, mock_settings, mock_templates, mock_metadata, mock_db_settings): + """Test credentials page shows db source when credential is in database.""" + from app.views.settings import credentials_page + + mock_db_settings.return_value = {"openai_api_key": "sk-from-db"} + mock_metadata.items.return_value = [ + ("openai_api_key", {"sensitive": True, "category": "AI", "description": "API key"}), + ] + mock_settings.openai_api_key = "sk-from-env" + mock_settings.version = "1.0.0" + + mock_request = Mock() + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + mock_db = Mock() + + await credentials_page(mock_request, mock_db) + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + ai_creds = context["categories"].get("AI", []) + if ai_creds: + assert ai_creds[0]["source"] == "db" + assert ai_creds[0]["configured"] is True + + @patch("app.views.settings.get_all_settings_from_db") + @patch("app.views.settings.SETTING_METADATA") + @patch("app.views.settings.templates") + @patch("app.views.settings.settings") + @pytest.mark.asyncio + async def test_credentials_page_source_env(self, mock_settings, mock_templates, mock_metadata, mock_db_settings): + """Test credentials page shows env source when credential is only in env.""" + from app.views.settings import credentials_page + + mock_db_settings.return_value = {} + mock_metadata.items.return_value = [ + ("openai_api_key", {"sensitive": True, "category": "AI", "description": "API key"}), + ] + mock_settings.openai_api_key = "sk-from-env" + mock_settings.version = "1.0.0" + + mock_request = Mock() + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + mock_db = Mock() + + await credentials_page(mock_request, mock_db) + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + ai_creds = context["categories"].get("AI", []) + if ai_creds: + assert ai_creds[0]["source"] == "env" + + @patch("app.views.settings.get_all_settings_from_db") + @patch("app.views.settings.SETTING_METADATA") + @patch("app.views.settings.templates") + @patch("app.views.settings.settings") + @pytest.mark.asyncio + async def test_credentials_page_not_configured( + self, mock_settings, mock_templates, mock_metadata, mock_db_settings + ): + """Test credentials page shows not configured when no credential value.""" + from app.views.settings import credentials_page + + mock_db_settings.return_value = {} + mock_metadata.items.return_value = [ + ("openai_api_key", {"sensitive": True, "category": "AI", "description": "API key"}), + ] + mock_settings.openai_api_key = None + mock_settings.version = "1.0.0" + + mock_request = Mock() + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + mock_db = Mock() + + await credentials_page(mock_request, mock_db) + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + ai_creds = context["categories"].get("AI", []) + if ai_creds: + assert ai_creds[0]["source"] is None + assert ai_creds[0]["configured"] is False + + @patch("app.views.settings.get_all_settings_from_db") + @pytest.mark.asyncio + async def test_credentials_page_raises_500_on_error(self, mock_db_settings): + """Test credentials page raises 500 on error.""" + from fastapi import HTTPException + + from app.views.settings import credentials_page + + mock_db_settings.side_effect = Exception("DB error") + mock_request = Mock() + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + mock_db = Mock() + + with pytest.raises(HTTPException) as exc_info: + await credentials_page(mock_request, mock_db) + assert exc_info.value.status_code == 500 + + def test_credentials_page_redirects_non_admin(self, client): + """Test that non-admin users are redirected from credentials page.""" + response = client.get("/admin/credentials", follow_redirects=False) + assert response.status_code in (200, 302, 303) + + def test_credentials_page_accessible_with_admin(self, client): + """Test credentials page is accessible with admin session.""" + client.cookies.set("session", _make_admin_session_cookie()) + response = client.get("/admin/credentials", follow_redirects=False) + assert response.status_code == 200 + + +@pytest.mark.unit +class TestAuditLogPage: + """Tests for audit_log_page endpoint.""" + + @patch("app.utils.settings_service.get_audit_log") + @patch("app.views.settings.templates") + @patch("app.views.settings.settings") + @pytest.mark.asyncio + async def test_audit_log_page_success(self, mock_settings, mock_templates, mock_audit_log): + """Test audit log page renders successfully.""" + from app.views.settings import audit_log_page + + mock_audit_log.return_value = [] + mock_settings.version = "1.0.0" + + mock_request = Mock() + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + mock_db = Mock() + + await audit_log_page(mock_request, mock_db) + mock_templates.TemplateResponse.assert_called_once() + call_args = mock_templates.TemplateResponse.call_args + assert call_args[0][0] == "audit_log.html" + + @patch("app.utils.settings_service.get_audit_log") + @patch("app.views.settings.templates") + @patch("app.views.settings.settings") + @pytest.mark.asyncio + async def test_audit_log_page_with_entries(self, mock_settings, mock_templates, mock_audit_log): + """Test audit log page with actual entries.""" + from app.views.settings import audit_log_page + + mock_entries = [ + {"key": "workdir", "old_value": "/old", "new_value": "/new", "changed_by": "admin"}, + ] + mock_audit_log.return_value = mock_entries + mock_settings.version = "1.0.0" + + mock_request = Mock() + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + mock_db = Mock() + + await audit_log_page(mock_request, mock_db) + call_args = mock_templates.TemplateResponse.call_args + context = call_args[0][1] + assert len(context["entries"]) == 1 + + @patch("app.utils.settings_service.get_audit_log") + @pytest.mark.asyncio + async def test_audit_log_page_raises_500_on_error(self, mock_audit_log): + """Test audit log page raises 500 on error.""" + from fastapi import HTTPException + + from app.views.settings import audit_log_page + + mock_audit_log.side_effect = Exception("DB error") + mock_request = Mock() + mock_request.session = {"user": {"id": "admin", "is_admin": True}} + mock_db = Mock() + + with pytest.raises(HTTPException) as exc_info: + await audit_log_page(mock_request, mock_db) + assert exc_info.value.status_code == 500 + + def test_audit_log_redirects_non_admin(self, client): + """Test that non-admin users are redirected.""" + response = client.get("/admin/settings/audit-log", follow_redirects=False) + assert response.status_code in (200, 302, 303) + + def test_audit_log_accessible_with_admin(self, client): + """Test audit log page is accessible with admin session.""" + client.cookies.set("session", _make_admin_session_cookie()) + response = client.get("/admin/settings/audit-log", follow_redirects=False) + assert response.status_code == 200