Konstantin Fickel
d76951fe47
The regenerate command accepts one or more target names and forces them to be rebuilt even if they are up to date. This respects the archive_folder setting, archiving previous versions before overwriting.
346 lines
12 KiB
Python
346 lines
12 KiB
Python
"""Integration tests for hokusai.cli.
|
|
|
|
Patching ``Path.cwd()`` produces Any-typed return values from mock objects.
|
|
"""
|
|
# pyright: reportAny=false
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
import yaml
|
|
from typer.testing import CliRunner
|
|
|
|
from hokusai.builder import BuildResult
|
|
from hokusai.cli import app
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
@pytest.fixture
|
|
def cli_project(tmp_path: Path) -> Path:
|
|
"""Create a minimal project directory with a config file."""
|
|
config = {
|
|
"targets": {
|
|
"output.txt": {"prompt": "Generate text"},
|
|
"image.png": {"prompt": "Generate image"},
|
|
}
|
|
}
|
|
_ = (tmp_path / "project.hokusai.yaml").write_text(
|
|
yaml.dump(config, default_flow_style=False)
|
|
)
|
|
return tmp_path
|
|
|
|
|
|
class TestFindConfig:
|
|
"""Test config file discovery."""
|
|
|
|
def test_no_config_file(self, tmp_path: Path) -> None:
|
|
with patch("hokusai.cli.Path") as mock_path_cls:
|
|
mock_path_cls.cwd.return_value = tmp_path
|
|
result = runner.invoke(app, ["build"])
|
|
assert result.exit_code != 0
|
|
assert "No .hokusai.yaml/.yml file found" in result.output
|
|
|
|
def test_multiple_config_files(self, tmp_path: Path) -> None:
|
|
_ = (tmp_path / "a.hokusai.yaml").write_text(
|
|
yaml.dump({"targets": {"x.txt": {"prompt": "a"}}})
|
|
)
|
|
_ = (tmp_path / "b.hokusai.yaml").write_text(
|
|
yaml.dump({"targets": {"y.txt": {"prompt": "b"}}})
|
|
)
|
|
with patch("hokusai.cli.Path") as mock_path_cls:
|
|
mock_path_cls.cwd.return_value = tmp_path
|
|
result = runner.invoke(app, ["build"])
|
|
assert result.exit_code != 0
|
|
assert "Multiple .hokusai.yaml/.yml files found" in result.output
|
|
|
|
|
|
class TestBuildCommand:
|
|
"""Test the build CLI command."""
|
|
|
|
def test_build_success(self, cli_project: Path) -> None:
|
|
build_result = BuildResult(
|
|
built=["output.txt", "image.png"], skipped=[], failed={}
|
|
)
|
|
with (
|
|
patch("hokusai.cli.Path") as mock_path_cls,
|
|
patch(
|
|
"hokusai.cli.run_build",
|
|
new_callable=AsyncMock,
|
|
return_value=build_result,
|
|
),
|
|
):
|
|
mock_path_cls.cwd.return_value = cli_project
|
|
result = runner.invoke(app, ["build"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "2 built" in result.output
|
|
|
|
def test_build_with_skipped(self, cli_project: Path) -> None:
|
|
build_result = BuildResult(
|
|
built=[], skipped=["output.txt", "image.png"], failed={}
|
|
)
|
|
with (
|
|
patch("hokusai.cli.Path") as mock_path_cls,
|
|
patch(
|
|
"hokusai.cli.run_build",
|
|
new_callable=AsyncMock,
|
|
return_value=build_result,
|
|
),
|
|
):
|
|
mock_path_cls.cwd.return_value = cli_project
|
|
result = runner.invoke(app, ["build"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "2 skipped" in result.output
|
|
|
|
def test_build_with_failures(self, cli_project: Path) -> None:
|
|
build_result = BuildResult(
|
|
built=["output.txt"],
|
|
skipped=[],
|
|
failed={"image.png": "Missing BFL_API_KEY"},
|
|
)
|
|
with (
|
|
patch("hokusai.cli.Path") as mock_path_cls,
|
|
patch(
|
|
"hokusai.cli.run_build",
|
|
new_callable=AsyncMock,
|
|
return_value=build_result,
|
|
),
|
|
):
|
|
mock_path_cls.cwd.return_value = cli_project
|
|
result = runner.invoke(app, ["build"])
|
|
|
|
assert result.exit_code == 1
|
|
assert "1 failed" in result.output
|
|
|
|
def test_build_specific_target(self, cli_project: Path) -> None:
|
|
build_result = BuildResult(built=["output.txt"], skipped=[], failed={})
|
|
with (
|
|
patch("hokusai.cli.Path") as mock_path_cls,
|
|
patch(
|
|
"hokusai.cli.run_build",
|
|
new_callable=AsyncMock,
|
|
return_value=build_result,
|
|
) as mock_run,
|
|
):
|
|
mock_path_cls.cwd.return_value = cli_project
|
|
result = runner.invoke(app, ["build", "output.txt"])
|
|
|
|
assert result.exit_code == 0
|
|
call_args = mock_run.call_args
|
|
# positional args: (config, project_dir, project_name, target)
|
|
assert call_args[0][3] == "output.txt"
|
|
|
|
|
|
class TestRegenerateCommand:
|
|
"""Test the regenerate CLI command."""
|
|
|
|
def test_regenerate_forces_rebuild(self, cli_project: Path) -> None:
|
|
build_result = BuildResult(
|
|
built=["output.txt"], skipped=["image.png"], failed={}
|
|
)
|
|
with (
|
|
patch("hokusai.cli.Path") as mock_path_cls,
|
|
patch(
|
|
"hokusai.cli.run_build",
|
|
new_callable=AsyncMock,
|
|
return_value=build_result,
|
|
) as mock_run,
|
|
):
|
|
mock_path_cls.cwd.return_value = cli_project
|
|
result = runner.invoke(app, ["regenerate", "output.txt"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "1 regenerated" in result.output
|
|
# Check force_dirty was passed
|
|
call_kwargs = mock_run.call_args[1]
|
|
assert call_kwargs["force_dirty"] == {"output.txt"}
|
|
|
|
def test_regenerate_multiple_targets(self, cli_project: Path) -> None:
|
|
build_result = BuildResult(
|
|
built=["output.txt", "image.png"], skipped=[], failed={}
|
|
)
|
|
with (
|
|
patch("hokusai.cli.Path") as mock_path_cls,
|
|
patch(
|
|
"hokusai.cli.run_build",
|
|
new_callable=AsyncMock,
|
|
return_value=build_result,
|
|
) as mock_run,
|
|
):
|
|
mock_path_cls.cwd.return_value = cli_project
|
|
result = runner.invoke(app, ["regenerate", "output.txt", "image.png"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "2 regenerated" in result.output
|
|
call_kwargs = mock_run.call_args[1]
|
|
assert call_kwargs["force_dirty"] == {"output.txt", "image.png"}
|
|
|
|
def test_regenerate_unknown_target(self, cli_project: Path) -> None:
|
|
with patch("hokusai.cli.Path") as mock_path_cls:
|
|
mock_path_cls.cwd.return_value = cli_project
|
|
result = runner.invoke(app, ["regenerate", "nonexistent.txt"])
|
|
|
|
assert result.exit_code == 1
|
|
assert "Unknown target(s): nonexistent.txt" in result.output
|
|
|
|
|
|
class TestCleanCommand:
|
|
"""Test the clean CLI command."""
|
|
|
|
def test_clean_removes_targets(self, cli_project: Path) -> None:
|
|
_ = (cli_project / "output.txt").write_text("generated")
|
|
_ = (cli_project / "image.png").write_bytes(b"\x89PNG")
|
|
state_file = ".project.hokusai-state.yaml"
|
|
_ = (cli_project / state_file).write_text("targets: {}")
|
|
|
|
with patch("hokusai.cli.Path") as mock_path_cls:
|
|
mock_path_cls.cwd.return_value = cli_project
|
|
result = runner.invoke(app, ["clean"])
|
|
|
|
assert result.exit_code == 0
|
|
assert not (cli_project / "output.txt").exists()
|
|
assert not (cli_project / "image.png").exists()
|
|
assert not (cli_project / state_file).exists()
|
|
assert "Cleaned 2 artifact(s)" in result.output
|
|
|
|
def test_clean_no_artifacts(self, cli_project: Path) -> None:
|
|
with patch("hokusai.cli.Path") as mock_path_cls:
|
|
mock_path_cls.cwd.return_value = cli_project
|
|
result = runner.invoke(app, ["clean"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Cleaned 0 artifact(s)" in result.output
|
|
|
|
def test_clean_partial_artifacts(self, cli_project: Path) -> None:
|
|
_ = (cli_project / "output.txt").write_text("generated")
|
|
|
|
with patch("hokusai.cli.Path") as mock_path_cls:
|
|
mock_path_cls.cwd.return_value = cli_project
|
|
result = runner.invoke(app, ["clean"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Cleaned 1 artifact(s)" in result.output
|
|
assert not (cli_project / "output.txt").exists()
|
|
|
|
def test_clean_archives_when_archive_folder_set(self, tmp_path: Path) -> None:
|
|
config = {
|
|
"archive_folder": "archive",
|
|
"targets": {
|
|
"output.txt": {"prompt": "Generate text"},
|
|
"image.png": {"prompt": "Generate image"},
|
|
},
|
|
}
|
|
_ = (tmp_path / "project.hokusai.yaml").write_text(
|
|
yaml.dump(config, default_flow_style=False)
|
|
)
|
|
_ = (tmp_path / "output.txt").write_text("generated text")
|
|
_ = (tmp_path / "image.png").write_bytes(b"\x89PNG")
|
|
|
|
with patch("hokusai.cli.Path") as mock_path_cls:
|
|
mock_path_cls.cwd.return_value = tmp_path
|
|
result = runner.invoke(app, ["clean"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Archived 2 artifact(s)" in result.output
|
|
assert "mv" in result.output
|
|
assert not (tmp_path / "output.txt").exists()
|
|
assert not (tmp_path / "image.png").exists()
|
|
assert (tmp_path / "archive" / "output.01.txt").read_text() == "generated text"
|
|
assert (tmp_path / "archive" / "image.01.png").read_bytes() == b"\x89PNG"
|
|
|
|
def test_clean_archive_preserves_subfolders(self, tmp_path: Path) -> None:
|
|
config = {
|
|
"archive_folder": "archive",
|
|
"targets": {"img/photo.png": {"prompt": "photo"}},
|
|
}
|
|
_ = (tmp_path / "project.hokusai.yaml").write_text(
|
|
yaml.dump(config, default_flow_style=False)
|
|
)
|
|
(tmp_path / "img").mkdir()
|
|
_ = (tmp_path / "img" / "photo.png").write_bytes(b"img")
|
|
|
|
with patch("hokusai.cli.Path") as mock_path_cls:
|
|
mock_path_cls.cwd.return_value = tmp_path
|
|
result = runner.invoke(app, ["clean"])
|
|
|
|
assert result.exit_code == 0
|
|
assert (tmp_path / "archive" / "img" / "photo.01.png").exists()
|
|
|
|
def test_clean_archive_still_deletes_state(self, tmp_path: Path) -> None:
|
|
config = {
|
|
"archive_folder": "archive",
|
|
"targets": {"output.txt": {"prompt": "text"}},
|
|
}
|
|
_ = (tmp_path / "project.hokusai.yaml").write_text(
|
|
yaml.dump(config, default_flow_style=False)
|
|
)
|
|
_ = (tmp_path / "output.txt").write_text("data")
|
|
state_file = ".project.hokusai-state.yaml"
|
|
_ = (tmp_path / state_file).write_text("targets: {}")
|
|
|
|
with patch("hokusai.cli.Path") as mock_path_cls:
|
|
mock_path_cls.cwd.return_value = tmp_path
|
|
result = runner.invoke(app, ["clean"])
|
|
|
|
assert result.exit_code == 0
|
|
assert not (tmp_path / state_file).exists()
|
|
|
|
|
|
class TestGraphCommand:
|
|
"""Test the graph CLI command."""
|
|
|
|
def test_graph_single_target(self, tmp_path: Path) -> None:
|
|
config = {"targets": {"out.txt": {"prompt": "hello"}}}
|
|
_ = (tmp_path / "test.hokusai.yaml").write_text(
|
|
yaml.dump(config, default_flow_style=False)
|
|
)
|
|
with patch("hokusai.cli.Path") as mock_path_cls:
|
|
mock_path_cls.cwd.return_value = tmp_path
|
|
result = runner.invoke(app, ["graph"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "out.txt" in result.output
|
|
|
|
def test_graph_with_dependencies(self, tmp_path: Path) -> None:
|
|
_ = (tmp_path / "input.txt").write_text("data")
|
|
config = {
|
|
"targets": {
|
|
"step1.md": {"prompt": "x", "inputs": ["input.txt"]},
|
|
"step2.txt": {"prompt": "y", "inputs": ["step1.md"]},
|
|
}
|
|
}
|
|
_ = (tmp_path / "test.hokusai.yaml").write_text(
|
|
yaml.dump(config, default_flow_style=False)
|
|
)
|
|
with patch("hokusai.cli.Path") as mock_path_cls:
|
|
mock_path_cls.cwd.return_value = tmp_path
|
|
result = runner.invoke(app, ["graph"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "input.txt" in result.output
|
|
assert "step1.md" in result.output
|
|
assert "step2.txt" in result.output
|
|
assert "<-" in result.output
|
|
|
|
def test_graph_shows_stages(self, tmp_path: Path) -> None:
|
|
_ = (tmp_path / "data.txt").write_text("data")
|
|
config = {
|
|
"targets": {
|
|
"a.txt": {"prompt": "x", "inputs": ["data.txt"]},
|
|
"b.txt": {"prompt": "y", "inputs": ["a.txt"]},
|
|
}
|
|
}
|
|
_ = (tmp_path / "test.hokusai.yaml").write_text(
|
|
yaml.dump(config, default_flow_style=False)
|
|
)
|
|
with patch("hokusai.cli.Path") as mock_path_cls:
|
|
mock_path_cls.cwd.return_value = tmp_path
|
|
result = runner.invoke(app, ["graph"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Stage" in result.output
|