Loading README.md +15 −0 Changes for README.md: 15 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -245,3 +245,18 @@ backend. A mismatch raises non-retryable InvalidTargetRef. This applies to direct concrete adapter calls as well as factory construction. The check uses an opaque identity; topology relationships and placement remain in SRM. It does not establish authorization or the provenance of an arbitrary backend ID. ## One file bootstrap for discovery and execution Load `BootstrapTopology.from_file(path)` once in the composition root. Its four existing discovery methods return normalized topology. The separate `get_control_path_configurations()` accessor returns defensive copies of runtime entries with `ref`, `adapter_type` and `infrastructure`. Map those entries into the registry, deriving `config["control_path_ref"]` from `ref` before calling the existing SDK factory. Do not repeat identity inside bootstrap infrastructure. Duplicate/missing refs, malformed entries, unknown kinds and unresolved binding refs fail locally. Discovery does not construct execution adapters or resolve credentials. Keep runtime config out of normalized responses and persisted topology. The deployment owns mounts; the SDK does not parse a secret URI. See `tests/contract/bootstrap_example.yaml` for an explicit file-mode example. src/openop_tfsdk/contract/topology.py +51 −7 Changes for src/openop_tfsdk/contract/topology.py: 51 added lines, 7 removed lines. Original line number Diff line number Diff line Loading @@ -12,12 +12,17 @@ and supported_runtime_kinds, which are not topology and come from the adapter behind that control path. """ from copy import deepcopy from typing import List, Optional, Tuple import yaml from openop_tfsdk.contract.adapters import capability_surface from openop_tfsdk.contract.errors import InvalidTargetRef from openop_tfsdk.contract.adapters import ADAPTERS_BY_TYPE, capability_surface from openop_tfsdk.contract.errors import ( InvalidTargetRef, UnknownAdapterKind, ValidationError, ) from openop_tfsdk.contract.interfaces import TopologyDiscoveryAdapter from openop_tfsdk.contract.models import ( Capability, Loading @@ -36,15 +41,39 @@ class BootstrapTopology(TopologyDiscoveryAdapter): """ def __init__(self, bootstrap: dict): if not isinstance(bootstrap, dict): raise ValidationError("Bootstrap must be a mapping") bootstrap = deepcopy(bootstrap) self._zones: List[Zone] = [] self._domains: List[Tuple[str, Domain]] = [] self._capabilities: List[Tuple[str, Capability]] = [] self._bindings: List[Tuple[str, ControlPathBinding]] = [] self._control_paths = { raw["ref"]: (raw.get("adapter_type"), raw.get("infrastructure")) for raw in bootstrap.get("control_paths") or [] } self._control_paths = {} entries = bootstrap.get("control_paths", []) if not isinstance(entries, list): raise ValidationError("Bootstrap control_paths must be a list") for raw in entries: if not isinstance(raw, dict): raise ValidationError("Each bootstrap control path must be a mapping") ref = raw.get("ref") if not isinstance(ref, str) or not ref.strip(): raise ValidationError("Each control path requires a nonempty ref") if ref in self._control_paths: raise ValidationError("Duplicate control-path ref") adapter_type = raw.get("adapter_type") if not isinstance(adapter_type, str) or adapter_type not in ADAPTERS_BY_TYPE: raise UnknownAdapterKind("Bootstrap contains an unregistered adapter kind") infrastructure = raw.get("infrastructure") if infrastructure is None: infrastructure = {} if not isinstance(infrastructure, dict): raise ValidationError("Control-path infrastructure must be a mapping") if "control_path_ref" in infrastructure: raise ValidationError( "Declare identity once in ref, not infrastructure.control_path_ref" ) self._control_paths[ref] = (adapter_type, infrastructure) for raw_zone in bootstrap.get("zones") or []: self._add_zone(raw_zone) Loading @@ -54,6 +83,18 @@ class BootstrapTopology(TopologyDiscoveryAdapter): with open(path, encoding="utf-8") as handle: return cls(yaml.safe_load(handle) or {}) def get_control_path_configurations(self) -> List[dict]: """Runtime-only construction entries from this loaded revision. This file-bootstrap accessor is separate from normalized discovery. The consumer derives instance identity from ref; credentials and config must not be persisted as topology or logged. """ return [ {"ref": ref, "adapter_type": kind, "infrastructure": deepcopy(infrastructure)} for ref, (kind, infrastructure) in self._control_paths.items() ] def _add_zone(self, raw: dict) -> None: self._zones.append( Zone( Loading Loading @@ -110,7 +151,10 @@ class BootstrapTopology(TopologyDiscoveryAdapter): def _add_binding(self, capability_ref: str, capability_kind: str, raw: dict) -> None: # Not read from the YAML: TF-SDK reports what a control path can execute # for its adapter kind and configuration, without building an adapter. control_path = self._control_paths.get(raw["control_path_ref"]) ref = raw.get("control_path_ref") if not isinstance(ref, str) or not ref.strip(): raise InvalidTargetRef("Binding requires a nonempty control_path_ref") control_path = self._control_paths.get(ref) if control_path is None: raise InvalidTargetRef(raw["control_path_ref"]) adapter_type, infrastructure = control_path Loading tests/contract/bootstrap_example.yaml +2 −9 Changes for tests/contract/bootstrap_example.yaml: 2 added lines, 9 removed lines. Original line number Diff line number Diff line Loading @@ -18,20 +18,13 @@ control_paths: adapter_type: kubernetes adapter_contract_version: v1 invocation: adapter_class: openop_tfsdk.adapters.kubernetes.KubernetesComputeAdapter infrastructure: api_server: https://k8s-athens-edge.example.local:6443 kubeconfig_ref: secret://openop/k8s-athens-edge/kubeconfig # Ops resolves the reference and mounts this file before adapter execution. # Deployment mounts a protected Secret at this explicit runtime path. kubeconfig_path: /etc/openop/kubeconfigs/athens-edge.yaml kube_context: athens-edge namespace: openop-services default_storage_class: fast-edge-ssd registry_mirror: registry.athens-edge.example.local tls_verify: true timeout_seconds: 30 runtime_directory: /var/run/openop-tfsdk retry: transient_attempts: 3 Loading tests/contract/test_bootstrap_runtime_configuration.py 0 → 100644 +130 −0 Changes for tests/contract/test_bootstrap_runtime_configuration.py: 130 added lines, 0 removed lines. Original line number Diff line number Diff line """One SDK-loaded declaration supplies discovery and isolated runtime entries.""" import copy import json import pytest from openop_tfsdk.contract.adapters import ADAPTERS_BY_TYPE from openop_tfsdk.contract.errors import ( InvalidTargetRef, UnknownAdapterKind, ValidationError, ) from openop_tfsdk.contract.topology import BootstrapTopology def declaration(): return { "control_paths": [ { "ref": "athens", "adapter_type": "kubernetes", "infrastructure": { "in_cluster": True, "namespace": "applications", "nested": {"value": "original"}, }, }, { "ref": "berlin", "adapter_type": "kubernetes", "infrastructure": { "in_cluster": True, "namespace": "other-applications", }, }, ], "zones": [ { "ref": "z", "name": "zone", "kind": "resource", "domains": [ { "ref": "d", "name": "domain", "kind": "compute", "capabilities": [ { "ref": "c", "name": "workload", "kind": "workload", "control_path_bindings": [ {"ref": "binding", "control_path_ref": "athens"}, ], }, ], }, ], } ], } def test_runtime_config_is_separate_from_discovery_and_defensively_copied(monkeypatch): source = declaration() expected = copy.deepcopy(source["control_paths"]) for adapter in ADAPTERS_BY_TYPE.values(): monkeypatch.setattr( adapter, "__init__", lambda *a, **k: pytest.fail("discovery constructed a backend") ) topology = BootstrapTopology(source) source["control_paths"][0]["infrastructure"]["nested"]["value"] = "mutated input" returned = topology.get_control_path_configurations() assert returned == expected returned[0]["infrastructure"]["nested"]["value"] = "mutated result" assert topology.get_control_path_configurations() == expected assert topology.get_control_path_bindings()[0].control_path_ref == "athens" normalized = json.dumps( [m.model_dump(mode="json") for m in topology.get_control_path_bindings()] ) assert "infrastructure" not in normalized and "applications" not in normalized assert "get_control_path_configurations" not in topology.__class__.__bases__[0].__dict__ @pytest.mark.parametrize("ref", [None, "", " ", 10]) def test_missing_or_invalid_control_path_ref_is_rejected(ref): source = declaration() source["control_paths"][0]["ref"] = ref with pytest.raises(ValidationError): BootstrapTopology(source) def test_duplicate_control_paths_fail_before_last_entry_can_overwrite_the_first(): source = declaration() source["control_paths"][1]["ref"] = "athens" with pytest.raises(ValidationError, match="Duplicate"): BootstrapTopology(source) def test_identity_is_not_repeated_inside_infrastructure(): source = declaration() source["control_paths"][0]["infrastructure"]["control_path_ref"] = "athens" with pytest.raises(ValidationError, match="control_path_ref"): BootstrapTopology(source) def test_unknown_adapter_is_rejected_even_when_no_binding_uses_it(): source = declaration() source["control_paths"][1]["adapter_type"] = "unknown-kind" with pytest.raises(UnknownAdapterKind): BootstrapTopology(source) def test_binding_cannot_name_an_undeclared_control_path(): source = declaration() source["zones"][0]["domains"][0]["capabilities"][0]["control_path_bindings"][0][ "control_path_ref" ] = "missing" with pytest.raises(InvalidTargetRef): BootstrapTopology(source) @pytest.mark.parametrize( "entries", ["not-a-list", [None], [{"ref": "a", "adapter_type": "kubernetes", "infrastructure": "bad"}]], ) def test_malformed_control_path_configuration_is_a_typed_error(entries): with pytest.raises(ValidationError): BootstrapTopology({"control_paths": entries}) Loading
README.md +15 −0 Changes for README.md: 15 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -245,3 +245,18 @@ backend. A mismatch raises non-retryable InvalidTargetRef. This applies to direct concrete adapter calls as well as factory construction. The check uses an opaque identity; topology relationships and placement remain in SRM. It does not establish authorization or the provenance of an arbitrary backend ID. ## One file bootstrap for discovery and execution Load `BootstrapTopology.from_file(path)` once in the composition root. Its four existing discovery methods return normalized topology. The separate `get_control_path_configurations()` accessor returns defensive copies of runtime entries with `ref`, `adapter_type` and `infrastructure`. Map those entries into the registry, deriving `config["control_path_ref"]` from `ref` before calling the existing SDK factory. Do not repeat identity inside bootstrap infrastructure. Duplicate/missing refs, malformed entries, unknown kinds and unresolved binding refs fail locally. Discovery does not construct execution adapters or resolve credentials. Keep runtime config out of normalized responses and persisted topology. The deployment owns mounts; the SDK does not parse a secret URI. See `tests/contract/bootstrap_example.yaml` for an explicit file-mode example.
src/openop_tfsdk/contract/topology.py +51 −7 Changes for src/openop_tfsdk/contract/topology.py: 51 added lines, 7 removed lines. Original line number Diff line number Diff line Loading @@ -12,12 +12,17 @@ and supported_runtime_kinds, which are not topology and come from the adapter behind that control path. """ from copy import deepcopy from typing import List, Optional, Tuple import yaml from openop_tfsdk.contract.adapters import capability_surface from openop_tfsdk.contract.errors import InvalidTargetRef from openop_tfsdk.contract.adapters import ADAPTERS_BY_TYPE, capability_surface from openop_tfsdk.contract.errors import ( InvalidTargetRef, UnknownAdapterKind, ValidationError, ) from openop_tfsdk.contract.interfaces import TopologyDiscoveryAdapter from openop_tfsdk.contract.models import ( Capability, Loading @@ -36,15 +41,39 @@ class BootstrapTopology(TopologyDiscoveryAdapter): """ def __init__(self, bootstrap: dict): if not isinstance(bootstrap, dict): raise ValidationError("Bootstrap must be a mapping") bootstrap = deepcopy(bootstrap) self._zones: List[Zone] = [] self._domains: List[Tuple[str, Domain]] = [] self._capabilities: List[Tuple[str, Capability]] = [] self._bindings: List[Tuple[str, ControlPathBinding]] = [] self._control_paths = { raw["ref"]: (raw.get("adapter_type"), raw.get("infrastructure")) for raw in bootstrap.get("control_paths") or [] } self._control_paths = {} entries = bootstrap.get("control_paths", []) if not isinstance(entries, list): raise ValidationError("Bootstrap control_paths must be a list") for raw in entries: if not isinstance(raw, dict): raise ValidationError("Each bootstrap control path must be a mapping") ref = raw.get("ref") if not isinstance(ref, str) or not ref.strip(): raise ValidationError("Each control path requires a nonempty ref") if ref in self._control_paths: raise ValidationError("Duplicate control-path ref") adapter_type = raw.get("adapter_type") if not isinstance(adapter_type, str) or adapter_type not in ADAPTERS_BY_TYPE: raise UnknownAdapterKind("Bootstrap contains an unregistered adapter kind") infrastructure = raw.get("infrastructure") if infrastructure is None: infrastructure = {} if not isinstance(infrastructure, dict): raise ValidationError("Control-path infrastructure must be a mapping") if "control_path_ref" in infrastructure: raise ValidationError( "Declare identity once in ref, not infrastructure.control_path_ref" ) self._control_paths[ref] = (adapter_type, infrastructure) for raw_zone in bootstrap.get("zones") or []: self._add_zone(raw_zone) Loading @@ -54,6 +83,18 @@ class BootstrapTopology(TopologyDiscoveryAdapter): with open(path, encoding="utf-8") as handle: return cls(yaml.safe_load(handle) or {}) def get_control_path_configurations(self) -> List[dict]: """Runtime-only construction entries from this loaded revision. This file-bootstrap accessor is separate from normalized discovery. The consumer derives instance identity from ref; credentials and config must not be persisted as topology or logged. """ return [ {"ref": ref, "adapter_type": kind, "infrastructure": deepcopy(infrastructure)} for ref, (kind, infrastructure) in self._control_paths.items() ] def _add_zone(self, raw: dict) -> None: self._zones.append( Zone( Loading Loading @@ -110,7 +151,10 @@ class BootstrapTopology(TopologyDiscoveryAdapter): def _add_binding(self, capability_ref: str, capability_kind: str, raw: dict) -> None: # Not read from the YAML: TF-SDK reports what a control path can execute # for its adapter kind and configuration, without building an adapter. control_path = self._control_paths.get(raw["control_path_ref"]) ref = raw.get("control_path_ref") if not isinstance(ref, str) or not ref.strip(): raise InvalidTargetRef("Binding requires a nonempty control_path_ref") control_path = self._control_paths.get(ref) if control_path is None: raise InvalidTargetRef(raw["control_path_ref"]) adapter_type, infrastructure = control_path Loading
tests/contract/bootstrap_example.yaml +2 −9 Changes for tests/contract/bootstrap_example.yaml: 2 added lines, 9 removed lines. Original line number Diff line number Diff line Loading @@ -18,20 +18,13 @@ control_paths: adapter_type: kubernetes adapter_contract_version: v1 invocation: adapter_class: openop_tfsdk.adapters.kubernetes.KubernetesComputeAdapter infrastructure: api_server: https://k8s-athens-edge.example.local:6443 kubeconfig_ref: secret://openop/k8s-athens-edge/kubeconfig # Ops resolves the reference and mounts this file before adapter execution. # Deployment mounts a protected Secret at this explicit runtime path. kubeconfig_path: /etc/openop/kubeconfigs/athens-edge.yaml kube_context: athens-edge namespace: openop-services default_storage_class: fast-edge-ssd registry_mirror: registry.athens-edge.example.local tls_verify: true timeout_seconds: 30 runtime_directory: /var/run/openop-tfsdk retry: transient_attempts: 3 Loading
tests/contract/test_bootstrap_runtime_configuration.py 0 → 100644 +130 −0 Changes for tests/contract/test_bootstrap_runtime_configuration.py: 130 added lines, 0 removed lines. Original line number Diff line number Diff line """One SDK-loaded declaration supplies discovery and isolated runtime entries.""" import copy import json import pytest from openop_tfsdk.contract.adapters import ADAPTERS_BY_TYPE from openop_tfsdk.contract.errors import ( InvalidTargetRef, UnknownAdapterKind, ValidationError, ) from openop_tfsdk.contract.topology import BootstrapTopology def declaration(): return { "control_paths": [ { "ref": "athens", "adapter_type": "kubernetes", "infrastructure": { "in_cluster": True, "namespace": "applications", "nested": {"value": "original"}, }, }, { "ref": "berlin", "adapter_type": "kubernetes", "infrastructure": { "in_cluster": True, "namespace": "other-applications", }, }, ], "zones": [ { "ref": "z", "name": "zone", "kind": "resource", "domains": [ { "ref": "d", "name": "domain", "kind": "compute", "capabilities": [ { "ref": "c", "name": "workload", "kind": "workload", "control_path_bindings": [ {"ref": "binding", "control_path_ref": "athens"}, ], }, ], }, ], } ], } def test_runtime_config_is_separate_from_discovery_and_defensively_copied(monkeypatch): source = declaration() expected = copy.deepcopy(source["control_paths"]) for adapter in ADAPTERS_BY_TYPE.values(): monkeypatch.setattr( adapter, "__init__", lambda *a, **k: pytest.fail("discovery constructed a backend") ) topology = BootstrapTopology(source) source["control_paths"][0]["infrastructure"]["nested"]["value"] = "mutated input" returned = topology.get_control_path_configurations() assert returned == expected returned[0]["infrastructure"]["nested"]["value"] = "mutated result" assert topology.get_control_path_configurations() == expected assert topology.get_control_path_bindings()[0].control_path_ref == "athens" normalized = json.dumps( [m.model_dump(mode="json") for m in topology.get_control_path_bindings()] ) assert "infrastructure" not in normalized and "applications" not in normalized assert "get_control_path_configurations" not in topology.__class__.__bases__[0].__dict__ @pytest.mark.parametrize("ref", [None, "", " ", 10]) def test_missing_or_invalid_control_path_ref_is_rejected(ref): source = declaration() source["control_paths"][0]["ref"] = ref with pytest.raises(ValidationError): BootstrapTopology(source) def test_duplicate_control_paths_fail_before_last_entry_can_overwrite_the_first(): source = declaration() source["control_paths"][1]["ref"] = "athens" with pytest.raises(ValidationError, match="Duplicate"): BootstrapTopology(source) def test_identity_is_not_repeated_inside_infrastructure(): source = declaration() source["control_paths"][0]["infrastructure"]["control_path_ref"] = "athens" with pytest.raises(ValidationError, match="control_path_ref"): BootstrapTopology(source) def test_unknown_adapter_is_rejected_even_when_no_binding_uses_it(): source = declaration() source["control_paths"][1]["adapter_type"] = "unknown-kind" with pytest.raises(UnknownAdapterKind): BootstrapTopology(source) def test_binding_cannot_name_an_undeclared_control_path(): source = declaration() source["zones"][0]["domains"][0]["capabilities"][0]["control_path_bindings"][0][ "control_path_ref" ] = "missing" with pytest.raises(InvalidTargetRef): BootstrapTopology(source) @pytest.mark.parametrize( "entries", ["not-a-list", [None], [{"ref": "a", "adapter_type": "kubernetes", "infrastructure": "bad"}]], ) def test_malformed_control_path_configuration_is_a_typed_error(entries): with pytest.raises(ValidationError): BootstrapTopology({"control_paths": entries})
