froxxxy/imagepipeline
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: resume fixes, OpenRouter templates, and project context

Browse Source 
Delegate expected output filenames to modules so resume works for rembg
and composite; normalize G'MIC multi-frame output; add OpenRouter style
reference support with tests. Add Crusaders, orange, and team gallery
pipelines plus SOUL/AGENTS context files.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Frank Schwenk
2026-07-12 10:45:00 +02:00
parent 980c7f3b8b
commit ce431d7eec
22 changed files with 1616 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_ai_modules.py
+87
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import io
import shutil
from pathlib import Path
from unittest.mock import MagicMock, patch
@@ -73,6 +74,92 @@ class TestAIParameters:
assert params["model"] == "black-forest-labs/flux.2-klein-4b"
assert params["strength"] == 0.3
assert params["api_key_env"] == "OPENROUTER_API_KEY"
assert params["template_image"] is None
def test_openrouter_accepts_template_image(self, tmp_path: Path) -> None:
template = tmp_path / "ref.png"
make_png(template)
params = OpenRouterEditModule.validate_module_params(
{"prompt": "match style", "template_image": template}
)
assert params["template_image"] == template
def test_build_payload_with_template(self) -> None:
payload = OpenRouterEditModule._build_payload(
"data:image/jpeg;base64,abc",
"match colors",
"google/gemini-3-pro-image",
0.3,
template_data_url="data:image/jpeg;base64,ref",
)
assert payload["modalities"] == ["image", "text"]
assert "image_config" not in payload
content = payload["messages"][0]["content"]
assert content[0]["type"] == "text"
assert "FIRST image" in content[0]["text"]
assert content[1]["image_url"]["url"] == "data:image/jpeg;base64,ref"
assert content[2]["image_url"]["url"] == "data:image/jpeg;base64,abc"
def test_build_payload_flux_keeps_strength(self) -> None:
payload = OpenRouterEditModule._build_payload(
"data:image/jpeg;base64,abc",
"brighten",
"black-forest-labs/flux.2-klein-4b",
0.25,
)
assert payload["modalities"] == ["image"]
assert payload["image_config"] == {"strength": 0.25}
def test_save_result_matching_source_preserves_png_size(
self, tmp_path: Path
) -> None:
try:
from PIL import Image
except ImportError:
pytest.skip("Pillow not installed")
source = tmp_path / "source.png"
dest = tmp_path / "out.png"
with Image.new("RGBA", (16, 12), (10, 20, 30, 128)) as image:
image.save(source, format="PNG")
with Image.new("RGB", (8, 6), (200, 100, 50)) as edited:
buffer = io.BytesIO()
edited.save(buffer, format="PNG")
result_bytes = buffer.getvalue()
OpenRouterEditModule._save_result_matching_source(source, result_bytes, dest)
with Image.open(dest) as saved:
assert saved.size == (16, 12)
assert saved.mode == "RGBA"
assert saved.getchannel("A").getextrema() == (128, 128)
def test_missing_template_image_raises(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("OPENROUTER_API_KEY", "test-key")
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=OpenRouterEditModule.validate_module_params(
{
"prompt": "match",
"template_image": tmp_path / "missing.png",
"max_edge": 0,
}
),
pipeline_output_root=tmp_path,
step_id="openrouter_edit_01",
logger=None,
)
with pytest.raises(FileNotFoundError, match="Template image not found"):
OpenRouterEditModule().run(ctx)
def test_comfy_requires_prompt(self) -> None:
with pytest.raises(ValueError, match="required"):
tests/test_resume.py
+98 -2
View File
@@ -6,11 +6,16 @@ from pathlib import Path
import pytest
from imagepipeline.core.pipeline import Pipeline
from imagepipeline.core.resume import materialize_external_outputs, step_outputs_complete
from imagepipeline.core.resume import (
expected_output_filenames,
materialize_external_outputs,
step_outputs_complete,
)
from imagepipeline.core.step import StepDefinition
from imagepipeline.modules.imagemagick_grayscale import ImageMagickGrayscale
from imagepipeline.modules.registry import get_module
from imagepipeline.utils.gmic import split_gmic_command
from imagepipeline.modules.rembg import RembgModule
from imagepipeline.utils.gmic import finalize_gmic_output, split_gmic_command
from tests.conftest import make_png
@@ -26,6 +31,97 @@ class TestGmicCommandSplit:
assert parts == ["-fx_custom_gradient", "0,0,0,,1,0"]
class TestFinalizeGmicOutput:
def test_keeps_frame_000001_and_removes_000000(self, tmp_path: Path) -> None:
output_dir = tmp_path / "out"
output_dir.mkdir()
intended = output_dir / "photo.png"
frame_000000 = output_dir / "photo_000000.png"
frame_000001 = output_dir / "photo_000001.png"
frame_000000.write_bytes(b"discard")
frame_000001.write_bytes(b"keep")
result = finalize_gmic_output(output_dir, intended)
assert result == intended
assert intended.read_bytes() == b"keep"
assert not frame_000000.exists()
assert not frame_000001.exists()
def test_leaves_single_output_unchanged(self, tmp_path: Path) -> None:
output_dir = tmp_path / "out"
output_dir.mkdir()
intended = output_dir / "photo.png"
intended.write_bytes(b"single")
result = finalize_gmic_output(output_dir, intended)
assert result == intended
assert intended.read_bytes() == b"single"
def test_renames_only_000000_when_000001_missing(self, tmp_path: Path) -> None:
output_dir = tmp_path / "out"
output_dir.mkdir()
intended = output_dir / "photo.png"
frame_000000 = output_dir / "photo_000000.png"
frame_000000.write_bytes(b"only")
result = finalize_gmic_output(output_dir, intended)
assert result == intended
assert intended.read_bytes() == b"only"
assert not frame_000000.exists()
class TestExpectedOutputFilenames:
def test_rembg_maps_jpg_inputs_to_png_outputs(self, tmp_path: Path) -> None:
jpg = tmp_path / "photo.jpg"
jpg.write_bytes(b"jpeg")
step = StepDefinition(
step_id="rembg_01",
module_name="rembg",
module=RembgModule,
input_refs=["input"],
params={},
output_dir_name="rembg_01",
)
names = expected_output_filenames(
step,
matched_groups=[[jpg]],
input_paths=[jpg],
params=RembgModule.validate_module_params({}),
)
assert names == ["photo.png"]
def test_rembg_resume_detects_existing_png_outputs(self, tmp_path: Path) -> None:
output_dir = tmp_path / "rembg_01"
output_dir.mkdir()
png = output_dir / "photo.png"
make_png(png)
jpg = tmp_path / "input" / "photo.jpg"
jpg.parent.mkdir()
jpg.write_bytes(b"jpeg")
step = StepDefinition(
step_id="rembg_01",
module_name="rembg",
module=RembgModule,
input_refs=["input"],
params={},
output_dir_name="rembg_01",
)
params = RembgModule.validate_module_params({})
expected = expected_output_filenames(
step,
matched_groups=[[jpg]],
input_paths=[jpg],
params=params,
)
assert step_outputs_complete([output_dir / name for name in expected])
class TestPipelineResume:
@pytest.mark.skipif(not shutil.which("magick"), reason="ImageMagick not installed")
def test_continue_skips_completed_steps(self, input_dir: Path, output_base: Path, capsys) -> None: