diff --git a/src/lokf/cli.py b/src/lokf/cli.py index 260acab..892d9a8 100644 --- a/src/lokf/cli.py +++ b/src/lokf/cli.py @@ -2,6 +2,7 @@ lokf new my-kb # scaffold a knowledge base lokf convert path/to/concept.md --format ttl # markdown -> RDF + lokf validate knowledge # assemble + validate a bundle lokf query examples/acme-knowledge "SELECT ..." # SPARQL over a bundle lokf serve examples/acme-knowledge # local SPARQL endpoint + viz lokf propose examples/acme-knowledge --apply # typed relations from links @@ -94,6 +95,62 @@ def convert( typer.echo(data, nl=False) +# --------------------------------------------------------------------------- +# validate +# --------------------------------------------------------------------------- +@app.command() +def validate( + bundle_dir: Path = typer.Argument( + Path("knowledge"), exists=True, file_okay=False, + help="A LOKF bundle directory to assemble and validate.", + ), + schema: Optional[Path] = typer.Option( + None, "--schema", "-s", + help="Schema file (default: a local lokf.yaml checkout, else the copy " + "packaged with lokf).", + ), +) -> None: + """Assemble a bundle and validate it against the LOKF schema. + + Works on any bundle directory. The schema is resolved from ``--schema`` if + passed, otherwise from a ``lokf.yaml`` found in the current directory or an + ancestor, otherwise from the copy shipped inside the installed ``lokf`` + package - so a bundle needs no schema file of its own to be validated. + """ + import os + import subprocess + import tempfile + + from lokf.model import load_bundle + from lokf.schema import schema_path + + try: + sch = schema_path(schema) + except FileNotFoundError as exc: + _err(str(exc)) + raise typer.Exit(1) + + bundle = load_bundle(bundle_dir) + doc = dict(bundle.meta) + doc["concepts"] = bundle.docs() + + fd, tmp = tempfile.mkstemp(suffix=".bundle.json") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(doc, f) + proc = subprocess.run( + ["linkml-validate", "-s", str(sch), "-C", "KnowledgeBundle", tmp] + ) + finally: + os.remove(tmp) + if proc.returncode != 0: + raise typer.Exit(proc.returncode) + typer.echo( + f"OK — {len(bundle.concepts)} concepts in {bundle_dir} validate " + "against KnowledgeBundle." + ) + + # --------------------------------------------------------------------------- # tables # --------------------------------------------------------------------------- diff --git a/src/lokf/schema.py b/src/lokf/schema.py index 953a45d..764c476 100644 --- a/src/lokf/schema.py +++ b/src/lokf/schema.py @@ -67,6 +67,17 @@ def _load_context(resolved: str) -> dict: return json.loads(pathlib.Path(resolved).read_text(encoding="utf-8"))["@context"] +def schema_path(path: str | pathlib.Path | None = None) -> pathlib.Path: + """Resolve the LOKF schema file (``lokf.yaml``) to a filesystem path. + + Resolution order (same as every other reader here): an explicit *path*, a + ``lokf.yaml`` found in the current directory or an ancestor, the the copy + shipped inside the installed ``lokf`` package - so a bundle can be + validated without a schema file of its own. + """ + return _resolve(_SCHEMA_NAME, path) + + def load_schema(path: str | pathlib.Path | None = None) -> dict: """Return the LOKF LinkML schema as a plain dict (cached per file).""" return _load_schema(str(_resolve(_SCHEMA_NAME, path))) diff --git a/tests/test_cli.py b/tests/test_cli.py index 47907af..65f9fd2 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -65,6 +65,20 @@ def test_convert_output_writes_file(tmp_path): assert "lokf:Metric" in text +# -- validate --------------------------------------------------------------- +def test_validate_reference_bundle_ok(): + """validate assembles and validates against KnowledgeBundle.""" + result = runner.invoke(app, ["validate", str(BUNDLE)]) + assert result.exit_code == 0 + assert "validate against KnowledgeBundle" in result.stdout + + +def test_validate_missing_bundle_nonzero_exit(): + """A non-existent bundle directory is rejected before validation.""" + result = runner.invoke(app, ["validate", "does-not-exist"]) + assert result.exit_code != 0 + + # -- query ------------------------------------------------------------------ def test_query_select_table_contains_wau(): """query