| from __future__ import annotations |
|
|
| from typing import BinaryIO |
| from pathlib import Path |
| from types import SimpleNamespace |
|
|
| from app.domain.schemas import PipelineDefaultsModel |
| from app.domain.types import RuntimeSettings |
|
|
|
|
| class FakeProcessRemoteImage: |
| def __init__(self) -> None: |
| self.calls: list[object] = [] |
|
|
| def process(self, request) -> dict[str, object]: |
| self.calls.append(request) |
| return {"jobId": "test-remote-job"} |
|
|
|
|
| class FakeProcessUploadedImage: |
| def __init__(self) -> None: |
| self.calls: list[tuple[str | None, bytes, object]] = [] |
|
|
| def process(self, file_name: str | None, content_stream: BinaryIO, request) -> dict[str, object]: |
| if hasattr(content_stream, "seek"): |
| content_stream.seek(0) |
| self.calls.append((file_name, content_stream.read(), request)) |
| return {"jobId": "test-upload-job"} |
|
|
|
|
| class FailingProcessRemoteImage: |
| def process(self, request) -> dict[str, object]: |
| raise RuntimeError("remote failed") |
|
|
|
|
| class InvalidProcessRemoteImage: |
| def process(self, request) -> dict[str, object]: |
| raise ValueError("invalid source") |
|
|
|
|
| def make_container( |
| root: Path, |
| *, |
| process_remote_image, |
| process_uploaded_image, |
| api_key: str | None = None, |
| ): |
| return SimpleNamespace( |
| settings=RuntimeSettings( |
| host="127.0.0.1", |
| port=8000, |
| api_key=api_key, |
| output_dir=root / "output", |
| save_output=True, |
| device="cpu", |
| max_input_side=1024, |
| restore_original_size=True, |
| jpeg_quality=90, |
| brightness_skip_threshold=0.6, |
| hvi_model_id="Fediory/HVI-CIDNet-FiveK", |
| hvi_gamma=0.8, |
| hvi_alpha_s=1.0, |
| hvi_alpha_i=1.0, |
| default_formats=("avif", "jpg"), |
| config_path=root / "pipeline.config.json", |
| r2_config=None, |
| ), |
| defaults=PipelineDefaultsModel(sizes=[]), |
| process_remote_image=process_remote_image, |
| process_uploaded_image=process_uploaded_image, |
| ) |
|
|