From 0a3fc2dd38b28a04decdf68407b9b5662ee86a56 Mon Sep 17 00:00:00 2001 From: Hugo Durand Date: Tue, 22 Sep 2026 14:37:37 -0400 Subject: [PATCH] feat(cli): resolve a listener placement from the workspace listeners --- libs/cli/langgraph_cli/deploy.py | 88 +++++++++++ .../tests/unit_tests/test_deploy_helpers.py | 139 ++++++++++++++++++ 2 files changed, 227 insertions(+) diff --git a/libs/cli/langgraph_cli/deploy.py b/libs/cli/langgraph_cli/deploy.py index 14b653f64..73d12f802 100644 --- a/libs/cli/langgraph_cli/deploy.py +++ b/libs/cli/langgraph_cli/deploy.py @@ -178,6 +178,94 @@ class Listener: ) +@dataclass(frozen=True, slots=True) +class Unplaced: + def source_config(self) -> dict[str, object]: + return {} + + +@dataclass(frozen=True, slots=True) +class OnListener: + listener_id: str + k8s_namespace: str + + def source_config(self) -> dict[str, object]: + return { + "listener_id": self.listener_id, + "listener_config": {"k8s_namespace": self.k8s_namespace}, + } + + +Placement = Unplaced | OnListener + + +@dataclass(frozen=True, slots=True) +class RequestedPlacement: + listener_id: str | None = None + k8s_namespace: str | None = None + + @property + def requested(self) -> bool: + return self.listener_id is not None or self.k8s_namespace is not None + + def resolve(self, listeners: Sequence[Listener], *, required: bool) -> Placement: + if not listeners: + if self.requested: + raise click.UsageError( + "This workspace has no listeners, so --listener-id and " + "--k8s-namespace do not apply." + ) + return Unplaced() + if not required and not self.requested: + return Unplaced() + listener = self._listener(listeners) + return OnListener(listener.id, self._namespace(listener)) + + def _listener(self, listeners: Sequence[Listener]) -> Listener: + if self.listener_id is None: + if len(listeners) == 1: + return listeners[0] + raise click.UsageError( + "This workspace has several listeners. Choose one with " + f"--listener-id:\n{_describe(listeners)}" + ) + for listener in listeners: + if listener.id == self.listener_id: + return listener + raise click.UsageError( + f"Listener {self.listener_id} was not found in this workspace. " + f"Available listeners:\n{_describe(listeners)}" + ) + + def _namespace(self, listener: Listener) -> str: + if not listener.namespaces: + raise click.UsageError( + f"Listener {listener.id} serves no namespaces. Check its configuration." + ) + if self.k8s_namespace is None: + if len(listener.namespaces) == 1: + return listener.namespaces[0] + raise click.UsageError( + f"Listener {listener.id} serves several namespaces. Choose one with " + f"--k8s-namespace: {', '.join(listener.namespaces)}" + ) + if self.k8s_namespace not in listener.namespaces: + raise click.UsageError( + f"Listener {listener.id} does not serve namespace " + f"'{self.k8s_namespace}'. Choose one of: " + f"{', '.join(listener.namespaces)}" + ) + return self.k8s_namespace + + +def _describe(listeners: Sequence[Listener]) -> str: + return "\n".join( + f" {listener.id} cluster {listener.compute_id} " + f"namespaces: {', '.join(listener.namespaces)}" + for listener in listeners + ) + + @dataclass(frozen=True, slots=True) class ExistingDeployment: id: str diff --git a/libs/cli/tests/unit_tests/test_deploy_helpers.py b/libs/cli/tests/unit_tests/test_deploy_helpers.py index cdd74e16c..8b369a6db 100644 --- a/libs/cli/tests/unit_tests/test_deploy_helpers.py +++ b/libs/cli/tests/unit_tests/test_deploy_helpers.py @@ -17,7 +17,10 @@ from langgraph_cli.deploy import ( DockerBuildCommand, Listener, ManagedRegistrySource, + OnListener, RemoteBuildSource, + RequestedPlacement, + Unplaced, _call_host_backend_with_optional_tenant, _create_host_backend_client, _docker_config_for_token, @@ -927,3 +930,139 @@ class TestListener: ) def test_from_resource_reads_the_control_plane_shape(self, resource, expected): assert Listener.from_resource(resource) == expected + + +ONE_NAMESPACE = Listener("listener-1", "prod-cluster", ("agents",)) +TWO_NAMESPACES = Listener("listener-2", "multi-cluster", ("agents", "agents-staging")) +NO_NAMESPACE = Listener("listener-3", "broken-cluster", ()) + + +class TestRequestedPlacement: + @pytest.mark.parametrize( + ("request_", "listeners", "required", "expected"), + [ + pytest.param( + RequestedPlacement(), (), True, Unplaced(), id="no_listeners_no_request" + ), + pytest.param( + RequestedPlacement(), + (ONE_NAMESPACE,), + True, + OnListener("listener-1", "agents"), + id="cloud_uses_the_only_possible_answer", + ), + pytest.param( + RequestedPlacement(), + (ONE_NAMESPACE,), + False, + Unplaced(), + id="self_hosted_keeps_its_bundled_operator", + ), + pytest.param( + RequestedPlacement(listener_id="listener-1"), + (ONE_NAMESPACE,), + False, + OnListener("listener-1", "agents"), + id="self_hosted_places_when_asked", + ), + pytest.param( + RequestedPlacement(k8s_namespace="agents-staging"), + (TWO_NAMESPACES,), + True, + OnListener("listener-2", "agents-staging"), + id="namespace_alone_picks_the_only_listener", + ), + pytest.param( + RequestedPlacement(listener_id="listener-1"), + (ONE_NAMESPACE, TWO_NAMESPACES), + True, + OnListener("listener-1", "agents"), + id="listener_alone_picks_its_only_namespace", + ), + pytest.param( + RequestedPlacement(listener_id="listener-2", k8s_namespace="agents"), + (ONE_NAMESPACE, TWO_NAMESPACES), + True, + OnListener("listener-2", "agents"), + id="both_given", + ), + ], + ) + def test_resolves_to_a_placement(self, request_, listeners, required, expected): + assert request_.resolve(listeners, required=required) == expected + + @pytest.mark.parametrize( + ("request_", "listeners", "message"), + [ + pytest.param( + RequestedPlacement(listener_id="listener-1"), + (), + "no listeners", + id="workspace_has_no_listeners", + ), + pytest.param( + RequestedPlacement(), + (ONE_NAMESPACE, TWO_NAMESPACES), + "--listener-id", + id="several_listeners_need_a_choice", + ), + pytest.param( + RequestedPlacement(k8s_namespace="agents"), + (ONE_NAMESPACE, TWO_NAMESPACES), + "--listener-id", + id="namespace_alone_is_ambiguous_with_several_listeners", + ), + pytest.param( + RequestedPlacement(listener_id="listener-9"), + (ONE_NAMESPACE,), + "was not found", + id="unknown_listener", + ), + pytest.param( + RequestedPlacement(), + (TWO_NAMESPACES,), + "--k8s-namespace", + id="several_namespaces_need_a_choice", + ), + pytest.param( + RequestedPlacement(listener_id="listener-2", k8s_namespace="nope"), + (TWO_NAMESPACES,), + "does not serve namespace", + id="unknown_namespace", + ), + pytest.param( + RequestedPlacement(listener_id="listener-3"), + (NO_NAMESPACE,), + "serves no namespaces", + id="listener_without_namespaces", + ), + ], + ) + def test_refuses_and_names_the_choices(self, request_, listeners, message): + with pytest.raises(click.UsageError, match=message): + request_.resolve(listeners, required=True) + + def test_the_error_lists_every_listener_with_its_cluster_and_namespaces(self): + with pytest.raises(click.UsageError) as error: + RequestedPlacement().resolve((ONE_NAMESPACE, TWO_NAMESPACES), required=True) + + assert "listener-1" in error.value.message + assert "prod-cluster" in error.value.message + assert "agents-staging" in error.value.message + + @pytest.mark.parametrize( + ("placement", "expected"), + [ + pytest.param(Unplaced(), {}, id="unplaced_adds_nothing"), + pytest.param( + OnListener("listener-1", "agents"), + { + "listener_id": "listener-1", + "listener_config": {"k8s_namespace": "agents"}, + }, + id="placed_carries_listener_and_namespace", + ), + ], + ) + def test_source_config_matches_the_control_plane_shape(self, placement, expected): + assert placement.source_config() == expected