amayer5125 is savage
This commit is contained in:
Frank Schwenk
@@ -0,0 +1 @@
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
import struct
|
||||
import zlib
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
import imagepipeline.modules # noqa: F401
|
||||
|
||||
|
||||
def make_png(
|
||||
path: Path,
|
||||
width: int = 2,
|
||||
height: int = 2,
|
||||
rgb: tuple[int, int, int] = (200, 100, 50),
|
||||
) -> None:
|
||||
"""Write a minimal valid PNG without external dependencies."""
|
||||
r, g, b = rgb
|
||||
|
||||
def chunk(tag: bytes, data: bytes) -> bytes:
|
||||
crc = zlib.crc32(tag + data) & 0xFFFFFFFF
|
||||
return struct.pack(">I", len(data)) + tag + data + struct.pack(">I", crc)
|
||||
|
||||
raw = b"".join(
|
||||
b"\x00" + bytes([r, g, b] * width)
|
||||
for _ in range(height)
|
||||
)
|
||||
compressed = zlib.compress(raw, 9)
|
||||
ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)
|
||||
png = (
|
||||
b"\x89PNG\r\n\x1a\n"
|
||||
+ chunk(b"IHDR", ihdr)
|
||||
+ chunk(b"IDAT", compressed)
|
||||
+ chunk(b"IEND", b"")
|
||||
)
|
||||
path.write_bytes(png)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _ensure_builtin_modules() -> None:
|
||||
import imagepipeline.modules # noqa: F401
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def input_dir(tmp_path: Path) -> Path:
|
||||
directory = tmp_path / "input"
|
||||
directory.mkdir()
|
||||
make_png(directory / "photo_a.png")
|
||||
make_png(directory / "photo_b.png", rgb=(10, 20, 30))
|
||||
return directory
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def output_base(tmp_path: Path) -> Path:
|
||||
base = tmp_path / "output"
|
||||
base.mkdir()
|
||||
return base
|
||||
@@ -0,0 +1,222 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from imagepipeline.core.context import ModuleContext
|
||||
from imagepipeline.core.exceptions import DependencyError
|
||||
from imagepipeline.modules.ai_exposure import AIExposureModule
|
||||
from imagepipeline.modules.ai_tone_map import AIToneMapModule
|
||||
from imagepipeline.modules.comfy_flux_edit import ComfyFluxEditModule
|
||||
from imagepipeline.modules.openrouter_edit import OpenRouterEditModule
|
||||
from imagepipeline.modules.registry import get_module, list_modules
|
||||
from tests.conftest import make_png
|
||||
|
||||
try:
|
||||
import numpy # noqa: F401
|
||||
|
||||
has_numpy = True
|
||||
except ImportError:
|
||||
has_numpy = False
|
||||
|
||||
try:
|
||||
import torch # noqa: F401
|
||||
|
||||
has_torch = True
|
||||
except ImportError:
|
||||
has_torch = False
|
||||
|
||||
has_magick = bool(shutil.which("magick") or shutil.which("convert"))
|
||||
|
||||
|
||||
class TestAIModuleRegistration:
|
||||
def test_ai_modules_registered(self) -> None:
|
||||
names = list_modules()
|
||||
for name in (
|
||||
"ai_exposure",
|
||||
"ai_tone_map",
|
||||
"openrouter_edit",
|
||||
"comfy_flux_edit",
|
||||
):
|
||||
assert name in names
|
||||
|
||||
def test_get_ai_modules(self) -> None:
|
||||
assert get_module("ai_exposure") is AIExposureModule
|
||||
assert get_module("ai_tone_map") is AIToneMapModule
|
||||
assert get_module("openrouter_edit") is OpenRouterEditModule
|
||||
assert get_module("comfy_flux_edit") is ComfyFluxEditModule
|
||||
|
||||
|
||||
class TestAIParameters:
|
||||
def test_ai_exposure_defaults(self) -> None:
|
||||
params = AIExposureModule.validate_module_params({})
|
||||
assert params["skip_existing"] is True
|
||||
assert params["max_edge"] == 2048
|
||||
assert params["device"] == "cpu"
|
||||
assert params["strength"] == 1.0
|
||||
|
||||
def test_ai_tone_map_defaults(self) -> None:
|
||||
params = AIToneMapModule.validate_module_params({})
|
||||
assert params["checkpoint"] == ""
|
||||
assert params["strength"] == 1.0
|
||||
assert params["net_input_size"] == 256
|
||||
|
||||
def test_openrouter_requires_prompt(self) -> None:
|
||||
with pytest.raises(ValueError, match="required"):
|
||||
OpenRouterEditModule.validate_module_params({})
|
||||
|
||||
def test_openrouter_defaults(self) -> None:
|
||||
params = OpenRouterEditModule.validate_module_params({"prompt": "brighten shadows"})
|
||||
assert params["model"] == "black-forest-labs/flux.2-klein-4b"
|
||||
assert params["strength"] == 0.3
|
||||
assert params["api_key_env"] == "OPENROUTER_API_KEY"
|
||||
|
||||
def test_comfy_requires_prompt(self) -> None:
|
||||
with pytest.raises(ValueError, match="required"):
|
||||
ComfyFluxEditModule.validate_module_params({})
|
||||
|
||||
|
||||
class TestAIToneMapFallback:
|
||||
@pytest.mark.skipif(not has_numpy, reason="numpy not installed")
|
||||
def test_clahe_fallback_writes_output(self, tmp_path: Path) -> None:
|
||||
src = tmp_path / "photo.png"
|
||||
make_png(src, width=8, height=8, rgb=(40, 80, 120))
|
||||
output_dir = tmp_path / "out"
|
||||
output_dir.mkdir()
|
||||
ctx = ModuleContext(
|
||||
input_paths=[src],
|
||||
matched_groups=[],
|
||||
output_dir=output_dir,
|
||||
params=AIToneMapModule.validate_module_params({"strength": 1.0}),
|
||||
pipeline_output_root=tmp_path,
|
||||
step_id="ai_tone_map_01",
|
||||
logger=None,
|
||||
)
|
||||
AIToneMapModule().run(ctx)
|
||||
dst = output_dir / "photo.png"
|
||||
assert dst.is_file()
|
||||
assert dst.stat().st_size > 0
|
||||
|
||||
|
||||
class TestSkipExisting:
|
||||
@pytest.mark.skipif(not has_numpy, reason="numpy not installed")
|
||||
def test_second_run_skips_existing_outputs(self, tmp_path: Path, capsys) -> None:
|
||||
src = tmp_path / "photo.png"
|
||||
make_png(src, width=8, height=8)
|
||||
output_dir = tmp_path / "out"
|
||||
output_dir.mkdir()
|
||||
params = AIToneMapModule.validate_module_params({"checkpoint": ""})
|
||||
from imagepipeline.core.log import PipelineLogger
|
||||
|
||||
logger = PipelineLogger(verbose=True)
|
||||
ctx = ModuleContext(
|
||||
input_paths=[src],
|
||||
matched_groups=[],
|
||||
output_dir=output_dir,
|
||||
params=params,
|
||||
pipeline_output_root=tmp_path,
|
||||
step_id="ai_tone_map_01",
|
||||
logger=logger,
|
||||
)
|
||||
module = AIToneMapModule()
|
||||
module.run(ctx)
|
||||
assert (output_dir / "photo.png").is_file()
|
||||
|
||||
module.run(ctx)
|
||||
output = capsys.readouterr().out
|
||||
assert "Skipped module ai_tone_map" in output
|
||||
|
||||
|
||||
@pytest.mark.skipif(not has_torch, reason="torch not installed")
|
||||
class TestAIExposure:
|
||||
def test_ai_exposure_processes_image(self, tmp_path: Path) -> None:
|
||||
src = tmp_path / "photo.png"
|
||||
make_png(src, width=4, height=4, rgb=(30, 60, 90))
|
||||
output_dir = tmp_path / "out"
|
||||
output_dir.mkdir()
|
||||
|
||||
mock_model = MagicMock()
|
||||
mock_device = MagicMock()
|
||||
|
||||
def fake_enhance(_model, image, *, device, strength):
|
||||
return image
|
||||
|
||||
with (
|
||||
patch.object(AIExposureModule, "_get_model", return_value=(mock_model, mock_device)),
|
||||
patch(
|
||||
"imagepipeline.ai.zero_dce.enhance_image",
|
||||
side_effect=fake_enhance,
|
||||
),
|
||||
):
|
||||
ctx = ModuleContext(
|
||||
input_paths=[src],
|
||||
matched_groups=[],
|
||||
output_dir=output_dir,
|
||||
params=AIExposureModule.validate_module_params({"max_edge": 0}),
|
||||
pipeline_output_root=tmp_path,
|
||||
step_id="ai_exposure_01",
|
||||
logger=None,
|
||||
)
|
||||
AIExposureModule().run(ctx)
|
||||
|
||||
assert (output_dir / "photo.png").is_file()
|
||||
|
||||
|
||||
class TestComfyFluxEdit:
|
||||
def test_server_unreachable_raises(self, tmp_path: Path) -> None:
|
||||
src = tmp_path / "photo.png"
|
||||
make_png(src)
|
||||
output_dir = tmp_path / "out"
|
||||
output_dir.mkdir()
|
||||
workflow = tmp_path / "workflow.json"
|
||||
workflow.write_text('{"1": {"class_type": "LoadImage", "inputs": {"image": "x"}}}')
|
||||
|
||||
ctx = ModuleContext(
|
||||
input_paths=[src],
|
||||
matched_groups=[],
|
||||
output_dir=output_dir,
|
||||
params=ComfyFluxEditModule.validate_module_params(
|
||||
{
|
||||
"prompt": "test",
|
||||
"server_url": "http://127.0.0.1:1",
|
||||
"workflow_path": workflow,
|
||||
}
|
||||
),
|
||||
pipeline_output_root=tmp_path,
|
||||
step_id="comfy_flux_edit_01",
|
||||
logger=None,
|
||||
)
|
||||
with pytest.raises(DependencyError, match="not reachable"):
|
||||
ComfyFluxEditModule().run(ctx)
|
||||
|
||||
def test_missing_workflow_raises(self, tmp_path: Path) -> None:
|
||||
src = tmp_path / "photo.png"
|
||||
make_png(src)
|
||||
output_dir = tmp_path / "out"
|
||||
output_dir.mkdir()
|
||||
ctx = ModuleContext(
|
||||
input_paths=[src],
|
||||
matched_groups=[],
|
||||
output_dir=output_dir,
|
||||
params=ComfyFluxEditModule.validate_module_params(
|
||||
{
|
||||
"prompt": "test",
|
||||
"workflow_path": tmp_path / "missing.json",
|
||||
}
|
||||
),
|
||||
pipeline_output_root=tmp_path,
|
||||
step_id="comfy_flux_edit_01",
|
||||
logger=None,
|
||||
)
|
||||
with pytest.raises(DependencyError, match="workflow not found"):
|
||||
ComfyFluxEditModule().run(ctx)
|
||||
|
||||
|
||||
class TestOpenRouterDependencies:
|
||||
def test_missing_api_key_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
|
||||
with pytest.raises(DependencyError, match="OPENROUTER_API_KEY"):
|
||||
OpenRouterEditModule.check_dependencies()
|
||||
@@ -0,0 +1,232 @@
|
||||
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()
|
||||
@@ -0,0 +1,298 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from imagepipeline.core.params import validate_params
|
||||
from imagepipeline.modules.composite import CompositeModule
|
||||
from imagepipeline.modules.crop_square import CropSquareModule
|
||||
from imagepipeline.modules.darktable_style import DarktableStyleModule
|
||||
from imagepipeline.modules.imagemagick_fill import (
|
||||
ImageMagickFillModule,
|
||||
build_fill_arguments,
|
||||
)
|
||||
from imagepipeline.modules.imagemagick_grayscale import ImageMagickGrayscale
|
||||
from imagepipeline.modules.gmic_grayscale import GmicGrayscale
|
||||
from imagepipeline.modules.registry import get_module, list_modules
|
||||
from imagepipeline.modules.rembg import RembgModule
|
||||
|
||||
has_magick = bool(shutil.which("magick") or shutil.which("convert"))
|
||||
|
||||
|
||||
class TestModuleRegistration:
|
||||
def test_workflow_modules_registered(self) -> None:
|
||||
names = list_modules()
|
||||
for name in (
|
||||
"rembg",
|
||||
"gmic_grayscale",
|
||||
"composite",
|
||||
"darktable_style",
|
||||
"imagemagick_grayscale",
|
||||
"imagemagick_fill",
|
||||
"crop_square",
|
||||
):
|
||||
assert name in names
|
||||
|
||||
def test_get_module_returns_class(self) -> None:
|
||||
assert get_module("rembg") is RembgModule
|
||||
assert get_module("gmic_grayscale") is GmicGrayscale
|
||||
assert get_module("composite") is CompositeModule
|
||||
assert get_module("darktable_style") is DarktableStyleModule
|
||||
assert get_module("crop_square") is CropSquareModule
|
||||
assert get_module("imagemagick_fill") is ImageMagickFillModule
|
||||
|
||||
|
||||
class TestImageMagickFill:
|
||||
def test_solid_fill_arguments(self) -> None:
|
||||
args = build_fill_arguments(
|
||||
800,
|
||||
600,
|
||||
color1="#d7fd00",
|
||||
color2=None,
|
||||
gradient=False,
|
||||
radial=False,
|
||||
angle=None,
|
||||
)
|
||||
assert args == ["-size", "800x600", "xc:#d7fd00"]
|
||||
|
||||
def test_linear_gradient_arguments(self) -> None:
|
||||
args = build_fill_arguments(
|
||||
100,
|
||||
50,
|
||||
color1="#d7fd00",
|
||||
color2="#fc0ade",
|
||||
gradient=True,
|
||||
radial=False,
|
||||
angle=45.0,
|
||||
)
|
||||
assert args == [
|
||||
"-size",
|
||||
"100x50",
|
||||
"-define",
|
||||
"gradient:angle=45.0",
|
||||
"gradient:#d7fd00-#fc0ade",
|
||||
]
|
||||
|
||||
def test_radial_gradient_arguments(self) -> None:
|
||||
args = build_fill_arguments(
|
||||
100,
|
||||
50,
|
||||
color1="d7fd00",
|
||||
color2="fc0ade",
|
||||
gradient=True,
|
||||
radial=True,
|
||||
angle=90.0,
|
||||
)
|
||||
assert args == ["-size", "100x50", "radial-gradient:#d7fd00-#fc0ade"]
|
||||
|
||||
def test_requires_color1(self) -> None:
|
||||
with pytest.raises(ValueError, match="required"):
|
||||
ImageMagickFillModule.validate_module_params({})
|
||||
|
||||
@pytest.mark.skipif(not has_magick, reason="ImageMagick not installed")
|
||||
def test_output_matches_input_size(self, tmp_path: Path) -> None:
|
||||
from imagepipeline.core.context import ModuleContext
|
||||
from imagepipeline.utils.subprocess import run_command
|
||||
|
||||
from tests.conftest import make_png
|
||||
|
||||
src = tmp_path / "ref.png"
|
||||
make_png(src, width=40, height=20)
|
||||
output_dir = tmp_path / "out"
|
||||
output_dir.mkdir()
|
||||
ctx = ModuleContext(
|
||||
input_paths=[src],
|
||||
matched_groups=[],
|
||||
output_dir=output_dir,
|
||||
params=ImageMagickFillModule.validate_module_params(
|
||||
{
|
||||
"color1": "#d7fd00",
|
||||
"color2": "#fc0ade",
|
||||
"gradient": True,
|
||||
"angle": 45,
|
||||
}
|
||||
),
|
||||
pipeline_output_root=tmp_path,
|
||||
step_id="imagemagick_fill_01",
|
||||
logger=None,
|
||||
)
|
||||
ImageMagickFillModule().run(ctx)
|
||||
|
||||
dst = output_dir / "ref.png"
|
||||
assert dst.is_file()
|
||||
magick = shutil.which("magick") or shutil.which("convert")
|
||||
result = run_command([magick, "-format", "%w %h", str(dst), "info:"])
|
||||
assert result.stdout.strip() == "40 20"
|
||||
|
||||
|
||||
class TestCropSquare:
|
||||
@pytest.mark.skipif(not has_magick, reason="ImageMagick not installed")
|
||||
def test_center_crops_to_square(self, tmp_path: Path) -> None:
|
||||
from imagepipeline.core.context import ModuleContext
|
||||
from imagepipeline.utils.subprocess import run_command
|
||||
|
||||
from tests.conftest import make_png
|
||||
|
||||
src = tmp_path / "wide.png"
|
||||
make_png(src, width=40, height=20)
|
||||
output_dir = tmp_path / "out"
|
||||
output_dir.mkdir()
|
||||
ctx = ModuleContext(
|
||||
input_paths=[src],
|
||||
matched_groups=[],
|
||||
output_dir=output_dir,
|
||||
params={},
|
||||
pipeline_output_root=tmp_path,
|
||||
step_id="crop_square_01",
|
||||
logger=None,
|
||||
)
|
||||
CropSquareModule().run(ctx)
|
||||
|
||||
dst = output_dir / "wide.png"
|
||||
assert dst.is_file()
|
||||
magick = shutil.which("magick") or shutil.which("convert")
|
||||
result = run_command(
|
||||
[magick, "-format", "%w %h", str(dst), "info:"],
|
||||
)
|
||||
width, height = map(int, result.stdout.strip().split())
|
||||
assert width == height == 20
|
||||
|
||||
|
||||
class TestModuleParameters:
|
||||
def test_darktable_style_requires_style(self) -> None:
|
||||
with pytest.raises(ValueError, match="required"):
|
||||
DarktableStyleModule.validate_module_params({})
|
||||
|
||||
def test_darktable_style_accepts_style(self) -> None:
|
||||
params = DarktableStyleModule.validate_module_params(
|
||||
{"style": "Watermark F12.rocks"}
|
||||
)
|
||||
assert params["style"] == "Watermark F12.rocks"
|
||||
assert params["style_overwrite"] is True
|
||||
|
||||
def test_composite_mode_choices(self) -> None:
|
||||
with pytest.raises(ValueError, match="must be one of"):
|
||||
CompositeModule.validate_module_params({"mode": "invalid"})
|
||||
|
||||
def test_rembg_defaults(self) -> None:
|
||||
params = RembgModule.validate_module_params({})
|
||||
assert params["model"] == "birefnet-general"
|
||||
assert params["alpha_matting"] is True
|
||||
|
||||
def test_darktable_export_conf_jpeg(self) -> None:
|
||||
from imagepipeline.modules.darktable_style import export_conf_options
|
||||
|
||||
assert export_conf_options("jpeg") == [
|
||||
"plugins/imageio/format/jpeg/quality=90"
|
||||
]
|
||||
|
||||
def test_darktable_export_conf_png(self) -> None:
|
||||
from imagepipeline.modules.darktable_style import export_conf_options
|
||||
|
||||
assert export_conf_options("png") == [
|
||||
"plugins/imageio/format/png/bpp=8",
|
||||
"plugins/imageio/format/png/compression=5",
|
||||
]
|
||||
|
||||
def test_darktable_resolve_format_from_input(self) -> None:
|
||||
from imagepipeline.modules.darktable_style import resolve_export_format
|
||||
|
||||
format_name, conf = resolve_export_format(Path("photo.jpg"), "")
|
||||
assert format_name == "jpeg"
|
||||
assert conf == ["plugins/imageio/format/jpeg/quality=90"]
|
||||
|
||||
format_name, conf = resolve_export_format(Path("photo.png"), "")
|
||||
assert format_name == "png"
|
||||
assert "plugins/imageio/format/png/compression=5" in conf
|
||||
|
||||
def test_darktable_rejects_unsupported_format(self) -> None:
|
||||
from imagepipeline.modules.darktable_style import resolve_export_format
|
||||
|
||||
with pytest.raises(ValueError, match="Unsupported output format"):
|
||||
resolve_export_format(Path("photo.tiff"), "tiff")
|
||||
|
||||
|
||||
@pytest.mark.skipif(not has_magick, reason="ImageMagick not installed")
|
||||
class TestCompositeColor:
|
||||
def test_preserves_color_over_grayscale_background(self, tmp_path: Path) -> None:
|
||||
from imagepipeline.core.context import ModuleContext
|
||||
from imagepipeline.utils.subprocess import run_command
|
||||
|
||||
from tests.conftest import make_png
|
||||
|
||||
src = tmp_path / "src.png"
|
||||
make_png(src, width=40, height=40, rgb=(200, 50, 50))
|
||||
background = tmp_path / "bg.png"
|
||||
foreground = tmp_path / "fg.png"
|
||||
output_dir = tmp_path / "composite_out"
|
||||
output_dir.mkdir()
|
||||
magick = shutil.which("magick") or shutil.which("convert")
|
||||
|
||||
run_command([magick, str(src), "-colorspace", "Gray", str(background)])
|
||||
run_command(
|
||||
[
|
||||
magick,
|
||||
"-size",
|
||||
"40x40",
|
||||
"xc:none",
|
||||
"-fill",
|
||||
"rgba(255,0,0,0.8)",
|
||||
"-draw",
|
||||
"circle 20,20 20,2",
|
||||
str(foreground),
|
||||
]
|
||||
)
|
||||
|
||||
module = CompositeModule()
|
||||
ctx = ModuleContext(
|
||||
input_paths=[background, foreground],
|
||||
output_dir=output_dir,
|
||||
params=CompositeModule.validate_module_params({}),
|
||||
pipeline_output_root=tmp_path,
|
||||
step_id="composite_01",
|
||||
matched_groups=[[background, foreground]],
|
||||
)
|
||||
module.run(ctx)
|
||||
|
||||
output = output_dir / "fg.png"
|
||||
assert output.is_file()
|
||||
result = run_command(
|
||||
[magick, "identify", "-format", "%[type]", str(output)]
|
||||
)
|
||||
assert result.stdout.strip() != "Grayscale"
|
||||
|
||||
|
||||
has_workflow_tools = all(
|
||||
shutil.which(name)
|
||||
for name in ("rembg", "gmic", "magick", "darktable-cli")
|
||||
) or all(shutil.which(name) for name in ("rembg", "gmic", "convert", "darktable-cli"))
|
||||
|
||||
|
||||
@pytest.mark.skipif(not has_workflow_tools, reason="Workflow CLI tools not installed")
|
||||
class TestWorkflowIntegration:
|
||||
def test_watermark_pipeline(self, input_dir, output_base) -> None:
|
||||
from imagepipeline import Pipeline
|
||||
|
||||
with Pipeline(
|
||||
name="workflow_test",
|
||||
input_dir=input_dir,
|
||||
output_base=output_base,
|
||||
verbose=False,
|
||||
) as p:
|
||||
rembg_out = p.step("rembg", inputs="input")
|
||||
grayscale = p.step("gmic_grayscale", inputs="input")
|
||||
combined = p.step("composite", inputs=[grayscale, rembg_out])
|
||||
p.step(
|
||||
"darktable_style",
|
||||
inputs=combined,
|
||||
style="Watermark F12.rocks",
|
||||
)
|
||||
root = p.run()
|
||||
|
||||
assert (root / "rembg_01").is_dir()
|
||||
assert (root / "gmic_grayscale_01").is_dir()
|
||||
assert (root / "composite_01").is_dir()
|
||||
assert (root / "darktable_style_01").is_dir()
|
||||
assert list((root / "darktable_style_01").iterdir())
|
||||
Reference in New Issue
Block a user