Implement whole-Knowledge-Base validation per MASTER_SPEC §15.10, §26.4: compiler/common/kb_validation.py (importable by the loader) plus a CLI at scripts/validate_knowledge_base.py. validate_knowledge_base(entries) collects every problem (never stops at the first); each Message carries a code, the offending id, and a description: - duplicate canonical id -> SC0005 - invalid category -> SC0008 (drawn from the shared 19-category constant) - circular expansion (incl. self-expansion) -> SC0003, via DFS cycle detection - duplicate alias, alias colliding with a canonical id (no chains), missing expansion target, and per-entry schema nonconformance (empty tags, unknown fields, ...) -> umbrella SC0004 Findings are deterministically ordered. The CLI loads every *.json (array of entries) from a directory, reports invalid JSON / non-array files as SC0004, writes JSON findings to stdout, and exits non-zero when invalid. Tests: a passing fixture plus a fixture per failure mode, multi-problem reporting, determinism, and the CLI on valid/invalid/broken-JSON inputs. Co-authored-by: kunail0804 <kunail0804@gmail.com> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Crespin Alexandre
co-authored by
kunail0804
Claude Opus 4.8
kunail0804
Claude Opus 4.8
parent
07b0f9a18b
commit
d6dcabe4f7
@@ -0,0 +1,221 @@
|
||||
"""Knowledge Base validation tooling (MASTER_SPEC §15.10, §26.4).
|
||||
|
||||
Validates a whole Knowledge Base — both per-entry structure and cross-entry
|
||||
integrity — and returns every problem found (it does not stop at the first).
|
||||
Each problem is a :class:`Message` carrying a code, the offending ``id`` in its
|
||||
context, and a description.
|
||||
|
||||
This core is importable by the Knowledge Base loader (#7) and driven by the CLI
|
||||
in ``scripts/validate_knowledge_base.py``.
|
||||
|
||||
Codes (Appendix B): duplicate canonical id → ``SC0005``; invalid category →
|
||||
``SC0008``; circular expansion → ``SC0003``. The remaining Knowledge-Base
|
||||
structural problems have no dedicated Appendix B code and are reported under the
|
||||
umbrella ``SC0004`` (Knowledge Base Load Failure): duplicate alias, an alias
|
||||
colliding with a canonical id (which would create a resolution chain), a missing
|
||||
expansion target, and any per-entry schema nonconformance (including empty tags
|
||||
and unknown fields).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable, Mapping, Sequence
|
||||
from typing import Any
|
||||
|
||||
from compiler.common.categories import is_valid_category
|
||||
from compiler.common.result import Message, Severity
|
||||
from schemas.validation import validate_document
|
||||
|
||||
_ENTRY_SCHEMA = "knowledge_base_entry"
|
||||
|
||||
|
||||
def _sc0004(title: str, description: str, entry_id: str, **context: Any) -> Message:
|
||||
return Message(
|
||||
code="SC0004",
|
||||
severity=Severity.FATAL,
|
||||
title=title,
|
||||
description=description,
|
||||
context={"id": entry_id, **context},
|
||||
)
|
||||
|
||||
|
||||
def _schema_messages(entry: Mapping[str, Any], entry_id: str) -> list[Message]:
|
||||
"""Per-entry schema conformance, minus the category (checked explicitly)."""
|
||||
messages: list[Message] = []
|
||||
for issue in validate_document(entry, _ENTRY_SCHEMA):
|
||||
if issue.path == "category":
|
||||
continue # reported as SC0008 by the explicit category check
|
||||
messages.append(
|
||||
_sc0004(
|
||||
title="Invalid Knowledge Base entry",
|
||||
description=f"Entry '{entry_id}' failed schema validation: {issue.message}",
|
||||
entry_id=entry_id,
|
||||
path=issue.path,
|
||||
)
|
||||
)
|
||||
return messages
|
||||
|
||||
|
||||
def _category_messages(entry: Mapping[str, Any], entry_id: str) -> list[Message]:
|
||||
category = entry.get("category")
|
||||
if category is not None and not is_valid_category(category):
|
||||
return [
|
||||
Message(
|
||||
code="SC0008",
|
||||
severity=Severity.ERROR,
|
||||
title="Invalid Category",
|
||||
description=f"Entry '{entry_id}' references unknown category '{category}'.",
|
||||
context={"id": entry_id, "category": category},
|
||||
)
|
||||
]
|
||||
return []
|
||||
|
||||
|
||||
def _duplicate_id_messages(entries: Sequence[Mapping[str, Any]]) -> list[Message]:
|
||||
seen: set[str] = set()
|
||||
messages: list[Message] = []
|
||||
for entry in entries:
|
||||
entry_id = entry.get("id")
|
||||
if not isinstance(entry_id, str):
|
||||
continue
|
||||
if entry_id in seen:
|
||||
messages.append(
|
||||
Message(
|
||||
code="SC0005",
|
||||
severity=Severity.ERROR,
|
||||
title="Duplicate Canonical ID",
|
||||
description=f"Canonical id '{entry_id}' is defined more than once.",
|
||||
context={"id": entry_id},
|
||||
)
|
||||
)
|
||||
seen.add(entry_id)
|
||||
return messages
|
||||
|
||||
|
||||
def _alias_messages(entries: Sequence[Mapping[str, Any]]) -> list[Message]:
|
||||
"""Aliases must be globally unique and never collide with a canonical id."""
|
||||
ids = {e["id"] for e in entries if isinstance(e.get("id"), str)}
|
||||
messages: list[Message] = []
|
||||
seen_aliases: set[str] = set()
|
||||
for entry in entries:
|
||||
entry_id = entry.get("id", "<unknown>")
|
||||
for alias in entry.get("aliases", []):
|
||||
if alias in ids:
|
||||
messages.append(
|
||||
_sc0004(
|
||||
title="Alias collides with a canonical id",
|
||||
description=(
|
||||
f"Alias '{alias}' on entry '{entry_id}' equals a canonical id; "
|
||||
"aliases must resolve directly (no chains)."
|
||||
),
|
||||
entry_id=entry_id,
|
||||
alias=alias,
|
||||
)
|
||||
)
|
||||
if alias in seen_aliases:
|
||||
messages.append(
|
||||
_sc0004(
|
||||
title="Duplicate alias",
|
||||
description=(
|
||||
f"Alias '{alias}' (entry '{entry_id}') is defined more than once."
|
||||
),
|
||||
entry_id=entry_id,
|
||||
alias=alias,
|
||||
)
|
||||
)
|
||||
seen_aliases.add(alias)
|
||||
return messages
|
||||
|
||||
|
||||
def _expansion_messages(entries: Sequence[Mapping[str, Any]]) -> list[Message]:
|
||||
"""Expansion targets must exist and must not form cycles."""
|
||||
ids = {e["id"] for e in entries if isinstance(e.get("id"), str)}
|
||||
graph = {
|
||||
e["id"]: [t for t in e.get("expand", [])] for e in entries if isinstance(e.get("id"), str)
|
||||
}
|
||||
messages: list[Message] = []
|
||||
|
||||
# Missing targets.
|
||||
for entry in entries:
|
||||
entry_id = entry.get("id", "<unknown>")
|
||||
for target in entry.get("expand", []):
|
||||
if target not in ids:
|
||||
messages.append(
|
||||
_sc0004(
|
||||
title="Expansion target does not exist",
|
||||
description=(
|
||||
f"Entry '{entry_id}' expands to unknown canonical id '{target}'."
|
||||
),
|
||||
entry_id=entry_id,
|
||||
target=target,
|
||||
)
|
||||
)
|
||||
|
||||
# Cycles (only over targets that exist), reported once per distinct cycle.
|
||||
reported: set[frozenset[str]] = set()
|
||||
for cycle in _find_cycles(graph, ids):
|
||||
key = frozenset(cycle)
|
||||
if key in reported:
|
||||
continue
|
||||
reported.add(key)
|
||||
messages.append(
|
||||
Message(
|
||||
code="SC0003",
|
||||
severity=Severity.ERROR,
|
||||
title="Circular Expansion",
|
||||
description=f"Expansion cycle detected: {' -> '.join(cycle)}.",
|
||||
context={"id": cycle[0], "cycle": cycle},
|
||||
)
|
||||
)
|
||||
return messages
|
||||
|
||||
|
||||
def _find_cycles(graph: Mapping[str, list[str]], ids: set[str]) -> list[list[str]]:
|
||||
"""Return the ids of each expansion cycle, deterministically."""
|
||||
cycles: list[list[str]] = []
|
||||
on_stack: dict[str, int] = {}
|
||||
stack: list[str] = []
|
||||
visited: set[str] = set()
|
||||
|
||||
def dfs(node: str) -> None:
|
||||
visited.add(node)
|
||||
on_stack[node] = len(stack)
|
||||
stack.append(node)
|
||||
for target in graph.get(node, []):
|
||||
if target not in ids:
|
||||
continue
|
||||
if target in on_stack:
|
||||
cycles.append(stack[on_stack[target] :])
|
||||
elif target not in visited:
|
||||
dfs(target)
|
||||
stack.pop()
|
||||
del on_stack[node]
|
||||
|
||||
for node in graph: # insertion order -> deterministic
|
||||
if node not in visited:
|
||||
dfs(node)
|
||||
return cycles
|
||||
|
||||
|
||||
def validate_knowledge_base(entries: Iterable[Mapping[str, Any]]) -> list[Message]:
|
||||
"""Validate a whole Knowledge Base and return every problem found.
|
||||
|
||||
Args:
|
||||
entries: All Knowledge Base entries (parsed JSON), across every file.
|
||||
|
||||
Returns:
|
||||
A deterministically-ordered list of :class:`Message`; empty when the
|
||||
Knowledge Base is valid.
|
||||
"""
|
||||
entries = list(entries)
|
||||
messages: list[Message] = []
|
||||
|
||||
for entry in entries:
|
||||
entry_id = entry.get("id", "<unknown>")
|
||||
messages.extend(_schema_messages(entry, entry_id))
|
||||
messages.extend(_category_messages(entry, entry_id))
|
||||
|
||||
messages.extend(_duplicate_id_messages(entries))
|
||||
messages.extend(_alias_messages(entries))
|
||||
messages.extend(_expansion_messages(entries))
|
||||
return messages
|
||||
@@ -0,0 +1,79 @@
|
||||
"""CLI to validate a Knowledge Base directory (issue #9).
|
||||
|
||||
Loads every ``*.json`` file under the given directory (each a JSON array of
|
||||
entries), validates the whole Knowledge Base, and writes the findings as JSON to
|
||||
stdout. Exit code is 0 when the Knowledge Base is valid, 1 otherwise.
|
||||
|
||||
Usage:
|
||||
python scripts/validate_knowledge_base.py knowledge_base/
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Allow running directly (python scripts/validate_knowledge_base.py) without install.
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from compiler.common.kb_validation import validate_knowledge_base # noqa: E402
|
||||
from compiler.common.result import Message, Severity # noqa: E402
|
||||
|
||||
|
||||
def load_entries(directory: Path) -> tuple[list[dict], list[Message]]:
|
||||
"""Load all entries from ``*.json`` files; return (entries, load_errors)."""
|
||||
entries: list[dict] = []
|
||||
errors: list[Message] = []
|
||||
for path in sorted(directory.glob("*.json")):
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
except json.JSONDecodeError as exc:
|
||||
errors.append(
|
||||
Message(
|
||||
code="SC0004",
|
||||
severity=Severity.FATAL,
|
||||
title="Invalid JSON",
|
||||
description=f"{path.name}: {exc}",
|
||||
context={"id": path.name},
|
||||
)
|
||||
)
|
||||
continue
|
||||
if isinstance(data, list):
|
||||
entries.extend(data)
|
||||
else:
|
||||
errors.append(
|
||||
Message(
|
||||
code="SC0004",
|
||||
severity=Severity.FATAL,
|
||||
title="Invalid Knowledge Base file",
|
||||
description=f"{path.name}: expected a JSON array of entries.",
|
||||
context={"id": path.name},
|
||||
)
|
||||
)
|
||||
return entries, errors
|
||||
|
||||
|
||||
def main(argv: list[str]) -> int:
|
||||
if len(argv) != 1:
|
||||
sys.stderr.write("usage: validate_knowledge_base.py <knowledge_base_dir>\n")
|
||||
return 2
|
||||
directory = Path(argv[0])
|
||||
if not directory.is_dir():
|
||||
sys.stderr.write(f"not a directory: {directory}\n")
|
||||
return 2
|
||||
|
||||
entries, findings = load_entries(directory)
|
||||
findings = findings + validate_knowledge_base(entries)
|
||||
|
||||
report = {
|
||||
"valid": not findings,
|
||||
"entry_count": len(entries),
|
||||
"findings": [m.to_json() for m in findings],
|
||||
}
|
||||
sys.stdout.write(json.dumps(report, indent=2, sort_keys=True) + "\n")
|
||||
return 0 if not findings else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main(sys.argv[1:]))
|
||||
@@ -0,0 +1,128 @@
|
||||
"""Tests for the Knowledge Base validation tooling (issue #9, §15.10, §26.4)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from compiler.common.kb_validation import validate_knowledge_base
|
||||
from compiler.common.result import Severity
|
||||
|
||||
|
||||
def entry(id_: str, **overrides) -> dict:
|
||||
base = {"id": id_, "tags": [id_.replace("_", " ")], "category": "hair"}
|
||||
base.update(overrides)
|
||||
return base
|
||||
|
||||
|
||||
def codes(messages) -> list[str]:
|
||||
return [m.code for m in messages]
|
||||
|
||||
|
||||
# --- passing ---------------------------------------------------------------
|
||||
|
||||
|
||||
def test_valid_knowledge_base_passes_cleanly() -> None:
|
||||
entries = [
|
||||
entry("long_hair", aliases=["lengthy hair"]),
|
||||
entry("twin_braids", aliases=["double braids"], expand=["long_hair"]),
|
||||
]
|
||||
assert validate_knowledge_base(entries) == []
|
||||
|
||||
|
||||
def test_empty_knowledge_base_passes() -> None:
|
||||
assert validate_knowledge_base([]) == []
|
||||
|
||||
|
||||
# --- duplicate canonical id (SC0005) ---------------------------------------
|
||||
|
||||
|
||||
def test_duplicate_id_reports_sc0005() -> None:
|
||||
messages = validate_knowledge_base([entry("long_hair"), entry("long_hair")])
|
||||
assert "SC0005" in codes(messages)
|
||||
dup = next(m for m in messages if m.code == "SC0005")
|
||||
assert dup.context["id"] == "long_hair"
|
||||
|
||||
|
||||
# --- invalid category (SC0008) ---------------------------------------------
|
||||
|
||||
|
||||
def test_invalid_category_reports_sc0008() -> None:
|
||||
messages = validate_knowledge_base([entry("x", category="hairstyle")])
|
||||
assert "SC0008" in codes(messages)
|
||||
|
||||
|
||||
def test_valid_category_accepted() -> None:
|
||||
assert validate_knowledge_base([entry("x", category="eyes")]) == []
|
||||
|
||||
|
||||
# --- circular expansion (SC0003) -------------------------------------------
|
||||
|
||||
|
||||
def test_circular_expansion_reports_sc0003() -> None:
|
||||
entries = [entry("a", expand=["b"]), entry("b", expand=["a"])]
|
||||
messages = validate_knowledge_base(entries)
|
||||
assert "SC0003" in codes(messages)
|
||||
|
||||
|
||||
def test_self_expansion_is_circular() -> None:
|
||||
messages = validate_knowledge_base([entry("a", expand=["a"])])
|
||||
assert "SC0003" in codes(messages)
|
||||
|
||||
|
||||
# --- other KB-structural failures (SC0004) ---------------------------------
|
||||
|
||||
|
||||
def test_duplicate_alias_reports_sc0004() -> None:
|
||||
entries = [entry("a", aliases=["shared"]), entry("b", aliases=["shared"])]
|
||||
assert "SC0004" in codes(validate_knowledge_base(entries))
|
||||
|
||||
|
||||
def test_alias_colliding_with_id_reports_sc0004() -> None:
|
||||
# "long_hair" is a canonical id; using it as an alias would create a chain.
|
||||
entries = [entry("long_hair"), entry("b", aliases=["long_hair"])]
|
||||
assert "SC0004" in codes(validate_knowledge_base(entries))
|
||||
|
||||
|
||||
def test_missing_expansion_target_reports_sc0004() -> None:
|
||||
messages = validate_knowledge_base([entry("a", expand=["does_not_exist"])])
|
||||
assert "SC0004" in codes(messages)
|
||||
|
||||
|
||||
def test_empty_tags_reported() -> None:
|
||||
messages = validate_knowledge_base([entry("a", tags=[])])
|
||||
assert messages != []
|
||||
|
||||
|
||||
def test_unknown_field_reported() -> None:
|
||||
messages = validate_knowledge_base([entry("a", colour="brown")])
|
||||
assert messages != []
|
||||
|
||||
|
||||
# --- reporting behaviour ---------------------------------------------------
|
||||
|
||||
|
||||
def test_reports_all_problems_not_just_first() -> None:
|
||||
entries = [
|
||||
entry("dup"),
|
||||
entry("dup"), # duplicate id (SC0005)
|
||||
entry("bad", category="nope"), # invalid category (SC0008)
|
||||
]
|
||||
found = set(codes(validate_knowledge_base(entries)))
|
||||
assert {"SC0005", "SC0008"}.issubset(found)
|
||||
|
||||
|
||||
def test_every_message_has_code_id_and_description() -> None:
|
||||
messages = validate_knowledge_base([entry("a", category="nope", expand=["missing"])])
|
||||
assert messages
|
||||
for m in messages:
|
||||
assert m.code
|
||||
assert m.description
|
||||
assert "id" in m.context
|
||||
|
||||
|
||||
def test_findings_are_deterministic() -> None:
|
||||
entries = [entry("dup"), entry("dup"), entry("bad", category="nope")]
|
||||
assert codes(validate_knowledge_base(entries)) == codes(validate_knowledge_base(entries))
|
||||
|
||||
|
||||
def test_findings_carry_error_or_fatal_severity() -> None:
|
||||
messages = validate_knowledge_base([entry("x", category="nope")])
|
||||
assert all(m.severity in (Severity.ERROR, Severity.FATAL) for m in messages)
|
||||
@@ -0,0 +1,49 @@
|
||||
"""Tests for the Knowledge Base validation CLI (issue #9)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
SCRIPT = REPO_ROOT / "scripts" / "validate_knowledge_base.py"
|
||||
|
||||
|
||||
def _load_cli():
|
||||
spec = importlib.util.spec_from_file_location("kb_cli", SCRIPT)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module) # type: ignore[union-attr]
|
||||
return module
|
||||
|
||||
|
||||
def test_cli_passes_on_valid_kb(tmp_path, capsys) -> None:
|
||||
(tmp_path / "hair.json").write_text(
|
||||
json.dumps([{"id": "long_hair", "tags": ["long hair"], "category": "hair"}]),
|
||||
encoding="utf-8",
|
||||
)
|
||||
exit_code = _load_cli().main([str(tmp_path)])
|
||||
report = json.loads(capsys.readouterr().out)
|
||||
assert exit_code == 0
|
||||
assert report["valid"] is True
|
||||
assert report["entry_count"] == 1
|
||||
|
||||
|
||||
def test_cli_fails_on_invalid_kb(tmp_path, capsys) -> None:
|
||||
(tmp_path / "bad.json").write_text(
|
||||
json.dumps([{"id": "x", "tags": ["x"], "category": "nope"}]),
|
||||
encoding="utf-8",
|
||||
)
|
||||
exit_code = _load_cli().main([str(tmp_path)])
|
||||
report = json.loads(capsys.readouterr().out)
|
||||
assert exit_code == 1
|
||||
assert report["valid"] is False
|
||||
assert any(f["code"] == "SC0008" for f in report["findings"])
|
||||
|
||||
|
||||
def test_cli_reports_invalid_json(tmp_path, capsys) -> None:
|
||||
(tmp_path / "broken.json").write_text("{not json", encoding="utf-8")
|
||||
exit_code = _load_cli().main([str(tmp_path)])
|
||||
report = json.loads(capsys.readouterr().out)
|
||||
assert exit_code == 1
|
||||
assert any(f["code"] == "SC0004" for f in report["findings"])
|
||||
Reference in New Issue
Block a user