# #region Test.Api.Storage [C:3] [TYPE Module] [SEMANTICS test,storage,routes] # @BRIEF Tests for storage API routes — list, upload, delete, download, get-by-path. # @RELATION BINDS_TO -> [storage_routes] import os os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") os.environ.setdefault("AUTH_DATABASE_URL", "sqlite:///:memory:") os.environ.setdefault("SECRET_KEY", "test-secret-key-for-tests") os.environ.setdefault("DEV_MODE", "true") import sys from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi import FastAPI from fastapi.testclient import TestClient _src = str(Path(__file__).resolve().parent.parent.parent / "src") if _src not in sys.path: sys.path.insert(0, _src) P = "/api/storage" def _make_client(overrides: dict | None = None) -> tuple[TestClient, MagicMock]: from src.api.routes.storage import router from src.dependencies import get_plugin_loader, get_current_user from src.schemas.auth import User, RoleSchema app = FastAPI() app.include_router(router, prefix="/api/storage") mock_user = User( id="admin-1", username="admin", email="admin@x.com", auth_source="LOCAL", created_at=__import__("datetime").datetime.now(), roles=[RoleSchema(id="r1", name="Admin", description="", permissions=[])], ) mock_plugin_loader = MagicMock() app.dependency_overrides[get_current_user] = lambda: mock_user app.dependency_overrides[get_plugin_loader] = lambda: mock_plugin_loader if overrides: for dep, fn in overrides.items(): app.dependency_overrides[dep] = fn return TestClient(app), mock_plugin_loader class TestListFiles: """GET /api/storage/files""" def test_success(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.list_files.return_value = [{ "name": "file1.txt", "path": "/files/file1.txt", "size": 1024, "created_at": "2024-01-01T00:00:00", "category": "backups", "mime_type": "text/plain", }] mock_loader.get_plugin.return_value = mock_plugin resp = client.get(f"{P}/files") assert resp.status_code == 200 def test_with_filters(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.list_files.return_value = [] mock_loader.get_plugin.return_value = mock_plugin resp = client.get(f"{P}/files?category=backups&path=/test&recursive=true") assert resp.status_code == 200 mock_plugin.list_files.assert_called_once_with("backups", "/test", True) def test_plugin_not_loaded(self): client, mock_loader = _make_client() mock_loader.get_plugin.return_value = None resp = client.get(f"{P}/files") assert resp.status_code == 500 class TestUploadFile: """POST /api/storage/upload""" def test_success(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.save_file = AsyncMock(return_value={ "name": "test.txt", "path": "/uploads/test.txt", "size": 5, "created_at": "2024-01-01T00:00:00", "category": "backups", "mime_type": "text/plain", }) mock_loader.get_plugin.return_value = mock_plugin resp = client.post(f"{P}/upload", data={"category": "backups"}, files={"file": ("test.txt", b"hello")}) assert resp.status_code == 201 def test_value_error(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.save_file = AsyncMock(side_effect=ValueError("bad")) mock_loader.get_plugin.return_value = mock_plugin resp = client.post(f"{P}/upload", data={"category": "backups"}, files={"file": ("test.txt", b"data")}) assert resp.status_code == 400 def test_plugin_not_loaded(self): client, mock_loader = _make_client() mock_loader.get_plugin.return_value = None resp = client.post(f"{P}/upload", data={"category": "backups"}, files={"file": ("test.txt", b"data")}) assert resp.status_code == 500 class TestDeleteFile: """DELETE /api/storage/files/{category}/{path}""" def test_success(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_loader.get_plugin.return_value = mock_plugin resp = client.delete(f"{P}/files/backups/test.txt") assert resp.status_code == 204 def test_not_found(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.delete_file.side_effect = FileNotFoundError mock_loader.get_plugin.return_value = mock_plugin resp = client.delete(f"{P}/files/backups/missing.txt") assert resp.status_code == 404 def test_value_error(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.delete_file.side_effect = ValueError("bad") mock_loader.get_plugin.return_value = mock_plugin resp = client.delete(f"{P}/files/backups/bad.txt") assert resp.status_code == 400 def test_plugin_not_loaded(self): client, mock_loader = _make_client() mock_loader.get_plugin.return_value = None resp = client.delete(f"{P}/files/backups/test.txt") assert resp.status_code == 500 class TestDownloadFile: """GET /api/storage/download/{category}/{path}""" def test_success(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.get_file_path.return_value = "/tmp/storage/backups/test.txt" mock_loader.get_plugin.return_value = mock_plugin with patch("src.api.routes.storage.FileResponse") as MockFR: MockFR.return_value = {"status": "ok"} resp = client.get(f"{P}/download/backups/test.txt") assert resp.status_code == 200 mock_plugin.get_file_path.assert_called_once_with("backups", "test.txt") def test_not_found(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.get_file_path.side_effect = FileNotFoundError mock_loader.get_plugin.return_value = mock_plugin resp = client.get(f"{P}/download/backups/missing.txt") assert resp.status_code == 404 def test_value_error(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.get_file_path.side_effect = ValueError("bad") mock_loader.get_plugin.return_value = mock_plugin resp = client.get(f"{P}/download/backups/bad.txt") assert resp.status_code == 400 def test_plugin_not_loaded(self): client, mock_loader = _make_client() mock_loader.get_plugin.return_value = None resp = client.get(f"{P}/download/backups/test.txt") assert resp.status_code == 500 class TestGetFileByPath: """GET /api/storage/file""" def test_success_relative_path(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.get_storage_root.return_value = Path("/storage/root") mock_plugin.validate_path.return_value = Path("/storage/root/test.txt") mock_loader.get_plugin.return_value = mock_plugin with patch("src.api.routes.storage.FileResponse") as MockFR, \ patch("pathlib.Path.exists", return_value=True), \ patch("pathlib.Path.is_file", return_value=True): MockFR.return_value = {"status": "ok"} resp = client.get(f"{P}/file?path=test.txt") assert resp.status_code == 200 def test_success_absolute_path(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.validate_path.return_value = Path("/valid/path/file.txt") mock_loader.get_plugin.return_value = mock_plugin with patch("src.api.routes.storage.FileResponse") as MockFR, \ patch("pathlib.Path.exists", return_value=True), \ patch("pathlib.Path.is_file", return_value=True), \ patch("pathlib.Path.is_absolute", return_value=True): MockFR.return_value = {"status": "ok"} resp = client.get(f"{P}/file?path=/absolute/path/file.txt") assert resp.status_code == 200 mock_plugin.validate_path.assert_called_once() def test_missing_path(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_loader.get_plugin.return_value = mock_plugin resp = client.get(f"{P}/file?path=") assert resp.status_code == 400 def test_file_not_found(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.validate_path.return_value = Path("/missing.txt") mock_loader.get_plugin.return_value = mock_plugin with patch("pathlib.Path.exists", return_value=False): resp = client.get(f"{P}/file?path=missing.txt") assert resp.status_code == 404 def test_value_error_from_validate(self): client, mock_loader = _make_client() mock_plugin = MagicMock() mock_plugin.validate_path.side_effect = ValueError("bad") mock_loader.get_plugin.return_value = mock_plugin resp = client.get(f"{P}/file?path=../../../etc/passwd") assert resp.status_code == 400 def test_plugin_not_loaded(self): client, mock_loader = _make_client() mock_loader.get_plugin.return_value = None resp = client.get(f"{P}/file?path=test.txt") assert resp.status_code == 500 # #endregion Test.Api.Storage