|
| 1 | +""" |
| 2 | +Unit tests for attachment storage and serving functionality. |
| 3 | +
|
| 4 | +Tests the AttachmentStorage class and the serve_attachment endpoint. |
| 5 | +""" |
| 6 | + |
| 7 | +import pytest |
| 8 | +from unittest.mock import Mock, patch |
| 9 | +import sys |
| 10 | +import os |
| 11 | +import base64 |
| 12 | +import tempfile |
| 13 | +from pathlib import Path |
| 14 | + |
| 15 | +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))) |
| 16 | + |
| 17 | + |
| 18 | +class TestAttachmentStorage: |
| 19 | + """Tests for the AttachmentStorage class.""" |
| 20 | + |
| 21 | + def test_save_attachment_stores_metadata(self): |
| 22 | + """Test that save_attachment correctly stores file and metadata.""" |
| 23 | + from core.attachment_storage import AttachmentStorage |
| 24 | + |
| 25 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 26 | + with patch("core.attachment_storage.STORAGE_DIR", Path(tmpdir)): |
| 27 | + storage = AttachmentStorage() |
| 28 | + |
| 29 | + # Create test data (base64 URL-safe encoded) |
| 30 | + test_content = b"test image content" |
| 31 | + base64_data = base64.urlsafe_b64encode(test_content).decode() |
| 32 | + |
| 33 | + file_id = storage.save_attachment( |
| 34 | + base64_data=base64_data, |
| 35 | + filename="test.png", |
| 36 | + mime_type="image/png", |
| 37 | + ) |
| 38 | + |
| 39 | + # Verify file_id is returned |
| 40 | + assert file_id is not None |
| 41 | + assert len(file_id) == 36 # UUID format |
| 42 | + |
| 43 | + # Verify metadata is stored |
| 44 | + metadata = storage.get_attachment_metadata(file_id) |
| 45 | + assert metadata is not None |
| 46 | + assert metadata["filename"] == "test.png" |
| 47 | + assert metadata["mime_type"] == "image/png" |
| 48 | + assert metadata["size"] == len(test_content) |
| 49 | + |
| 50 | + def test_get_attachment_metadata_returns_none_for_unknown_id(self): |
| 51 | + """Test that get_attachment_metadata returns None for unknown file_id.""" |
| 52 | + from core.attachment_storage import AttachmentStorage |
| 53 | + |
| 54 | + storage = AttachmentStorage() |
| 55 | + metadata = storage.get_attachment_metadata("nonexistent-uuid") |
| 56 | + |
| 57 | + assert metadata is None |
| 58 | + |
| 59 | + def test_get_attachment_path_returns_path_for_valid_id(self): |
| 60 | + """Test that get_attachment_path returns correct path.""" |
| 61 | + from core.attachment_storage import AttachmentStorage |
| 62 | + |
| 63 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 64 | + with patch("core.attachment_storage.STORAGE_DIR", Path(tmpdir)): |
| 65 | + storage = AttachmentStorage() |
| 66 | + |
| 67 | + test_content = b"test content" |
| 68 | + base64_data = base64.urlsafe_b64encode(test_content).decode() |
| 69 | + |
| 70 | + file_id = storage.save_attachment( |
| 71 | + base64_data=base64_data, |
| 72 | + filename="test.txt", |
| 73 | + mime_type="text/plain", |
| 74 | + ) |
| 75 | + |
| 76 | + path = storage.get_attachment_path(file_id) |
| 77 | + assert path is not None |
| 78 | + assert path.exists() |
| 79 | + |
| 80 | + def test_save_attachment_without_filename_uses_default(self): |
| 81 | + """Test that save_attachment uses default filename when not provided.""" |
| 82 | + from core.attachment_storage import AttachmentStorage |
| 83 | + |
| 84 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 85 | + with patch("core.attachment_storage.STORAGE_DIR", Path(tmpdir)): |
| 86 | + storage = AttachmentStorage() |
| 87 | + |
| 88 | + test_content = b"test content" |
| 89 | + base64_data = base64.urlsafe_b64encode(test_content).decode() |
| 90 | + |
| 91 | + file_id = storage.save_attachment(base64_data=base64_data) |
| 92 | + |
| 93 | + metadata = storage.get_attachment_metadata(file_id) |
| 94 | + assert metadata["filename"] == "attachment" |
| 95 | + assert metadata["mime_type"] == "application/octet-stream" |
| 96 | + |
| 97 | + |
| 98 | +class TestServeAttachmentEndpoint: |
| 99 | + """Tests for the serve_attachment endpoint.""" |
| 100 | + |
| 101 | + @pytest.mark.asyncio |
| 102 | + async def test_serve_attachment_extracts_file_id_from_path_params(self): |
| 103 | + """Test that serve_attachment correctly extracts file_id from request.path_params.""" |
| 104 | + # Import the function (need to mock the storage) |
| 105 | + from core.server import serve_attachment |
| 106 | + |
| 107 | + # Create a mock Request object with path_params |
| 108 | + mock_request = Mock() |
| 109 | + mock_request.path_params = {"file_id": "test-uuid-1234"} |
| 110 | + |
| 111 | + # Mock the attachment storage to return valid metadata |
| 112 | + mock_storage = Mock() |
| 113 | + mock_storage.get_attachment_metadata.return_value = { |
| 114 | + "filename": "test.png", |
| 115 | + "mime_type": "image/png", |
| 116 | + "size": 1024, |
| 117 | + } |
| 118 | + mock_storage.get_attachment_path.return_value = Path("/tmp/test.png") |
| 119 | + |
| 120 | + # Patch where get_attachment_storage is imported from (inside the function) |
| 121 | + with patch( |
| 122 | + "core.attachment_storage.get_attachment_storage", |
| 123 | + return_value=mock_storage, |
| 124 | + ): |
| 125 | + with patch("core.server.FileResponse") as mock_file_response: |
| 126 | + mock_file_response.return_value = Mock() |
| 127 | + |
| 128 | + await serve_attachment(mock_request) |
| 129 | + |
| 130 | + # Verify get_attachment_metadata was called with the correct file_id |
| 131 | + mock_storage.get_attachment_metadata.assert_called_once_with( |
| 132 | + "test-uuid-1234" |
| 133 | + ) |
| 134 | + |
| 135 | + @pytest.mark.asyncio |
| 136 | + async def test_serve_attachment_returns_400_for_missing_file_id(self): |
| 137 | + """Test that serve_attachment returns 400 when file_id is missing.""" |
| 138 | + from core.server import serve_attachment |
| 139 | + |
| 140 | + mock_request = Mock() |
| 141 | + mock_request.path_params = {} # No file_id |
| 142 | + |
| 143 | + response = await serve_attachment(mock_request) |
| 144 | + |
| 145 | + assert response.status_code == 400 |
| 146 | + assert b"Missing file_id" in response.body |
| 147 | + |
| 148 | + @pytest.mark.asyncio |
| 149 | + async def test_serve_attachment_returns_404_for_unknown_file(self): |
| 150 | + """Test that serve_attachment returns 404 for unknown file_id.""" |
| 151 | + from core.server import serve_attachment |
| 152 | + |
| 153 | + mock_request = Mock() |
| 154 | + mock_request.path_params = {"file_id": "unknown-uuid"} |
| 155 | + |
| 156 | + mock_storage = Mock() |
| 157 | + mock_storage.get_attachment_metadata.return_value = None |
| 158 | + |
| 159 | + with patch( |
| 160 | + "core.attachment_storage.get_attachment_storage", |
| 161 | + return_value=mock_storage, |
| 162 | + ): |
| 163 | + response = await serve_attachment(mock_request) |
| 164 | + |
| 165 | + assert response.status_code == 404 |
| 166 | + assert b"not found" in response.body.lower() |
| 167 | + |
| 168 | + @pytest.mark.asyncio |
| 169 | + async def test_serve_attachment_uses_request_object_not_string(self): |
| 170 | + """ |
| 171 | + Regression test: Verify serve_attachment receives Request object |
| 172 | + and extracts file_id from path_params, not treating Request as file_id. |
| 173 | +
|
| 174 | + This was the bug: FastMCP's custom_route passes Request as first arg, |
| 175 | + but the function signature expected file_id: str directly. |
| 176 | + """ |
| 177 | + from core.server import serve_attachment |
| 178 | + import inspect |
| 179 | + |
| 180 | + # Verify the function signature expects a request parameter |
| 181 | + sig = inspect.signature(serve_attachment) |
| 182 | + params = list(sig.parameters.keys()) |
| 183 | + |
| 184 | + # The first parameter should be named 'request' (not 'file_id') |
| 185 | + assert params[0] == "request", ( |
| 186 | + "serve_attachment should accept 'request' as first parameter, " |
| 187 | + f"but got '{params[0]}'. FastMCP's custom_route passes Request object." |
| 188 | + ) |
0 commit comments