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()