chore(xtest): Shared Scenario/Instance Pydantic schema in otdf-sdk-mgr

otdf-sdk-mgr/pyproject.toml

@@ -5,7 +5,9 @@ description = "SDK artifact management CLI for OpenTDF cross-client tests"
 requires-python = ">=3.11"
 dependencies = [
     "gitpython>=3.1.50",
+    "pydantic>=2.6.0",
     "rich>=13.7.0",
+    "ruamel.yaml>=0.18.0",
     "typer>=0.12.0",
 ]
 

otdf-sdk-mgr/src/otdf_sdk_mgr/schema.py

@@ -0,0 +1,227 @@
+"""Shared Pydantic models for OpenTDF scenarios and instances.
+
+Both `otdf-sdk-mgr` and `otdf-local` import from this module so the on-disk
+YAML formats (`scenarios.yaml`, `instance.yaml`) have exactly one canonical
+definition.
+"""
+
+from __future__ import annotations
+
+import sys
+from datetime import date
+from pathlib import Path
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+from ruamel.yaml import YAML
+
+API_VERSION = "opentdf.io/v1alpha1"
+
+KasMode = Literal["standard", "key_management"]
+SdkName = Literal["go", "java", "js"]
+ContainerKind = Literal["ztdf", "ztdf-ecwrap", "nano", "nano-with-policy"]
+
+
+class _StrictModel(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=False)
+
+
+class SourceRef(_StrictModel):
+    ref: str = Field(description="Git tag, branch, or SHA")
+    path: Path | None = Field(default=None, description="Optional local checkout path")
+
+
+class PlatformPin(_StrictModel):
+    """Version pin for the platform service.
+
+    `dist` references a built binary at `xtest/platform/dist/<dist>/service`
+    produced by `otdf-sdk-mgr install platform:<version>`. `source.ref` is a
+    git ref to build from on demand. `image` is reserved for forward-compat
+    once container images are published; rejected at run time today.
+    """
+
+    dist: str | None = None
+    source: SourceRef | None = None
+    image: str | None = None
+
+    @model_validator(mode="after")
+    def _exactly_one(self) -> PlatformPin:
+        set_fields = [k for k in ("dist", "source", "image") if getattr(self, k)]
+        if len(set_fields) != 1:
+            raise ValueError(
+                f"PlatformPin must set exactly one of dist|source|image (got {set_fields or 'none'})"
+            )
+        return self
+
+
+class KasPin(_StrictModel):
+    """Per-KAS-instance version + mode pin."""
+
+    dist: str | None = None
+    source: SourceRef | None = None
+    image: str | None = None
+    mode: KasMode = "standard"
+    features: dict[str, bool] = Field(default_factory=dict)
+
+    @model_validator(mode="after")
+    def _exactly_one(self) -> KasPin:
+        set_fields = [k for k in ("dist", "source", "image") if getattr(self, k)]
+        if len(set_fields) != 1:
+            raise ValueError(
+                f"KasPin must set exactly one of dist|source|image (got {set_fields or 'none'})"
+            )
+        return self
+
+
+class SdkPin(_StrictModel):
+    """SDK version pin (forwarded to otdf-sdk-mgr's existing resolve())."""
+
+    version: str
+    source: str | None = Field(
+        default=None,
+        description='For Go: "platform" to use the monorepo module path',
+    )
+
+
+class PortsConfig(_StrictModel):
+    base: int = Field(default=8080, ge=1024, le=60000)
+
+
+class Metadata(_StrictModel):
+    name: str | None = None
+    id: str | None = None
+    title: str | None = None
+    created: date | None = None
+
+
+class Fixtures(_StrictModel):
+    attributes: Path | None = None
+    policy: Path | None = None
+
+
+class Instance(_StrictModel):
+    """Standalone instance definition (one platform + N KAS).
+
+    Persisted to `tests/instances/<name>/instance.yaml`. Also embedded inside
+    Scenario to keep the "describe a bug-repro environment" entry point a
+    single file.
+    """
+
+    apiVersion: str = API_VERSION
+    kind: Literal["Instance"] = "Instance"
+    metadata: Metadata = Field(default_factory=Metadata)
+    platform: PlatformPin
+    ports: PortsConfig = Field(default_factory=PortsConfig)
+    kas: dict[str, KasPin] = Field(default_factory=dict)
+    features: dict[str, bool] = Field(default_factory=dict)
+    fixtures: Fixtures = Field(default_factory=Fixtures)
+
+
+class ScenarioSdks(_StrictModel):
+    """Encrypt/decrypt split mirrors xtest's --sdks-encrypt/--sdks-decrypt.
+
+    Listing the same SDK in both maps reproduces the legacy "all pairs" mode.
+    """
+
+    encrypt: dict[SdkName, SdkPin] = Field(default_factory=dict)
+    decrypt: dict[SdkName, SdkPin] = Field(default_factory=dict)
+
+    def union(self) -> dict[SdkName, SdkPin]:
+        """Return the union of encrypt+decrypt SDK pins (decrypt wins on conflict)."""
+        return {**self.encrypt, **self.decrypt}
+
+
+class Suite(_StrictModel):
+    """Pytest selection + flags."""
+
+    select: str = Field(description="Pytest -k or path::node selector")
+    containers: str | None = Field(default=None, description="Forwarded to --containers")
+    markers: str | None = Field(default=None, description="Forwarded to -m")
+    extra_args: list[str] = Field(default_factory=list)
+
+
+class Scenario(_StrictModel):
+    """Top-level scenarios.yaml model.
+
+    Composes an Instance with SDK pins and a pytest Suite selection.
+    """
+
+    apiVersion: str = API_VERSION
+    kind: Literal["Scenario"] = "Scenario"
+    metadata: Metadata = Field(default_factory=Metadata)
+    instance: Annotated[Instance, Field(description="Inline instance definition")]
+    sdks: ScenarioSdks = Field(default_factory=ScenarioSdks)
+    suite: Suite
+    expected: str | None = None
+    actual: str | None = None
+
+
+def _yaml() -> YAML:
+    y = YAML(typ="safe")
+    y.preserve_quotes = True
+    return y
+
+
+def load_scenario(path: str | Path) -> Scenario:
+    """Parse and validate a scenarios.yaml file."""
+    p = Path(path)
+    raw = _yaml().load(p.read_text())
+    if not isinstance(raw, dict):
+        raise ValueError(f"{p}: top-level YAML must be a mapping, got {type(raw).__name__}")
+    return Scenario.model_validate(raw)
+
+
+def load_instance(path: str | Path) -> Instance:
+    """Parse and validate an instance.yaml file."""
+    p = Path(path)
+    raw = _yaml().load(p.read_text())
+    if not isinstance(raw, dict):
+        raise ValueError(f"{p}: top-level YAML must be a mapping, got {type(raw).__name__}")
+    return Instance.model_validate(raw)
+
+
+def dump_instance(instance: Instance, path: str | Path) -> None:
+    """Serialize an Instance to YAML at `path`."""
+    p = Path(path)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    data = instance.model_dump(mode="json", exclude_none=True)
+    y = _yaml()
+    with p.open("w") as f:
+        y.dump(data, f)
+
+
+def _main(argv: list[str] | None = None) -> int:
+    """`python -m otdf_sdk_mgr.schema validate <path>` entry point."""
+    args = list(sys.argv[1:] if argv is None else argv)
+    if len(args) != 2 or args[0] != "validate":
+        print("usage: python -m otdf_sdk_mgr.schema validate <path>", file=sys.stderr)
+        return 2
+    path = Path(args[1])
+    try:
+        raw = _yaml().load(path.read_text())
+    except OSError as e:
+        print(f"error: cannot read {path}: {e}", file=sys.stderr)
+        return 1
+    if not isinstance(raw, dict):
+        print(f"error: {path} top-level YAML must be a mapping", file=sys.stderr)
+        return 1
+    kind = raw.get("kind")
+    model: type[BaseModel]
+    if kind == "Scenario":
+        model = Scenario
+    elif kind == "Instance":
+        model = Instance
+    else:
+        print(f"error: {path} has unknown kind {kind!r}; expected Scenario or Instance", file=sys.stderr)
+        return 1
+    try:
+        model.model_validate(raw)
+    except Exception as e:
+        print(f"invalid: {e}", file=sys.stderr)
+        return 1
+    print(f"ok: {path} ({kind})")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(_main())

otdf-sdk-mgr/tests/test_schema.py

@@ -0,0 +1,102 @@
+"""Schema smoke tests for Scenario/Instance/Pin models."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from otdf_sdk_mgr.schema import (
+    Instance,
+    KasPin,
+    PlatformPin,
+    Scenario,
+    ScenarioSdks,
+    SdkPin,
+    SourceRef,
+    Suite,
+    dump_instance,
+    load_instance,
+    load_scenario,
+)
+from pydantic import ValidationError
+
+
+def _minimal_scenario_yaml() -> str:
+    return """
+apiVersion: opentdf.io/v1alpha1
+kind: Scenario
+metadata:
+  id: smoke
+  title: "schema smoke"
+  created: 2026-05-15
+instance:
+  metadata: { name: smoke }
+  platform: { dist: v0.9.0 }
+  ports: { base: 9080 }
+  kas:
+    alpha: { dist: v0.9.0, mode: standard }
+sdks:
+  encrypt:
+    go: { version: lts }
+  decrypt:
+    java: { version: "0.7.8" }
+suite:
+  select: "xtest/test_tdfs.py::test_tdf_roundtrip"
+  containers: ztdf
+"""
+
+
+def test_scenario_roundtrip(tmp_path: Path) -> None:
+    path = tmp_path / "scenario.yaml"
+    path.write_text(_minimal_scenario_yaml())
+    scenario = load_scenario(path)
+    assert scenario.kind == "Scenario"
+    assert scenario.instance.platform.dist == "v0.9.0"
+    assert scenario.instance.ports.base == 9080
+    assert "alpha" in scenario.instance.kas
+    assert scenario.sdks.encrypt["go"].version == "lts"
+    assert scenario.sdks.decrypt["java"].version == "0.7.8"
+
+
+def test_platform_pin_requires_exactly_one_source() -> None:
+    with pytest.raises(ValidationError):
+        PlatformPin()  # no fields set
+    with pytest.raises(ValidationError):
+        PlatformPin(dist="v0.9.0", image="ghcr.io/x:v0.9.0")  # two set
+
+
+def test_kas_pin_features_pass_through() -> None:
+    pin = KasPin(dist="v0.9.0", mode="key_management", features={"ec_tdf_enabled": True})
+    assert pin.features["ec_tdf_enabled"] is True
+
+
+def test_scenario_sdks_union_dedupes() -> None:
+    sdks = ScenarioSdks(
+        encrypt={"go": SdkPin(version="lts")},
+        decrypt={"go": SdkPin(version="lts"), "java": SdkPin(version="0.7.8")},
+    )
+    union = sdks.union()
+    assert set(union.keys()) == {"go", "java"}
+
+
+def test_dump_load_instance_roundtrip(tmp_path: Path) -> None:
+    inst = Instance(
+        platform=PlatformPin(source=SourceRef(ref="main")),
+        kas={"alpha": KasPin(dist="v0.9.0")},
+    )
+    out = tmp_path / "instance.yaml"
+    dump_instance(inst, out)
+    loaded = load_instance(out)
+    assert loaded.platform.source is not None
+    assert loaded.platform.source.ref == "main"
+    assert loaded.kas["alpha"].dist == "v0.9.0"
+
+
+def test_unknown_kind_rejected_by_extra_forbid(tmp_path: Path) -> None:
+    bad = tmp_path / "bad.yaml"
+    bad.write_text(
+        "apiVersion: opentdf.io/v1alpha1\nkind: Scenario\nunknown_field: oops\n"
+        "instance:\n  platform: { dist: v0.9.0 }\nsuite:\n  select: foo\n"
+    )
+    with pytest.raises(ValidationError):
+        load_scenario(bad)