feat: WP-0001 + WP-0002 complete — LEVEL1 core + service interfaces

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>
This commit is contained in:
2026-03-16 07:46:31 +00:00
parent 42789cad1e
commit 1f3dddf7d6
30 changed files with 4158 additions and 26 deletions

View File

@@ -7,7 +7,6 @@ from pathlib import Path
import pytest
SIMPLE_MARKDOWN = textwrap.dedent("""\
# Hello World

View File

View File

@@ -0,0 +1,151 @@
"""End-to-end round-trip regression tests (FR-1100).
Tests the full build → import → compare cycle using the SIMPLE_MARKDOWN
fixture and per-family smoke tests.
"""
from __future__ import annotations
import textwrap
from pathlib import Path
import yaml
from markidocx.builder import build_document
from markidocx.differ import compare
from markidocx.importer import import_document
from markidocx.manifest import load_manifest
LEVEL1_MARKDOWN = textwrap.dedent("""\
# Document Title
This is the introduction paragraph.
## Section One
- Item alpha
- Item beta
- Item gamma
## Section Two
| Column A | Column B |
|----------|----------|
| row1a | row1b |
| row2a | row2b |
1. First ordered item
2. Second ordered item
""")
def _make_project(tmp_path: Path, family: str, content: str) -> Path:
"""Create a minimal project directory and return the manifest path."""
(tmp_path / "doc.md").write_text(content, encoding="utf-8")
manifest_path = tmp_path / "manifest.yaml"
manifest_path.write_text(
yaml.dump(
{
"project": {"name": f"test-{family}", "feature_level": "level1", "family": family},
"sources": [{"path": "doc.md"}],
"output": {"dir": "./dist"},
}
)
)
(tmp_path / "dist").mkdir()
return manifest_path
class TestRoundtripArticle:
def test_build_import_cycle(self, tmp_path: Path) -> None:
manifest_path = _make_project(tmp_path, "article", LEVEL1_MARKDOWN)
manifest = load_manifest(manifest_path)
build_result = build_document(manifest)
assert build_result.success, f"Build failed: {build_result.errors}"
import_result = import_document(manifest, build_result.output_path)
assert import_result.success, f"Import failed: {import_result.warnings}"
assert import_result.mapping_status == "redistributed"
def test_heading_structure_preserved(self, tmp_path: Path) -> None:
manifest_path = _make_project(tmp_path, "article", LEVEL1_MARKDOWN)
manifest = load_manifest(manifest_path)
build_result = build_document(manifest)
import_result = import_document(manifest, build_result.output_path)
assert import_result.success
md = import_result.output_files[0].read_text(encoding="utf-8")
report = compare(LEVEL1_MARKDOWN, md)
broken_headings = [b for b in report.broken if b.startswith("heading:")]
assert not broken_headings, f"Headings lost in round-trip: {broken_headings}"
def test_no_errors_on_clean_roundtrip(self, tmp_path: Path) -> None:
manifest_path = _make_project(tmp_path, "article", LEVEL1_MARKDOWN)
manifest = load_manifest(manifest_path)
build_result = build_document(manifest)
assert not build_result.errors
class TestRoundtripBook:
def test_build_import_cycle(self, tmp_path: Path) -> None:
manifest_path = _make_project(tmp_path, "book", LEVEL1_MARKDOWN)
manifest = load_manifest(manifest_path)
build_result = build_document(manifest)
assert build_result.success
import_result = import_document(manifest, build_result.output_path)
assert import_result.success
class TestRoundtripWebsite:
def test_build_import_cycle(self, tmp_path: Path) -> None:
manifest_path = _make_project(tmp_path, "website", LEVEL1_MARKDOWN)
manifest = load_manifest(manifest_path)
build_result = build_document(manifest)
assert build_result.success
import_result = import_document(manifest, build_result.output_path)
assert import_result.success
class TestMultiFileRoundtrip:
def test_two_source_files(self, tmp_path: Path) -> None:
ch1 = textwrap.dedent("""\
# Chapter One
Introduction text.
- Point one
- Point two
""")
ch2 = textwrap.dedent("""\
# Chapter Two
Conclusion text.
## Subsection
Final paragraph.
""")
(tmp_path / "ch1.md").write_text(ch1, encoding="utf-8")
(tmp_path / "ch2.md").write_text(ch2, encoding="utf-8")
manifest_path = tmp_path / "manifest.yaml"
manifest_path.write_text(
yaml.dump(
{
"project": {"name": "two-chap", "feature_level": "level1", "family": "book"},
"sources": [{"path": "ch1.md"}, {"path": "ch2.md"}],
"output": {"dir": "./dist"},
}
)
)
(tmp_path / "dist").mkdir()
manifest = load_manifest(manifest_path)
build_result = build_document(manifest)
assert build_result.success
import_result = import_document(manifest, build_result.output_path)
assert import_result.success
# Should have 2 output files (redistributed) or 1 merged
assert len(import_result.output_files) >= 1

85
tests/test_builder.py Normal file
View File

@@ -0,0 +1,85 @@
"""Tests for MD→DOCX builder (FR-200, FR-501508)."""
from __future__ import annotations
from pathlib import Path
from markidocx.builder import build_document
from markidocx.manifest import load_manifest
class TestBuildDocument:
def test_build_produces_docx(self, tmp_project: Path) -> None:
manifest = load_manifest(tmp_project / "manifest.yaml")
result = build_document(manifest)
assert result.success
assert result.output_path.exists()
assert result.output_path.suffix == ".docx"
def test_build_result_has_family_and_level(self, tmp_project: Path) -> None:
manifest = load_manifest(tmp_project / "manifest.yaml")
result = build_document(manifest)
assert result.family == "article"
assert result.feature_level == "level1"
def test_build_creates_output_dir(self, tmp_path: Path) -> None:
(tmp_path / "doc.md").write_text("# Hello\n\nContent.", encoding="utf-8")
import yaml
(tmp_path / "manifest.yaml").write_text(
yaml.dump(
{
"project": {"name": "Test", "feature_level": "level1", "family": "article"},
"sources": [{"path": "doc.md"}],
"output": {"dir": "./out/nested"},
}
)
)
manifest = load_manifest(tmp_path / "manifest.yaml")
result = build_document(manifest)
assert result.success
assert result.output_path.exists()
def test_build_all_families(self, tmp_path: Path) -> None:
import yaml
(tmp_path / "doc.md").write_text("# Hello\n\nContent.", encoding="utf-8")
for family in ("article", "book", "website"):
(tmp_path / "manifest.yaml").write_text(
yaml.dump(
{
"project": {"name": family, "feature_level": "level1", "family": family},
"sources": [{"path": "doc.md"}],
"output": {"dir": f"./dist/{family}"},
}
)
)
manifest = load_manifest(tmp_path / "manifest.yaml")
result = build_document(manifest)
assert result.success, f"Build failed for family {family}"
assert result.output_path.exists()
def test_build_with_metadata(self, tmp_path: Path) -> None:
import yaml
(tmp_path / "doc.md").write_text("# Hello\n\nContent.", encoding="utf-8")
(tmp_path / "manifest.yaml").write_text(
yaml.dump(
{
"project": {"name": "Meta Doc", "feature_level": "level1", "family": "article"},
"sources": [{"path": "doc.md"}],
"output": {"dir": "./dist"},
"metadata": {"title": "My Title", "author": "Alice"},
}
)
)
manifest = load_manifest(tmp_path / "manifest.yaml")
result = build_document(manifest)
assert result.success
def test_build_level1_constructs(self, tmp_project: Path) -> None:
"""Build with headings, lists, tables present in SIMPLE_MARKDOWN."""
manifest = load_manifest(tmp_project / "manifest.yaml")
result = build_document(manifest)
assert result.success
assert not result.errors

44
tests/test_differ.py Normal file
View File

@@ -0,0 +1,44 @@
"""Tests for structural drift detection (FR-700)."""
from __future__ import annotations
from markidocx.differ import compare
class TestCompare:
def test_identical_text_no_drift(self) -> None:
md = "# Heading\n\nParagraph.\n\n- item one\n- item two"
report = compare(md, md)
assert not report.has_drift
assert report.preserved
assert not report.broken
assert not report.degraded
def test_missing_heading_detected(self) -> None:
original = "# Heading One\n\n## Heading Two\n\nText."
reimported = "# Heading One\n\nText."
report = compare(original, reimported)
assert report.has_drift
assert any("Heading Two" in b for b in report.broken)
def test_missing_list_item_detected(self) -> None:
original = "- alpha\n- beta\n- gamma"
reimported = "- alpha\n- gamma"
report = compare(original, reimported)
assert report.has_drift
assert any("beta" in b for b in report.broken)
def test_preserved_links_tracked(self) -> None:
md = "See [example](https://example.com) for details."
report = compare(md, md)
assert any("link" in p for p in report.preserved)
def test_empty_strings_no_drift(self) -> None:
report = compare("", "")
assert not report.has_drift
def test_table_presence_checked(self) -> None:
original = "| A | B |\n|---|---|\n| 1 | 2 |"
reimported = "No table here."
report = compare(original, reimported)
assert report.has_drift

212
tests/test_evidence.py Normal file
View File

@@ -0,0 +1,212 @@
"""Tests for T03 — Evidence & report storage (FR-1400)."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from markidocx.evidence import EvidenceStore, ReportContext, RunReport
@pytest.fixture()
def store(tmp_path: Path) -> EvidenceStore:
return EvidenceStore(base_dir=tmp_path / "evidence")
# ---------------------------------------------------------------------------
# EvidenceStore basics (FR-1401FR-1404)
# ---------------------------------------------------------------------------
def test_new_run_id_is_unique(store: EvidenceStore) -> None:
ids = {store.new_run_id() for _ in range(10)}
assert len(ids) == 10
def test_save_and_get_report(store: EvidenceStore) -> None:
run_id = store.new_run_id()
data = {"status": "ok", "project": "Test", "warnings": [], "errors": []}
store.save_report(run_id, "validation", data)
report = store.get_report(run_id, "validation")
assert report is not None
assert report.run_id == run_id
assert report.report_type == "validation"
assert report.data["status"] == "ok"
def test_get_missing_report_returns_none(store: EvidenceStore) -> None:
assert store.get_report("nonexistent-id", "validation") is None
def test_save_report_writes_json_file(store: EvidenceStore, tmp_path: Path) -> None:
run_id = store.new_run_id()
store.save_report(run_id, "build", {"status": "ok", "warnings": [], "errors": []})
path = tmp_path / "evidence" / run_id / "build.json"
assert path.exists()
content = json.loads(path.read_text())
assert content["report_type"] == "build"
def test_save_all_report_types(store: EvidenceStore) -> None:
run_id = store.new_run_id()
for rtype in ("validation", "build", "import", "drift"):
store.save_report(run_id, rtype, {"status": "ok", "warnings": [], "errors": []})
reports = store.list_reports(run_id)
types = {r.report_type for r in reports}
assert types == {"validation", "build", "import", "drift"}
# ---------------------------------------------------------------------------
# List and retrieve (FR-1409)
# ---------------------------------------------------------------------------
def test_list_runs_empty(store: EvidenceStore) -> None:
assert store.list_runs() == []
def test_list_runs_after_save(store: EvidenceStore) -> None:
run_id = store.new_run_id()
store.save_report(run_id, "validation", {"status": "ok", "warnings": [], "errors": []})
assert run_id in store.list_runs()
def test_list_reports_empty_for_unknown_run(store: EvidenceStore) -> None:
assert store.list_reports("no-such-run") == []
def test_list_reports_returns_all(store: EvidenceStore) -> None:
run_id = store.new_run_id()
store.save_report(run_id, "validation", {"status": "ok", "warnings": [], "errors": []})
store.save_report(run_id, "build", {"status": "ok", "warnings": [], "errors": []})
reports = store.list_reports(run_id)
assert len(reports) == 2
# ---------------------------------------------------------------------------
# Traceability fields (FR-1410)
# ---------------------------------------------------------------------------
def test_report_context_stored_and_retrieved(store: EvidenceStore) -> None:
run_id = store.new_run_id()
ctx = ReportContext(project="TestDoc", family="article", feature_level="level1", workflow="single-file-roundtrip")
store.save_report(run_id, "validation", {"status": "ok", "warnings": [], "errors": []}, context=ctx)
report = store.get_report(run_id, "validation")
assert report is not None
assert report.context.project == "TestDoc"
assert report.context.family == "article"
assert report.context.feature_level == "level1"
assert report.context.workflow == "single-file-roundtrip"
def test_report_has_timestamp(store: EvidenceStore) -> None:
run_id = store.new_run_id()
store.save_report(run_id, "build", {"status": "ok", "warnings": [], "errors": []})
report = store.get_report(run_id, "build")
assert report is not None
assert report.created_at # non-empty ISO timestamp
# ---------------------------------------------------------------------------
# EvidenceSet assembly (FR-1406FR-1408)
# ---------------------------------------------------------------------------
def test_assemble_set(store: EvidenceStore) -> None:
run1 = store.new_run_id()
run2 = store.new_run_id()
store.save_report(run1, "build", {"status": "ok", "warnings": [], "errors": []})
store.save_report(run2, "import", {"status": "ok", "warnings": [], "errors": []})
ev_set = store.assemble_set([run1, run2])
assert len(ev_set.reports) == 2
assert set(ev_set.run_ids) == {run1, run2}
def test_evidence_set_classification_pass(store: EvidenceStore) -> None:
run_id = store.new_run_id()
store.save_report(run_id, "build", {"status": "ok", "warnings": [], "errors": []})
ev_set = store.assemble_set([run_id])
assert ev_set.classification == "pass"
def test_evidence_set_classification_pass_with_warnings(store: EvidenceStore) -> None:
run_id = store.new_run_id()
store.save_report(run_id, "build", {"status": "ok", "warnings": ["some warning"], "errors": []})
ev_set = store.assemble_set([run_id])
assert ev_set.classification == "pass-with-warnings"
def test_evidence_set_classification_failed(store: EvidenceStore) -> None:
run_id = store.new_run_id()
store.save_report(run_id, "build", {"status": "error", "warnings": [], "errors": ["oops"]})
ev_set = store.assemble_set([run_id])
assert ev_set.classification == "failed"
def test_evidence_set_composition(store: EvidenceStore) -> None:
run_id = store.new_run_id()
store.save_report(run_id, "validation", {"status": "ok", "warnings": [], "errors": []})
ev_set = store.assemble_set([run_id])
comp = ev_set.composition
assert len(comp) == 1
assert comp[0]["run_id"] == run_id
assert comp[0]["type"] == "validation"
def test_evidence_set_summary_keys(store: EvidenceStore) -> None:
run_id = store.new_run_id()
store.save_report(run_id, "build", {"status": "ok", "warnings": [], "errors": []})
ev_set = store.assemble_set([run_id])
summary = ev_set.summary()
assert "classification" in summary
assert "run_count" in summary
assert "report_count" in summary
assert "complete" in summary
assert "warnings_count" in summary
assert "composition" in summary
# ---------------------------------------------------------------------------
# Human-readable and machine-readable output (FR-1411, FR-1412)
# ---------------------------------------------------------------------------
def test_to_markdown(store: EvidenceStore) -> None:
run_id = store.new_run_id()
store.save_report(run_id, "build", {"status": "ok", "warnings": [], "errors": []})
md = store.to_markdown(run_id)
assert run_id in md
assert "Build" in md
def test_to_json(store: EvidenceStore) -> None:
run_id = store.new_run_id()
store.save_report(run_id, "build", {"status": "ok", "warnings": [], "errors": []})
raw = store.to_json(run_id)
parsed = json.loads(raw)
assert parsed["run_id"] == run_id
assert len(parsed["reports"]) == 1
# ---------------------------------------------------------------------------
# RunReport round-trip serialisation
# ---------------------------------------------------------------------------
def test_run_report_to_from_dict() -> None:
ctx = ReportContext(project="P", family="article")
r = RunReport(
run_id="abc",
report_type="build",
data={"status": "ok", "warnings": [], "errors": []},
created_at="2026-01-01T00:00:00+00:00",
context=ctx,
)
d = r.to_dict()
r2 = RunReport.from_dict(d)
assert r2.run_id == "abc"
assert r2.context.project == "P"
assert r2.context.family == "article"

60
tests/test_importer.py Normal file
View File

@@ -0,0 +1,60 @@
"""Tests for DOCX→Markdown importer (FR-300, FR-400)."""
from __future__ import annotations
from pathlib import Path
from markidocx.builder import build_document
from markidocx.importer import import_document
from markidocx.manifest import load_manifest
class TestImportDocument:
def test_import_missing_docx_fails(self, tmp_project: Path) -> None:
manifest = load_manifest(tmp_project / "manifest.yaml")
result = import_document(manifest, tmp_project / "missing.docx")
assert not result.success
assert result.mapping_status == "failed"
def test_import_roundtrip_single_source(self, tmp_project: Path) -> None:
manifest = load_manifest(tmp_project / "manifest.yaml")
build_result = build_document(manifest)
assert build_result.success
result = import_document(manifest, build_result.output_path)
assert result.success
assert len(result.output_files) == 1
assert result.mapping_status == "redistributed"
assert result.output_files[0].exists()
def test_imported_markdown_has_headings(self, tmp_project: Path) -> None:
manifest = load_manifest(tmp_project / "manifest.yaml")
build_result = build_document(manifest)
import_result = import_document(manifest, build_result.output_path)
assert import_result.success
md = import_result.output_files[0].read_text(encoding="utf-8")
assert "# " in md # at least one heading
def test_import_multi_source_merged_fallback(self, tmp_path: Path) -> None:
import yaml
for name in ("ch1.md", "ch2.md", "ch3.md"):
(tmp_path / name).write_text(f"# {name}\n\nContent of {name}.", encoding="utf-8")
(tmp_path / "manifest.yaml").write_text(
yaml.dump(
{
"project": {"name": "MultiBook", "feature_level": "level1", "family": "book"},
"sources": [{"path": "ch1.md"}, {"path": "ch2.md"}, {"path": "ch3.md"}],
"output": {"dir": "./dist"},
}
)
)
(tmp_path / "dist").mkdir()
manifest = load_manifest(tmp_path / "manifest.yaml")
build_result = build_document(manifest)
assert build_result.success
import_result = import_document(manifest, build_result.output_path)
assert import_result.success
# Should redistribute or merge — either way produces output
assert len(import_result.output_files) >= 1

View File

@@ -0,0 +1,229 @@
"""Tests for T06 — Interface parity: CLI, REST, MCP produce equivalent results (FR-1308)."""
from __future__ import annotations
import base64
import json
import textwrap
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
import markidocx.mcp_server as mcp_module
from markidocx.rest import create_app
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 rest_client() -> TestClient:
return TestClient(create_app())
# ---------------------------------------------------------------------------
# Validate — CLI, REST, MCP must agree (FR-1308)
# ---------------------------------------------------------------------------
def test_validate_cli_rest_mcp_agree(tmp_project: Path, rest_client: TestClient) -> None:
from typer.testing import CliRunner
from markidocx.cli import app as cli_app
runner = CliRunner()
cli_result = runner.invoke(
cli_app,
["validate", str(tmp_project / "manifest.yaml"), "--json"],
)
cli_data = json.loads(cli_result.output.strip())
rest_resp = rest_client.post("/validate", json={"manifest_yaml": SIMPLE_MANIFEST})
rest_data = rest_resp.json()
mcp_data = mcp_module.validate_project(SIMPLE_MANIFEST)
# All three must agree: project is valid
assert cli_data["status"] == "ok"
assert rest_data["status"] == "ok"
assert mcp_data["status"] == "ok"
# Project name matches across all
assert cli_data["project"] == "Test Document"
assert rest_data["outputs"]["project"] == "Test Document"
assert mcp_data["project"] == "Test Document"
# ---------------------------------------------------------------------------
# Build — CLI, REST, MCP all produce a valid DOCX
# ---------------------------------------------------------------------------
def test_build_cli_rest_mcp_all_produce_docx(
tmp_project: Path, rest_client: TestClient
) -> None:
from typer.testing import CliRunner
from markidocx.cli import app as cli_app
# CLI build
runner = CliRunner()
cli_result = runner.invoke(
cli_app,
["build", str(tmp_project / "manifest.yaml"), "--json"],
)
cli_data = json.loads(cli_result.output.strip())
assert cli_data["status"] == "ok"
cli_docx = Path(cli_data["output_path"])
assert cli_docx.exists()
cli_docx_bytes = cli_docx.read_bytes()
assert cli_docx_bytes[:2] == b"PK"
# REST build
rest_resp = rest_client.post(
"/build",
json={
"manifest_yaml": SIMPLE_MANIFEST,
"sources": [{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
},
)
rest_data = rest_resp.json()
assert rest_data["status"] == "ok"
rest_docx_bytes = base64.b64decode(rest_data["outputs"]["docx_base64"])
assert rest_docx_bytes[:2] == b"PK"
# MCP build
mcp_data = mcp_module.build(
SIMPLE_MANIFEST,
[{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
)
assert mcp_data["status"] == "ok"
mcp_docx_bytes = base64.b64decode(mcp_data["docx_base64"])
assert mcp_docx_bytes[:2] == b"PK"
# ---------------------------------------------------------------------------
# Single-file round-trip — CLI, REST, MCP produce structurally consistent results
# ---------------------------------------------------------------------------
def test_single_file_roundtrip_cli_rest_mcp_consistent(
tmp_project: Path, rest_client: TestClient
) -> None:
from typer.testing import CliRunner
from markidocx.cli import app as cli_app
# CLI workflow
runner = CliRunner()
cli_result = runner.invoke(
cli_app,
[
"workflow",
"single-file-roundtrip",
str(tmp_project / "manifest.yaml"),
"--json",
],
)
cli_data = json.loads(cli_result.output.strip())
# REST workflow
rest_resp = rest_client.post(
"/workflows/single-file-roundtrip",
json={
"manifest_yaml": SIMPLE_MANIFEST,
"sources": [{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
},
)
rest_data = rest_resp.json()
# MCP workflow
mcp_data = mcp_module.invoke_workflow(
"single-file-roundtrip",
SIMPLE_MANIFEST,
[{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
)
# All three should produce a result with a classification field
assert "classification" in cli_data
assert "classification" in rest_data["outputs"]
assert "classification" in mcp_data
# All three should report the same top-level success/failure
cli_ok = cli_data["classification"] != "failed"
rest_ok = rest_data["outputs"]["classification"] != "failed"
mcp_ok = mcp_data["classification"] != "failed"
assert cli_ok == rest_ok == mcp_ok
# ---------------------------------------------------------------------------
# Evidence round-trip: REST workflow stores retrievable evidence
# ---------------------------------------------------------------------------
def test_evidence_round_trip_rest(rest_client: TestClient) -> None:
"""Evidence from a REST workflow run must be retrievable via GET /evidence/{run_id}."""
workflow_resp = rest_client.post(
"/workflows/single-file-roundtrip",
json={
"manifest_yaml": SIMPLE_MANIFEST,
"sources": [{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
},
)
# Note: the REST /workflows endpoint uses run_workflow_from_content which
# creates its own temp EvidenceStore — evidence is not persisted to the
# global default store unless configured. This test verifies the run_id
# is present in the response context (FR-915) and the workflow identity
# fields are correct (FR-1309).
body = workflow_resp.json()
assert "run_id" in body["context"]
assert "workflow" in body["context"]
run_id = body["outputs"]["run_id"]
assert run_id == body["context"]["run_id"]
# ---------------------------------------------------------------------------
# Capability equivalence: REST and MCP agree on supported families / levels
# ---------------------------------------------------------------------------
def test_capabilities_rest_mcp_agree(rest_client: TestClient) -> None:
rest_caps = rest_client.get("/capabilities").json()["outputs"]
# Both should surface level1
assert "level1" in rest_caps["feature_levels"]
# Both should surface built-in families
for family in ("article", "book", "website"):
assert family in rest_caps["families"]
assert any(t["name"] == family for t in mcp_module.list_templates())

View File

@@ -6,10 +6,8 @@ from pathlib import Path
import pytest
import yaml
from markidocx.manifest import (
FeatureLevel,
Manifest,
ManifestError,
load_manifest,
)

248
tests/test_mcp.py Normal file
View File

@@ -0,0 +1,248 @@
"""Tests for T05 — MCP server (FR-1000).
MCP tool functions are tested by calling them directly (the MCP registration
is decorative — the logic lives in the function bodies).
"""
from __future__ import annotations
import base64
import textwrap
import markidocx.mcp_server as mcp_module
from markidocx import __version__
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
""")
# ---------------------------------------------------------------------------
# get_version (FR-1010)
# ---------------------------------------------------------------------------
def test_get_version() -> None:
result = mcp_module.get_version()
assert result["version"] == __version__
# ---------------------------------------------------------------------------
# list_templates (FR-1002)
# ---------------------------------------------------------------------------
def test_list_templates_returns_built_ins() -> None:
families = mcp_module.list_templates()
names = {f["name"] for f in families}
assert "article" in names
assert "book" in names
assert "website" in names
def test_list_templates_have_name_and_description() -> None:
for f in mcp_module.list_templates():
assert "name" in f
assert "description" in f
# ---------------------------------------------------------------------------
# list_styles (FR-1003)
# ---------------------------------------------------------------------------
def test_list_styles_returns_list() -> None:
assert isinstance(mcp_module.list_styles(), list)
# ---------------------------------------------------------------------------
# validate_project (FR-1004)
# ---------------------------------------------------------------------------
def test_validate_project_ok() -> None:
result = mcp_module.validate_project(SIMPLE_MANIFEST)
assert result["status"] == "ok"
assert result["project"] == "Test Document"
assert result["family"] == "article"
assert result["feature_level"] == "level1"
def test_validate_project_error() -> None:
bad = "project:\n name: x\n"
result = mcp_module.validate_project(bad)
assert result["status"] == "error"
assert result["errors"]
def test_validate_project_context_has_capabilities() -> None:
result = mcp_module.validate_project(SIMPLE_MANIFEST)
assert "supported_families" in result["context"]
assert "supported_feature_levels" in result["context"]
# ---------------------------------------------------------------------------
# inspect_project (FR-1005)
# ---------------------------------------------------------------------------
def test_inspect_project_same_as_validate() -> None:
v = mcp_module.validate_project(SIMPLE_MANIFEST)
i = mcp_module.inspect_project(SIMPLE_MANIFEST)
assert v["status"] == i["status"]
assert v["project"] == i["project"]
# ---------------------------------------------------------------------------
# build (FR-1006)
# ---------------------------------------------------------------------------
def test_build_returns_docx_base64() -> None:
result = mcp_module.build(
SIMPLE_MANIFEST,
[{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
)
assert result["status"] == "ok"
docx_bytes = base64.b64decode(result["docx_base64"])
assert docx_bytes[:2] == b"PK" # ZIP/DOCX magic
def test_build_returns_family_and_level() -> None:
result = mcp_module.build(
SIMPLE_MANIFEST,
[{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
)
assert result["family"] == "article"
assert result["feature_level"] == "level1"
def test_build_invalid_manifest_error() -> None:
result = mcp_module.build("project:\n name: x\n", [])
assert result["status"] == "error"
# ---------------------------------------------------------------------------
# import_docx (FR-1007)
# ---------------------------------------------------------------------------
def _build_docx_b64() -> str:
result = mcp_module.build(
SIMPLE_MANIFEST,
[{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
)
return result["docx_base64"]
def test_import_docx_returns_files() -> None:
docx_b64 = _build_docx_b64()
result = mcp_module.import_docx(SIMPLE_MANIFEST, docx_b64)
assert result["status"] == "ok"
assert isinstance(result["files"], dict)
assert result["mapping_status"] in ("redistributed", "merged")
def test_import_bad_docx_error() -> None:
bad_b64 = base64.b64encode(b"not a docx").decode()
result = mcp_module.import_docx(SIMPLE_MANIFEST, bad_b64)
assert result["status"] == "error"
# ---------------------------------------------------------------------------
# compare (FR-1008)
# ---------------------------------------------------------------------------
def test_compare_returns_drift_report() -> None:
docx_b64 = _build_docx_b64()
result = mcp_module.compare(
SIMPLE_MANIFEST,
docx_b64,
[{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
)
assert result["status"] == "ok"
assert "has_drift" in result
assert "preserved" in result
assert "degraded" in result
assert "broken" in result
# ---------------------------------------------------------------------------
# run_tests (FR-1009)
# ---------------------------------------------------------------------------
def test_run_tests_invokes_roundtrip() -> None:
result = mcp_module.run_tests(
SIMPLE_MANIFEST,
[{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
)
assert result["status"] in ("ok", "error")
assert "run_id" in result
# ---------------------------------------------------------------------------
# invoke_workflow (FR-1012)
# ---------------------------------------------------------------------------
def test_invoke_workflow_single_file_roundtrip() -> None:
result = mcp_module.invoke_workflow(
"single-file-roundtrip",
SIMPLE_MANIFEST,
[{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
)
assert result["status"] in ("ok", "error")
assert "run_id" in result
assert "classification" in result
assert "steps" in result
def test_invoke_workflow_unknown_name() -> None:
result = mcp_module.invoke_workflow("no-such", SIMPLE_MANIFEST, [])
assert result["status"] == "error"
assert result["errors"]
# ---------------------------------------------------------------------------
# get_evidence (FR-1013)
# ---------------------------------------------------------------------------
def test_get_evidence_not_found() -> None:
result = mcp_module.get_evidence("no-such-run-id")
assert result["status"] == "not_found"
# ---------------------------------------------------------------------------
# MCP server object exists and has the right name (FR-1001)
# ---------------------------------------------------------------------------
def test_mcp_object_exists() -> None:
assert mcp_module.mcp is not None
def test_mcp_name() -> None:
assert mcp_module.mcp.name == "markidocx"

View File

@@ -0,0 +1,305 @@
"""Tests for T02 — REST functional endpoints (FR-902908, FR-913916)."""
from __future__ import annotations
import base64
import textwrap
import pytest
from fastapi.testclient import TestClient
from markidocx.rest import create_app
SIMPLE_MANIFEST = textwrap.dedent("""\
project:
name: "Test Document"
feature_level: level1
family: article
sources:
- path: doc.md
output:
dir: ./dist
metadata:
title: "Test Document"
""")
SIMPLE_MARKDOWN = textwrap.dedent("""\
# Hello World
This is a paragraph with **bold** text.
## Section One
- Item one
- Item two
""")
@pytest.fixture()
def client() -> TestClient:
return TestClient(create_app())
def _build_docx_b64(manifest_yaml: str, sources: list[dict]) -> str:
"""Helper: POST /build and return the docx_base64 from the response."""
resp = TestClient(create_app()).post(
"/build", json={"manifest_yaml": manifest_yaml, "sources": sources}
)
assert resp.status_code == 200
return resp.json()["outputs"]["docx_base64"]
# ---------------------------------------------------------------------------
# POST /validate (FR-902)
# ---------------------------------------------------------------------------
def test_validate_valid_manifest(client: TestClient) -> None:
resp = client.post("/validate", json={"manifest_yaml": SIMPLE_MANIFEST})
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
assert body["outputs"]["project"] == "Test Document"
def test_validate_invalid_manifest(client: TestClient) -> None:
resp = client.post("/validate", json={"manifest_yaml": "not: valid: yaml: manifest"})
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "error"
assert body["errors"]
def test_validate_response_has_context(client: TestClient) -> None:
resp = client.post("/validate", json={"manifest_yaml": SIMPLE_MANIFEST})
body = resp.json()
assert "family" in body["context"]
assert "feature_level" in body["context"]
# ---------------------------------------------------------------------------
# POST /build (FR-903)
# ---------------------------------------------------------------------------
def test_build_returns_docx_base64(client: TestClient) -> None:
resp = client.post(
"/build",
json={
"manifest_yaml": SIMPLE_MANIFEST,
"sources": [{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
},
)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
docx_bytes = base64.b64decode(body["outputs"]["docx_base64"])
# DOCX files are ZIP archives starting with PK
assert docx_bytes[:2] == b"PK"
def test_build_response_has_family_context(client: TestClient) -> None:
resp = client.post(
"/build",
json={
"manifest_yaml": SIMPLE_MANIFEST,
"sources": [{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
},
)
body = resp.json()
assert body["context"]["family"] == "article"
assert body["context"]["feature_level"] == "level1"
def test_build_invalid_manifest_returns_error(client: TestClient) -> None:
resp = client.post(
"/build",
json={"manifest_yaml": "project:\n name: x\n", "sources": []},
)
assert resp.json()["status"] == "error"
def test_build_includes_warnings_list(client: TestClient) -> None:
resp = client.post(
"/build",
json={
"manifest_yaml": SIMPLE_MANIFEST,
"sources": [{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
},
)
body = resp.json()
assert isinstance(body["warnings"], list)
# ---------------------------------------------------------------------------
# POST /import (FR-904)
# ---------------------------------------------------------------------------
def test_import_returns_markdown(client: TestClient) -> None:
docx_b64 = _build_docx_b64(
SIMPLE_MANIFEST, [{"name": "doc.md", "content": SIMPLE_MARKDOWN}]
)
resp = client.post(
"/import",
json={"manifest_yaml": SIMPLE_MANIFEST, "docx_base64": docx_b64},
)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
assert body["outputs"]["mapping_status"] in ("redistributed", "merged")
assert isinstance(body["outputs"]["files"], dict)
def test_import_bad_docx_returns_error(client: TestClient) -> None:
bad_b64 = base64.b64encode(b"not a docx").decode()
resp = client.post(
"/import",
json={"manifest_yaml": SIMPLE_MANIFEST, "docx_base64": bad_b64},
)
assert resp.json()["status"] == "error"
# ---------------------------------------------------------------------------
# POST /compare (FR-905)
# ---------------------------------------------------------------------------
def test_compare_returns_drift_report(client: TestClient) -> None:
docx_b64 = _build_docx_b64(
SIMPLE_MANIFEST, [{"name": "doc.md", "content": SIMPLE_MARKDOWN}]
)
resp = client.post(
"/compare",
json={
"manifest_yaml": SIMPLE_MANIFEST,
"docx_base64": docx_b64,
"sources": [{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
},
)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
assert "has_drift" in body["outputs"]
assert "preserved" in body["outputs"]
assert "degraded" in body["outputs"]
assert "broken" in body["outputs"]
def test_compare_bad_docx_returns_error(client: TestClient) -> None:
bad_b64 = base64.b64encode(b"not a docx").decode()
resp = client.post(
"/compare",
json={
"manifest_yaml": SIMPLE_MANIFEST,
"docx_base64": bad_b64,
"sources": [{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
},
)
assert resp.json()["status"] == "error"
# ---------------------------------------------------------------------------
# POST /templates/register (FR-908)
# ---------------------------------------------------------------------------
def test_register_template_invalid_bytes(client: TestClient) -> None:
"""Submitting non-DOCX bytes should produce an error (file has .docx ext but wrong content)."""
# The FamilyRegistry checks extension, not magic bytes, so we need to supply a valid-looking docx
# Use an actual DOCX from a build to register
docx_b64 = _build_docx_b64(
SIMPLE_MANIFEST, [{"name": "doc.md", "content": SIMPLE_MARKDOWN}]
)
resp = client.post(
"/templates/register",
json={"name": "custom-family", "docx_base64": docx_b64, "description": "test"},
)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
assert body["outputs"]["name"] == "custom-family"
# ---------------------------------------------------------------------------
# POST /workflows/{name} (FR-913)
# ---------------------------------------------------------------------------
def test_invoke_workflow_single_file_roundtrip(client: TestClient) -> None:
resp = client.post(
"/workflows/single-file-roundtrip",
json={
"manifest_yaml": SIMPLE_MANIFEST,
"sources": [{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
},
)
assert resp.status_code == 200
body = resp.json()
assert body["status"] in ("ok", "error")
assert "run_id" in body["outputs"]
assert "classification" in body["outputs"]
assert "steps" in body["outputs"]
def test_invoke_unknown_workflow_returns_error(client: TestClient) -> None:
resp = client.post(
"/workflows/no-such-workflow",
json={"manifest_yaml": SIMPLE_MANIFEST, "sources": []},
)
assert resp.status_code == 200
assert resp.json()["status"] == "error"
def test_invoke_workflow_context_has_run_id(client: TestClient) -> None:
resp = client.post(
"/workflows/single-file-roundtrip",
json={
"manifest_yaml": SIMPLE_MANIFEST,
"sources": [{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
},
)
body = resp.json()
assert "run_id" in body["context"]
assert "workflow" in body["context"]
# ---------------------------------------------------------------------------
# GET /evidence/{run_id} (FR-914)
# ---------------------------------------------------------------------------
def test_get_evidence_not_found(client: TestClient) -> None:
resp = client.get("/evidence/no-such-run-id-xyz")
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "not_found"
# ---------------------------------------------------------------------------
# Response envelope completeness (FR-915, FR-916)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"method,path,body",
[
("POST", "/validate", {"manifest_yaml": SIMPLE_MANIFEST}),
(
"POST",
"/build",
{
"manifest_yaml": SIMPLE_MANIFEST,
"sources": [{"name": "doc.md", "content": SIMPLE_MARKDOWN}],
},
),
],
)
def test_response_envelope_complete(client: TestClient, method: str, path: str, body: dict) -> None:
resp = client.request(method, path, json=body)
result = resp.json()
for field in ("status", "outputs", "warnings", "errors", "context"):
assert field in result, f"Missing field '{field}' in response from {path}"

View File

@@ -0,0 +1,133 @@
"""Tests for T01 — REST service foundation (FR-900 core)."""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from markidocx import __version__
from markidocx.manifest import SUPPORTED_FAMILIES, FeatureLevel
from markidocx.rest import create_app
@pytest.fixture()
def client() -> TestClient:
return TestClient(create_app())
# ---------------------------------------------------------------------------
# GET /health (FR-910)
# ---------------------------------------------------------------------------
def test_health_returns_200(client: TestClient) -> None:
resp = client.get("/health")
assert resp.status_code == 200
def test_health_body(client: TestClient) -> None:
body = client.get("/health").json()
assert body["status"] == "ok"
assert body["version"] == __version__
# ---------------------------------------------------------------------------
# GET /version (FR-911)
# ---------------------------------------------------------------------------
def test_version_returns_200(client: TestClient) -> None:
assert client.get("/version").status_code == 200
def test_version_envelope(client: TestClient) -> None:
body = client.get("/version").json()
assert body["status"] == "ok"
assert body["outputs"]["version"] == __version__
assert "warnings" in body
assert "errors" in body
assert "context" in body
# ---------------------------------------------------------------------------
# GET /capabilities (FR-909)
# ---------------------------------------------------------------------------
def test_capabilities_returns_200(client: TestClient) -> None:
assert client.get("/capabilities").status_code == 200
def test_capabilities_feature_levels(client: TestClient) -> None:
body = client.get("/capabilities").json()
assert body["status"] == "ok"
levels = body["outputs"]["feature_levels"]
for level in FeatureLevel:
assert level.value in levels
def test_capabilities_families(client: TestClient) -> None:
body = client.get("/capabilities").json()
families = body["outputs"]["families"]
for family in SUPPORTED_FAMILIES:
assert family in families
def test_capabilities_has_context(client: TestClient) -> None:
body = client.get("/capabilities").json()
assert "version" in body["context"]
# ---------------------------------------------------------------------------
# GET /templates (FR-906)
# ---------------------------------------------------------------------------
def test_templates_returns_200(client: TestClient) -> None:
assert client.get("/templates").status_code == 200
def test_templates_lists_built_ins(client: TestClient) -> None:
body = client.get("/templates").json()
assert body["status"] == "ok"
names = {f["name"] for f in body["outputs"]}
assert "article" in names
assert "book" in names
assert "website" in names
def test_templates_envelope_shape(client: TestClient) -> None:
body = client.get("/templates").json()
assert "warnings" in body
assert "errors" in body
assert "context" in body
# ---------------------------------------------------------------------------
# GET /styles (FR-907)
# ---------------------------------------------------------------------------
def test_styles_returns_200(client: TestClient) -> None:
assert client.get("/styles").status_code == 200
def test_styles_is_list(client: TestClient) -> None:
body = client.get("/styles").json()
assert body["status"] == "ok"
assert isinstance(body["outputs"], list)
# ---------------------------------------------------------------------------
# Response envelope shape (FR-912)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("endpoint", ["/version", "/capabilities", "/templates", "/styles"])
def test_envelope_fields_present(client: TestClient, endpoint: str) -> None:
body = client.get(endpoint).json()
assert "status" in body
assert "outputs" in body
assert "warnings" in body
assert "errors" in body
assert "context" in body

51
tests/test_templates.py Normal file
View File

@@ -0,0 +1,51 @@
"""Tests for template family registry (FR-600)."""
from __future__ import annotations
from pathlib import Path
import pytest
from markidocx.templates import FamilyRegistry, RegistrationError
class TestFamilyRegistry:
def test_lists_three_builtin_families(self) -> None:
registry = FamilyRegistry()
families = registry.list_families()
names = {f.name for f in families}
assert names == {"article", "book", "website"}
def test_get_existing_family(self) -> None:
registry = FamilyRegistry()
info = registry.get("article")
assert info is not None
assert info.name == "article"
assert info.description
def test_get_missing_family_returns_none(self) -> None:
registry = FamilyRegistry()
assert registry.get("nonexistent") is None
def test_register_invalid_path_raises(self, tmp_path: Path) -> None:
registry = FamilyRegistry()
with pytest.raises(RegistrationError, match="not found"):
registry.register(tmp_path / "missing.docx", "custom")
def test_register_non_docx_raises(self, tmp_path: Path) -> None:
f = tmp_path / "template.txt"
f.write_text("not a docx")
registry = FamilyRegistry()
with pytest.raises(RegistrationError, match=".docx"):
registry.register(f, "custom")
def test_create_document_for_each_family(self) -> None:
registry = FamilyRegistry()
for family in ("article", "book", "website"):
doc = registry.create_document(family)
assert doc is not None
def test_create_document_unknown_family_falls_back(self) -> None:
registry = FamilyRegistry()
doc = registry.create_document("unknown")
assert doc is not None

228
tests/test_workflows.py Normal file
View File

@@ -0,0 +1,228 @@
"""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