Preserve existing Helm observation behavior in the control-path candidate (1b778a3c) · Commits · OOP / code / transformation-function-sdk

README.md

+0 −9

Changes for README.md: 0 added lines, 9 removed lines.
Original line number	Diff line number	Diff line
		@@ -260,12 +260,3 @@ 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.

		## Runtime observation failures

		A nonzero Helm status command or malformed status response raises the existing
		DomainExecutionFailed. CLI stderr alone is not used as authoritative absence
		evidence. Valid status responses keep the existing runtime-state mapping.
		SRM can distinguish an observation failure from an observed state; it must not
		interpret an outage as successful deletion. Wider SRM UNKNOWN handling remains
		separate from this SDK correction.

src/openop_tfsdk/contract/adapters/kubernetes.py

+5 −6

Changes for src/openop_tfsdk/contract/adapters/kubernetes.py: 5 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -249,14 +249,13 @@ class KubernetesCapabilityAdapter:
		error_code=DomainExecutionFailed.error_code,
		**helm_diagnostics(result),
		)
		raise DomainExecutionFailed("helm status failed")
		return CapabilityRuntimeState(
		external_id=release_name,
		capability_kind=request.capability_kind,
		state=RuntimeState.UNKNOWN,
		)

		try:
		helm_status = json.loads(result.stdout)["info"]["status"]
		if not isinstance(helm_status, str) or not helm_status:
		raise ValueError("Missing Helm status")
		except (ValueError, KeyError, TypeError) as error:
		raise DomainExecutionFailed("Invalid helm status response") from error
		if helm_status == "deployed":
		state = RuntimeState.ACTIVE
		elif helm_status == "failed":

tests/contract/test_kubernetes_status_failures.py

deleted100644 → 0

+0 −34

Changes for tests/contract/test_kubernetes_status_failures.py: 0 added lines, 34 removed lines.
Original line number	Diff line number	Diff line
		"""A failed observation cannot establish that a realization is absent."""

		from types import SimpleNamespace

		import pytest

		from openop_tfsdk.contract.adapters import build_adapter
		from openop_tfsdk.contract.errors import DomainExecutionFailed

		from .test_control_path_guard import request_for


		@pytest.mark.parametrize(
		"stderr", ["Kubernetes cluster unreachable", "Error: release: not found", "unknown failure"]
		)
		def test_nonzero_status_is_a_typed_observation_failure(kubernetes_settings, monkeypatch, stderr):
		instance = build_adapter("kubernetes", {**kubernetes_settings, "control_path_ref": "athens"})
		monkeypatch.setattr(
		"subprocess.run", lambda a, *k: SimpleNamespace(returncode=1, stderr=stderr)
		)
		with pytest.raises(DomainExecutionFailed, match="helm status failed"):
		instance.read_runtime_state(request_for("read_runtime_state", "athens"))


		@pytest.mark.parametrize(
		"stdout", ["bad-json", "null", "[]", "{}", '{"info":{}}', '{"info":{"status":null}}']
		)
		def test_malformed_success_cannot_become_a_runtime_state(kubernetes_settings, monkeypatch, stdout):
		instance = build_adapter("kubernetes", {**kubernetes_settings, "control_path_ref": "athens"})
		monkeypatch.setattr(
		"subprocess.run", lambda a, *k: SimpleNamespace(returncode=0, stdout=stdout)
		)
		with pytest.raises(DomainExecutionFailed, match="status response"):
		instance.read_runtime_state(request_for("read_runtime_state", "athens"))

tests/contract/test_logging.py

+6 −5

Changes for tests/contract/test_logging.py: 6 added lines, 5 removed lines.
Original line number	Diff line number	Diff line
		@@ -18,6 +18,7 @@ from openop_tfsdk.contract.errors import (
		ResourceNotFound,
		ValidationError,
		)
		from openop_tfsdk.contract.execution_models import RuntimeState
		from openop_tfsdk.edgecloud.adapters.kubernetes.client import EdgeApplicationManager
		from openop_tfsdk.logger import get_logger, setup_logger
		from openop_tfsdk.network.core.common import CoreHttpError
		@@ -70,7 +71,7 @@ def test_logging_preserves_typed_failures_without_raw_backend_text(
		"wrapper",
		[structlog.make_filtering_bound_logger(0), structlog.BoundLogger, structlog.stdlib.BoundLogger],
		)
		def test_failed_helm_status_raises_without_logging_stderr(
		def test_failed_helm_status_returns_unknown_without_logging_stderr(
		kubernetes_settings, host_events, wrapper, monkeypatch
		):
		structlog.configure(wrapper_class=wrapper)
		@@ -82,14 +83,14 @@ def test_failed_helm_status_raises_without_logging_stderr(
		),
		)
		adapter = KubernetesCapabilityAdapter({kubernetes_settings, {"namespace": "test"}})
		with pytest.raises(DomainExecutionFailed):
		adapter.read_runtime_state(
		result = adapter.read_runtime_state(
		SimpleNamespace(
		external_id="known-release",
		capability_kind="workload",
		target=SimpleNamespace(control_path_ref="test-k8s"),
		)
		)
		assert result.state is RuntimeState.UNKNOWN
		assert marker not in str(host_events)
		assert host_events[0]["exit_code"] == 1
		assert host_events[0]["reason"] == "cluster_unreachable"
		@@ -262,14 +263,14 @@ def test_unknown_helm_diagnostics_keep_only_exit_code_and_generic_reason(
		lambda args, *kwargs: SimpleNamespace(returncode=9, stderr=marker),
		)
		adapter = KubernetesCapabilityAdapter(kubernetes_settings)
		with pytest.raises(DomainExecutionFailed):
		adapter.read_runtime_state(
		result = adapter.read_runtime_state(
		SimpleNamespace(
		external_id="known-release",
		capability_kind="workload",
		target=SimpleNamespace(control_path_ref="test-k8s"),
		)
		)
		assert result.state is RuntimeState.UNKNOWN
		assert host_events[0]["exit_code"] == 9
		assert host_events[0]["reason"] == "helm_command_failed"
		assert marker not in str(host_events)