198 lines
7.2 KiB
Python
198 lines
7.2 KiB
Python
# #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"}]
|
|
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"})
|
|
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_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_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
|