Konstantin Fickel
bb03975ece
Content targets write a string directly to the output file without
invoking any AI provider. They don't require API keys and are not
archived when overwritten.
Example usage in .hokusai.yaml:
file.txt:
content: ABC
753 lines
27 KiB
Python
753 lines
27 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 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
|