orchestrator/tests/test_frontmatter.py

"""ORCH-076 (ORCH-52c): unit tests for the unified frontmatter contract.

Covers TC-01..TC-07 of docs/work-items/ORCH-076/04-test-plan.yaml:
  * writer (render/round-trip), validator (full / partial schema, strict on/off),
  * reader contract preserved (read_frontmatter_value), never-raise on bad input.

The whole module honours a never-raise contract (NFR-2): no input shape may raise.
"""

import os
import tempfile

os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")

from src import frontmatter as fm  # noqa: E402
from src.frontmatter import (  # noqa: E402
    REQUIRED_FIELDS,
    FrontmatterParse,
    SchemaValidation,
    maybe_warn_schema,
    parse_frontmatter,
    parse_frontmatter_dict,
    read_frontmatter,
    read_frontmatter_value,
    render_frontmatter,
    strip_frontmatter,
    validate_schema,
    write_frontmatter,
)


def _full_schema():
    return {
        "work_item": "ORCH-076",
        "stage": "review",
        "author_agent": "reviewer",
        "status": "APPROVED",
        "created_at": "2026-06-09",
        "model_used": "claude-opus-4-8",
    }


# --------------------------------------------------------------------------- #
# TC-01 — writer serialises a mapping into canonical leading YAML-frontmatter
#         readable by the existing parsers (split("---", 2) + yaml.safe_load).
# --------------------------------------------------------------------------- #
def test_tc01_render_frontmatter_is_canonical_and_reparseable():
    out = render_frontmatter({"verdict": "APPROVED", "work_item": "ORCH-076"}, "body text")
    assert out.startswith("---\n")
    # Existing parser shape: split on '---' into 3 segments + yaml.safe_load.
    parts = out.split("---", 2)
    assert len(parts) == 3
    import yaml
    data = yaml.safe_load(parts[1])
    assert data["verdict"] == "APPROVED"
    assert data["work_item"] == "ORCH-076"
    # Body is preserved verbatim after the closing fence.
    assert parts[2].lstrip("\n") == "body text"
    # And our own primitive round-trips it.
    assert parse_frontmatter(out).data == {"verdict": "APPROVED", "work_item": "ORCH-076"}


def test_tc01_render_empty_body_default():
    out = render_frontmatter({"a": 1})
    assert out == "---\na: 1\n---\n"


# --------------------------------------------------------------------------- #
# TC-02 — round-trip: writer -> reader read_frontmatter_value yields same values.
# --------------------------------------------------------------------------- #
def test_tc02_write_then_read_roundtrip():
    data = _full_schema()
    with tempfile.TemporaryDirectory() as d:
        path = os.path.join(d, "12-review.md")
        assert write_frontmatter(path, data, "# Review body") is True
        for key, val in data.items():
            assert read_frontmatter_value(path, key) == val
        # Whole-mapping read matches too.
        assert read_frontmatter(path) == data


def test_tc02_render_parse_dict_roundtrip():
    data = _full_schema()
    rendered = render_frontmatter(data, "body")
    assert parse_frontmatter_dict(rendered) == data


# --------------------------------------------------------------------------- #
# TC-03 — validator: full schema -> valid=True, no missing fields.
# --------------------------------------------------------------------------- #
def test_tc03_validate_full_schema_valid():
    res = validate_schema(_full_schema())
    assert isinstance(res, SchemaValidation)
    assert res.valid is True
    assert res.missing == []


# --------------------------------------------------------------------------- #
# TC-04 — validator: partial schema -> valid=False with the missing list,
#         WITHOUT raising (warning-only by default).
# --------------------------------------------------------------------------- #
def test_tc04_validate_partial_schema_lists_missing():
    res = validate_schema({"work_item": "ORCH-076", "stage": "review"})
    assert res.valid is False
    # The four absent required fields are reported (order = REQUIRED_FIELDS).
    assert set(res.missing) == set(REQUIRED_FIELDS) - {"work_item", "stage"}
    assert res.missing == [f for f in REQUIRED_FIELDS if f in res.missing]


def test_tc04_blank_and_none_count_as_missing():
    data = _full_schema()
    data["status"] = ""        # blank -> missing
    data["model_used"] = None  # None -> missing
    res = validate_schema(data)
    assert res.valid is False
    assert set(res.missing) == {"status", "model_used"}


# --------------------------------------------------------------------------- #
# TC-05 — never-raise: writer + validator on broken input return a safe value.
# --------------------------------------------------------------------------- #
def test_tc05_validate_non_mapping_never_raises():
    for bad in (None, "not a mapping", 123, ["a", "b"]):
        res = validate_schema(bad)  # type: ignore[arg-type]
        assert res.valid is False
        assert set(res.missing) == set(REQUIRED_FIELDS)


def test_tc05_parse_broken_inputs_never_raise():
    # No frontmatter.
    p = parse_frontmatter("just prose, no fence")
    assert p == FrontmatterParse()
    assert p.data == {} and p.has_block is False
    # Unterminated block.
    p = parse_frontmatter("---\nkey: val\nno closing fence")
    assert p.has_block is True and p.malformed is True and p.data == {}
    # Bad YAML.
    p = parse_frontmatter("---\nkey: : :\n bad\n---\n")
    assert p.has_block is True and p.yaml_error is not None and p.data == {}
    # Non-mapping scalar frontmatter.
    p = parse_frontmatter("---\njust a string\n---\nbody")
    assert p.has_block is True and p.data == {}
    # Non-string input.
    assert parse_frontmatter(None).data == {}  # type: ignore[arg-type]
    assert parse_frontmatter_dict(12345) == {}  # type: ignore[arg-type]


def test_tc05_write_to_unwritable_path_returns_false():
    # A path under a non-existent directory cannot be opened -> False, no raise.
    ok = write_frontmatter("/nonexistent-dir-xyz/cannot/12-review.md", {"a": 1})
    assert ok is False


def test_tc05_render_unserialisable_degrades_to_body():
    class Bad:
        pass

    out = render_frontmatter({"x": Bad()}, "fallback-body")
    # yaml cannot serialise an arbitrary object -> degrade to the body, never raise.
    assert out == "fallback-body"


def test_tc05_read_missing_file_returns_empty():
    assert read_frontmatter("/no/such/file.md") == {}
    assert read_frontmatter_value("/no/such/file.md", "verdict") is None


# --------------------------------------------------------------------------- #
# TC-06 — reader read_frontmatter_value keeps its previous contract.
# --------------------------------------------------------------------------- #
def test_tc06_reader_contract_preserved():
    with tempfile.TemporaryDirectory() as d:
        path = os.path.join(d, "doc.md")
        with open(path, "w", encoding="utf-8") as f:
            f.write("---\nverdict: Approved\nempty:\n---\nbody\n")
        # strip + case preserved.
        assert read_frontmatter_value(path, "verdict") == "Approved"
        # empty value -> None.
        assert read_frontmatter_value(path, "empty") is None
        # absent key -> None.
        assert read_frontmatter_value(path, "missing") is None

    # No frontmatter -> None.
    with tempfile.TemporaryDirectory() as d:
        path = os.path.join(d, "doc.md")
        with open(path, "w", encoding="utf-8") as f:
            f.write("no frontmatter here\n")
        assert read_frontmatter_value(path, "verdict") is None


def test_tc06_reader_strips_whitespace():
    with tempfile.TemporaryDirectory() as d:
        path = os.path.join(d, "doc.md")
        with open(path, "w", encoding="utf-8") as f:
            f.write('---\nverdict: "  PASS  "\n---\n')
        assert read_frontmatter_value(path, "verdict") == "PASS"


# --------------------------------------------------------------------------- #
# TC-07 — kill-switch: strict False (default) is inert; strict True signals
#         invalidity. maybe_warn_schema never changes a verdict either way.
# --------------------------------------------------------------------------- #
def test_tc07_maybe_warn_schema_default_warning_only(monkeypatch, caplog):
    monkeypatch.setattr(fm, "logger", fm.logger)
    from src.config import settings
    monkeypatch.setattr(settings, "frontmatter_validation_strict", False)
    incomplete = render_frontmatter({"verdict": "APPROVED"}, "body")
    res = maybe_warn_schema(incomplete, "review report")
    # Validation still reports invalidity (the data IS incomplete)...
    assert res.valid is False
    assert "model_used" in res.missing
    # ...but the helper is inert: it returns a value, it does not raise / block.


def test_tc07_strict_flag_visible_to_helper(monkeypatch):
    from src.config import settings
    # Full schema -> valid regardless of the flag.
    monkeypatch.setattr(settings, "frontmatter_validation_strict", True)
    res_full = maybe_warn_schema(render_frontmatter(_full_schema(), "b"), "doc")
    assert res_full.valid is True
    # Incomplete -> invalid; strict True does not raise, just signals.
    res_partial = maybe_warn_schema(render_frontmatter({"stage": "review"}, "b"), "doc")
    assert res_partial.valid is False


def test_tc07_maybe_warn_schema_on_garbage_is_inert():
    # Never-raise: a non-string / no-frontmatter input returns a SchemaValidation
    # (reporting the missing fields) WITHOUT raising — the gate verdict is untouched.
    for bad in ("no frontmatter", None, 123):
        res = maybe_warn_schema(bad, "doc")  # type: ignore[arg-type]
        assert isinstance(res, SchemaValidation)


# --------------------------------------------------------------------------- #
# strip_frontmatter helper parity.
# --------------------------------------------------------------------------- #
def test_strip_frontmatter_parity():
    assert strip_frontmatter("---\na: 1\n---\nbody") == "\nbody"
    # No well-formed block -> unchanged.
    assert strip_frontmatter("no fence") == "no fence"
    assert strip_frontmatter("---\nunterminated") == "---\nunterminated"
    # Never-raise on non-string.
    assert strip_frontmatter(None) is None  # type: ignore[arg-type]