Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 46 additions & 3 deletions scripts/dekc_common.py
Original file line number Diff line number Diff line change
Expand Up @@ -420,13 +420,25 @@ def resolve_knowledge_root(repo_root: Path, override: str | None = None) -> Path
return root if root.is_absolute() else (repo_root / root)
cfg = load_config(repo_root)
name = cfg.get("knowledge_root") or "knowledge"
intended = repo_root / name
for candidate in (
repo_root / name,
intended,
repo_root / "sample-knowledge",
repo_root / ".okf",
repo_root / "knowledge",
):
if candidate.is_dir() and (candidate / "index.md").is_file():
# Say so when we did NOT land on the intended root. There are 16
# call sites and only dekc_doctor announced the bundle it used, so
# with any other command you could not tell. This repo ships a
# sample-knowledge/, so a capture run inside a clone before
# initializing a bundle silently wrote there.
if candidate != intended:
print(
f"dekc: '{intended}' is not an initialized bundle; "
f"using '{candidate}' instead. Pass --bundle to be explicit.",
file=sys.stderr,
)
return candidate
return repo_root / name

Expand Down Expand Up @@ -594,7 +606,34 @@ def ensure_catalog_index(bundle: Path, catalog: str, title: str | None = None) -
return index


# Catalogs the sibling capture plugins also declare. Their renderers emit a bare
# `- [label](path)` with no annotation, so adding ours to a shared catalog makes
# the file flip on every alternation between plugins. Scope the annotation to
# catalogs only this plugin owns; correctness of a shared bundle beats a nicety.
_SAC_SHARED_CATALOGS = frozenset(
{"agents", "diagrams", "domains", "glossary", "packs", "products",
"storage", "workflows"}
)


def _escape_link_label(label: str) -> str:
"""Make a concept title safe to use as a Markdown link label.

An unescaped `[AREA]` title renders as `[[AREA]](/cat/x.md)`, which the OKF
graph reader's link regex cannot match. That yields a MISSING edge rather
than a broken one, and validate reports only broken edges -- so the concept
silently loses its catalog backlink.
"""
return label.replace("[", "\\[").replace("]", "\\]")


def refresh_catalog_index(bundle: Path, catalog: str) -> None:
# Refuse catalogs this plugin does not declare, so an outside caller cannot
# drive this renderer over a sibling plugin's catalog. Note this alone does
# NOT stabilise a shared bundle -- for a catalog two plugins both declare it
# passes in both. That is what the annotation scoping below is for.
if catalog not in CATALOGS:
return
cat_dir = bundle / catalog
if not cat_dir.is_dir():
return
Expand All @@ -609,9 +648,13 @@ def refresh_catalog_index(bundle: Path, catalog: str) -> None:
if p.name == "index.md":
continue
fm_c, _ = parse_frontmatter(p.read_text(encoding="utf-8"))
label = fm_c.get("title") or p.stem
label = _escape_link_label(fm_c.get("title") or p.stem)
layer = fm_c.get("layer")
suffix = f" · {layer}" if layer else ""
# Only annotate catalogs no sibling plugin renders. On a shared catalog
# the annotation is what makes the file churn back and forth.
suffix = (
f" · {layer}" if layer and catalog not in _SAC_SHARED_CATALOGS else ""
)
entries.append(f"- [{label}](/{catalog}/{p.name}){suffix}")
body = f"# {title}\n\nConcepts in this catalog:\n\n"
body += "\n".join(entries) + ("\n" if entries else "_None yet._\n")
Expand Down
74 changes: 73 additions & 1 deletion tests/test_dekc.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

import json
import subprocess
import re
import sys
import tempfile
import unittest
Expand All @@ -14,13 +15,84 @@
SCRIPTS = ROOT / "scripts"
sys.path.insert(0, str(SCRIPTS))

from dekc_common import slugify, list_concepts, parse_frontmatter, dump_frontmatter # noqa: E402
from dekc_common import ( # noqa: E402
CATALOGS,
dump_frontmatter,
list_concepts,
parse_frontmatter,
refresh_catalog_index,
resolve_knowledge_root,
slugify,
)
from dekc_lineage import build_graph # noqa: E402
from dekc_validate import validate_bundle # noqa: E402
from dekc_index import build_index, search_index # noqa: E402
from dekc_walk import extract_sql_tables # noqa: E402


class TestCatalogIndex(unittest.TestCase):
STRICT = re.compile(r"\[([^\]]+)\]\(([^)]+)\)")
AWARE = re.compile(r"\[((?:\\.|\[[^\[\]]*\]|[^\]])+)\]\(([^)]+)\)")

def _bundle(self, td, catalog, title, layer=None):
b = Path(td)
(b / "index.md").write_text("---\ntype: Bundle\ntitle: T\n---\n\n# T\n", encoding="utf-8")
(b / catalog).mkdir(parents=True, exist_ok=True)
lay = f"layer: {layer}\n" if layer else ""
(b / catalog / "a.md").write_text(
f"---\ntype: Table\ntitle: {title}\n{lay}---\n\n# A\n", encoding="utf-8")
refresh_catalog_index(b, catalog)
body = (b / catalog / "index.md").read_text(encoding="utf-8")
return [l for l in body.splitlines() if l.startswith("- [")][0]

def test_bracketed_title_is_escaped(self):
with tempfile.TemporaryDirectory() as td:
line = self._bundle(td, "tables", "Fact [Sales]")
self.assertIn(r"\[Sales\]", line, f"label not escaped: {line!r}")
self.assertEqual(self.AWARE.findall(line)[0][1], "/tables/a.md")
# Escaping alone does not rescue a `[^\]]+` reader; that class has no
# notion of an escape. This half depends on the reader change landing.
self.assertFalse(self.STRICT.findall(line))

def test_layer_annotation_only_on_catalogs_we_alone_own(self):
"""A shared catalog must render byte-identically to the sibling
plugins, or the file flips every time the other one runs."""
with tempfile.TemporaryDirectory() as td:
shared = self._bundle(td, "glossary", "Alpha", layer="gold")
self.assertNotIn("\u00b7 gold", shared, f"annotated a shared catalog: {shared!r}")
with tempfile.TemporaryDirectory() as td:
own = self._bundle(td, "layers", "Gold tier", layer="gold")
self.assertIn("\u00b7 gold", own, f"lost the annotation on our own catalog: {own!r}")

def test_refuses_a_catalog_this_plugin_does_not_declare(self):
self.assertNotIn("adrs", CATALOGS)
with tempfile.TemporaryDirectory() as td:
b = Path(td)
(b / "adrs").mkdir()
marker = "- [Untouched](/adrs/a.md)\n"
(b / "adrs" / "index.md").write_text(marker, encoding="utf-8")
refresh_catalog_index(b, "adrs")
self.assertEqual((b / "adrs" / "index.md").read_text(encoding="utf-8"), marker)


class TestResolveKnowledgeRoot(unittest.TestCase):
def test_configured_root_wins_when_initialized(self):
with tempfile.TemporaryDirectory() as td:
repo = Path(td)
for name in ("knowledge", "sample-knowledge"):
(repo / name).mkdir()
(repo / name / "index.md").write_text("# x\n", encoding="utf-8")
self.assertEqual(resolve_knowledge_root(repo).name, "knowledge")

def test_falls_back_only_when_intended_root_is_not_a_bundle(self):
with tempfile.TemporaryDirectory() as td:
repo = Path(td)
(repo / "knowledge").mkdir()
(repo / "sample-knowledge").mkdir()
(repo / "sample-knowledge" / "index.md").write_text("# x\n", encoding="utf-8")
self.assertEqual(resolve_knowledge_root(repo).name, "sample-knowledge")


class TestFrontmatterRoundTrip(unittest.TestCase):
"""parse(dump(x)) == x.

Expand Down
Loading