Tany Nguyen
refacoring
3bc48ea
Raw
History Blame Contribute Delete
2.11 kB
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,
)