from __future__ import annotations import asyncio import sys import tempfile import unittest import zipfile from io import BytesIO from pathlib import Path import httpx sys.path.insert(0, str(Path(__file__).resolve().parents[3])) from backend.app.dependencies import get_file_ops_service from backend.app.fs.filesystem_adapter import FilesystemAdapter from backend.app.main import app from backend.app.security.path_guard import PathGuard from backend.app.services.file_ops_service import FileOpsService class DownloadApiGoldenTest(unittest.TestCase): def setUp(self) -> None: self.temp_dir = tempfile.TemporaryDirectory() self.root = Path(self.temp_dir.name) / "root" self.root.mkdir(parents=True, exist_ok=True) path_guard = PathGuard({"storage1": str(self.root), "storage2": str(self.root)}) service = FileOpsService(path_guard=path_guard, filesystem=FilesystemAdapter()) async def _override_file_ops_service() -> FileOpsService: return service app.dependency_overrides[get_file_ops_service] = _override_file_ops_service def tearDown(self) -> None: app.dependency_overrides.clear() self.temp_dir.cleanup() def _get(self, url: str) -> httpx.Response: async def _run() -> httpx.Response: transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: return await client.get(url) return asyncio.run(_run()) def test_download_success_for_allowed_file(self) -> None: src = self.root / "report.txt" src.write_text("hello download", encoding="utf-8") response = self._get("/api/files/download?path=storage1/report.txt") self.assertEqual(response.status_code, 200) self.assertEqual(response.content, b"hello download") self.assertIn('attachment; filename="report.txt"', response.headers.get("content-disposition", "")) self.assertEqual(response.headers.get("content-type"), "text/plain; charset=utf-8") def test_download_directory_type_conflict(self) -> None: (self.root / "docs").mkdir() (self.root / "docs" / "a.txt").write_text("a", encoding="utf-8") response = self._get("/api/files/download?path=storage1/docs") self.assertEqual(response.status_code, 200) self.assertIn('attachment; filename="docs.zip"', response.headers.get("content-disposition", "")) with zipfile.ZipFile(BytesIO(response.content)) as archive: self.assertIn("docs/", archive.namelist()) self.assertIn("docs/a.txt", archive.namelist()) self.assertEqual(archive.read("docs/a.txt"), b"a") def test_download_multi_file_selection_as_zip(self) -> None: (self.root / "a.txt").write_text("A", encoding="utf-8") (self.root / "b.txt").write_text("B", encoding="utf-8") response = self._get("/api/files/download?path=storage1/a.txt&path=storage1/b.txt") self.assertEqual(response.status_code, 200) self.assertRegex( response.headers.get("content-disposition", ""), r'attachment; filename="kodidownload-\d{8}-\d{6}\.zip"', ) with zipfile.ZipFile(BytesIO(response.content)) as archive: self.assertIn("a.txt", archive.namelist()) self.assertIn("b.txt", archive.namelist()) self.assertEqual(archive.read("a.txt"), b"A") self.assertEqual(archive.read("b.txt"), b"B") def test_download_multi_directory_selection_as_zip(self) -> None: (self.root / "dir1" / "sub").mkdir(parents=True) (self.root / "dir2").mkdir() (self.root / "dir1" / "sub" / "a.txt").write_text("A", encoding="utf-8") (self.root / "dir2" / "b.txt").write_text("B", encoding="utf-8") response = self._get("/api/files/download?path=storage1/dir1&path=storage1/dir2") self.assertEqual(response.status_code, 200) self.assertRegex( response.headers.get("content-disposition", ""), r'attachment; filename="kodidownload-\d{8}-\d{6}\.zip"', ) with zipfile.ZipFile(BytesIO(response.content)) as archive: self.assertIn("dir1/", archive.namelist()) self.assertIn("dir1/sub/", archive.namelist()) self.assertIn("dir1/sub/a.txt", archive.namelist()) self.assertIn("dir2/b.txt", archive.namelist()) def test_download_mixed_file_and_directory_selection_as_zip(self) -> None: (self.root / "readme.txt").write_text("R", encoding="utf-8") (self.root / "photos" / "nested").mkdir(parents=True) (self.root / "photos" / "nested" / "img.txt").write_text("P", encoding="utf-8") response = self._get("/api/files/download?path=storage1/readme.txt&path=storage1/photos") self.assertEqual(response.status_code, 200) self.assertRegex( response.headers.get("content-disposition", ""), r'attachment; filename="kodidownload-\d{8}-\d{6}\.zip"', ) with zipfile.ZipFile(BytesIO(response.content)) as archive: self.assertIn("readme.txt", archive.namelist()) self.assertIn("photos/", archive.namelist()) self.assertIn("photos/nested/img.txt", archive.namelist()) def test_download_directory_with_symlink_rejected(self) -> None: target = self.root / "real.txt" target.write_text("x", encoding="utf-8") (self.root / "docs").mkdir() (self.root / "docs" / "link.txt").symlink_to(target) response = self._get("/api/files/download?path=storage1/docs") self.assertEqual(response.status_code, 409) self.assertEqual(response.json()["error"]["code"], "type_conflict") def test_download_path_not_found(self) -> None: response = self._get("/api/files/download?path=storage1/missing.txt") self.assertEqual(response.status_code, 404) self.assertEqual(response.json()["error"]["code"], "path_not_found") def test_download_invalid_root_alias(self) -> None: response = self._get("/api/files/download?path=unknown/file.txt") self.assertEqual(response.status_code, 403) self.assertEqual(response.json()["error"]["code"], "invalid_root_alias") def test_download_traversal_blocked(self) -> None: response = self._get("/api/files/download?path=storage1/../etc/passwd") self.assertEqual(response.status_code, 403) self.assertEqual(response.json()["error"]["code"], "path_traversal_detected") if __name__ == "__main__": unittest.main()