llm-ingestion-okf/tests/test_import_flow.py

"""Door C import flow against a stub import gate (Phase 2 step 5).

The stub stands in for the pinned `llm-ingestion-guard` `okf.import_bundle`
surface so the merge, quarantine and reject branches run deterministically
without the dependency installed. It is a TEST DOUBLE and lives here on
purpose — never in `src/`.

What the flow owes the caller, and what this suite pins:

- the guard decides per concept; the library only branches on the verdict and
  never re-derives one (an unrecognised disposition, and a concept the gate
  returned no verdict for, must both fail CLOSED);
- a merged concept is written VERBATIM — exactly the text handed to the gate,
  with no frontmatter of ours merged into it (round-tripping the concept's own
  frontmatter through this library's line-oriented parser would silently
  destroy the block lists the guard's parser accepts);
- ownership is proven by content identity, not by a stamp: a target name held
  by identical bytes is a no-op re-merge, and a target name held by ANYTHING
  else is refused — curated content and a changed external concept alike;
- one bad concept never aborts the run;
- the guard's log body is returned to the caller, never persisted (the
  reserved-file policy is Phase 3's).
"""

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path

import pytest

from llm_ingestion_okf.importer import BundleDecision, ImportDecision, ImportResult, import_bundle

INGESTED_AT = "2026-07-25T12:00:00Z"

# The guard's Origin/Channel/Disposition are `str, Enum`s; these are their
# VALUES, restated here independently of the library constants so a drift in
# either is visible.
EXTERNAL = "external"
INTERNAL = "internal"
AUTOMATIC = "automatic"
MANUAL = "manual"
WARN = "warn"
QUARANTINE_REVIEW = "quarantine_review"
FAIL_SECURE = "fail_secure"


# --- the stub gate --------------------------------------------------------


@dataclass
class StubImportGate:
    """A bundle gate whose per-concept verdict is scripted.

    `by_marker` maps a substring to the disposition any concept containing it
    gets; `errors` maps a concept path to a hard-reject reason (the guard's
    `ConceptResult.error`). Everything else passes at the non-blocking floor.
    """

    by_marker: dict[str, str] = field(default_factory=dict)
    errors: dict[str, str] = field(default_factory=dict)
    omit: frozenset[str] = frozenset()
    log: str = ""
    calls: list[dict[str, str]] = field(default_factory=list)
    provenance: list[tuple[str, str]] = field(default_factory=list)

    def __call__(self, bundle: dict[str, str], *, origin: str, channel: str) -> BundleDecision:
        self.calls.append(dict(bundle))
        self.provenance.append((origin, channel))
        decisions = []
        for path in sorted(bundle):
            if path in self.omit:
                continue
            error = self.errors.get(path)
            disposition = FAIL_SECURE if error is not None else WARN
            for marker, scripted in self.by_marker.items():
                if marker in bundle[path]:
                    disposition = scripted
            decisions.append(
                ImportDecision(
                    path=path,
                    disposition=disposition,
                    error=error,
                    reasons=(f"scanned {path}",),
                )
            )
        return BundleDecision(concepts=tuple(decisions), log=self.log)


def place(source: Path, relpath: str, content: str) -> Path:
    path = source / relpath
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8", newline="")
    return path


def run(
    tmp_path: Path,
    gate: StubImportGate,
    *,
    origin: str = EXTERNAL,
    channel: str = AUTOMATIC,
    ingested_at: str = INGESTED_AT,
) -> tuple[ImportResult, Path]:
    bundle = tmp_path / "bundle"
    result = import_bundle(
        tmp_path / "source",
        bundle,
        ingested_at,
        origin=origin,
        channel=channel,
        gate=gate,
    )
    return result, bundle


CONCEPT = "---\ntype: dataset\ntitle: Users\n---\n\nA table of users.\n"


# --- the merge branch -----------------------------------------------------


def test_an_accepted_concept_is_merged(tmp_path: Path) -> None:
    place(tmp_path / "source", "tables/users.md", CONCEPT)

    result, bundle = run(tmp_path, StubImportGate())

    assert [entry.concept_path for entry in result.merged] == ["tables/users.md"]
    assert (bundle / "import-tables-users.md").is_file()


def test_the_merged_bytes_are_exactly_the_bytes_the_gate_saw(tmp_path: Path) -> None:
    """Verbatim is the whole contract at this door: the guard validated a
    document, so the document is what may be persisted. Nothing of ours is
    merged into its frontmatter — this library's line-oriented parser cannot
    round-trip the block lists the guard's parser accepts, so writing back a
    re-rendered concept would silently drop data the operator sent us.
    """
    document = "---\ntags:\n  - alpha\n  - beta\n---\n\nBody with a [link](other.md).\n"
    place(tmp_path / "source", "notes/tagged.md", document)

    result, bundle = run(tmp_path, gate := StubImportGate())

    written = (bundle / "import-notes-tagged.md").read_bytes()
    assert written == document.encode("utf-8")
    assert gate.calls[0]["notes/tagged.md"] == document
    assert result.failed == ()


def test_the_index_links_every_merged_concept_by_concept_id(tmp_path: Path) -> None:
    place(tmp_path / "source", "tables/users.md", CONCEPT)

    _, bundle = run(tmp_path, StubImportGate())

    index = (bundle / "index.md").read_text(encoding="utf-8")
    assert index == "- [tables/users](import-tables-users.md)\n"


def test_nested_paths_keep_their_directory_in_the_generated_name(tmp_path: Path) -> None:
    """Flattening on the stem alone would collide two distinct concepts; the
    whole bundle-relative path is what reduces to the slug.
    """
    place(tmp_path / "source", "tables/users.md", "one\n")
    place(tmp_path / "source", "views/users.md", "two\n")

    result, bundle = run(tmp_path, StubImportGate())

    assert [entry.path.name for entry in result.merged] == [
        "import-tables-users.md",
        "import-views-users.md",
    ]
    assert (bundle / "import-tables-users.md").read_text(encoding="utf-8") == "one\n"
    assert (bundle / "import-views-users.md").read_text(encoding="utf-8") == "two\n"


def test_the_gate_is_called_once_with_the_whole_bundle(tmp_path: Path) -> None:
    """`okf.import_bundle` is a bundle-level call — it resolves the cross-link
    graph across concepts — so the flow hands it the whole mapping at once,
    keyed by POSIX-separated bundle-relative paths.
    """
    place(tmp_path / "source", "b.md", "second\n")
    place(tmp_path / "source", "nested/a.md", "first\n")

    _, _ = run(tmp_path, gate := StubImportGate())

    assert len(gate.calls) == 1
    assert gate.calls[0] == {"b.md": "second\n", "nested/a.md": "first\n"}


def test_origin_and_channel_reach_the_gate_verbatim(tmp_path: Path) -> None:
    """Trust follows origin, never channel — the caller declares both at the
    door and the library only carries them; it never defaults or derives them.
    """
    place(tmp_path / "source", "a.md", "body\n")

    run(tmp_path, gate := StubImportGate(), origin=INTERNAL, channel=MANUAL)

    assert gate.provenance == [(INTERNAL, MANUAL)]


# --- the blocking branches ------------------------------------------------


def test_a_hard_rejected_concept_is_not_merged(tmp_path: Path) -> None:
    place(tmp_path / "source", "bad.md", "body\n")

    result, bundle = run(tmp_path, StubImportGate(errors={"bad.md": "reserved filename"}))

    assert result.merged == ()
    assert [entry.concept_path for entry in result.rejected] == ["bad.md"]
    assert result.rejected[0].error == "reserved filename"
    assert not (bundle / "import-bad.md").exists()


def test_quarantine_review_is_reported_separately_from_rejection(tmp_path: Path) -> None:
    """Quarantine means "hold for human review" and is the operator's queue; a
    fail-secure verdict is a decision, not a queue. Same split as Door B.
    """
    place(tmp_path / "source", "suspect.md", "hold me\n")

    result, bundle = run(tmp_path, StubImportGate(by_marker={"hold me": QUARANTINE_REVIEW}))

    assert result.merged == ()
    assert [entry.concept_path for entry in result.quarantined] == ["suspect.md"]
    assert result.rejected == ()
    assert not (bundle / "import-suspect.md").exists()


@pytest.mark.parametrize("disposition", ["", "pass", "ok", "WARN", "warn "])
def test_an_unrecognised_disposition_fails_closed(tmp_path: Path, disposition: str) -> None:
    place(tmp_path / "source", "a.md", "body\n")

    result, bundle = run(tmp_path, StubImportGate(by_marker={"body": disposition}))

    assert result.merged == ()
    assert [entry.concept_path for entry in result.rejected] == ["a.md"]
    assert list(bundle.glob("*.md")) == []


def test_an_error_with_a_non_blocking_disposition_still_fails_closed(tmp_path: Path) -> None:
    """The guard pairs a hard reject with FAIL_SECURE, but the library must not
    depend on that pairing: an error is a refusal on its own terms.
    """
    place(tmp_path / "source", "a.md", "body\n")
    gate = StubImportGate(errors={"a.md": "unsafe resource"}, by_marker={"body": WARN})

    result, bundle = run(tmp_path, gate)

    assert result.merged == ()
    assert [entry.concept_path for entry in result.rejected] == ["a.md"]
    assert list(bundle.glob("*.md")) == []


def test_a_concept_the_gate_returned_no_verdict_for_is_not_merged(tmp_path: Path) -> None:
    """No verdict is not consent. A gate that drops a concept from its result —
    an adapter bug, or a guard that stops reporting one — must not be read as
    approval.
    """
    place(tmp_path / "source", "a.md", "body\n")
    place(tmp_path / "source", "b.md", "body\n")

    result, bundle = run(tmp_path, StubImportGate(omit=frozenset({"a.md"})))

    assert [entry.concept_path for entry in result.merged] == ["b.md"]
    assert [entry.concept_path for entry in result.rejected] == ["a.md"]
    assert not (bundle / "import-a.md").exists()


def test_a_rejected_concept_leaves_the_bundle_byte_identical(tmp_path: Path) -> None:
    """The persist-gate proof: a rejected concept produces ZERO new files and
    does not touch the ones already there — not even the index.
    """
    place(tmp_path / "source", "ok.md", "body\n")
    result, bundle = run(tmp_path, StubImportGate())
    assert result.merged != ()
    before = sorted((path.name, path.read_bytes()) for path in bundle.iterdir())

    (tmp_path / "source" / "ok.md").unlink()
    place(tmp_path / "source", "poisoned.md", "body\n")
    run(tmp_path, StubImportGate(errors={"poisoned.md": "injection"}))

    after = sorted((path.name, path.read_bytes()) for path in bundle.iterdir())
    assert after == before


def test_a_mixed_bundle_merges_only_the_accepted_concepts(tmp_path: Path) -> None:
    """Partial merge is the point of a per-concept gate: one poisoned concept
    must not cost the operator the rest of the bundle, and must not sneak in.
    """
    place(tmp_path / "source", "good.md", "harmless\n")
    place(tmp_path / "source", "held.md", "hold me\n")
    place(tmp_path / "source", "bad.md", "poison\n")

    result, bundle = run(
        tmp_path,
        StubImportGate(by_marker={"poison": FAIL_SECURE, "hold me": QUARANTINE_REVIEW}),
    )

    assert [entry.concept_path for entry in result.merged] == ["good.md"]
    assert [entry.concept_path for entry in result.quarantined] == ["held.md"]
    assert [entry.concept_path for entry in result.rejected] == ["bad.md"]
    assert sorted(path.name for path in bundle.glob("*.md")) == ["import-good.md", "index.md"]
    assert (bundle / "index.md").read_text(encoding="utf-8") == "- [good](import-good.md)\n"


# --- ownership by content identity ----------------------------------------


def test_a_re_import_of_an_unchanged_bundle_is_a_no_op(tmp_path: Path) -> None:
    """Identical bytes at the target name are proof enough that the file is
    ours: the re-import is reported as merged, writes nothing new, and does not
    duplicate the index link.
    """
    place(tmp_path / "source", "tables/users.md", CONCEPT)

    result, bundle = run(tmp_path, StubImportGate())
    first = sorted((path.name, path.read_bytes()) for path in bundle.iterdir())

    result, bundle = run(tmp_path, StubImportGate())
    second = sorted((path.name, path.read_bytes()) for path in bundle.iterdir())

    assert [entry.concept_path for entry in result.merged] == ["tables/users.md"]
    assert result.failed == ()
    assert second == first


def test_a_changed_concept_is_refused_without_overwriting(tmp_path: Path) -> None:
    """Without a stamp the library cannot prove the existing file is a previous
    import rather than an edit the operator made — so a changed concept is
    refused, not merged. The operator resolves it by removing the old file.
    """
    place(tmp_path / "source", "note.md", "first version\n")
    run(tmp_path, StubImportGate())
    place(tmp_path / "source", "note.md", "second version\n")

    result, bundle = run(tmp_path, StubImportGate())

    assert result.merged == ()
    assert [entry.concept_path for entry in result.failed] == ["note.md"]
    assert result.failed[0].error.code == "collision_unstamped"
    assert (bundle / "import-note.md").read_text(encoding="utf-8") == "first version\n"


def test_curated_content_at_the_target_name_is_never_overwritten(tmp_path: Path) -> None:
    bundle = tmp_path / "bundle"
    bundle.mkdir()
    curated = bundle / "import-note.md"
    curated.write_text("curated by hand\n", encoding="utf-8")
    place(tmp_path / "source", "note.md", "external\n")
    place(tmp_path / "source", "other.md", "external\n")

    result, _ = run(tmp_path, StubImportGate())

    assert curated.read_text(encoding="utf-8") == "curated by hand\n"
    assert [entry.error.code for entry in result.failed] == ["collision_unstamped"]
    assert [entry.concept_path for entry in result.merged] == ["other.md"]


def test_the_collision_gate_is_evaluated_before_any_write(tmp_path: Path) -> None:
    """Pre-mutation, exactly as at Door A and Door B: occupancy is judged
    against the bundle as it was BEFORE this run.
    """
    bundle = tmp_path / "bundle"
    bundle.mkdir()
    (bundle / "import-zzz.md").write_text("curated\n", encoding="utf-8")
    place(tmp_path / "source", "aaa.md", "body\n")
    place(tmp_path / "source", "zzz.md", "body\n")

    result, _ = run(tmp_path, StubImportGate())

    assert [entry.error.code for entry in result.failed] == ["collision_unstamped"]
    assert [entry.concept_path for entry in result.merged] == ["aaa.md"]
    assert (bundle / "import-zzz.md").read_text(encoding="utf-8") == "curated\n"


def test_two_concept_paths_that_slug_alike_are_both_refused(tmp_path: Path) -> None:
    """`a/b.md` and `a-b.md` both reduce to `import-a-b.md`. The library will
    not pick a winner: both are refused and the sender renames one.
    """
    place(tmp_path / "source", "a/b.md", "one\n")
    place(tmp_path / "source", "a-b.md", "two\n")
    place(tmp_path / "source", "other.md", "three\n")

    result, bundle = run(tmp_path, StubImportGate())

    assert [entry.concept_path for entry in result.failed] == ["a-b.md", "a/b.md"]
    assert {entry.error.code for entry in result.failed} == {"import_slug_collision"}
    assert not (bundle / "import-a-b.md").exists()
    assert [entry.concept_path for entry in result.merged] == ["other.md"]


# --- one bad concept never aborts the run ---------------------------------


def test_a_concept_path_that_reduces_to_an_empty_slug_is_a_per_concept_failure(
    tmp_path: Path,
) -> None:
    place(tmp_path / "source", "!!!.md", "body\n")
    place(tmp_path / "source", "good.md", "body\n")

    result, _ = run(tmp_path, StubImportGate())

    assert [entry.concept_path for entry in result.merged] == ["good.md"]
    assert [entry.error.code for entry in result.failed] == ["import_path_empty"]


def test_an_over_long_concept_path_is_a_per_concept_failure(tmp_path: Path) -> None:
    """A bundle-relative path has no 255-byte limit of its own — only its
    individual segments do — so a deep enough tree reduces to a filename the
    filesystem cannot hold. Refused, never truncated.
    """
    deep = "/".join(["segment"] * 40) + ".md"
    place(tmp_path / "source", deep, "body\n")
    place(tmp_path / "source", "good.md", "body\n")

    result, _ = run(tmp_path, StubImportGate())

    assert [entry.concept_path for entry in result.merged] == ["good.md"]
    assert [entry.error.code for entry in result.failed] == ["import_path_too_long"]


def test_a_concept_path_that_would_break_an_index_link_is_a_per_concept_failure(
    tmp_path: Path,
) -> None:
    """The guard's path gate permits brackets; this library's index link does
    not. The concept-ID is rendered verbatim as the link label, so it meets the
    same fail-fast rule Door A and Door B apply to a title.
    """
    place(tmp_path / "source", "report [v2].md", "body\n")

    result, _ = run(tmp_path, StubImportGate())

    assert result.merged == ()
    assert [entry.error.code for entry in result.failed] == ["import_label_invalid"]


def test_a_non_utf8_concept_is_a_per_concept_failure(tmp_path: Path) -> None:
    """Read failures happen before the gate — an undecodable concept is never
    handed to it, and never counted as approved.
    """
    source = tmp_path / "source"
    source.mkdir()
    (source / "broken.md").write_bytes(b"\xffnot utf-8\n")
    place(source, "good.md", "body\n")

    result, _ = run(tmp_path, gate := StubImportGate())

    assert [entry.concept_path for entry in result.merged] == ["good.md"]
    assert [entry.error.code for entry in result.failed] == ["extractor_decode_error"]
    assert "broken.md" not in gate.calls[0]


def test_an_uppercase_md_suffix_is_still_a_concept(tmp_path: Path) -> None:
    """Determinism across filesystems: a `*.md` glob is case-sensitive on ext4
    and case-insensitive on APFS, so the same bundle would import differently
    per platform. The suffix test is explicit and case-folded, as the guard's
    own path gate is.
    """
    place(tmp_path / "source", "NOTE.MD", "body\n")

    result, bundle = run(tmp_path, gate := StubImportGate())

    assert list(gate.calls[0]) == ["NOTE.MD"]
    assert [entry.concept_path for entry in result.merged] == ["NOTE.MD"]
    assert (bundle / "import-note.md").read_text(encoding="utf-8") == "body\n"


def test_non_markdown_files_are_not_concepts(tmp_path: Path) -> None:
    """An OKF concept is a `.md` document by definition (the guard's path gate
    says so too), so anything else in the source tree is not this door's to
    merge — it is neither gated nor written.
    """
    place(tmp_path / "source", "a.md", "body\n")
    place(tmp_path / "source", "data.csv", "a,b\n1,2\n")

    result, bundle = run(tmp_path, gate := StubImportGate())

    assert list(gate.calls[0]) == ["a.md"]
    assert [entry.concept_path for entry in result.merged] == ["a.md"]
    assert sorted(path.name for path in bundle.iterdir()) == ["import-a.md", "index.md"]


# --- the import report ----------------------------------------------------


def test_the_guards_log_is_returned_and_not_persisted(tmp_path: Path) -> None:
    """Deliverable 3: the caller gets the log body; writing it into the bundle
    is a reserved-file decision that differs per consumer and lands in Phase 3.
    """
    place(tmp_path / "source", "a.md", "body\n")

    result, bundle = run(tmp_path, StubImportGate(log="a\texternal\tautomatic\tuntrusted\twarn"))

    assert result.log == "a\texternal\tautomatic\tuntrusted\twarn"
    assert not (bundle / "log.md").exists()


def test_the_result_carries_the_run_timestamp(tmp_path: Path) -> None:
    """`ingested_at` reaches no file at this door — a merged concept is written
    verbatim — but the run is still timestamped explicitly rather than by wall
    clock, so a caller persisting the log has a deterministic timestamp to use.
    """
    place(tmp_path / "source", "a.md", "body\n")

    result, _ = run(tmp_path, StubImportGate())

    assert result.ingested_at == INGESTED_AT


def test_the_per_concept_reasons_are_carried_verbatim(tmp_path: Path) -> None:
    place(tmp_path / "source", "a.md", "poison\n")

    result, _ = run(tmp_path, StubImportGate(by_marker={"poison": FAIL_SECURE}))

    assert result.rejected[0].reasons == ("scanned a.md",)


# --- run-level fail-fast --------------------------------------------------


def test_an_invalid_ingested_at_fails_the_whole_run(tmp_path: Path) -> None:
    from llm_ingestion_okf import MaterializationError

    place(tmp_path / "source", "a.md", "body\n")

    with pytest.raises(MaterializationError) as excinfo:
        run(tmp_path, StubImportGate(), ingested_at="2026-07-25")
    assert excinfo.value.code == "ingested_at_invalid"


@pytest.mark.parametrize(
    ("origin", "channel"),
    [("extenal", AUTOMATIC), ("", AUTOMATIC), (EXTERNAL, "auto"), (EXTERNAL, "")],
)
def test_an_unknown_origin_or_channel_fails_the_whole_run(
    tmp_path: Path, origin: str, channel: str
) -> None:
    """A value outside the guard's pinned vocabulary would reach the guard as a
    plain string, miss its enum identity check, and be silently downgraded to
    untrusted. Fail-fast instead: the library refuses to carry a provenance
    declaration it cannot recognise, rather than letting a typo decide trust.
    """
    from llm_ingestion_okf import MaterializationError

    place(tmp_path / "source", "a.md", "body\n")

    with pytest.raises(MaterializationError) as excinfo:
        run(tmp_path, StubImportGate(), origin=origin, channel=channel)
    assert excinfo.value.code == "import_provenance_invalid"


def test_a_missing_source_directory_is_refused(tmp_path: Path) -> None:
    from llm_ingestion_okf import SourceError

    with pytest.raises(SourceError) as excinfo:
        import_bundle(
            tmp_path / "nope",
            tmp_path / "bundle",
            INGESTED_AT,
            origin=EXTERNAL,
            channel=AUTOMATIC,
            gate=StubImportGate(),
        )
    assert excinfo.value.code == "source_root_missing"


def test_an_empty_source_bundle_writes_nothing_and_never_calls_the_gate(tmp_path: Path) -> None:
    (tmp_path / "source").mkdir()

    result, bundle = run(tmp_path, gate := StubImportGate())

    assert result == ImportResult(
        merged=(), quarantined=(), rejected=(), failed=(), log="", ingested_at=INGESTED_AT
    )
    assert gate.calls == []
    assert not bundle.exists() or list(bundle.iterdir()) == []