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.
143 lines
4.8 KiB
143 lines
4.8 KiB
from pathlib import Path
|
|
import shutil
|
|
|
|
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")
|
|
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
|
|
|