Konstantin Fickel
d565329e16
Replace singular reference_image field with reference_images list to support an arbitrary number of reference images. Map them to the correct BFL API parameter names based on model family: - flux-2-*: input_image, input_image_2, ..., input_image_8 - flux-kontext-*: input_image, input_image_2, ..., input_image_4 - flux 1.x: image_prompt (single) BREAKING CHANGE: reference_image config field renamed to reference_images (list).
382 lines
14 KiB
Python
382 lines
14 KiB
Python
"""Integration tests for bulkgen.providers (image and text).
|
|
|
|
Mock-heavy tests produce many Any-typed expressions from MagicMock.
|
|
"""
|
|
# pyright: reportAny=false
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from bulkgen.config import TargetConfig
|
|
from bulkgen.providers.bfl import BFLResult
|
|
from bulkgen.providers.image import ImageProvider
|
|
from bulkgen.providers.image import (
|
|
_encode_image_b64 as encode_image_b64, # pyright: ignore[reportPrivateUsage]
|
|
)
|
|
from bulkgen.providers.text import TextProvider
|
|
|
|
|
|
def _make_bfl_mocks(
|
|
image_bytes: bytes,
|
|
) -> tuple[BFLResult, MagicMock]:
|
|
"""Return (bfl_result, mock_http) for BFL image generation tests."""
|
|
bfl_result = BFLResult(
|
|
task_id="test-task-id", sample_url="https://example.com/img.png"
|
|
)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.content = image_bytes
|
|
mock_response.raise_for_status.return_value = None
|
|
|
|
mock_http = AsyncMock()
|
|
mock_http.get.return_value = mock_response
|
|
mock_http.__aenter__ = AsyncMock(return_value=mock_http)
|
|
mock_http.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
return bfl_result, mock_http
|
|
|
|
|
|
def _make_mistral_mock(response: MagicMock) -> AsyncMock:
|
|
"""Return a mock Mistral client."""
|
|
mock_client = AsyncMock()
|
|
mock_client.chat.complete_async.return_value = response
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
return mock_client
|
|
|
|
|
|
def _make_text_response(content: str | None) -> MagicMock:
|
|
"""Return a mock Mistral response with one choice."""
|
|
choice = MagicMock()
|
|
choice.message.content = content
|
|
response = MagicMock()
|
|
response.choices = [choice]
|
|
return response
|
|
|
|
|
|
class TestImageProvider:
|
|
"""Test ImageProvider with mocked BFL client and HTTP."""
|
|
|
|
@pytest.fixture
|
|
def image_bytes(self) -> bytes:
|
|
return b"\x89PNG\r\n\x1a\n" + b"\x00" * 100
|
|
|
|
async def test_basic_image_generation(
|
|
self, project_dir: Path, image_bytes: bytes
|
|
) -> None:
|
|
target_config = TargetConfig(prompt="A red square")
|
|
bfl_result, mock_http = _make_bfl_mocks(image_bytes)
|
|
|
|
with (
|
|
patch("bulkgen.providers.image.BFLClient") as mock_cls,
|
|
patch("bulkgen.providers.image.httpx.AsyncClient") as mock_http_cls,
|
|
):
|
|
mock_cls.return_value.generate = AsyncMock(return_value=bfl_result)
|
|
mock_http_cls.return_value = mock_http
|
|
|
|
provider = ImageProvider(api_key="test-key")
|
|
await provider.generate(
|
|
target_name="out.png",
|
|
target_config=target_config,
|
|
resolved_prompt="A red square",
|
|
resolved_model="flux-pro-1.1",
|
|
project_dir=project_dir,
|
|
)
|
|
|
|
output = project_dir / "out.png"
|
|
assert output.exists()
|
|
assert output.read_bytes() == image_bytes
|
|
|
|
async def test_image_with_dimensions(
|
|
self, project_dir: Path, image_bytes: bytes
|
|
) -> None:
|
|
target_config = TargetConfig(prompt="A banner", width=1920, height=480)
|
|
bfl_result, mock_http = _make_bfl_mocks(image_bytes)
|
|
|
|
with (
|
|
patch("bulkgen.providers.image.BFLClient") as mock_cls,
|
|
patch("bulkgen.providers.image.httpx.AsyncClient") as mock_http_cls,
|
|
):
|
|
mock_generate = AsyncMock(return_value=bfl_result)
|
|
mock_cls.return_value.generate = mock_generate
|
|
mock_http_cls.return_value = mock_http
|
|
|
|
provider = ImageProvider(api_key="test-key")
|
|
await provider.generate(
|
|
target_name="banner.png",
|
|
target_config=target_config,
|
|
resolved_prompt="A banner",
|
|
resolved_model="flux-pro-1.1",
|
|
project_dir=project_dir,
|
|
)
|
|
|
|
call_args = mock_generate.call_args
|
|
inputs = call_args[0][1]
|
|
assert inputs["width"] == 1920
|
|
assert inputs["height"] == 480
|
|
|
|
async def test_image_with_reference_image_flux1(
|
|
self, project_dir: Path, image_bytes: bytes
|
|
) -> None:
|
|
ref_path = project_dir / "ref.png"
|
|
_ = ref_path.write_bytes(b"reference image data")
|
|
|
|
target_config = TargetConfig(prompt="Like this", reference_images=["ref.png"])
|
|
bfl_result, mock_http = _make_bfl_mocks(image_bytes)
|
|
|
|
with (
|
|
patch("bulkgen.providers.image.BFLClient") as mock_cls,
|
|
patch("bulkgen.providers.image.httpx.AsyncClient") as mock_http_cls,
|
|
):
|
|
mock_generate = AsyncMock(return_value=bfl_result)
|
|
mock_cls.return_value.generate = mock_generate
|
|
mock_http_cls.return_value = mock_http
|
|
|
|
provider = ImageProvider(api_key="test-key")
|
|
await provider.generate(
|
|
target_name="out.png",
|
|
target_config=target_config,
|
|
resolved_prompt="Like this",
|
|
resolved_model="flux-pro-1.1",
|
|
project_dir=project_dir,
|
|
)
|
|
|
|
call_args = mock_generate.call_args
|
|
inputs = call_args[0][1]
|
|
assert "image_prompt" in inputs
|
|
assert inputs["image_prompt"] == encode_image_b64(ref_path)
|
|
|
|
async def test_image_with_reference_images_flux2(
|
|
self, project_dir: Path, image_bytes: bytes
|
|
) -> None:
|
|
ref1 = project_dir / "ref1.png"
|
|
ref2 = project_dir / "ref2.png"
|
|
_ = ref1.write_bytes(b"ref1 data")
|
|
_ = ref2.write_bytes(b"ref2 data")
|
|
|
|
target_config = TargetConfig(
|
|
prompt="Combine", reference_images=["ref1.png", "ref2.png"]
|
|
)
|
|
bfl_result, mock_http = _make_bfl_mocks(image_bytes)
|
|
|
|
with (
|
|
patch("bulkgen.providers.image.BFLClient") as mock_cls,
|
|
patch("bulkgen.providers.image.httpx.AsyncClient") as mock_http_cls,
|
|
):
|
|
mock_generate = AsyncMock(return_value=bfl_result)
|
|
mock_cls.return_value.generate = mock_generate
|
|
mock_http_cls.return_value = mock_http
|
|
|
|
provider = ImageProvider(api_key="test-key")
|
|
await provider.generate(
|
|
target_name="out.png",
|
|
target_config=target_config,
|
|
resolved_prompt="Combine",
|
|
resolved_model="flux-2-pro",
|
|
project_dir=project_dir,
|
|
)
|
|
|
|
call_args = mock_generate.call_args
|
|
inputs = call_args[0][1]
|
|
assert inputs["input_image"] == encode_image_b64(ref1)
|
|
assert inputs["input_image_2"] == encode_image_b64(ref2)
|
|
|
|
async def test_image_with_reference_image_kontext(
|
|
self, project_dir: Path, image_bytes: bytes
|
|
) -> None:
|
|
ref_path = project_dir / "ref.png"
|
|
_ = ref_path.write_bytes(b"reference image data")
|
|
|
|
target_config = TargetConfig(prompt="Edit", reference_images=["ref.png"])
|
|
bfl_result, mock_http = _make_bfl_mocks(image_bytes)
|
|
|
|
with (
|
|
patch("bulkgen.providers.image.BFLClient") as mock_cls,
|
|
patch("bulkgen.providers.image.httpx.AsyncClient") as mock_http_cls,
|
|
):
|
|
mock_generate = AsyncMock(return_value=bfl_result)
|
|
mock_cls.return_value.generate = mock_generate
|
|
mock_http_cls.return_value = mock_http
|
|
|
|
provider = ImageProvider(api_key="test-key")
|
|
await provider.generate(
|
|
target_name="out.png",
|
|
target_config=target_config,
|
|
resolved_prompt="Edit",
|
|
resolved_model="flux-kontext-pro",
|
|
project_dir=project_dir,
|
|
)
|
|
|
|
call_args = mock_generate.call_args
|
|
inputs = call_args[0][1]
|
|
assert inputs["input_image"] == encode_image_b64(ref_path)
|
|
|
|
async def test_image_no_sample_url_raises(self, project_dir: Path) -> None:
|
|
target_config = TargetConfig(prompt="x")
|
|
|
|
with patch("bulkgen.providers.image.BFLClient") as mock_cls:
|
|
from bulkgen.providers.bfl import BFLError
|
|
|
|
mock_cls.return_value.generate = AsyncMock(
|
|
side_effect=BFLError("BFL task test ready but no sample URL: {}")
|
|
)
|
|
|
|
provider = ImageProvider(api_key="test-key")
|
|
with pytest.raises(BFLError, match="no sample URL"):
|
|
await provider.generate(
|
|
target_name="fail.png",
|
|
target_config=target_config,
|
|
resolved_prompt="x",
|
|
resolved_model="flux-pro-1.1",
|
|
project_dir=project_dir,
|
|
)
|
|
|
|
def test_encode_image_b64(self, project_dir: Path) -> None:
|
|
data = b"test image bytes"
|
|
f = project_dir / "test.png"
|
|
_ = f.write_bytes(data)
|
|
|
|
encoded = encode_image_b64(f)
|
|
assert base64.b64decode(encoded) == data
|
|
|
|
|
|
class TestTextProvider:
|
|
"""Test TextProvider with mocked Mistral client."""
|
|
|
|
async def test_basic_text_generation(self, project_dir: Path) -> None:
|
|
target_config = TargetConfig(prompt="Write a poem")
|
|
response = _make_text_response("Roses are red...")
|
|
|
|
with patch("bulkgen.providers.text.Mistral") as mock_cls:
|
|
mock_cls.return_value = _make_mistral_mock(response)
|
|
|
|
provider = TextProvider(api_key="test-key")
|
|
await provider.generate(
|
|
target_name="poem.txt",
|
|
target_config=target_config,
|
|
resolved_prompt="Write a poem",
|
|
resolved_model="mistral-large-latest",
|
|
project_dir=project_dir,
|
|
)
|
|
|
|
output = project_dir / "poem.txt"
|
|
assert output.exists()
|
|
assert output.read_text() == "Roses are red..."
|
|
|
|
async def test_text_with_text_input(self, project_dir: Path) -> None:
|
|
_ = (project_dir / "source.txt").write_text("Source material here")
|
|
target_config = TargetConfig(prompt="Summarize", inputs=["source.txt"])
|
|
response = _make_text_response("Summary: ...")
|
|
|
|
with patch("bulkgen.providers.text.Mistral") as mock_cls:
|
|
mock_client = _make_mistral_mock(response)
|
|
mock_cls.return_value = mock_client
|
|
|
|
provider = TextProvider(api_key="test-key")
|
|
await provider.generate(
|
|
target_name="summary.md",
|
|
target_config=target_config,
|
|
resolved_prompt="Summarize",
|
|
resolved_model="mistral-large-latest",
|
|
project_dir=project_dir,
|
|
)
|
|
|
|
call_args = mock_client.chat.complete_async.call_args
|
|
messages = call_args.kwargs["messages"]
|
|
prompt_text = messages[0].content
|
|
assert "--- Contents of source.txt ---" in prompt_text
|
|
assert "Source material here" in prompt_text
|
|
|
|
async def test_text_with_image_input(self, project_dir: Path) -> None:
|
|
_ = (project_dir / "photo.png").write_bytes(b"\x89PNG")
|
|
target_config = TargetConfig(prompt="Describe this image", inputs=["photo.png"])
|
|
response = _make_text_response("A beautiful photo")
|
|
|
|
with patch("bulkgen.providers.text.Mistral") as mock_cls:
|
|
mock_client = _make_mistral_mock(response)
|
|
mock_cls.return_value = mock_client
|
|
|
|
provider = TextProvider(api_key="test-key")
|
|
await provider.generate(
|
|
target_name="desc.txt",
|
|
target_config=target_config,
|
|
resolved_prompt="Describe this image",
|
|
resolved_model="mistral-large-latest",
|
|
project_dir=project_dir,
|
|
)
|
|
|
|
call_args = mock_client.chat.complete_async.call_args
|
|
messages = call_args.kwargs["messages"]
|
|
prompt_text = messages[0].content
|
|
assert "[Attached image: photo.png]" in prompt_text
|
|
|
|
async def test_text_no_choices_raises(self, project_dir: Path) -> None:
|
|
target_config = TargetConfig(prompt="x")
|
|
response = MagicMock()
|
|
response.choices = []
|
|
|
|
with patch("bulkgen.providers.text.Mistral") as mock_cls:
|
|
mock_cls.return_value = _make_mistral_mock(response)
|
|
|
|
provider = TextProvider(api_key="test-key")
|
|
with pytest.raises(RuntimeError, match="no choices"):
|
|
await provider.generate(
|
|
target_name="fail.txt",
|
|
target_config=target_config,
|
|
resolved_prompt="x",
|
|
resolved_model="mistral-large-latest",
|
|
project_dir=project_dir,
|
|
)
|
|
|
|
async def test_text_empty_content_raises(self, project_dir: Path) -> None:
|
|
target_config = TargetConfig(prompt="x")
|
|
response = _make_text_response(None)
|
|
|
|
with patch("bulkgen.providers.text.Mistral") as mock_cls:
|
|
mock_cls.return_value = _make_mistral_mock(response)
|
|
|
|
provider = TextProvider(api_key="test-key")
|
|
with pytest.raises(RuntimeError, match="empty content"):
|
|
await provider.generate(
|
|
target_name="fail.txt",
|
|
target_config=target_config,
|
|
resolved_prompt="x",
|
|
resolved_model="mistral-large-latest",
|
|
project_dir=project_dir,
|
|
)
|
|
|
|
async def test_text_with_multiple_inputs(self, project_dir: Path) -> None:
|
|
_ = (project_dir / "a.txt").write_text("content A")
|
|
_ = (project_dir / "b.txt").write_text("content B")
|
|
_ = (project_dir / "c.png").write_bytes(b"\x89PNG")
|
|
|
|
target_config = TargetConfig(
|
|
prompt="Combine all", inputs=["a.txt", "b.txt", "c.png"]
|
|
)
|
|
response = _make_text_response("Combined")
|
|
|
|
with patch("bulkgen.providers.text.Mistral") as mock_cls:
|
|
mock_client = _make_mistral_mock(response)
|
|
mock_cls.return_value = mock_client
|
|
|
|
provider = TextProvider(api_key="test-key")
|
|
await provider.generate(
|
|
target_name="out.md",
|
|
target_config=target_config,
|
|
resolved_prompt="Combine all",
|
|
resolved_model="mistral-large-latest",
|
|
project_dir=project_dir,
|
|
)
|
|
|
|
call_args = mock_client.chat.complete_async.call_args
|
|
prompt_text = call_args.kwargs["messages"][0].content
|
|
assert "--- Contents of a.txt ---" in prompt_text
|
|
assert "content A" in prompt_text
|
|
assert "--- Contents of b.txt ---" in prompt_text
|
|
assert "content B" in prompt_text
|
|
assert "[Attached image: c.png]" in prompt_text
|