"""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()) == []