Konstantin Fickel
d8e0ed561d
When a target is present in the state file but no longer in the config, its output file is deleted (or archived if archive_folder is set) and its state entry is removed. This runs at the start of every build.
944 lines
34 KiB
Python
944 lines
34 KiB
Python
"""Integration tests for hokusai.builder."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import override
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from hokusai.builder import (
|
|
_collect_all_deps, # pyright: ignore[reportPrivateUsage]
|
|
_collect_dep_files, # pyright: ignore[reportPrivateUsage]
|
|
_collect_extra_params, # pyright: ignore[reportPrivateUsage]
|
|
run_build,
|
|
)
|
|
from hokusai.config import GenerateTargetConfig, ProjectConfig
|
|
from hokusai.providers import Provider
|
|
from hokusai.providers.models import Capability, ModelInfo
|
|
from hokusai.state import load_state
|
|
|
|
WriteConfig = Callable[[dict[str, object]], ProjectConfig]
|
|
|
|
_PROJECT = "project"
|
|
|
|
|
|
_FAKE_TEXT_MODELS = [
|
|
ModelInfo(
|
|
name="pixtral-large-latest",
|
|
provider="Fake",
|
|
type="text",
|
|
capabilities=[Capability.TEXT_GENERATION, Capability.VISION],
|
|
),
|
|
]
|
|
|
|
_FAKE_IMAGE_MODELS = [
|
|
ModelInfo(
|
|
name="flux-2-pro",
|
|
provider="Fake",
|
|
type="image",
|
|
capabilities=[Capability.TEXT_TO_IMAGE, Capability.REFERENCE_IMAGES],
|
|
),
|
|
]
|
|
|
|
|
|
class FakeTextProvider(Provider):
|
|
"""A text provider that writes a marker file instead of calling an API."""
|
|
|
|
@staticmethod
|
|
@override
|
|
def get_provided_models() -> list[ModelInfo]:
|
|
return _FAKE_TEXT_MODELS
|
|
|
|
@override
|
|
async def generate(
|
|
self,
|
|
target_name: str,
|
|
target_config: GenerateTargetConfig,
|
|
resolved_prompt: str,
|
|
resolved_model: ModelInfo,
|
|
project_dir: Path,
|
|
) -> None:
|
|
output = project_dir / target_name
|
|
_ = output.write_text(f"generated:{target_name}:{resolved_prompt}")
|
|
|
|
|
|
class FakeImageProvider(Provider):
|
|
"""An image provider that writes a marker file instead of calling an API."""
|
|
|
|
@staticmethod
|
|
@override
|
|
def get_provided_models() -> list[ModelInfo]:
|
|
return _FAKE_IMAGE_MODELS
|
|
|
|
@override
|
|
async def generate(
|
|
self,
|
|
target_name: str,
|
|
target_config: GenerateTargetConfig,
|
|
resolved_prompt: str,
|
|
resolved_model: ModelInfo,
|
|
project_dir: Path,
|
|
) -> None:
|
|
output = project_dir / target_name
|
|
_ = output.write_text(f"generated:{target_name}:{resolved_prompt}")
|
|
|
|
|
|
class FailingTextProvider(Provider):
|
|
"""A text provider that always raises."""
|
|
|
|
@staticmethod
|
|
@override
|
|
def get_provided_models() -> list[ModelInfo]:
|
|
return _FAKE_TEXT_MODELS
|
|
|
|
@override
|
|
async def generate(
|
|
self,
|
|
target_name: str,
|
|
target_config: GenerateTargetConfig,
|
|
resolved_prompt: str,
|
|
resolved_model: ModelInfo,
|
|
project_dir: Path,
|
|
) -> None:
|
|
msg = f"Simulated failure for {target_name}"
|
|
raise RuntimeError(msg)
|
|
|
|
|
|
def _fake_providers() -> list[Provider]:
|
|
return [FakeTextProvider(), FakeImageProvider()]
|
|
|
|
|
|
class TestCollectHelpers:
|
|
"""Test dependency collection helpers."""
|
|
|
|
def test_collect_dep_files(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
_ = (project_dir / "input.txt").write_text("data")
|
|
_ = (project_dir / "ref.png").write_bytes(b"ref")
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"out.png": {
|
|
"prompt": "x",
|
|
"inputs": ["input.txt"],
|
|
"reference_images": ["ref.png"],
|
|
"control_images": [],
|
|
}
|
|
}
|
|
}
|
|
)
|
|
deps = _collect_dep_files("out.png", config, project_dir)
|
|
dep_names = [d.name for d in deps]
|
|
assert "input.txt" in dep_names
|
|
assert "ref.png" in dep_names
|
|
|
|
def test_collect_extra_params(self, write_config: WriteConfig) -> None:
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"out.png": {
|
|
"prompt": "x",
|
|
"width": 512,
|
|
"height": 256,
|
|
"reference_images": ["ref.png"],
|
|
}
|
|
}
|
|
}
|
|
)
|
|
params = _collect_extra_params("out.png", config)
|
|
assert params["width"] == 512
|
|
assert params["height"] == 256
|
|
assert params["reference_images"] == ("ref.png",)
|
|
|
|
def test_collect_extra_params_empty(self, write_config: WriteConfig) -> None:
|
|
config = write_config({"targets": {"out.txt": {"prompt": "x"}}})
|
|
assert _collect_extra_params("out.txt", config) == {}
|
|
|
|
def test_collect_all_deps(self, write_config: WriteConfig) -> None:
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"out.png": {
|
|
"prompt": "x",
|
|
"inputs": ["a.txt"],
|
|
"reference_images": ["ref.png"],
|
|
"control_images": ["c1.png", "c2.png"],
|
|
}
|
|
}
|
|
}
|
|
)
|
|
deps = _collect_all_deps("out.png", config)
|
|
assert deps == ["a.txt", "ref.png", "c1.png", "c2.png"]
|
|
|
|
|
|
class TestRunBuild:
|
|
"""Integration tests for the full build pipeline with fake providers."""
|
|
|
|
async def test_build_single_text_target(
|
|
self, project_dir: Path, simple_text_config: ProjectConfig
|
|
) -> None:
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
result = await run_build(simple_text_config, project_dir, _PROJECT)
|
|
|
|
assert result.built == ["output.txt"]
|
|
assert result.skipped == []
|
|
assert result.failed == {}
|
|
assert (project_dir / "output.txt").exists()
|
|
|
|
async def test_build_chain_dependency(
|
|
self, project_dir: Path, multi_target_config: ProjectConfig
|
|
) -> None:
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
result = await run_build(multi_target_config, project_dir, _PROJECT)
|
|
|
|
assert "summary.md" in result.built
|
|
assert "final.txt" in result.built
|
|
assert "hero.png" in result.built
|
|
assert result.failed == {}
|
|
|
|
assert (project_dir / "summary.md").exists()
|
|
assert (project_dir / "final.txt").exists()
|
|
assert (project_dir / "hero.png").exists()
|
|
|
|
async def test_incremental_build_skips_clean_targets(
|
|
self, project_dir: Path, simple_text_config: ProjectConfig
|
|
) -> None:
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
result1 = await run_build(simple_text_config, project_dir, _PROJECT)
|
|
assert result1.built == ["output.txt"]
|
|
|
|
result2 = await run_build(simple_text_config, project_dir, _PROJECT)
|
|
assert result2.skipped == ["output.txt"]
|
|
assert result2.built == []
|
|
|
|
async def test_rebuild_after_prompt_change(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config1 = write_config({"targets": {"out.txt": {"prompt": "version 1"}}})
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
r1 = await run_build(config1, project_dir, _PROJECT)
|
|
assert r1.built == ["out.txt"]
|
|
|
|
config2 = write_config({"targets": {"out.txt": {"prompt": "version 2"}}})
|
|
r2 = await run_build(config2, project_dir, _PROJECT)
|
|
assert r2.built == ["out.txt"]
|
|
|
|
async def test_rebuild_after_input_change(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
_ = (project_dir / "data.txt").write_text("original")
|
|
config = write_config(
|
|
{"targets": {"out.md": {"prompt": "x", "inputs": ["data.txt"]}}}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
r1 = await run_build(config, project_dir, _PROJECT)
|
|
assert r1.built == ["out.md"]
|
|
|
|
_ = (project_dir / "data.txt").write_text("modified")
|
|
r2 = await run_build(config, project_dir, _PROJECT)
|
|
assert r2.built == ["out.md"]
|
|
|
|
async def test_selective_build_single_target(
|
|
self, project_dir: Path, multi_target_config: ProjectConfig
|
|
) -> None:
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
result = await run_build(
|
|
multi_target_config, project_dir, _PROJECT, target="summary.md"
|
|
)
|
|
|
|
assert "summary.md" in result.built
|
|
assert "hero.png" not in result.built
|
|
assert "final.txt" not in result.built
|
|
|
|
async def test_selective_build_unknown_target_raises(
|
|
self, project_dir: Path, simple_text_config: ProjectConfig
|
|
) -> None:
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
with pytest.raises(ValueError, match="Unknown target"):
|
|
_ = await run_build(
|
|
simple_text_config, project_dir, _PROJECT, target="nonexistent.txt"
|
|
)
|
|
|
|
async def test_failed_target_isolates_independent(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"fail.txt": {"prompt": "will fail"},
|
|
"ok.txt": {"prompt": "will succeed"},
|
|
}
|
|
}
|
|
)
|
|
fail_provider = FailingTextProvider()
|
|
fake_provider = FakeTextProvider()
|
|
|
|
async def selective_generate(
|
|
target_name: str,
|
|
target_config: GenerateTargetConfig,
|
|
resolved_prompt: str,
|
|
resolved_model: ModelInfo,
|
|
project_dir: Path,
|
|
) -> None:
|
|
if target_name == "fail.txt":
|
|
await fail_provider.generate(
|
|
target_name,
|
|
target_config,
|
|
resolved_prompt,
|
|
resolved_model,
|
|
project_dir,
|
|
)
|
|
else:
|
|
await fake_provider.generate(
|
|
target_name,
|
|
target_config,
|
|
resolved_prompt,
|
|
resolved_model,
|
|
project_dir,
|
|
)
|
|
|
|
routing_provider = FakeTextProvider()
|
|
routing_provider.generate = selective_generate # type: ignore[assignment]
|
|
|
|
with patch(
|
|
"hokusai.builder._create_providers",
|
|
return_value=[routing_provider, FakeImageProvider()],
|
|
):
|
|
result = await run_build(config, project_dir, _PROJECT)
|
|
|
|
assert "fail.txt" in result.failed
|
|
assert "ok.txt" in result.built
|
|
|
|
async def test_failed_dep_cascades(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"base.txt": {"prompt": "base"},
|
|
"child.txt": {"prompt": "child", "inputs": ["base.txt"]},
|
|
}
|
|
}
|
|
)
|
|
|
|
with patch(
|
|
"hokusai.builder._create_providers",
|
|
return_value=[FailingTextProvider(), FakeImageProvider()],
|
|
):
|
|
result = await run_build(config, project_dir, _PROJECT)
|
|
|
|
assert "base.txt" in result.failed
|
|
assert "child.txt" in result.failed
|
|
assert "Dependency failed" in result.failed["child.txt"]
|
|
|
|
async def test_missing_provider_records_failure(
|
|
self, project_dir: Path, simple_text_config: ProjectConfig
|
|
) -> None:
|
|
with patch(
|
|
"hokusai.builder._create_providers",
|
|
return_value=[],
|
|
):
|
|
result = await run_build(simple_text_config, project_dir, _PROJECT)
|
|
|
|
assert "output.txt" in result.failed
|
|
assert "No provider available" in result.failed["output.txt"]
|
|
|
|
async def test_state_saved_after_each_generation(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"a.txt": {"prompt": "first"},
|
|
"b.txt": {"prompt": "second", "inputs": ["a.txt"]},
|
|
}
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
_ = await run_build(config, project_dir, _PROJECT)
|
|
|
|
state = load_state(project_dir, _PROJECT)
|
|
assert "a.txt" in state.targets
|
|
assert "b.txt" in state.targets
|
|
|
|
async def test_prompt_file_resolution_in_build(
|
|
self, project_dir: Path, prompt_file: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config({"targets": {"out.txt": {"prompt": prompt_file.name}}})
|
|
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
result = await run_build(config, project_dir, _PROJECT)
|
|
|
|
assert result.built == ["out.txt"]
|
|
content = (project_dir / "out.txt").read_text()
|
|
assert "This prompt comes from a file" in content
|
|
|
|
async def test_rebuild_after_output_deleted(
|
|
self, project_dir: Path, simple_text_config: ProjectConfig
|
|
) -> None:
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
r1 = await run_build(simple_text_config, project_dir, _PROJECT)
|
|
assert r1.built == ["output.txt"]
|
|
|
|
(project_dir / "output.txt").unlink()
|
|
|
|
r2 = await run_build(simple_text_config, project_dir, _PROJECT)
|
|
assert r2.built == ["output.txt"]
|
|
|
|
async def test_diamond_dependency_all_built(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
_ = (project_dir / "root.txt").write_text("root data")
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"left.md": {"prompt": "left", "inputs": ["root.txt"]},
|
|
"right.md": {"prompt": "right", "inputs": ["root.txt"]},
|
|
"merge.txt": {
|
|
"prompt": "merge",
|
|
"inputs": ["left.md", "right.md"],
|
|
},
|
|
}
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
result = await run_build(config, project_dir, _PROJECT)
|
|
|
|
assert set(result.built) == {"left.md", "right.md", "merge.txt"}
|
|
assert result.failed == {}
|
|
|
|
|
|
class TestArchiveOnBuild:
|
|
"""Test that build archives existing artifacts when archive_folder is set."""
|
|
|
|
async def test_build_archives_existing_file(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config(
|
|
{
|
|
"archive_folder": "archive",
|
|
"targets": {"out.txt": {"prompt": "version 1"}},
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
r1 = await run_build(config, project_dir, _PROJECT)
|
|
assert r1.built == ["out.txt"]
|
|
v1_content = (project_dir / "out.txt").read_text()
|
|
|
|
config2 = write_config(
|
|
{
|
|
"archive_folder": "archive",
|
|
"targets": {"out.txt": {"prompt": "version 2"}},
|
|
}
|
|
)
|
|
r2 = await run_build(config2, project_dir, _PROJECT)
|
|
assert r2.built == ["out.txt"]
|
|
|
|
# v1 should be archived, v2 should be current
|
|
archived = project_dir / "archive" / "out.01.txt"
|
|
assert archived.exists()
|
|
assert archived.read_text() == v1_content
|
|
assert (project_dir / "out.txt").exists()
|
|
|
|
async def test_build_no_archive_without_setting(
|
|
self, project_dir: Path, simple_text_config: ProjectConfig
|
|
) -> None:
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
r1 = await run_build(simple_text_config, project_dir, _PROJECT)
|
|
assert r1.built == ["output.txt"]
|
|
|
|
assert not (project_dir / "archive").exists()
|
|
|
|
async def test_build_archives_increment(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config_raw: dict[str, object] = {
|
|
"archive_folder": "archive",
|
|
"targets": {"out.txt": {"prompt": "v"}},
|
|
}
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
for i in range(1, 4):
|
|
cfg = write_config(
|
|
{**config_raw, "targets": {"out.txt": {"prompt": f"v{i}"}}}
|
|
)
|
|
_ = await run_build(cfg, project_dir, _PROJECT)
|
|
|
|
assert (project_dir / "archive" / "out.01.txt").exists()
|
|
assert (project_dir / "archive" / "out.02.txt").exists()
|
|
assert not (project_dir / "archive" / "out.03.txt").exists()
|
|
|
|
|
|
class TestDownloadTarget:
|
|
"""Tests for download-type targets that fetch files from URLs."""
|
|
|
|
async def test_download_target_fetches_url(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"fish.png": {"download": "https://example.com/fish.png"},
|
|
}
|
|
}
|
|
)
|
|
mock_response = MagicMock()
|
|
mock_response.content = b"fake image bytes"
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with (
|
|
patch("hokusai.builder._create_providers", return_value=_fake_providers()),
|
|
patch("hokusai.builder.httpx.AsyncClient") as mock_client_cls,
|
|
):
|
|
mock_client = AsyncMock()
|
|
mock_client.get = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
result = await run_build(config, project_dir, _PROJECT)
|
|
|
|
assert result.built == ["fish.png"]
|
|
assert (project_dir / "fish.png").read_bytes() == b"fake image bytes"
|
|
mock_client.get.assert_called_once_with( # pyright: ignore[reportAny]
|
|
"https://example.com/fish.png", follow_redirects=True
|
|
)
|
|
|
|
async def test_download_target_incremental_skip(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"fish.png": {"download": "https://example.com/fish.png"},
|
|
}
|
|
}
|
|
)
|
|
mock_response = MagicMock()
|
|
mock_response.content = b"fake image bytes"
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with (
|
|
patch("hokusai.builder._create_providers", return_value=_fake_providers()),
|
|
patch("hokusai.builder.httpx.AsyncClient") as mock_client_cls,
|
|
):
|
|
mock_client = AsyncMock()
|
|
mock_client.get = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
r1 = await run_build(config, project_dir, _PROJECT)
|
|
assert r1.built == ["fish.png"]
|
|
|
|
r2 = await run_build(config, project_dir, _PROJECT)
|
|
assert r2.skipped == ["fish.png"]
|
|
assert r2.built == []
|
|
|
|
async def test_download_target_rebuild_on_url_change(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config1 = write_config(
|
|
{
|
|
"targets": {
|
|
"fish.png": {"download": "https://example.com/fish-v1.png"},
|
|
}
|
|
}
|
|
)
|
|
mock_response = MagicMock()
|
|
mock_response.content = b"v1 bytes"
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with (
|
|
patch("hokusai.builder._create_providers", return_value=_fake_providers()),
|
|
patch("hokusai.builder.httpx.AsyncClient") as mock_client_cls,
|
|
):
|
|
mock_client = AsyncMock()
|
|
mock_client.get = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
r1 = await run_build(config1, project_dir, _PROJECT)
|
|
assert r1.built == ["fish.png"]
|
|
|
|
config2 = write_config(
|
|
{
|
|
"targets": {
|
|
"fish.png": {"download": "https://example.com/fish-v2.png"},
|
|
}
|
|
}
|
|
)
|
|
mock_response.content = b"v2 bytes"
|
|
|
|
r2 = await run_build(config2, project_dir, _PROJECT)
|
|
assert r2.built == ["fish.png"]
|
|
assert (project_dir / "fish.png").read_bytes() == b"v2 bytes"
|
|
|
|
async def test_download_target_as_dependency(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"fish.png": {"download": "https://example.com/fish.png"},
|
|
"description.txt": {
|
|
"prompt": "Describe the fish",
|
|
"inputs": ["fish.png"],
|
|
},
|
|
}
|
|
}
|
|
)
|
|
mock_response = MagicMock()
|
|
mock_response.content = b"fake fish image"
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with (
|
|
patch("hokusai.builder._create_providers", return_value=_fake_providers()),
|
|
patch("hokusai.builder.httpx.AsyncClient") as mock_client_cls,
|
|
):
|
|
mock_client = AsyncMock()
|
|
mock_client.get = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
result = await run_build(config, project_dir, _PROJECT)
|
|
|
|
assert "fish.png" in result.built
|
|
assert "description.txt" in result.built
|
|
assert (project_dir / "fish.png").read_bytes() == b"fake fish image"
|
|
assert (project_dir / "description.txt").exists()
|
|
|
|
|
|
class TestContentTarget:
|
|
"""Tests for content-type targets that write literal text."""
|
|
|
|
async def test_content_target_writes_file(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config({"targets": {"file.txt": {"content": "ABC"}}})
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
result = await run_build(config, project_dir, _PROJECT)
|
|
|
|
assert result.built == ["file.txt"]
|
|
assert (project_dir / "file.txt").read_text() == "ABC"
|
|
|
|
async def test_content_target_incremental_skip(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config({"targets": {"file.txt": {"content": "ABC"}}})
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
r1 = await run_build(config, project_dir, _PROJECT)
|
|
assert r1.built == ["file.txt"]
|
|
|
|
r2 = await run_build(config, project_dir, _PROJECT)
|
|
assert r2.skipped == ["file.txt"]
|
|
assert r2.built == []
|
|
|
|
async def test_content_target_rebuild_on_change(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config1 = write_config({"targets": {"file.txt": {"content": "ABC"}}})
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
r1 = await run_build(config1, project_dir, _PROJECT)
|
|
assert r1.built == ["file.txt"]
|
|
|
|
config2 = write_config({"targets": {"file.txt": {"content": "XYZ"}}})
|
|
r2 = await run_build(config2, project_dir, _PROJECT)
|
|
assert r2.built == ["file.txt"]
|
|
assert (project_dir / "file.txt").read_text() == "XYZ"
|
|
|
|
async def test_content_target_no_archive(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config1 = write_config(
|
|
{
|
|
"archive_folder": "archive",
|
|
"targets": {"file.txt": {"content": "v1"}},
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
r1 = await run_build(config1, project_dir, _PROJECT)
|
|
assert r1.built == ["file.txt"]
|
|
|
|
config2 = write_config(
|
|
{
|
|
"archive_folder": "archive",
|
|
"targets": {"file.txt": {"content": "v2"}},
|
|
}
|
|
)
|
|
r2 = await run_build(config2, project_dir, _PROJECT)
|
|
assert r2.built == ["file.txt"]
|
|
|
|
assert (project_dir / "file.txt").read_text() == "v2"
|
|
assert not (project_dir / "archive").exists()
|
|
|
|
async def test_content_target_as_dependency(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"data.txt": {"content": "some data"},
|
|
"output.md": {
|
|
"prompt": "Process the data",
|
|
"inputs": ["data.txt"],
|
|
},
|
|
}
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
result = await run_build(config, project_dir, _PROJECT)
|
|
|
|
assert "data.txt" in result.built
|
|
assert "output.md" in result.built
|
|
assert (project_dir / "data.txt").read_text() == "some data"
|
|
|
|
async def test_content_target_no_provider_needed(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config({"targets": {"file.txt": {"content": "ABC"}}})
|
|
with patch("hokusai.builder._create_providers", return_value=[]):
|
|
result = await run_build(config, project_dir, _PROJECT)
|
|
|
|
assert result.built == ["file.txt"]
|
|
assert result.failed == {}
|
|
|
|
|
|
class TestStaleTargetCleanup:
|
|
"""Tests for cleanup of targets removed from config."""
|
|
|
|
async def test_stale_target_file_deleted(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config1 = write_config(
|
|
{
|
|
"targets": {
|
|
"a.txt": {"prompt": "first"},
|
|
"b.txt": {"prompt": "second"},
|
|
}
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
r1 = await run_build(config1, project_dir, _PROJECT)
|
|
assert set(r1.built) == {"a.txt", "b.txt"}
|
|
|
|
# Remove b.txt from config
|
|
config2 = write_config({"targets": {"a.txt": {"prompt": "first"}}})
|
|
r2 = await run_build(config2, project_dir, _PROJECT)
|
|
|
|
assert not (project_dir / "b.txt").exists()
|
|
assert "a.txt" in r2.skipped
|
|
|
|
async def test_stale_target_archived(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config1 = write_config(
|
|
{
|
|
"archive_folder": "archive",
|
|
"targets": {
|
|
"a.txt": {"prompt": "first"},
|
|
"b.txt": {"prompt": "second"},
|
|
},
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
r1 = await run_build(config1, project_dir, _PROJECT)
|
|
assert set(r1.built) == {"a.txt", "b.txt"}
|
|
|
|
config2 = write_config(
|
|
{
|
|
"archive_folder": "archive",
|
|
"targets": {"a.txt": {"prompt": "first"}},
|
|
}
|
|
)
|
|
_ = await run_build(config2, project_dir, _PROJECT)
|
|
|
|
assert not (project_dir / "b.txt").exists()
|
|
assert (project_dir / "archive" / "b.01.txt").exists()
|
|
|
|
async def test_stale_state_entry_removed(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config1 = write_config(
|
|
{
|
|
"targets": {
|
|
"a.txt": {"prompt": "first"},
|
|
"b.txt": {"prompt": "second"},
|
|
}
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
_ = await run_build(config1, project_dir, _PROJECT)
|
|
|
|
state = load_state(project_dir, _PROJECT)
|
|
assert "b.txt" in state.targets
|
|
|
|
config2 = write_config({"targets": {"a.txt": {"prompt": "first"}}})
|
|
_ = await run_build(config2, project_dir, _PROJECT)
|
|
|
|
state = load_state(project_dir, _PROJECT)
|
|
assert "b.txt" not in state.targets
|
|
assert "a.txt" in state.targets
|
|
|
|
async def test_stale_target_file_already_gone(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config1 = write_config(
|
|
{
|
|
"targets": {
|
|
"a.txt": {"prompt": "first"},
|
|
"b.txt": {"prompt": "second"},
|
|
}
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
_ = await run_build(config1, project_dir, _PROJECT)
|
|
(project_dir / "b.txt").unlink()
|
|
|
|
config2 = write_config({"targets": {"a.txt": {"prompt": "first"}}})
|
|
_ = await run_build(config2, project_dir, _PROJECT)
|
|
|
|
# State entry should still be cleaned even if file is gone
|
|
state = load_state(project_dir, _PROJECT)
|
|
assert "b.txt" not in state.targets
|
|
|
|
async def test_stale_cleanup_emits_progress(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
from hokusai.builder import BuildEvent
|
|
|
|
config1 = write_config(
|
|
{
|
|
"targets": {
|
|
"a.txt": {"prompt": "first"},
|
|
"b.txt": {"prompt": "second"},
|
|
}
|
|
}
|
|
)
|
|
events: list[tuple[BuildEvent, str, str]] = []
|
|
|
|
def recorder(event: BuildEvent, name: str, detail: str) -> None:
|
|
events.append((event, name, detail))
|
|
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
_ = await run_build(config1, project_dir, _PROJECT, on_progress=recorder)
|
|
events.clear()
|
|
|
|
config2 = write_config({"targets": {"a.txt": {"prompt": "first"}}})
|
|
_ = await run_build(config2, project_dir, _PROJECT, on_progress=recorder)
|
|
|
|
removed_events = [
|
|
(e, n) for e, n, _ in events if e is BuildEvent.TARGET_REMOVED
|
|
]
|
|
assert ("b.txt",) in [(n,) for _, n in removed_events]
|
|
|
|
|
|
class TestLoopExpansion:
|
|
"""End-to-end tests for loop-expanded targets in builds."""
|
|
|
|
async def test_loop_content_targets_build(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config(
|
|
{
|
|
"loops": {"n": ["1", "2", "3"]},
|
|
"targets": {"file-[n].txt": {"content": "Value [n]"}},
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
result = await run_build(config, project_dir, _PROJECT)
|
|
|
|
assert set(result.built) == {"file-1.txt", "file-2.txt", "file-3.txt"}
|
|
assert (project_dir / "file-1.txt").read_text() == "Value 1"
|
|
assert (project_dir / "file-2.txt").read_text() == "Value 2"
|
|
assert (project_dir / "file-3.txt").read_text() == "Value 3"
|
|
|
|
async def test_loop_incremental_skip(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config(
|
|
{
|
|
"loops": {"n": ["1", "2"]},
|
|
"targets": {"file-[n].txt": {"content": "Value [n]"}},
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
r1 = await run_build(config, project_dir, _PROJECT)
|
|
assert len(r1.built) == 2
|
|
|
|
r2 = await run_build(config, project_dir, _PROJECT)
|
|
assert r2.built == []
|
|
assert set(r2.skipped) == {"file-1.txt", "file-2.txt"}
|
|
|
|
async def test_loop_with_dependency_chain(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config(
|
|
{
|
|
"loops": {"id": ["a", "b"]},
|
|
"targets": {
|
|
"data-[id].txt": {"content": "Data for [id]"},
|
|
"summary-[id].txt": {
|
|
"prompt": "Summarize [id]",
|
|
"inputs": ["data-[id].txt"],
|
|
},
|
|
},
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
result = await run_build(config, project_dir, _PROJECT)
|
|
|
|
assert "data-a.txt" in result.built
|
|
assert "data-b.txt" in result.built
|
|
assert "summary-a.txt" in result.built
|
|
assert "summary-b.txt" in result.built
|
|
assert result.failed == {}
|
|
|
|
|
|
class TestPlaceholderPrompts:
|
|
"""Tests for prompt placeholder substitution in builds."""
|
|
|
|
async def test_placeholder_in_prompt_triggers_rebuild(
|
|
self, project_dir: Path, write_config: WriteConfig
|
|
) -> None:
|
|
_ = (project_dir / "style.txt").write_text("impressionist")
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"out.txt": {"prompt": "Paint in {style.txt} style"},
|
|
}
|
|
}
|
|
)
|
|
with patch("hokusai.builder._create_providers", return_value=_fake_providers()):
|
|
r1 = await run_build(config, project_dir, _PROJECT)
|
|
assert r1.built == ["out.txt"]
|
|
content1 = (project_dir / "out.txt").read_text()
|
|
assert "impressionist" in content1
|
|
|
|
# Change the placeholder file
|
|
_ = (project_dir / "style.txt").write_text("cubist")
|
|
r2 = await run_build(config, project_dir, _PROJECT)
|
|
assert r2.built == ["out.txt"]
|
|
|
|
async def test_placeholder_deps_in_collect_all(
|
|
self, write_config: WriteConfig
|
|
) -> None:
|
|
config = write_config(
|
|
{
|
|
"targets": {
|
|
"out.txt": {
|
|
"prompt": "Use {a.txt} and {b.txt}",
|
|
"inputs": ["c.txt"],
|
|
},
|
|
}
|
|
}
|
|
)
|
|
deps = _collect_all_deps("out.txt", config)
|
|
assert "a.txt" in deps
|
|
assert "b.txt" in deps
|
|
assert "c.txt" in deps
|