Frank Schwenk
a60a18a253
Introduce ruff lint/format config, expand .gitignore, and reformat Python sources. Co-authored-by: Cursor <cursoragent@cursor.com>
258 lines
9.1 KiB
Python
258 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from imagepipeline.core.context import ModuleContext
|
|
from imagepipeline.modules.registry import get_module, list_modules
|
|
from imagepipeline.modules.xcf_stack import XcfStackModule
|
|
from imagepipeline.utils.files import find_image_by_stem
|
|
from tests.conftest import make_png
|
|
|
|
has_gimp = bool(shutil.which("gimp-console") or shutil.which("gimp"))
|
|
has_magick = bool(shutil.which("magick") or shutil.which("convert"))
|
|
|
|
|
|
class TestFindImageByStem:
|
|
def test_finds_matching_stem_case_insensitively(self, tmp_path: Path) -> None:
|
|
make_png(tmp_path / "Photo.PNG")
|
|
found = find_image_by_stem(tmp_path, "photo")
|
|
assert found == tmp_path / "Photo.PNG"
|
|
|
|
def test_returns_none_when_missing(self, tmp_path: Path) -> None:
|
|
make_png(tmp_path / "other.png")
|
|
assert find_image_by_stem(tmp_path, "photo") is None
|
|
|
|
def test_handles_extension_mismatch(self, tmp_path: Path) -> None:
|
|
make_png(tmp_path / "photo.png")
|
|
# Input may be photo.jpg while step output is photo.png (stem-only match).
|
|
found = find_image_by_stem(tmp_path, Path("photo.jpg").stem)
|
|
assert found == tmp_path / "photo.png"
|
|
|
|
|
|
class TestModuleRegistration:
|
|
def test_xcf_stack_registered(self) -> None:
|
|
assert "xcf_stack" in list_modules()
|
|
|
|
def test_get_module_returns_xcf_stack_class(self) -> None:
|
|
assert get_module("xcf_stack") is XcfStackModule
|
|
|
|
def test_xcf_stack_waits_for_layer_inputs(self) -> None:
|
|
from imagepipeline.core.runner import PipelineRunner
|
|
from imagepipeline.core.step import StepDefinition
|
|
from imagepipeline.modules.imagemagick_grayscale import ImageMagickGrayscale
|
|
|
|
steps = [
|
|
StepDefinition(
|
|
step_id="xcf_stack_01",
|
|
module_name="xcf_stack",
|
|
module=XcfStackModule,
|
|
input_refs=["input", "imagemagick_grayscale_01"],
|
|
params={},
|
|
output_dir_name="xcf_stack_01",
|
|
),
|
|
StepDefinition(
|
|
step_id="imagemagick_grayscale_01",
|
|
module_name="imagemagick_grayscale",
|
|
module=ImageMagickGrayscale,
|
|
input_refs=["input"],
|
|
params={},
|
|
output_dir_name="imagemagick_grayscale_01",
|
|
),
|
|
]
|
|
runner = PipelineRunner(
|
|
name="order",
|
|
input_dir=Path("/tmp/unused"),
|
|
output_base=Path("/tmp/unused"),
|
|
steps=steps,
|
|
)
|
|
ordered = runner._topological_sort()
|
|
assert [step.step_id for step in ordered] == [
|
|
"imagemagick_grayscale_01",
|
|
"xcf_stack_01",
|
|
]
|
|
|
|
|
|
class TestExpectedOutputFilenames:
|
|
def test_returns_xcf_for_jpg_input(self) -> None:
|
|
names = XcfStackModule.expected_output_filenames(
|
|
matched_groups=[],
|
|
input_paths=[Path("photo.jpg")],
|
|
params={},
|
|
)
|
|
assert names == ["photo.xcf"]
|
|
|
|
def test_default_timeout(self) -> None:
|
|
params = XcfStackModule.validate_module_params({})
|
|
assert params["timeout"] == 120.0
|
|
|
|
def test_custom_timeout(self) -> None:
|
|
params = XcfStackModule.validate_module_params({"timeout": 45})
|
|
assert params["timeout"] == 45.0
|
|
|
|
|
|
def _make_stack_fixture(tmp_path: Path) -> dict[str, Path]:
|
|
root = tmp_path
|
|
input_dir = root / "input"
|
|
step_a = root / "step_a"
|
|
step_b = root / "step_b"
|
|
output_dir = root / "xcf_out"
|
|
for directory in (input_dir, step_a, step_b, output_dir):
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
|
|
make_png(input_dir / "photo.png")
|
|
make_png(step_a / "photo.png", rgb=(10, 20, 30))
|
|
make_png(step_b / "photo.png", rgb=(30, 20, 10))
|
|
|
|
return {
|
|
"root": root,
|
|
"input_dir": input_dir,
|
|
"step_a": step_a,
|
|
"step_b": step_b,
|
|
"output_dir": output_dir,
|
|
}
|
|
|
|
|
|
class TestXcfStackRun:
|
|
@patch("imagepipeline.modules.xcf_stack.stack_images_to_xcf")
|
|
def test_collects_layers_from_explicit_inputs(self, mock_stack: object, tmp_path: Path) -> None:
|
|
paths = _make_stack_fixture(tmp_path)
|
|
input_path = paths["root"] / "refs" / "photo.jpg"
|
|
input_path.parent.mkdir()
|
|
input_path.touch()
|
|
|
|
ctx = ModuleContext(
|
|
input_paths=[input_path],
|
|
output_dir=paths["output_dir"],
|
|
params=XcfStackModule.validate_module_params({}),
|
|
pipeline_output_root=paths["root"],
|
|
step_id="xcf_stack_01",
|
|
input_layer_dirs=[
|
|
("input", paths["input_dir"]),
|
|
("step_a", paths["step_a"]),
|
|
("step_b", paths["step_b"]),
|
|
],
|
|
logger=None,
|
|
)
|
|
XcfStackModule().run(ctx)
|
|
|
|
mock_stack.assert_called_once()
|
|
layers, outfile = mock_stack.call_args[0]
|
|
assert outfile == paths["output_dir"] / "photo.xcf"
|
|
assert layers == [
|
|
("input", paths["input_dir"] / "photo.png"),
|
|
("step_a", paths["step_a"] / "photo.png"),
|
|
("step_b", paths["step_b"] / "photo.png"),
|
|
]
|
|
|
|
@patch("imagepipeline.modules.xcf_stack.stack_images_to_xcf")
|
|
def test_skip_missing_true_skips_missing_step(self, mock_stack: object, tmp_path: Path) -> None:
|
|
paths = _make_stack_fixture(tmp_path)
|
|
(paths["step_b"] / "photo.png").unlink()
|
|
input_path = paths["root"] / "photo.jpg"
|
|
|
|
ctx = ModuleContext(
|
|
input_paths=[input_path],
|
|
output_dir=paths["output_dir"],
|
|
params=XcfStackModule.validate_module_params({"skip_missing": True}),
|
|
pipeline_output_root=paths["root"],
|
|
step_id="xcf_stack_01",
|
|
input_layer_dirs=[
|
|
("input", paths["input_dir"]),
|
|
("step_a", paths["step_a"]),
|
|
("step_b", paths["step_b"]),
|
|
],
|
|
logger=None,
|
|
)
|
|
XcfStackModule().run(ctx)
|
|
|
|
layers, _outfile = mock_stack.call_args[0]
|
|
step_ids = [step_id for step_id, _ in layers]
|
|
assert step_ids == ["input", "step_a"]
|
|
|
|
@patch("imagepipeline.modules.xcf_stack.stack_images_to_xcf")
|
|
def test_skip_missing_false_raises_on_missing_step(
|
|
self, mock_stack: object, tmp_path: Path
|
|
) -> None:
|
|
paths = _make_stack_fixture(tmp_path)
|
|
(paths["step_b"] / "photo.png").unlink()
|
|
input_path = paths["root"] / "photo.jpg"
|
|
|
|
ctx = ModuleContext(
|
|
input_paths=[input_path],
|
|
output_dir=paths["output_dir"],
|
|
params=XcfStackModule.validate_module_params({"skip_missing": False}),
|
|
pipeline_output_root=paths["root"],
|
|
step_id="xcf_stack_01",
|
|
input_layer_dirs=[
|
|
("input", paths["input_dir"]),
|
|
("step_a", paths["step_a"]),
|
|
("step_b", paths["step_b"]),
|
|
],
|
|
logger=None,
|
|
)
|
|
with pytest.raises(ValueError, match="no image with stem 'photo' in step 'step_b'"):
|
|
XcfStackModule().run(ctx)
|
|
|
|
mock_stack.assert_not_called()
|
|
|
|
|
|
@pytest.mark.skipif(not has_gimp, reason="GIMP not installed")
|
|
@pytest.mark.skipif(not has_magick, reason="ImageMagick not installed")
|
|
class TestStackImagesToXcfIntegration:
|
|
def test_stacks_pngs_into_xcf(self, tmp_path: Path) -> None:
|
|
from imagepipeline.utils.gimp import stack_images_to_xcf
|
|
|
|
bottom = tmp_path / "bottom.png"
|
|
top = tmp_path / "top.png"
|
|
magick = shutil.which("magick") or shutil.which("convert")
|
|
assert magick is not None
|
|
from imagepipeline.utils.subprocess import run_command
|
|
|
|
run_command([magick, "-size", "8x8", "xc:#ff0000", str(bottom)])
|
|
run_command([magick, "-size", "8x8", "xc:#0000ff", str(top)])
|
|
|
|
outfile = tmp_path / "stack.xcf"
|
|
stack_images_to_xcf(
|
|
[("bottom", bottom), ("top", top)],
|
|
outfile,
|
|
timeout=120.0,
|
|
)
|
|
|
|
assert outfile.is_file()
|
|
assert outfile.stat().st_size > 0
|
|
names = outfile.read_bytes()
|
|
assert b"bottom" in names
|
|
assert b"top" in names
|
|
|
|
|
|
class TestGimpTimeoutHandling:
|
|
def test_accepts_outfile_when_gimp_hangs_on_exit(self, tmp_path: Path) -> None:
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from imagepipeline.utils.gimp import stack_images_to_xcf
|
|
|
|
layer = tmp_path / "layer.png"
|
|
layer.write_bytes(b"png")
|
|
outfile = tmp_path / "stack.xcf"
|
|
outfile.write_bytes(b"xcf-data")
|
|
|
|
def fake_popen(*_args, **_kwargs):
|
|
process = MagicMock()
|
|
process.communicate.side_effect = [
|
|
subprocess.TimeoutExpired(cmd="gimp", timeout=1),
|
|
("", ""),
|
|
]
|
|
process.kill = MagicMock()
|
|
return process
|
|
|
|
with patch("imagepipeline.utils.gimp.subprocess.Popen", side_effect=fake_popen):
|
|
with patch("imagepipeline.utils.gimp.require_gimp", return_value="gimp-console"):
|
|
stack_images_to_xcf([("layer", layer)], outfile, timeout=1.0)
|
|
|
|
assert outfile.read_bytes() == b"xcf-data"
|