diff --git a/docs/registry.md b/docs/registry.md new file mode 100644 index 00000000..0e2ba239 --- /dev/null +++ b/docs/registry.md @@ -0,0 +1,249 @@ +# Spec Registry + +## Why this exists + +When a VSS-based model grows and gets consumed by multiple tools, teams, and systems, a common problem emerges: how do you refer to a concept — a signal, a branch, a struct — in a way that is stable across time and across different modeling projects? + +The fully qualified name (FQN) like `Vehicle.Cabin.Seat.Row1.DriverSide.Position` is human-readable and useful during development, but it is fragile as a long-term identifier. If a node gets renamed, restructured, or merged into another branch, any system that stored the FQN as a reference now has a broken link. There is no way to know that `Vehicle.Cabin.Seat.Row1.DriverSide.Position` and its successor `Vehicle.Cabin.Seat.Row1.Left.Position` refer to the same real-world concept. + +The spec registry solves this by assigning a **stable, short numeric ID** to every node the first time it appears in the model. That ID never changes, even if the node is renamed or restructured. Systems downstream can store `covss:142` instead of the full FQN and always resolve it back to the current name via the registry. + +This pattern is widely used in ontology engineering (RDF/OWL), standards bodies (ISO, W3C), and linked data systems. The registry tool brings the same discipline to VSS-based modeling projects. + +--- + +## Important note: IDs are not deterministic + +The IDs assigned by this tool depend on the **order in which nodes first appear** in the registry and the **current state of the modeling project at the time of the first sync**. Two projects that independently sync the same vspec will produce different numeric IDs for the same FQNs if they use different namespace prefixes or run the sync at different points in the model's history. + +This is by design. Each project owns its own namespace and is responsible for when it first mints IDs. The IDs are only meaningful within a given namespace. + +As a direct consequence, **the registry files must be treated as governed artifacts**: + +- They should be committed to version control and never edited manually. +- Write access in CI should be restricted to tagged release pipelines only. +- The `vspec registry validate` command should run as a pre-commit hook and in every CI pipeline to catch any corruption early. + +If the registry is corrupted or lost, previously published IDs cannot be reconstructed automatically. Treat the `registry.csv` file with the same care as a database migration history or a package lock file. + +--- + +## Concepts + +### Namespace + +A namespace is a URI that scopes a set of IDs. Each project declares exactly one namespace it **owns** — the namespace under which it mints new IDs. It may also declare additional namespaces it **imports** — external namespaces it references but does not mint into. + +### Prefix + +A short label used as a human-readable abbreviation for a namespace URI. For example, `covss` is the prefix for `https://www.covesa.global/model/vss#`. IDs are written in CURIE form: `covss:42`. + +### Composed ID + +An ID in the form `:`, for example `covss:0`, `covss:1`, `bmwvss:0`. The integer is assigned incrementally starting from 0, independently per prefix. + +### Registry + +A CSV file that acts as the append-only ledger mapping composed IDs to FQNs. Each row has a SHA-256 hash of its immutable fields, making manual edits detectable. + +--- + +## File formats + +### `namespaces.yaml` + +Declares the namespace this project owns and any external namespaces it references. + +```yaml +namespace: + prefix: covss + uri: "https://www.covesa.global/model/vss#" + description: "COVESA Vehicle Signal Specification" # optional + +imports: # optional + myns: + uri: "https://www.example.org/myModel#" + description: "Private model extension" +``` + +Rules: +- `namespace` is required. It must have a `prefix` and a `uri`. +- `prefix` must match `^[a-z][a-z0-9]*$` — lowercase alphanumeric, no hyphens. +- `imports` is optional. A project cannot import its own prefix. +- All URIs should end with `#` or `/` so that composed IDs form valid full URIs when concatenated. + +### `registry.csv` + +The ledger file. Managed entirely by the tool — do not edit manually. + +``` +composed_id,fqn,int_id,status,row_hash +covss:0,Vehicle,0,active,e3b0c44298fc1c149afb... +covss:1,Vehicle.Cabin,1,active,a87ff679a2f3e71d9181... +covss:2,Vehicle.Cabin.Seat,2,active,1679091c5a880faf6fb... +``` + +Column meanings: +- `composed_id` — the stable identifier, e.g. `covss:0` +- `fqn` — the fully qualified node name at the time the ID was minted +- `int_id` — the numeric part of `composed_id`, per-prefix counter +- `status` — `active` or `deprecated`. If a concept is not present in a future release of the model, then it is because it was deleted from the model. However, the id is persisted forever in the registry. +- `row_hash` — SHA-256 of `"{composed_id}|{fqn}"`, used to detect tampering + +--- + +## Commands + +### `vspec registry sync` + +Reads the current vspec model, finds any FQNs not yet in the registry, and mints new IDs for them. Safe to run repeatedly — existing rows are never modified. + +```bash +vspec registry sync \ + -s spec.vspec \ + -u units.yaml \ + -q quantities.yaml \ + -n namespaces.yaml \ + -r registry.csv +``` + +With include directories and type files (for models split across multiple files): + +```bash +vspec registry sync \ + -s spec.vspec \ + -I includes/ \ + -t types.vspec \ + -u units.yaml \ + -q quantities.yaml \ + -n namespaces.yaml \ + -r registry.csv +``` + +With optional JSON-LD sidecar export: + +```bash +vspec registry sync \ + -s spec.vspec \ + -u units.yaml \ + -q quantities.yaml \ + -n namespaces.yaml \ + -r registry.csv \ + --export-jsonld registry.jsonld +``` + +The JSON-LD sidecar is a derived artifact — it is regenerated from the CSV on every run and is safe to discard and regenerate at any time. + +### `vspec registry validate` + +Checks both files for integrity without writing anything. Exits with a non-zero status code if any check fails. Suitable for use as a pre-commit hook or CI step. + +```bash +vspec registry validate \ + -n namespaces.yaml \ + -r registry.csv +``` + +--- + +## Scenario: COVESA mints the base VSS model + +A COVESA project maintains the canonical VSS model. On every release they sync the registry to assign stable IDs to all nodes: + +**`namespaces.yaml`** (COVESA project): +```yaml +namespace: + prefix: covss + uri: "https://www.covesa.global/model/vss#" + description: "COVESA Vehicle Signal Specification" +``` + +```bash +vspec registry sync \ + -s Vehicle.vspec \ + -u units.yaml \ + -q quantities.yaml \ + -n namespaces.yaml \ + -r registry.csv +``` + +After the first sync, `registry.csv` contains one row per VSS node: + +``` +composed_id,fqn,int_id,status,row_hash +covss:0,Vehicle,0,active,e3b0c44298fc... +covss:1,Vehicle.ADAS,1,active,a87ff679a2... +covss:2,Vehicle.Cabin,2,active,1679091c5a... +... +``` + +This file is committed to the COVESA repo and published alongside the vspec release. + +--- + +## Scenario: A private project extends the model using an overlay + +A private modeling project builds on top of the COVESA VSS model by overlaying additional signals. They want to: +1. Reuse the stable COVESA IDs for all COVESA nodes. +2. Mint their own IDs under a different prefix for project-specific nodes. + +**Step 1**: The project copies the COVESA `registry.csv` as its starting point. This file already contains all the `covss:N` rows. + +**Step 2**: The project creates its own `namespaces.yaml`: + +```yaml +namespace: + prefix: myns + uri: "https://www.example.org/myModel#" + description: "Private VSS extension" + +imports: + covss: + uri: "https://www.covesa.global/model/vss#" + description: "COVESA VSS — imported reference" +``` + +**Step 3**: They run sync with their overlay applied: + +```bash +vspec registry sync \ + -s Vehicle.vspec \ + -l my_extension.vspec \ + -u units.yaml \ + -q quantities.yaml \ + -n namespaces.yaml \ + -r registry.csv +``` + +Because all COVESA FQNs are already in the registry (with `covss:N` IDs), the tool only mints IDs for nodes that are genuinely new — those introduced by the private overlay: + +``` +composed_id,fqn,int_id,status,row_hash +covss:0,Vehicle,0,active,e3b0c44298fc... +covss:1,Vehicle.ADAS,1,active,a87ff679a2... +... +myns:0,Vehicle.MyProject.FeatureA,0,active,7f8f4af0c8... +myns:1,Vehicle.MyProject.FeatureA.Status,1,active,9bf31c7ff0... +``` + +The result is a single registry file that: +- Preserves all COVESA stable IDs unchanged. +- Adds project-specific IDs starting from `myns:0`. +- Makes it unambiguous which concepts belong to which namespace at a glance. + +The optional JSON-LD sidecar makes both namespaces resolvable to full URIs for any toolchain that consumes linked data: + +```json +{ + "@context": { + "covss": "https://www.covesa.global/model/vss#", + "myns": "https://www.example.org/myModel#" + }, + "@graph": [ + {"@id": "covss:0", "fqn": "Vehicle", "status": "active"}, + {"@id": "covss:1", "fqn": "Vehicle.ADAS", "status": "active"}, + {"@id": "myns:0", "fqn": "Vehicle.MyProject.FeatureA", "status": "active"}, + {"@id": "myns:1", "fqn": "Vehicle.MyProject.FeatureA.Status", "status": "active"} + ] +} +``` diff --git a/src/vss_tools/cli.py b/src/vss_tools/cli.py index de84d00d..4ab6f455 100644 --- a/src/vss_tools/cli.py +++ b/src/vss_tools/cli.py @@ -61,3 +61,19 @@ def export(ctx: click.Context): """ if ctx.invoked_subcommand is None: click.echo(ctx.get_help()) + + +@cli.group( + cls=LazyGroup, + lazy_subcommands={ + "sync": "vss_tools.registry:sync_cli", + "validate": "vss_tools.registry:validate_cli", + }, +) +@click.pass_context +def registry(ctx: click.Context): + """ + Manage the stable identifier registry for a VSS modeling project. + """ + if ctx.invoked_subcommand is None: + click.echo(ctx.get_help()) diff --git a/src/vss_tools/registry.py b/src/vss_tools/registry.py new file mode 100644 index 00000000..22ec3ca9 --- /dev/null +++ b/src/vss_tools/registry.py @@ -0,0 +1,160 @@ +# Copyright (c) 2025 Contributors to COVESA +# +# This program and the accompanying materials are made available under the +# terms of the Mozilla Public License 2.0 which is available at +# https://www.mozilla.org/en-US/MPL/2.0/ +# +# SPDX-License-Identifier: MPL-2.0 + +"""Registry CLI commands: sync and validate.""" + +from pathlib import Path + +import rich_click as click +from anytree import PreOrderIter # type: ignore[import] + +import vss_tools.cli_options as clo +from vss_tools import log +from vss_tools.main import get_trees +from vss_tools.utils.registry_utils import ( + RegistryConfigException, + RegistryIntegrityException, + empty_registry, + load_namespaces, + load_registry, + sync_registry, + to_jsonld, + validate_immutability, + write_jsonld, +) + +_namespaces_opt = click.option( + "--namespaces", + "-n", + required=True, + type=click.Path(exists=True, dir_okay=False, readable=True, path_type=Path), + help="YAML file declaring namespace prefixes and URIs.", +) + +_registry_opt = click.option( + "--registry", + "-r", + required=True, + type=click.Path(dir_okay=False, writable=True, path_type=Path), + help="Registry CSV file (created on first run if absent).", +) + + +@click.command() +@clo.vspec_opt +@_namespaces_opt +@_registry_opt +@click.option( + "--export-jsonld", + type=click.Path(dir_okay=False, writable=True, path_type=Path), + default=None, + help="Optional path to write a JSON-LD sidecar of the full registry.", +) +@clo.include_dirs_opt +@clo.quantities_opt +@clo.units_opt +@clo.types_opt +@clo.overlays_opt +@clo.aborts_opt +@clo.strict_opt +@clo.strict_exceptions_opt +def sync_cli( + vspec: Path, + namespaces: Path, + registry: Path, + export_jsonld: Path | None, + include_dirs: tuple[Path, ...], + quantities: tuple[Path, ...], + units: tuple[Path, ...], + types: tuple[Path, ...], + overlays: tuple[Path, ...], + aborts: tuple[str, ...], + strict: bool, + strict_exceptions: Path | None, +) -> None: + """ + Synchronize the identifier registry with the current VSS spec. + + Mints stable IDs for any FQNs not yet in the registry under the prefix + declared in the namespaces file. Safe to run repeatedly — existing IDs + are never changed or removed. + """ + try: + ns_config = load_namespaces(namespaces) + except RegistryConfigException as e: + raise click.ClickException(str(e)) + + prefix = ns_config.namespace.prefix + + if registry.exists(): + try: + existing_df = load_registry(registry) + except RegistryIntegrityException as e: + raise click.ClickException(f"Registry integrity check failed: {e}") + else: + log.info(f"No registry found at {registry} — starting fresh.") + existing_df = empty_registry() + + tree, datatype_tree = get_trees( + vspec=vspec, + include_dirs=include_dirs, + aborts=aborts, + strict=strict, + extended_attributes=(), + quantities=quantities, + units=units, + types=types, + overlays=overlays, + ) + + fqns: list[str] = sorted(node.get_fqn() for node in PreOrderIter(tree)) + if datatype_tree: + fqns = sorted(fqns + [node.get_fqn() for node in PreOrderIter(datatype_tree)]) + + updated_df, minted = sync_registry(existing_df, fqns, prefix) + + try: + validate_immutability(existing_df, updated_df) + except RegistryIntegrityException as e: + raise click.ClickException(f"Immutability violation: {e}") + + updated_df.to_csv(registry, index=False) + log.info(f"Registry written to {registry} ({minted} new IDs minted under '{prefix}:').") + + if export_jsonld: + write_jsonld(to_jsonld(updated_df, ns_config), export_jsonld) + log.info(f"JSON-LD sidecar written to {export_jsonld}.") + + +@click.command() +@_namespaces_opt +@_registry_opt +def validate_cli(namespaces: Path, registry: Path) -> None: + """ + Validate the integrity of the registry and namespaces files. + + Checks schema validity and per-row hash integrity without writing anything. + Exits with a non-zero status code if any check fails. + """ + try: + ns_config = load_namespaces(namespaces) + owned = ns_config.namespace + imports = ns_config.imports + log.info(f"Namespaces OK — owned: '{owned.prefix}' ({owned.uri})") + if imports: + log.info(f" Imports: {', '.join(sorted(imports))}") + except RegistryConfigException as e: + raise click.ClickException(str(e)) + + try: + df = load_registry(registry) + log.info( + f"Registry OK — {len(df)} row(s) across {df['composed_id'].str.split(':').str[0].nunique()} prefix(es)." + ) + except RegistryIntegrityException as e: + raise click.ClickException(f"Registry integrity check failed: {e}") diff --git a/src/vss_tools/utils/registry_utils.py b/src/vss_tools/utils/registry_utils.py new file mode 100644 index 00000000..4d5a29ad --- /dev/null +++ b/src/vss_tools/utils/registry_utils.py @@ -0,0 +1,207 @@ +# Copyright (c) 2025 Contributors to COVESA +# +# This program and the accompanying materials are made available under the +# terms of the Mozilla Public License 2.0 which is available at +# https://www.mozilla.org/en-US/MPL/2.0/ +# +# SPDX-License-Identifier: MPL-2.0 + +"""Utilities for the VSS spec identifier registry.""" + +from __future__ import annotations + +import hashlib +import json +import re +from pathlib import Path +from typing import Literal + +import pandas as pd +import yaml +from pydantic import BaseModel, field_validator + +REGISTRY_COLUMNS = ["composed_id", "fqn", "int_id", "status", "row_hash"] +COMPOSED_ID_RE = re.compile(r"^[a-z][a-z0-9]*:\d+$") + + +class RegistryIntegrityException(Exception): + """Indicates a hash mismatch or an attempt to mutate an immutable registry field.""" + + +class RegistryConfigException(Exception): + """Indicates an invalid namespaces configuration file.""" + + +class NamespaceEntry(BaseModel): + uri: str + description: str | None = None + + +class OwnedNamespace(NamespaceEntry): + """The single namespace this project owns and mints IDs under.""" + + prefix: str + + +class NamespacesConfig(BaseModel): + """Parsed content of a namespaces.yaml file.""" + + namespace: OwnedNamespace + imports: dict[str, NamespaceEntry] = {} + + +class RegistryRow(BaseModel): + composed_id: str + fqn: str + int_id: int + status: Literal["active", "deprecated"] + row_hash: str + + @field_validator("composed_id") + @classmethod + def valid_composed_id(cls, v: str) -> str: + if not COMPOSED_ID_RE.match(v): + raise ValueError(f"Invalid composed_id format: '{v}' (expected e.g. 'covss:0')") + return v + + +def _hash_row(composed_id: str, fqn: str) -> str: + """Returns SHA256 hex digest of the immutable fields of a registry row.""" + return hashlib.sha256(f"{composed_id}|{fqn}".encode()).hexdigest() + + +def load_namespaces(path: Path) -> NamespacesConfig: + """ + Loads and validates a namespaces YAML file. + + Expected format:: + + namespace: + prefix: eg + uri: "https://www.example.org/myModel#" + description: "Optional description" + + imports: + covss: + uri: "https://www.covesa.global/model/vss#" + """ + try: + raw = yaml.safe_load(path.read_text()) + except yaml.YAMLError as e: + raise RegistryConfigException(f"Failed to parse {path}: {e}") from e + + if not isinstance(raw, dict): + raise RegistryConfigException(f"{path} must be a YAML mapping") + + if "namespace" not in raw: + raise RegistryConfigException(f"{path} must have a 'namespace' key") + + try: + owned = OwnedNamespace(**raw["namespace"]) + except Exception as e: + raise RegistryConfigException(f"Invalid 'namespace' entry in {path}: {e}") from e + + imports: dict[str, NamespaceEntry] = {} + for prefix, entry in (raw.get("imports") or {}).items(): + if not isinstance(entry, dict): + raise RegistryConfigException(f"Import entry for '{prefix}' must be a mapping") + try: + imports[prefix] = NamespaceEntry(**entry) + except Exception as e: + raise RegistryConfigException(f"Invalid import entry for '{prefix}': {e}") from e + + if owned.prefix in imports: + raise RegistryConfigException( + f"Owned prefix '{owned.prefix}' also appears in 'imports' — a project cannot import its own namespace" + ) + + return NamespacesConfig(namespace=owned, imports=imports) + + +def empty_registry() -> pd.DataFrame: + """Returns an empty registry DataFrame with the correct column schema.""" + return pd.DataFrame(columns=REGISTRY_COLUMNS) + + +def load_registry(path: Path) -> pd.DataFrame: + """ + Loads a registry CSV and validates schema and row hashes. + Raises RegistryIntegrityException on any hash mismatch or schema error. + """ + df = pd.read_csv(path, dtype={"composed_id": str, "fqn": str, "int_id": int, "status": str, "row_hash": str}) + + for _, row in df.iterrows(): + try: + RegistryRow(**row.to_dict()) + except Exception as e: + raise RegistryIntegrityException(f"Schema validation failed for row {row.to_dict()}: {e}") from e + + expected = _hash_row(str(row["composed_id"]), str(row["fqn"])) + if row["row_hash"] != expected: + raise RegistryIntegrityException( + f"Hash mismatch for '{row['composed_id']}' — registry may have been manually edited" + ) + + return df + + +def sync_registry(df: pd.DataFrame, fqns: list[str], prefix: str) -> tuple[pd.DataFrame, int]: + """ + Mints new IDs for FQNs not already present in the registry under any prefix. + IDs for the given prefix start at 0 and increment per entry in that namespace. + Returns (updated_df, number_of_newly_minted_ids). + """ + existing_fqns: set[str] = set(df["fqn"]) if not df.empty else set() + new_fqns = [f for f in fqns if f not in existing_fqns] + + if not new_fqns: + return df, 0 + + prefix_mask = df["composed_id"].str.startswith(f"{prefix}:") if not df.empty else pd.Series([], dtype=bool) + next_int_id = int(prefix_mask.sum()) + + new_rows: list[dict] = [] + for i, fqn in enumerate(new_fqns): + int_id = next_int_id + i + composed_id = f"{prefix}:{int_id}" + new_rows.append( + { + "composed_id": composed_id, + "fqn": fqn, + "int_id": int_id, + "status": "active", + "row_hash": _hash_row(composed_id, fqn), + } + ) + + updated = pd.concat([df, pd.DataFrame(new_rows)], ignore_index=True) + return updated, len(new_rows) + + +def validate_immutability(old: pd.DataFrame, new: pd.DataFrame) -> None: + """ + Asserts that no pre-existing row had its immutable fields (fqn, int_id, row_hash) changed. + Raises RegistryIntegrityException on any violation. + """ + if old.empty: + return + + merged = old.merge(new, on="composed_id", suffixes=("_old", "_new")) + for col in ("fqn", "int_id", "row_hash"): + mismatched = merged[merged[f"{col}_old"] != merged[f"{col}_new"]] + if not mismatched.empty: + bad_ids = mismatched["composed_id"].tolist() + raise RegistryIntegrityException(f"Immutable field '{col}' was modified for: {bad_ids}") + + +def to_jsonld(df: pd.DataFrame, config: NamespacesConfig) -> dict: + """Converts the registry DataFrame to a JSON-LD document.""" + context: dict[str, str] = {config.namespace.prefix: config.namespace.uri} + context.update({prefix: ns.uri for prefix, ns in config.imports.items()}) + graph = df[["composed_id", "fqn", "status"]].rename(columns={"composed_id": "@id"}).to_dict(orient="records") + return {"@context": context, "@graph": graph} + + +def write_jsonld(data: dict, path: Path) -> None: + """Writes a JSON-LD document to a file.""" + path.write_text(json.dumps(data, indent=2) + "\n") diff --git a/tests/test_registry.py b/tests/test_registry.py new file mode 100644 index 00000000..91724f75 --- /dev/null +++ b/tests/test_registry.py @@ -0,0 +1,342 @@ +# Copyright (c) 2025 Contributors to COVESA +# +# This program and the accompanying materials are made available under the +# terms of the Mozilla Public License 2.0 which is available at +# https://www.mozilla.org/en-US/MPL/2.0/ +# +# SPDX-License-Identifier: MPL-2.0 + +from pathlib import Path + +import pytest +import yaml +from vss_tools.utils.registry_utils import ( + REGISTRY_COLUMNS, + NamespacesConfig, + RegistryConfigException, + RegistryIntegrityException, + _hash_row, + empty_registry, + load_namespaces, + load_registry, + sync_registry, + to_jsonld, + validate_immutability, +) + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +VSPEC = Path("tests/vspec/test_s2dm/example_seat.vspec") +QUANTITIES = Path("tests/vspec/test_s2dm/test_quantities.yaml") +UNITS = Path("tests/vspec/test_s2dm/test_units.yaml") + +SAMPLE_FQNS = [ + "Vehicle", + "Vehicle.Cabin", + "Vehicle.Cabin.DriverPosition", + "Vehicle.Cabin.Seat", + "Vehicle.VehicleIdentification", + "Vehicle.VehicleIdentification.VIN", +] + + +@pytest.fixture +def sample_fqns() -> list[str]: + """A representative sorted list of FQNs (subset matching example_seat.vspec branches).""" + return sorted(SAMPLE_FQNS) + + +@pytest.fixture +def sample_namespaces() -> dict: + return { + "namespace": { + "prefix": "eg", + "uri": "https://www.example.org/myModel#", + "description": "Example overlay model", + }, + "imports": { + "covss": {"uri": "https://www.covesa.global/model/vss#", "description": "COVESA VSS"}, + }, + } + + +# --------------------------------------------------------------------------- +# Unit tests — pure functions +# --------------------------------------------------------------------------- + + +class TestHashRow: + def test_deterministic(self): + assert _hash_row("covss:0", "Vehicle") == _hash_row("covss:0", "Vehicle") + + def test_different_inputs_differ(self): + assert _hash_row("covss:0", "Vehicle") != _hash_row("covss:0", "Vehicle.Cabin") + + def test_hex_64_chars(self): + h = _hash_row("covss:0", "Vehicle") + assert len(h) == 64 + assert all(c in "0123456789abcdef" for c in h) + + +class TestEmptyRegistry: + def test_columns(self): + df = empty_registry() + assert list(df.columns) == REGISTRY_COLUMNS + + def test_empty(self): + assert len(empty_registry()) == 0 + + +class TestLoadNamespaces: + def test_valid(self, tmp_path: Path, sample_namespaces: dict): + ns_file = tmp_path / "namespaces.yaml" + ns_file.write_text(yaml.dump(sample_namespaces)) + config = load_namespaces(ns_file) + assert isinstance(config, NamespacesConfig) + assert config.namespace.prefix == "eg" + assert config.namespace.uri == "https://www.example.org/myModel#" + assert "covss" in config.imports + assert config.imports["covss"].uri == "https://www.covesa.global/model/vss#" + + def test_no_imports_is_valid(self, tmp_path: Path): + ns_file = tmp_path / "namespaces.yaml" + ns_file.write_text(yaml.dump({"namespace": {"prefix": "eg", "uri": "https://example.org/"}})) + config = load_namespaces(ns_file) + assert config.imports == {} + + def test_missing_namespace_key_raises(self, tmp_path: Path): + ns_file = tmp_path / "namespaces.yaml" + ns_file.write_text(yaml.dump({"imports": {"covss": {"uri": "https://covesa.global/"}}})) + with pytest.raises(RegistryConfigException): + load_namespaces(ns_file) + + def test_missing_uri_raises(self, tmp_path: Path): + ns_file = tmp_path / "namespaces.yaml" + ns_file.write_text(yaml.dump({"namespace": {"prefix": "eg", "description": "no uri"}})) + with pytest.raises(RegistryConfigException): + load_namespaces(ns_file) + + def test_missing_prefix_raises(self, tmp_path: Path): + ns_file = tmp_path / "namespaces.yaml" + ns_file.write_text(yaml.dump({"namespace": {"uri": "https://example.org/"}})) + with pytest.raises(RegistryConfigException): + load_namespaces(ns_file) + + def test_owned_prefix_in_imports_raises(self, tmp_path: Path): + ns_file = tmp_path / "namespaces.yaml" + ns_file.write_text( + yaml.dump( + { + "namespace": {"prefix": "eg", "uri": "https://example.org/"}, + "imports": {"eg": {"uri": "https://example.org/"}}, + } + ) + ) + with pytest.raises(RegistryConfigException): + load_namespaces(ns_file) + + def test_not_a_mapping_raises(self, tmp_path: Path): + ns_file = tmp_path / "namespaces.yaml" + ns_file.write_text("- item1\n- item2\n") + with pytest.raises(RegistryConfigException): + load_namespaces(ns_file) + + def test_invalid_yaml_raises(self, tmp_path: Path): + ns_file = tmp_path / "namespaces.yaml" + ns_file.write_text("key: [unclosed") + with pytest.raises(RegistryConfigException): + load_namespaces(ns_file) + + +class TestSyncRegistry: + def test_first_run_mints_all(self, sample_fqns: list[str]): + df = empty_registry() + updated, minted = sync_registry(df, sample_fqns, "covss") + assert minted == len(sample_fqns) + assert len(updated) == len(sample_fqns) + + def test_ids_start_at_zero(self, sample_fqns: list[str]): + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "covss") + assert updated["int_id"].min() == 0 + + def test_ids_are_contiguous(self, sample_fqns: list[str]): + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "covss") + ids = sorted(updated[updated["composed_id"].str.startswith("covss:")]["int_id"].tolist()) + assert ids == list(range(len(sample_fqns))) + + def test_idempotent(self, sample_fqns: list[str]): + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "covss") + updated2, minted2 = sync_registry(updated, sample_fqns, "covss") + assert minted2 == 0 + assert len(updated2) == len(sample_fqns) + + def test_new_fqn_mints_one(self, sample_fqns: list[str]): + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "covss") + new_fqns = sample_fqns + ["Vehicle.NewBranch"] + updated2, minted2 = sync_registry(updated, sorted(new_fqns), "covss") + assert minted2 == 1 + assert len(updated2) == len(sample_fqns) + 1 + + def test_composed_id_format(self, sample_fqns: list[str]): + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "covss") + for cid in updated["composed_id"]: + prefix, int_part = cid.split(":", 1) + assert prefix == "covss" + assert int_part.isdigit() + + def test_prefix_isolation(self, sample_fqns: list[str]): + """Each namespace prefix starts from 0 independently.""" + df = empty_registry() + df, _ = sync_registry(df, sample_fqns, "covss") + + eg_fqns = ["Other.Branch", "Other.Branch.Leaf"] + df, minted = sync_registry(df, eg_fqns, "eg") + assert minted == 2 + + eg_rows = df[df["composed_id"].str.startswith("eg:")] + assert sorted(eg_rows["int_id"].tolist()) == [0, 1] + assert eg_rows["composed_id"].tolist() == ["eg:0", "eg:1"] + + def test_row_hashes_are_correct(self, sample_fqns: list[str]): + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "covss") + for _, row in updated.iterrows(): + assert row["row_hash"] == _hash_row(row["composed_id"], row["fqn"]) + + +class TestValidateImmutability: + def test_passes_on_equal(self, sample_fqns: list[str]): + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "covss") + validate_immutability(updated, updated.copy()) # should not raise + + def test_passes_on_empty_old(self, sample_fqns: list[str]): + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "covss") + validate_immutability(empty_registry(), updated) # should not raise + + def test_raises_on_fqn_change(self, sample_fqns: list[str]): + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "covss") + corrupted = updated.copy() + corrupted.loc[0, "fqn"] = "Tampered.FQN" + with pytest.raises(RegistryIntegrityException): + validate_immutability(updated, corrupted) + + def test_raises_on_int_id_change(self, sample_fqns: list[str]): + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "covss") + corrupted = updated.copy() + corrupted.loc[0, "int_id"] = 9999 + with pytest.raises(RegistryIntegrityException): + validate_immutability(updated, corrupted) + + +class TestLoadRegistry: + def test_roundtrip(self, tmp_path: Path, sample_fqns: list[str]): + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "covss") + csv_path = tmp_path / "registry.csv" + updated.to_csv(csv_path, index=False) + + loaded = load_registry(csv_path) + assert len(loaded) == len(sample_fqns) + assert list(loaded.columns) == REGISTRY_COLUMNS + + def test_hash_mismatch_raises(self, tmp_path: Path, sample_fqns: list[str]): + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "covss") + csv_path = tmp_path / "registry.csv" + updated.to_csv(csv_path, index=False) + + # Corrupt a row's fqn after writing + lines = csv_path.read_text().splitlines() + lines[1] = lines[1].replace(sample_fqns[0], "Tampered.FQN") + csv_path.write_text("\n".join(lines) + "\n") + + with pytest.raises(RegistryIntegrityException): + load_registry(csv_path) + + +class TestToJsonLD: + def test_shape(self, sample_fqns: list[str], sample_namespaces: dict, tmp_path: Path): + ns_file = tmp_path / "namespaces.yaml" + ns_file.write_text(yaml.dump(sample_namespaces)) + config = load_namespaces(ns_file) + + df = empty_registry() + updated, _ = sync_registry(df, sample_fqns, "eg") + + doc = to_jsonld(updated, config) + assert "@context" in doc + assert "@graph" in doc + + # Owned namespace appears in context + assert "eg" in doc["@context"] + assert doc["@context"]["eg"] == "https://www.example.org/myModel#" + + # Imported namespaces also appear in context + assert "covss" in doc["@context"] + assert doc["@context"]["covss"] == "https://www.covesa.global/model/vss#" + + graph = doc["@graph"] + assert len(graph) == len(sample_fqns) + for entry in graph: + assert "@id" in entry + assert "fqn" in entry + assert "status" in entry + + +# --------------------------------------------------------------------------- +# Integration test — sync end-to-end with real vspec +# --------------------------------------------------------------------------- + + +class TestSyncEndToEnd: + def test_sync_real_vspec(self, tmp_path: Path): + """Full end-to-end: load real vspec, sync registry, reload and validate.""" + from anytree import PreOrderIter + from vss_tools.main import get_trees + + tree, _ = get_trees( + vspec=VSPEC, + include_dirs=(), + aborts=(), + strict=False, + extended_attributes=(), + quantities=(QUANTITIES,), + units=(UNITS,), + overlays=(), + expand=False, + ) + fqns = sorted(node.get_fqn() for node in PreOrderIter(tree)) + + ns_file = tmp_path / "namespaces.yaml" + ns_file.write_text( + yaml.dump( + { + "namespace": {"prefix": "covss", "uri": "https://www.covesa.global/model/vss#"}, + } + ) + ) + registry_path = tmp_path / "registry.csv" + + df, minted = sync_registry(empty_registry(), fqns, "covss") + assert minted == len(fqns) + df.to_csv(registry_path, index=False) + + # Reload and verify integrity + loaded = load_registry(registry_path) + assert len(loaded) == len(fqns) + + # Second sync is idempotent + df2, minted2 = sync_registry(loaded, fqns, "covss") + assert minted2 == 0 + assert len(df2) == len(fqns)