You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
240 lines
8.5 KiB
240 lines
8.5 KiB
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()
|
|
|