generated from coulomb/repo-seed
Bernd Worsch
1f3dddf7d6
WP-0001 (Foundation & LEVEL1 Core):
- manifest model (FR-100), MD→DOCX builder (FR-200), DOCX→MD importer
(FR-300/400), template family registry (FR-600), drift detector (FR-700),
CLI wiring, pre-commit config, CI skeleton, regression harness
WP-0002 (Service Interfaces & Workflow Orchestration):
- REST service via FastAPI (FR-900): /health, /version, /capabilities,
/templates, /styles, /validate, /build, /import, /compare,
/templates/register, /workflows/{name}, /evidence/{run_id}
- Evidence & report store (FR-1400): JSON-backed, per-run, retrievable
through all interfaces, classification (pass/warnings/failed)
- Composite workflow orchestration (FR-1300): single-file-roundtrip,
multi-file-roundtrip, release-regression, family-switch-build
- MCP server via FastMCP (FR-1000): all tools + resources
- CLI additions: `markidocx serve`, `markidocx workflow`, `markidocx mcp`
- Interface parity tests: CLI / REST / MCP produce equivalent results
135 tests passing, ruff + mypy clean.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
229 lines
7.9 KiB
Python
229 lines
7.9 KiB
Python
"""Tests for T04 — Composite workflow orchestration (FR-1300)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import textwrap
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from markidocx.evidence import EvidenceStore
|
|
from markidocx.workflows import (
|
|
SUPPORTED_WORKFLOWS,
|
|
WorkflowError,
|
|
WorkflowResult,
|
|
run_workflow,
|
|
run_workflow_from_content,
|
|
)
|
|
|
|
SIMPLE_MANIFEST = textwrap.dedent("""\
|
|
project:
|
|
name: "Test Document"
|
|
feature_level: level1
|
|
family: article
|
|
|
|
sources:
|
|
- path: doc.md
|
|
|
|
output:
|
|
dir: ./dist
|
|
""")
|
|
|
|
SIMPLE_MARKDOWN = textwrap.dedent("""\
|
|
# Hello World
|
|
|
|
A paragraph with **bold** text.
|
|
|
|
## Section
|
|
|
|
- Item one
|
|
- Item two
|
|
""")
|
|
|
|
|
|
@pytest.fixture()
|
|
def tmp_project(tmp_path: Path) -> Path:
|
|
(tmp_path / "doc.md").write_text(SIMPLE_MARKDOWN, encoding="utf-8")
|
|
(tmp_path / "manifest.yaml").write_text(SIMPLE_MANIFEST, encoding="utf-8")
|
|
(tmp_path / "dist").mkdir()
|
|
return tmp_path
|
|
|
|
|
|
@pytest.fixture()
|
|
def store(tmp_path: Path) -> EvidenceStore:
|
|
return EvidenceStore(base_dir=tmp_path / "evidence")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Known / unknown workflow names
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_supported_workflows_set() -> None:
|
|
assert "single-file-roundtrip" in SUPPORTED_WORKFLOWS
|
|
assert "multi-file-roundtrip" in SUPPORTED_WORKFLOWS
|
|
assert "release-regression" in SUPPORTED_WORKFLOWS
|
|
assert "family-switch-build" in SUPPORTED_WORKFLOWS
|
|
|
|
|
|
def test_unknown_workflow_raises(tmp_project: Path) -> None:
|
|
with pytest.raises(WorkflowError, match="Unknown workflow"):
|
|
run_workflow("no-such-workflow", tmp_project / "manifest.yaml")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WorkflowResult shape (FR-1303, FR-1309)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_single_file_roundtrip_result_shape(tmp_project: Path, store: EvidenceStore) -> None:
|
|
result = run_workflow("single-file-roundtrip", tmp_project / "manifest.yaml", store)
|
|
assert isinstance(result, WorkflowResult)
|
|
assert result.run_id
|
|
assert result.workflow_name == "single-file-roundtrip"
|
|
assert result.timestamp
|
|
assert result.classification in ("full", "with-fallback", "partial", "failed")
|
|
|
|
|
|
def test_result_has_steps(tmp_project: Path, store: EvidenceStore) -> None:
|
|
result = run_workflow("single-file-roundtrip", tmp_project / "manifest.yaml", store)
|
|
assert len(result.steps) >= 1
|
|
for step in result.steps:
|
|
assert step.name
|
|
assert step.status in ("executed", "skipped", "failed")
|
|
|
|
|
|
def test_result_has_aggregate_output(tmp_project: Path, store: EvidenceStore) -> None:
|
|
result = run_workflow("single-file-roundtrip", tmp_project / "manifest.yaml", store)
|
|
assert isinstance(result.aggregate_output, dict)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Step visibility (FR-1304)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_single_file_roundtrip_step_names(tmp_project: Path, store: EvidenceStore) -> None:
|
|
result = run_workflow("single-file-roundtrip", tmp_project / "manifest.yaml", store)
|
|
step_names = [s.name for s in result.steps]
|
|
assert "validate" in step_names
|
|
assert "build" in step_names
|
|
assert "import" in step_names
|
|
assert "compare" in step_names
|
|
|
|
|
|
def test_failed_validate_stops_early(tmp_path: Path, store: EvidenceStore) -> None:
|
|
"""If validation fails, subsequent steps should not execute."""
|
|
bad_manifest = tmp_path / "manifest.yaml"
|
|
bad_manifest.write_text("project:\n name: x\n", encoding="utf-8")
|
|
(tmp_path / "dist").mkdir()
|
|
result = run_workflow("single-file-roundtrip", bad_manifest, store)
|
|
assert result.classification == "failed"
|
|
step_names = [s.name for s in result.steps]
|
|
assert "validate" in step_names
|
|
# build should not appear after a validation failure
|
|
assert "build" not in step_names
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Workflow classification (FR-1305)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_successful_roundtrip_classification(tmp_project: Path, store: EvidenceStore) -> None:
|
|
result = run_workflow("single-file-roundtrip", tmp_project / "manifest.yaml", store)
|
|
assert result.classification in ("full", "with-fallback")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Evidence stored (FR-1309)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_evidence_written_to_store(tmp_project: Path, store: EvidenceStore) -> None:
|
|
result = run_workflow("single-file-roundtrip", tmp_project / "manifest.yaml", store)
|
|
runs = store.list_runs()
|
|
assert result.run_id in runs
|
|
|
|
|
|
def test_validation_report_in_store(tmp_project: Path, store: EvidenceStore) -> None:
|
|
result = run_workflow("single-file-roundtrip", tmp_project / "manifest.yaml", store)
|
|
report = store.get_report(result.run_id, "validation")
|
|
assert report is not None
|
|
assert report.data["status"] == "ok"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# multi-file-roundtrip and family-switch-build
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_multi_file_roundtrip_returns_result(tmp_project: Path, store: EvidenceStore) -> None:
|
|
result = run_workflow("multi-file-roundtrip", tmp_project / "manifest.yaml", store)
|
|
assert result.workflow_name == "multi-file-roundtrip"
|
|
assert result.classification in ("full", "with-fallback", "partial", "failed")
|
|
|
|
|
|
def test_family_switch_build_produces_multiple_steps(tmp_project: Path, store: EvidenceStore) -> None:
|
|
result = run_workflow("family-switch-build", tmp_project / "manifest.yaml", store)
|
|
# Should have one build step per family
|
|
build_steps = [s for s in result.steps if s.name.startswith("build:")]
|
|
assert len(build_steps) == 3 # article, book, website
|
|
assert "builds" in result.aggregate_output
|
|
|
|
|
|
def test_release_regression_returns_result(tmp_project: Path, store: EvidenceStore) -> None:
|
|
result = run_workflow("release-regression", tmp_project / "manifest.yaml", store)
|
|
assert result.workflow_name == "release-regression"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run_workflow_from_content (REST/MCP path)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_run_from_content_single_file(store: EvidenceStore) -> None:
|
|
result = run_workflow_from_content(
|
|
"single-file-roundtrip",
|
|
SIMPLE_MANIFEST,
|
|
[{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
|
|
store,
|
|
)
|
|
assert result.workflow_name == "single-file-roundtrip"
|
|
assert result.run_id
|
|
|
|
|
|
def test_run_from_content_unknown_workflow() -> None:
|
|
with pytest.raises(WorkflowError):
|
|
run_workflow_from_content("bad-name", SIMPLE_MANIFEST, [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI: markidocx workflow (FR-1308 — CLI interface)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_cli_workflow_command(tmp_project: Path) -> None:
|
|
from typer.testing import CliRunner
|
|
|
|
from markidocx.cli import app
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(app, ["workflow", "single-file-roundtrip", str(tmp_project / "manifest.yaml"), "--json"])
|
|
assert result.exit_code in (0, 1)
|
|
import json
|
|
|
|
data = json.loads(result.output.strip())
|
|
assert "run_id" in data
|
|
assert "classification" in data
|
|
|
|
|
|
def test_cli_workflow_unknown_name(tmp_project: Path) -> None:
|
|
from typer.testing import CliRunner
|
|
|
|
from markidocx.cli import app
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(app, ["workflow", "no-such", str(tmp_project / "manifest.yaml")])
|
|
assert result.exit_code == 1
|