diff --git a/api/tests/unit_tests/controllers/service_api/app/test_file_preview.py b/api/tests/unit_tests/controllers/service_api/app/test_file_preview.py index 14a00a9af82..33b6e83094e 100644 --- a/api/tests/unit_tests/controllers/service_api/app/test_file_preview.py +++ b/api/tests/unit_tests/controllers/service_api/app/test_file_preview.py @@ -1,17 +1,31 @@ -""" -Unit tests for Service API File Preview endpoint +"""Unit tests for the Service API file-preview endpoint. + +Ownership checks run against persisted message, file, app, and upload rows so the +tests exercise the same SQLAlchemy statements and tenant boundary as production. +Storage remains mocked because it is the external I/O boundary of the endpoint. """ import logging -import uuid +from collections.abc import Iterator +from dataclasses import dataclass +from datetime import datetime +from decimal import Decimal from typing import Protocol, cast from unittest.mock import Mock, patch +from uuid import uuid4 import pytest +from sqlalchemy import event +from sqlalchemy.engine import Engine +from sqlalchemy.orm import Session from controllers.service_api.app.error import FileAccessDeniedError, FileNotFoundError from controllers.service_api.app.file_preview import FilePreviewApi -from models.model import App, EndUser, Message, MessageFile, UploadFile +from extensions.storage.storage_type import StorageType +from graphon.file import FileTransferMethod, FileType +from models.base import TypeBase +from models.enums import ConversationFromSource, CreatorUserRole +from models.model import App, AppMode, Message, MessageFile, UploadFile class _FilePreviewLogRecord(Protocol): @@ -20,367 +34,252 @@ class _FilePreviewLogRecord(Protocol): error: str +@dataclass(frozen=True) +class _Database: + """Expose the real test session through the interface used by the controller.""" + + session: Session + + +@dataclass(frozen=True) +class _PreviewRecords: + app: App + message: Message + message_file: MessageFile + upload_file: UploadFile + + +@pytest.fixture +def database(sqlite_engine: Engine) -> Iterator[_Database]: + """Create only the tables required by file ownership validation.""" + + models = (App, Message, MessageFile, UploadFile) + tables = [TypeBase.metadata.tables[model.__tablename__] for model in models] + TypeBase.metadata.create_all(sqlite_engine, tables=tables) + with Session(sqlite_engine, expire_on_commit=False) as session: + yield _Database(session) + + +@pytest.fixture +def file_preview_api() -> FilePreviewApi: + """Create the resource instance under test.""" + + return FilePreviewApi() + + +def _upload_file(*, tenant_id: str, file_id: str | None = None) -> UploadFile: + upload_file = UploadFile( + tenant_id=tenant_id, + storage_type=StorageType.LOCAL, + key="storage/key/test_file.jpg", + name="test_file.jpg", + size=1024, + extension="jpg", + mime_type="image/jpeg", + created_by_role=CreatorUserRole.ACCOUNT, + created_by=str(uuid4()), + created_at=datetime(2026, 1, 1), + used=True, + ) + if file_id is not None: + upload_file.id = file_id + return upload_file + + +def _persist_preview_records( + session: Session, + *, + app_id: str | None = None, + app_tenant_id: str | None = None, + upload_tenant_id: str | None = None, +) -> _PreviewRecords: + app_id = app_id or str(uuid4()) + app_tenant_id = app_tenant_id or str(uuid4()) + upload_file = _upload_file(tenant_id=upload_tenant_id or app_tenant_id) + app = App( + id=app_id, + tenant_id=app_tenant_id, + name="Preview app", + description="", + mode=AppMode.CHAT, + icon_type=None, + icon="", + icon_background=None, + enable_site=True, + enable_api=True, + ) + message = Message( + id=str(uuid4()), + app_id=app_id, + conversation_id=str(uuid4()), + _inputs={}, + query="preview", + message={}, + message_unit_price=Decimal(0), + answer="answer", + answer_unit_price=Decimal(0), + currency="USD", + from_source=ConversationFromSource.API, + ) + message_file = MessageFile( + message_id=message.id, + type=FileType.IMAGE, + transfer_method=FileTransferMethod.LOCAL_FILE, + created_by_role=CreatorUserRole.ACCOUNT, + created_by=str(uuid4()), + upload_file_id=upload_file.id, + ) + session.add_all([app, message, message_file, upload_file]) + session.commit() + return _PreviewRecords(app=app, message=message, message_file=message_file, upload_file=upload_file) + + class TestFilePreviewApi: - """Test suite for FilePreviewApi""" + """Exercise ownership validation and response construction.""" - @pytest.fixture - def file_preview_api(self): - """Create FilePreviewApi instance for testing""" - return FilePreviewApi() + def test_validate_file_ownership_success(self, file_preview_api: FilePreviewApi, database: _Database): + records = _persist_preview_records(database.session) - @pytest.fixture - def mock_app(self): - """Mock App model""" - app = Mock(spec=App) - app.id = str(uuid.uuid4()) - app.tenant_id = str(uuid.uuid4()) - return app + with patch("controllers.service_api.app.file_preview.db", database): + message_file, upload_file = file_preview_api._validate_file_ownership( + records.upload_file.id, records.app.id + ) - @pytest.fixture - def mock_end_user(self): - """Mock EndUser model""" - end_user = Mock(spec=EndUser) - end_user.id = str(uuid.uuid4()) - return end_user + assert message_file.id == records.message_file.id + assert upload_file.id == records.upload_file.id + assert upload_file.tenant_id == records.app.tenant_id - @pytest.fixture - def mock_upload_file(self): - """Mock UploadFile model""" - upload_file = Mock(spec=UploadFile) - upload_file.id = str(uuid.uuid4()) - upload_file.name = "test_file.jpg" - upload_file.extension = "jpg" - upload_file.mime_type = "image/jpeg" - upload_file.size = 1024 - upload_file.key = "storage/key/test_file.jpg" - upload_file.tenant_id = str(uuid.uuid4()) - return upload_file + def test_validate_file_ownership_file_not_found(self, file_preview_api: FilePreviewApi, database: _Database): + with patch("controllers.service_api.app.file_preview.db", database): + with pytest.raises(FileNotFoundError, match="File not found in message context"): + file_preview_api._validate_file_ownership(str(uuid4()), str(uuid4())) - @pytest.fixture - def mock_message_file(self): - """Mock MessageFile model""" - message_file = Mock(spec=MessageFile) - message_file.id = str(uuid.uuid4()) - message_file.upload_file_id = str(uuid.uuid4()) - message_file.message_id = str(uuid.uuid4()) - return message_file + def test_validate_file_ownership_access_denied(self, file_preview_api: FilePreviewApi, database: _Database): + records = _persist_preview_records(database.session) - @pytest.fixture - def mock_message(self): - """Mock Message model""" - message = Mock(spec=Message) - message.id = str(uuid.uuid4()) - message.app_id = str(uuid.uuid4()) - return message + with patch("controllers.service_api.app.file_preview.db", database): + with pytest.raises(FileAccessDeniedError, match="not owned by requesting app"): + file_preview_api._validate_file_ownership(records.upload_file.id, str(uuid4())) - def test_validate_file_ownership_success( - self, file_preview_api: FilePreviewApi, mock_app, mock_upload_file, mock_message_file, mock_message - ): - """Test successful file ownership validation""" - file_id = str(uuid.uuid4()) - app_id = mock_app.id + def test_validate_file_ownership_upload_file_not_found(self, file_preview_api: FilePreviewApi, database: _Database): + records = _persist_preview_records(database.session) + database.session.delete(records.upload_file) + database.session.commit() - # Set up the mocks - mock_upload_file.tenant_id = mock_app.tenant_id - mock_message.app_id = app_id - mock_message_file.upload_file_id = file_id - mock_message_file.message_id = mock_message.id + with patch("controllers.service_api.app.file_preview.db", database): + with pytest.raises(FileNotFoundError, match="Upload file record not found"): + file_preview_api._validate_file_ownership(records.upload_file.id, records.app.id) - with patch("controllers.service_api.app.file_preview.db") as mock_db: - # Mock scalar() for MessageFile and Message queries - mock_db.session.scalar.side_effect = [ - mock_message_file, # MessageFile query - mock_message, # Message query - ] - # Mock get() for UploadFile and App PK lookups - mock_db.session.get.side_effect = [ - mock_upload_file, # UploadFile query - mock_app, # App query for tenant validation - ] + def test_validate_file_ownership_tenant_mismatch(self, file_preview_api: FilePreviewApi, database: _Database): + records = _persist_preview_records(database.session, upload_tenant_id=str(uuid4())) - # Execute the method - result_message_file, result_upload_file = file_preview_api._validate_file_ownership(file_id, app_id) - - # Assertions - assert result_message_file == mock_message_file - assert result_upload_file == mock_upload_file - - def test_validate_file_ownership_file_not_found(self, file_preview_api: FilePreviewApi): - """Test file ownership validation when MessageFile not found""" - file_id = str(uuid.uuid4()) - app_id = str(uuid.uuid4()) - - with patch("controllers.service_api.app.file_preview.db") as mock_db: - # Mock MessageFile not found via scalar() - mock_db.session.scalar.return_value = None - - # Execute and assert exception - with pytest.raises(FileNotFoundError) as exc_info: - file_preview_api._validate_file_ownership(file_id, app_id) - - assert "File not found in message context" in str(exc_info.value) - - def test_validate_file_ownership_access_denied(self, file_preview_api: FilePreviewApi, mock_message_file): - """Test file ownership validation when Message not owned by app""" - file_id = str(uuid.uuid4()) - app_id = str(uuid.uuid4()) - - with patch("controllers.service_api.app.file_preview.db") as mock_db: - # Mock MessageFile found but Message not owned by app via scalar() - mock_db.session.scalar.side_effect = [ - mock_message_file, # MessageFile query - found - None, # Message query - not found (access denied) - ] - - # Execute and assert exception - with pytest.raises(FileAccessDeniedError) as exc_info: - file_preview_api._validate_file_ownership(file_id, app_id) - - assert "not owned by requesting app" in str(exc_info.value) - - def test_validate_file_ownership_upload_file_not_found( - self, file_preview_api: FilePreviewApi, mock_message_file, mock_message - ): - """Test file ownership validation when UploadFile not found""" - file_id = str(uuid.uuid4()) - app_id = str(uuid.uuid4()) - - with patch("controllers.service_api.app.file_preview.db") as mock_db: - # Mock scalar() for MessageFile and Message - mock_db.session.scalar.side_effect = [ - mock_message_file, # MessageFile query - found - mock_message, # Message query - found - ] - # Mock get() for UploadFile - not found - mock_db.session.get.return_value = None - - # Execute and assert exception - with pytest.raises(FileNotFoundError) as exc_info: - file_preview_api._validate_file_ownership(file_id, app_id) - - assert "Upload file record not found" in str(exc_info.value) - - def test_validate_file_ownership_tenant_mismatch( - self, file_preview_api: FilePreviewApi, mock_app, mock_upload_file, mock_message_file, mock_message - ): - """Test file ownership validation with tenant mismatch""" - file_id = str(uuid.uuid4()) - app_id = mock_app.id - - # Set up tenant mismatch - mock_upload_file.tenant_id = "different_tenant_id" - mock_app.tenant_id = "app_tenant_id" - mock_message.app_id = app_id - mock_message_file.upload_file_id = file_id - mock_message_file.message_id = mock_message.id - - with patch("controllers.service_api.app.file_preview.db") as mock_db: - # Mock scalar() for MessageFile and Message queries - mock_db.session.scalar.side_effect = [ - mock_message_file, # MessageFile query - mock_message, # Message query - ] - # Mock get() for UploadFile and App PK lookups - mock_db.session.get.side_effect = [ - mock_upload_file, # UploadFile query - mock_app, # App query for tenant validation - ] - - # Execute and assert exception - with pytest.raises(FileAccessDeniedError) as exc_info: - file_preview_api._validate_file_ownership(file_id, app_id) - - assert "tenant mismatch" in str(exc_info.value) + with patch("controllers.service_api.app.file_preview.db", database): + with pytest.raises(FileAccessDeniedError, match="tenant mismatch"): + file_preview_api._validate_file_ownership(records.upload_file.id, records.app.id) def test_validate_file_ownership_invalid_input(self, file_preview_api: FilePreviewApi): - """Test file ownership validation with invalid input""" - - # Test with empty file_id - with pytest.raises(FileAccessDeniedError) as exc_info: + with pytest.raises(FileAccessDeniedError, match="Invalid file or app identifier"): file_preview_api._validate_file_ownership("", "app_id") - assert "Invalid file or app identifier" in str(exc_info.value) - # Test with empty app_id - with pytest.raises(FileAccessDeniedError) as exc_info: + with pytest.raises(FileAccessDeniedError, match="Invalid file or app identifier"): file_preview_api._validate_file_ownership("file_id", "") - assert "Invalid file or app identifier" in str(exc_info.value) - def test_build_file_response_basic(self, file_preview_api: FilePreviewApi, mock_upload_file): - """Test basic file response building""" - mock_generator = Mock() + @pytest.mark.parametrize( + ("as_attachment", "mime_type", "name", "extension", "size"), + [ + (False, "image/jpeg", "test_file.jpg", "jpg", 1024), + (True, "image/jpeg", "test_file.jpg", "jpg", 1024), + (False, "text/html", "unsafe.html", "html", 1024), + (False, "video/mp4", "test_file.mp4", "mp4", 1024), + (False, "image/jpeg", "test_file.jpg", "jpg", 0), + ], + ) + def test_build_file_response( + self, + file_preview_api: FilePreviewApi, + as_attachment: bool, + mime_type: str, + name: str, + extension: str, + size: int, + ): + upload_file = _upload_file(tenant_id=str(uuid4())) + upload_file.mime_type = mime_type + upload_file.name = name + upload_file.extension = extension + upload_file.size = size - response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False) + response = file_preview_api._build_file_response(Mock(), upload_file, as_attachment) - # Check response properties - assert response.mimetype == mock_upload_file.mime_type assert response.direct_passthrough is True - assert response.headers["Content-Length"] == str(mock_upload_file.size) assert "Cache-Control" in response.headers - - def test_build_file_response_as_attachment(self, file_preview_api: FilePreviewApi, mock_upload_file): - """Test file response building with attachment flag""" - mock_generator = Mock() - - response = file_preview_api._build_file_response(mock_generator, mock_upload_file, True) - - # Check attachment-specific headers - assert "attachment" in response.headers["Content-Disposition"] - assert mock_upload_file.name in response.headers["Content-Disposition"] - assert response.headers["Content-Type"] == "application/octet-stream" - - def test_build_file_response_html_forces_attachment(self, file_preview_api: FilePreviewApi, mock_upload_file): - """Test HTML files are forced to download""" - mock_generator = Mock() - mock_upload_file.mime_type = "text/html" - mock_upload_file.name = "unsafe.html" - mock_upload_file.extension = "html" - - response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False) - - assert "attachment" in response.headers["Content-Disposition"] - assert response.headers["Content-Type"] == "application/octet-stream" - assert response.headers["X-Content-Type-Options"] == "nosniff" - - def test_build_file_response_audio_video(self, file_preview_api: FilePreviewApi, mock_upload_file): - """Test file response building for audio/video files""" - mock_generator = Mock() - mock_upload_file.mime_type = "video/mp4" - - response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False) - - # Check Range support for media files - assert response.headers["Accept-Ranges"] == "bytes" - - def test_build_file_response_no_size(self, file_preview_api: FilePreviewApi, mock_upload_file): - """Test file response building when size is unknown""" - mock_generator = Mock() - mock_upload_file.size = 0 # Unknown size - - response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False) - - # Content-Length should not be set when size is unknown - assert "Content-Length" not in response.headers + assert ("Content-Length" in response.headers) is bool(size) + if as_attachment or mime_type == "text/html": + assert "attachment" in response.headers["Content-Disposition"] + assert response.headers["Content-Type"] == "application/octet-stream" + else: + assert response.mimetype == mime_type + if mime_type == "text/html": + assert response.headers["X-Content-Type-Options"] == "nosniff" + if mime_type.startswith("video/"): + assert response.headers["Accept-Ranges"] == "bytes" @patch("controllers.service_api.app.file_preview.storage") - def test_get_method_integration( - self, - mock_storage, - file_preview_api: FilePreviewApi, - mock_app, - mock_end_user, - mock_upload_file, - mock_message_file, - mock_message, + def test_components_use_validated_file( + self, mock_storage: Mock, file_preview_api: FilePreviewApi, database: _Database ): - """Test the full GET method integration (without decorator)""" - file_id = str(uuid.uuid4()) - app_id = mock_app.id + records = _persist_preview_records(database.session) + generator = Mock() - # Set up mocks - mock_upload_file.tenant_id = mock_app.tenant_id - mock_message.app_id = app_id - mock_message_file.upload_file_id = file_id - mock_message_file.message_id = mock_message.id + with patch("controllers.service_api.app.file_preview.db", database): + message_file, upload_file = file_preview_api._validate_file_ownership( + records.upload_file.id, records.app.id + ) + response = file_preview_api._build_file_response(generator, upload_file, False) - mock_generator = Mock() - mock_storage.load.return_value = mock_generator - - with patch("controllers.service_api.app.file_preview.db") as mock_db: - # Mock scalar() for MessageFile and Message queries - mock_db.session.scalar.side_effect = [ - mock_message_file, # MessageFile query - mock_message, # Message query - ] - # Mock get() for UploadFile and App PK lookups - mock_db.session.get.side_effect = [ - mock_upload_file, # UploadFile query - mock_app, # App query for tenant validation - ] - - # Test the core logic directly without Flask decorators - # Validate file ownership - result_message_file, result_upload_file = file_preview_api._validate_file_ownership(file_id, app_id) - assert result_message_file == mock_message_file - assert result_upload_file == mock_upload_file - - # Test file response building - response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False) - assert response is not None - - # Verify storage was called correctly - mock_storage.load.assert_not_called() # Since we're testing components separately + assert message_file.id == records.message_file.id + assert response.mimetype == "image/jpeg" + mock_storage.load.assert_not_called() @patch("controllers.service_api.app.file_preview.storage") - def test_storage_error_handling( - self, - mock_storage, - file_preview_api: FilePreviewApi, - mock_app, - mock_upload_file, - mock_message_file, - mock_message, + def test_storage_error_remains_external( + self, mock_storage: Mock, file_preview_api: FilePreviewApi, database: _Database ): - """Test storage error handling in the core logic""" - file_id = str(uuid.uuid4()) - app_id = mock_app.id + records = _persist_preview_records(database.session) + mock_storage.load.side_effect = OSError("Storage error") - # Set up mocks - mock_upload_file.tenant_id = mock_app.tenant_id - mock_message.app_id = app_id - mock_message_file.upload_file_id = file_id - mock_message_file.message_id = mock_message.id + with patch("controllers.service_api.app.file_preview.db", database): + _, upload_file = file_preview_api._validate_file_ownership(records.upload_file.id, records.app.id) - # Mock storage error - mock_storage.load.side_effect = Exception("Storage error") - - with patch("controllers.service_api.app.file_preview.db") as mock_db: - # Mock scalar() for MessageFile and Message queries - mock_db.session.scalar.side_effect = [ - mock_message_file, # MessageFile query - mock_message, # Message query - ] - # Mock get() for UploadFile and App PK lookups - mock_db.session.get.side_effect = [ - mock_upload_file, # UploadFile query - mock_app, # App query for tenant validation - ] - - # First validate file ownership works - result_message_file, result_upload_file = file_preview_api._validate_file_ownership(file_id, app_id) - assert result_message_file == mock_message_file - assert result_upload_file == mock_upload_file - - # Test storage error handling - with pytest.raises(Exception) as exc_info: - mock_storage.load(mock_upload_file.key, stream=True) - - assert "Storage error" in str(exc_info.value) + with pytest.raises(OSError, match="Storage error"): + mock_storage.load(upload_file.key, stream=True) def test_validate_file_ownership_unexpected_error_logging( - self, file_preview_api: FilePreviewApi, caplog: pytest.LogCaptureFixture + self, + file_preview_api: FilePreviewApi, + database: _Database, + sqlite_engine: Engine, + caplog: pytest.LogCaptureFixture, ): - """Test that unexpected errors are logged properly""" - file_id = str(uuid.uuid4()) - app_id = str(uuid.uuid4()) + file_id = str(uuid4()) + app_id = str(uuid4()) - with patch("controllers.service_api.app.file_preview.db") as mock_db: - # Mock database scalar to raise unexpected exception - mock_db.session.scalar.side_effect = Exception("Unexpected database error") + def fail_statement(*_args: object) -> None: + raise RuntimeError("Unexpected database error") - # Execute and assert exception - with caplog.at_level(logging.ERROR, logger="controllers.service_api.app.file_preview"): - with pytest.raises(FileAccessDeniedError) as exc_info: - file_preview_api._validate_file_ownership(file_id, app_id) + event.listen(sqlite_engine, "before_cursor_execute", fail_statement) + try: + with patch("controllers.service_api.app.file_preview.db", database): + with caplog.at_level(logging.ERROR, logger="controllers.service_api.app.file_preview"): + with pytest.raises(FileAccessDeniedError, match="File access validation failed"): + file_preview_api._validate_file_ownership(file_id, app_id) + finally: + event.remove(sqlite_engine, "before_cursor_execute", fail_statement) - # Verify error message - assert "File access validation failed" in str(exc_info.value) - - # Verify logging was called with the structured context fields. The ``extra`` keys - # are attached to the LogRecord as attributes, so they are not in ``caplog.text``. - assert len(caplog.records) == 1 - log_record = caplog.records[0] - assert log_record.getMessage() == "Unexpected error during file ownership validation" - record = cast(_FilePreviewLogRecord, log_record) - assert record.file_id == file_id - assert record.app_id == app_id - assert record.error == "Unexpected database error" + assert len(caplog.records) == 1 + log_record = caplog.records[0] + assert log_record.getMessage() == "Unexpected error during file ownership validation" + record = cast(_FilePreviewLogRecord, log_record) + assert record.file_id == file_id + assert record.app_id == app_id + assert record.error == "Unexpected database error"