import base64 import shutil import threading import xml.etree.ElementTree as ET from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import pytest from alembic import command from alembic.config import Config from httpx import ASGITransport, AsyncClient from app.core.config import get_settings from app.db.session import create_engine, create_session_factory, dispose_engine from app.main import create_app from app.adapters.storage.s3 import S3StorageAdapter from app.services.directory_service import DirectoryService from app.services.file_service import CreateFileMetadataInput, FileService @pytest.fixture def migrated_database_url(tmp_path: Path) -> str: database_path = tmp_path / "test.db" database_url = f"sqlite+aiosqlite:///{database_path}" alembic_config = Config("alembic.ini") alembic_config.set_main_option("sqlalchemy.url", database_url) command.upgrade(alembic_config, "head") return database_url @pytest.fixture def runtime_paths(tmp_path: Path) -> dict[str, str]: return { "upload_temp_dir": str(tmp_path / "uploads"), "local_storage_dir": str(tmp_path / "local-storage"), "cache_dir": str(tmp_path / "cache"), "fake_s3_dir": str(tmp_path / "fake-s3"), } @pytest.fixture async def test_app( monkeypatch: pytest.MonkeyPatch, migrated_database_url: str, runtime_paths: dict[str, str], ): monkeypatch.setenv("IRON_DATABASE_URL", migrated_database_url) monkeypatch.setenv("IRON_UPLOAD_TEMP_DIR", runtime_paths["upload_temp_dir"]) monkeypatch.setenv("IRON_LOCAL_STORAGE_DIR", runtime_paths["local_storage_dir"]) monkeypatch.setenv("IRON_CACHE_DIR", runtime_paths["cache_dir"]) monkeypatch.setenv("IRON_BOOTSTRAP_USERNAME", "admin") monkeypatch.setenv("IRON_BOOTSTRAP_PASSWORD", "changeme-iron") monkeypatch.setenv("IRON_SECRET_KEY", "test-secret-key") get_settings.cache_clear() create_engine.cache_clear() app = create_app() async with app.router.lifespan_context(app): yield app await dispose_engine() get_settings.cache_clear() @pytest.fixture async def raw_api_client(test_app): transport = ASGITransport(app=test_app) async with AsyncClient(transport=transport, base_url="http://testserver") as client: yield client @pytest.fixture async def api_client(raw_api_client: AsyncClient): login_response = await raw_api_client.post( "/api/auth/login", json={"username": "admin", "password": "changeme-iron"}, ) assert login_response.status_code == 200 token = login_response.json()["access_token"] raw_api_client.headers["Authorization"] = f"Bearer {token}" yield raw_api_client raw_api_client.headers.pop("Authorization", None) @pytest.fixture def session_factory(): return create_session_factory() @pytest.fixture async def create_file_metadata(session_factory): async def _create_file( *, directory_id: str = "dir_root", name: str = "movie.mp4", mime_type: str | None = "video/mp4", size_bytes: int = 1048576, content_hash: str = "hash_movie_mp4", ): async with session_factory() as session: directory_service = DirectoryService(session) await directory_service.ensure_root_directory() file_service = FileService(session) return await file_service.create_file_metadata( CreateFileMetadataInput( directory_id=directory_id, name=name, mime_type=mime_type, size_bytes=size_bytes, content_hash=content_hash, ) ) return _create_file @pytest.fixture def fake_s3(monkeypatch: pytest.MonkeyPatch, runtime_paths: dict[str, str]): base = Path(runtime_paths["fake_s3_dir"]) base.mkdir(parents=True, exist_ok=True) async def _check(self): return {"status": "healthy", "detail": self.config["bucket"]} async def _put_file(self, source_path: str, storage_key: str): destination = base / self.config["bucket"] / storage_key destination.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(source_path, destination) return {"etag": "fake-etag", "storage_key": storage_key} async def _download_file(self, storage_key: str, destination_path: Path): source = base / self.config["bucket"] / storage_key if not source.exists(): raise FileNotFoundError(storage_key) destination_path.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(source, destination_path) return destination_path async def _object_exists(self, storage_key: str): return (base / self.config["bucket"] / storage_key).exists() monkeypatch.setattr(S3StorageAdapter, "check", _check) monkeypatch.setattr(S3StorageAdapter, "put_file", _put_file) monkeypatch.setattr(S3StorageAdapter, "download_file", _download_file) monkeypatch.setattr(S3StorageAdapter, "object_exists", _object_exists) return base @pytest.fixture def fake_webdav(): files: dict[str, bytes] = {} directories = {"", "iron"} auth_header = "Basic " + base64.b64encode(b"iron:secret-password").decode("ascii") class Handler(BaseHTTPRequestHandler): def log_message(self, format, *args): # type: ignore[override] return def _require_auth(self) -> bool: if self.headers.get("Authorization") == auth_header: return True self.send_response(401) self.end_headers() return False def _relative_path(self) -> str: return self.path.removeprefix("/dav").strip("/") def do_PROPFIND(self): # type: ignore[override] if not self._require_auth(): return relative_path = self._relative_path() if relative_path not in directories and relative_path not in files: self.send_response(404) self.end_headers() return multistatus = ET.Element("{DAV:}multistatus") response = ET.SubElement(multistatus, "{DAV:}response") href = ET.SubElement(response, "{DAV:}href") href.text = f"/dav/{relative_path}" if relative_path else "/dav/" body = ET.tostring(multistatus, encoding="utf-8") self.send_response(207) self.send_header("Content-Type", "application/xml") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_MKCOL(self): # type: ignore[override] if not self._require_auth(): return relative_path = self._relative_path() directories.add(relative_path) self.send_response(201) self.end_headers() def do_PUT(self): # type: ignore[override] if not self._require_auth(): return relative_path = self._relative_path() length = int(self.headers.get("Content-Length", "0")) files[relative_path] = self.rfile.read(length) parent = "/".join(relative_path.split("/")[:-1]) directories.add(parent) self.send_response(201) self.send_header("ETag", "fake-webdav-etag") self.end_headers() def do_GET(self): # type: ignore[override] if not self._require_auth(): return relative_path = self._relative_path() if relative_path not in files: self.send_response(404) self.end_headers() return body = files[relative_path] self.send_response(200) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_HEAD(self): # type: ignore[override] if not self._require_auth(): return relative_path = self._relative_path() if relative_path in files: self.send_response(200) else: self.send_response(404) self.end_headers() server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() base_url = f"http://127.0.0.1:{server.server_address[1]}/dav" yield {"endpoint_url": base_url, "username": "iron", "password": "secret-password", "files": files, "directories": directories} server.shutdown() server.server_close()