Frank Schwenk
c29186a227
33 tests covering env/io helpers, config parsers, pipeline meta I/O, worker inbox logic, and FastAPI smoke routes. No gmic/rembg subprocesses. Co-authored-by: Cursor <cursoragent@cursor.com>
81 lines
2.2 KiB
Python
81 lines
2.2 KiB
Python
"""Tests for pipeline pure helpers and meta I/O."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from app import pipeline
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("raw", "expected"),
|
|
[
|
|
("photo.jpg", "photo"),
|
|
("../evil", "evil"),
|
|
("", "photo"),
|
|
(" spaced ", "spaced"),
|
|
("foo@bar", "foo_bar"),
|
|
],
|
|
)
|
|
def test_sanitize_stem(raw: str, expected: str) -> None:
|
|
assert pipeline.sanitize_stem(raw) == expected
|
|
|
|
|
|
def test_variant_filename() -> None:
|
|
assert pipeline.variant_filename("photo", "v1") == "photo_v1.png"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("command", "expected_tail"),
|
|
[
|
|
("fx_blur 3", ["fx_blur", "3"]),
|
|
("normalize", ["normalize"]),
|
|
],
|
|
)
|
|
def test_gmic_filter_args(tmp_path: Path, command: str, expected_tail: list[str]) -> None:
|
|
image = tmp_path / "in.png"
|
|
image.touch()
|
|
args = pipeline.gmic_filter_args(image, command)
|
|
assert args[0] == str(image)
|
|
assert args[1:] == expected_tail
|
|
|
|
|
|
def test_next_variant_id() -> None:
|
|
manifest = {"variants": [{"id": "v1"}, {"id": "v2"}]}
|
|
assert pipeline.next_variant_id(manifest) == "v3"
|
|
|
|
|
|
def test_next_variant_id_after_gap() -> None:
|
|
manifest = {"variants": [{"id": "v1"}, {"id": "v3"}]}
|
|
assert pipeline.next_variant_id(manifest) == "v4"
|
|
|
|
|
|
def test_strip_ansi() -> None:
|
|
assert pipeline.strip_ansi("\x1b[31merror\x1b[0m") == "error"
|
|
|
|
|
|
def test_read_write_meta_roundtrip(data_dir: Path) -> None:
|
|
paths = pipeline.job_paths("event01")
|
|
pipeline.write_status(paths, "processing", source_file="event01.jpg")
|
|
meta = pipeline.read_meta("event01")
|
|
assert meta is not None
|
|
assert meta["status"] == "processing"
|
|
assert meta["source_file"] == "event01.jpg"
|
|
|
|
status = pipeline.read_status("event01")
|
|
assert status is not None
|
|
assert status["status"] == "processing"
|
|
|
|
manifest = pipeline.read_manifest("event01")
|
|
assert manifest is not None
|
|
assert manifest["variants"] == []
|
|
|
|
|
|
def test_list_job_ids(data_dir: Path) -> None:
|
|
paths = pipeline.job_paths("a")
|
|
pipeline.write_status(paths, "done")
|
|
paths = pipeline.job_paths("b")
|
|
pipeline.write_status(paths, "processing")
|
|
assert sorted(pipeline.list_job_ids()) == ["a", "b"]
|