Frank Schwenk
233 lines
8.0 KiB
Python
233 lines
8.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from imagepipeline.core.exceptions import CycleError, ValidationError
|
|
from imagepipeline.core.manifest import PipelineManifest, write_manifest
|
|
from imagepipeline.core.params import Param, validate_params
|
|
from imagepipeline.core.pipeline import Pipeline
|
|
from imagepipeline.core.runner import PipelineRunner
|
|
from imagepipeline.core.step import StepDefinition
|
|
from imagepipeline.modules.base import BaseModule
|
|
from imagepipeline.modules.registry import get_module, register, unregister
|
|
from imagepipeline.utils.files import list_images, match_by_stem
|
|
from tests.conftest import make_png
|
|
|
|
|
|
class TestParams:
|
|
def test_defaults_applied(self) -> None:
|
|
schema = {"colorspace": Param("string", default="Gray")}
|
|
result = validate_params(schema, {})
|
|
assert result == {"colorspace": "Gray"}
|
|
|
|
def test_unknown_param_rejected(self) -> None:
|
|
schema = {"a": Param("int", default=1)}
|
|
with pytest.raises(ValueError, match="Unknown parameters"):
|
|
validate_params(schema, {"b": 2})
|
|
|
|
def test_choices_enforced(self) -> None:
|
|
schema = {"mode": Param("string", choices=("a", "b"))}
|
|
with pytest.raises(ValueError, match="must be one of"):
|
|
validate_params(schema, {"mode": "c"})
|
|
|
|
|
|
class TestFileMatching:
|
|
def test_match_by_stem_pairs_sources(self, tmp_path: Path) -> None:
|
|
a = tmp_path / "a"
|
|
b = tmp_path / "b"
|
|
a.mkdir()
|
|
b.mkdir()
|
|
make_png(a / "img1.png")
|
|
make_png(a / "img2.png")
|
|
make_png(b / "img1.png")
|
|
make_png(b / "img2.png")
|
|
|
|
groups = match_by_stem([list_images(a), list_images(b)])
|
|
assert len(groups) == 2
|
|
assert groups[0][0].parent == a
|
|
assert groups[0][1].parent == b
|
|
|
|
def test_missing_stem_raises(self, tmp_path: Path) -> None:
|
|
a = tmp_path / "a"
|
|
b = tmp_path / "b"
|
|
a.mkdir()
|
|
b.mkdir()
|
|
make_png(a / "only_a.png")
|
|
make_png(b / "only_b.png")
|
|
|
|
with pytest.raises(ValueError, match="Cannot match images by stem"):
|
|
match_by_stem([list_images(a), list_images(b)])
|
|
|
|
|
|
class TestPipelineRunner:
|
|
def test_cycle_detection(self, input_dir: Path, output_base: Path) -> None:
|
|
@register
|
|
class DummyModule(BaseModule):
|
|
name = "dummy_cycle"
|
|
|
|
def run(self, ctx) -> None:
|
|
pass
|
|
|
|
step_a = StepDefinition(
|
|
step_id="dummy_cycle_01",
|
|
module_name="dummy_cycle",
|
|
module=DummyModule,
|
|
input_refs=["dummy_cycle_02"],
|
|
params={},
|
|
output_dir_name="dummy_cycle_01",
|
|
)
|
|
step_b = StepDefinition(
|
|
step_id="dummy_cycle_02",
|
|
module_name="dummy_cycle",
|
|
module=DummyModule,
|
|
input_refs=["dummy_cycle_01"],
|
|
params={},
|
|
output_dir_name="dummy_cycle_02",
|
|
)
|
|
|
|
runner = PipelineRunner(
|
|
name="cycle_test",
|
|
input_dir=input_dir,
|
|
output_base=output_base,
|
|
steps=[step_a, step_b],
|
|
)
|
|
with pytest.raises(CycleError):
|
|
runner.run()
|
|
|
|
unregister("dummy_cycle")
|
|
|
|
def test_dag_execution_order(self, input_dir: Path, output_base: Path) -> None:
|
|
order: list[str] = []
|
|
|
|
@register
|
|
class OrderModule(BaseModule):
|
|
name = "order_tracker"
|
|
|
|
def run(self, ctx) -> None:
|
|
order.append(ctx.step_id)
|
|
ctx.output_dir.mkdir(parents=True, exist_ok=True)
|
|
for src in ctx.input_paths:
|
|
shutil.copy2(src, ctx.output_dir / src.name)
|
|
|
|
with Pipeline(name="order_test", input_dir=input_dir, output_base=output_base, verbose=False) as p:
|
|
step_b = p.step("order_tracker", inputs="input")
|
|
step_a = p.step("order_tracker", inputs=step_b)
|
|
p.run()
|
|
|
|
assert order == ["order_tracker_01", "order_tracker_02"]
|
|
|
|
unregister("order_tracker")
|
|
|
|
def test_duplicate_module_numbering(self, input_dir: Path, output_base: Path) -> None:
|
|
@register
|
|
class NumberModule(BaseModule):
|
|
name = "number_tracker"
|
|
|
|
def run(self, ctx) -> None:
|
|
ctx.output_dir.mkdir(parents=True, exist_ok=True)
|
|
for src in ctx.input_paths:
|
|
shutil.copy2(src, ctx.output_dir / src.name)
|
|
|
|
with Pipeline(name="dup_test", input_dir=input_dir, output_base=output_base, verbose=False) as p:
|
|
first = p.step("number_tracker", inputs="input")
|
|
p.step("number_tracker", inputs=first)
|
|
root = p.run()
|
|
|
|
assert (root / "number_tracker_01").is_dir()
|
|
assert (root / "number_tracker_02").is_dir()
|
|
|
|
unregister("number_tracker")
|
|
|
|
|
|
class TestManifest:
|
|
def test_write_manifest(self, tmp_path: Path) -> None:
|
|
manifest = PipelineManifest(
|
|
name="test",
|
|
output_root=str(tmp_path),
|
|
input_dir="/in",
|
|
started_at="2026-01-01T00:00:00+00:00",
|
|
finished_at="2026-01-01T00:00:01+00:00",
|
|
)
|
|
path = tmp_path / "pipeline_manifest.json"
|
|
write_manifest(path, manifest)
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
assert data["name"] == "test"
|
|
assert data["finished_at"] is not None
|
|
|
|
|
|
class TestPipelineValidation:
|
|
def test_empty_pipeline_rejected(self, input_dir: Path) -> None:
|
|
with Pipeline(name="empty", input_dir=input_dir, verbose=False) as p:
|
|
with pytest.raises(ValidationError, match="no steps"):
|
|
p.run()
|
|
|
|
def test_unknown_module_rejected(self, input_dir: Path) -> None:
|
|
with Pipeline(name="bad", input_dir=input_dir, verbose=False) as p:
|
|
with pytest.raises(KeyError, match="Unknown module"):
|
|
p.step("does_not_exist", inputs="input")
|
|
|
|
|
|
has_imagemagick = bool(shutil.which("magick") or shutil.which("convert"))
|
|
|
|
|
|
class TestPipelineLogging:
|
|
@pytest.mark.skipif(not has_imagemagick, reason="ImageMagick not installed")
|
|
def test_verbose_output(self, input_dir: Path, output_base: Path, capsys) -> None:
|
|
with Pipeline(
|
|
name="log_test",
|
|
input_dir=input_dir,
|
|
output_base=output_base,
|
|
verbose=True,
|
|
) as p:
|
|
p.step("imagemagick_grayscale", inputs="input")
|
|
p.run()
|
|
|
|
output = capsys.readouterr().out
|
|
assert "Found 2 photo(s)" in output
|
|
assert "Step 1/1: imagemagick_grayscale_01 (imagemagick_grayscale)" in output
|
|
assert "Applying module imagemagick_grayscale to image [1/2]" in output
|
|
assert "Pipeline finished." in output
|
|
|
|
@pytest.mark.skipif(not has_imagemagick, reason="ImageMagick not installed")
|
|
def test_quiet_output(self, input_dir: Path, output_base: Path, capsys) -> None:
|
|
with Pipeline(
|
|
name="quiet_test",
|
|
input_dir=input_dir,
|
|
output_base=output_base,
|
|
verbose=False,
|
|
) as p:
|
|
p.step("imagemagick_grayscale", inputs="input")
|
|
p.run()
|
|
|
|
assert capsys.readouterr().out == ""
|
|
|
|
|
|
@pytest.mark.skipif(not has_imagemagick, reason="ImageMagick not installed")
|
|
class TestImageMagickGrayscaleIntegration:
|
|
def test_grayscale_pipeline(self, input_dir: Path, output_base: Path) -> None:
|
|
with Pipeline(
|
|
name="gray_integration",
|
|
input_dir=input_dir,
|
|
output_base=output_base,
|
|
verbose=False,
|
|
) as p:
|
|
p.step("imagemagick_grayscale", inputs="input")
|
|
root = p.run()
|
|
|
|
manifest_path = root / "pipeline_manifest.json"
|
|
assert manifest_path.is_file()
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
assert len(manifest["steps"]) == 1
|
|
assert manifest["steps"][0]["module"] == "imagemagick_grayscale"
|
|
|
|
out_dir = root / "imagemagick_grayscale_01"
|
|
outputs = list_images(out_dir)
|
|
assert len(outputs) == 2
|
|
|
|
module = get_module("imagemagick_grayscale")
|
|
module.check_dependencies()
|