diff --git a/control_plane/product_config.py b/control_plane/product_config.py index 430a2f78..47c70fd0 100644 --- a/control_plane/product_config.py +++ b/control_plane/product_config.py @@ -4,7 +4,7 @@ import os from json import JSONDecodeError from pathlib import Path -from typing import Literal, Protocol, cast +from typing import Literal, Protocol, TypedDict, cast from control_plane import secrets as control_plane_secrets from control_plane.contracts.runtime_environment_record import RuntimeEnvironmentRecord @@ -45,6 +45,13 @@ def list_runtime_key_safety_policy_records( ) -> tuple[RuntimeKeySafetyPolicyRecord, ...]: ... +class _SecretBindingLookupKwargs(TypedDict, total=False): + integration: str + context_name: str + instance_name: str + limit: int | None + + class ProductConfigError(ValueError): """Operator-facing product config validation or planning failure.""" @@ -602,9 +609,16 @@ def _retire_disabled_runtime_secret_placeholders( return context_name = configured_binding.context.strip() instance_name = configured_binding.instance.strip() + lookup_kwargs: _SecretBindingLookupKwargs = { + "integration": configured_binding.integration, + "limit": None, + } + if context_name: + lookup_kwargs["context_name"] = context_name + if instance_name: + lookup_kwargs["instance_name"] = instance_name for binding in record_store.list_secret_bindings( - integration=configured_binding.integration, - limit=None, + **lookup_kwargs, ): if binding.binding_id == configured_binding.binding_id: continue diff --git a/tests/test_runtime_environments.py b/tests/test_runtime_environments.py index ed5fcb34..08cb3d90 100644 --- a/tests/test_runtime_environments.py +++ b/tests/test_runtime_environments.py @@ -123,6 +123,7 @@ def __init__(self) -> None: self.secret_records: dict[str, SecretRecord] = {} self.secret_versions: dict[str, SecretVersion] = {} self.secret_bindings: dict[str, SecretBinding] = {} + self.secret_binding_list_calls: list[dict[str, object]] = [] self.secret_audit_events: list[SecretAuditEvent] = [] self.runtime_key_safety_policy_records: tuple[RuntimeKeySafetyPolicyRecord, ...] = ( RuntimeKeySafetyPolicyRecord( @@ -216,6 +217,14 @@ def list_secret_bindings( instance_name: str = "", limit: int | None = None, ) -> tuple[SecretBinding, ...]: + self.secret_binding_list_calls.append( + { + "integration": integration, + "context_name": context_name, + "instance_name": instance_name, + "limit": limit, + } + ) bindings = tuple( binding for binding in self.secret_bindings.values() @@ -437,6 +446,11 @@ def test_product_config_apply_retires_disabled_runtime_secret_placeholder(self) self.assertEqual(len(active_bindings), 1) self.assertEqual(active_bindings[0].binding_key, "DISCORD_TOKEN") self.assertEqual(active_bindings[0].status, "configured") + retirement_lookup = store.secret_binding_list_calls[-1] + self.assertEqual(retirement_lookup["integration"], "runtime_environment") + self.assertEqual(retirement_lookup["context_name"], "discord-blue") + self.assertEqual(retirement_lookup["instance_name"], "prod") + self.assertIsNone(retirement_lookup["limit"]) retired_placeholder = store.secret_bindings["binding-discord-blue-placeholder"] self.assertEqual(retired_placeholder.integration, "retired:runtime_environment") self.assertEqual(retired_placeholder.binding_key, "DISCORD_TOKEN") @@ -490,6 +504,11 @@ def test_runtime_secret_placeholder_retirement_requires_exact_route(self) -> Non exact_placeholder = store.secret_bindings["binding-global-placeholder"] self.assertEqual(exact_placeholder.integration, "retired:runtime_environment") self.assertEqual(exact_placeholder.status, "disabled") + retirement_lookup = store.secret_binding_list_calls[-1] + self.assertEqual(retirement_lookup["integration"], "runtime_environment") + self.assertEqual(retirement_lookup["context_name"], "") + self.assertEqual(retirement_lookup["instance_name"], "") + self.assertIsNone(retirement_lookup["limit"]) unrelated_placeholder = store.secret_bindings["binding-unrelated-placeholder"] self.assertEqual(unrelated_placeholder.integration, "runtime_environment") self.assertEqual(unrelated_placeholder.status, "disabled") @@ -542,6 +561,11 @@ def test_runtime_secret_placeholder_retirement_supports_context_route(self) -> N context_placeholder = store.secret_bindings["binding-context-placeholder"] self.assertEqual(context_placeholder.integration, "retired:runtime_environment") self.assertEqual(context_placeholder.status, "disabled") + retirement_lookup = store.secret_binding_list_calls[-1] + self.assertEqual(retirement_lookup["integration"], "runtime_environment") + self.assertEqual(retirement_lookup["context_name"], "discord-blue") + self.assertEqual(retirement_lookup["instance_name"], "") + self.assertIsNone(retirement_lookup["limit"]) instance_placeholder = store.secret_bindings["binding-instance-placeholder"] self.assertEqual(instance_placeholder.integration, "runtime_environment") self.assertEqual(instance_placeholder.status, "disabled")