fix(cli): look up a deployment by exact name instead of a paged search

This commit is contained in:
Hugo Durand
2026-09-22 15:25:28 -04:00
parent beb3e9c1f1
commit 904478cbec
5 changed files with 93 additions and 13 deletions
+2 -2
View File
@@ -491,8 +491,8 @@ def _source_of(resource: object) -> str | None:
def find_deployment_by_name(
client: HostBackendClient, name: str
) -> ExistingDeployment | None:
for resource in client.list_deployments(name_contains=name):
if resource.get("name") == name and resource.get("id"):
for resource in client.list_deployments(name=name):
if resource.get("id"):
return ExistingDeployment(str(resource["id"]), _source_of(resource))
return None
+6 -7
View File
@@ -189,14 +189,13 @@ class HostBackendClient:
payload["secrets"] = secrets
return self._request("POST", "/v2/deployments", payload)
def list_deployments(self, name_contains: str = "") -> list[dict[str, Any]]:
return _resources(
self._request(
"GET",
"/v2/deployments",
params={"name_contains": name_contains},
)
def list_deployments(
self, *, name: str | None = None, name_contains: str | None = None
) -> list[dict[str, Any]]:
params = (
{"name": name} if name is not None else {"name_contains": name_contains}
)
return _resources(self._request("GET", "/v2/deployments", params=params))
def list_listeners(self) -> list[dict[str, Any]]:
return _resources(
@@ -85,13 +85,24 @@ class ControlPlaneDouble:
self.timeline.append(route)
if request.content:
self.bodies[route] = json.loads(request.content)
return self._respond(request.method, request.url.path)
return self._respond(request)
def _respond(self, method: str, path: str) -> httpx.Response:
def _respond(self, request: httpx.Request) -> httpx.Response:
method, path = request.method, request.url.path
if (method, path) == ("GET", "/v2/listeners"):
return httpx.Response(200, json={"resources": self.listeners})
if (method, path) == ("GET", "/v2/deployments"):
return httpx.Response(200, json={"resources": self.existing_deployments})
name = request.url.params.get("name")
return httpx.Response(
200,
json={
"resources": [
deployment
for deployment in self.existing_deployments
if name is None or deployment.get("name") == name
]
},
)
if (method, path) == ("POST", "/v2/deployments"):
if self.create_error is not None:
return httpx.Response(400, text=self.create_error)
@@ -15,6 +15,7 @@ import langgraph_cli.deploy as deploy_mod
from langgraph_cli.deploy import (
CustomerRegistrySource,
DockerBuildCommand,
ExistingDeployment,
Listener,
ManagedRegistrySource,
OnListener,
@@ -31,6 +32,7 @@ from langgraph_cli.deploy import (
_resolve_pushed_image_digest,
_select_source,
_validate_prebuilt_image,
find_deployment_by_name,
normalize_image_tag,
normalize_name,
)
@@ -1107,3 +1109,35 @@ class TestRequestedPlacement:
)
def test_source_config_matches_the_control_plane_shape(self, placement, expected):
assert placement.source_config() == expected
def test_finding_a_deployment_by_name_asks_the_server_for_an_exact_match():
seen: dict = {}
def handler(req: httpx.Request) -> httpx.Response:
seen["params"] = dict(req.url.params)
return httpx.Response(
200,
json={"resources": [{"id": "dep-1", "name": "agent", "source": "github"}]},
)
client = HostBackendClient(
"https://api.example.com", "key", transport=httpx.MockTransport(handler)
)
found = find_deployment_by_name(client, "agent")
assert seen["params"] == {"name": "agent"}
assert found == ExistingDeployment("dep-1", "github")
def test_finding_a_deployment_by_name_returns_none_when_the_server_has_no_match():
client = HostBackendClient(
"https://api.example.com",
"key",
transport=httpx.MockTransport(
lambda req: httpx.Response(200, json={"resources": []})
),
)
assert find_deployment_by_name(client, "agent") is None
+37 -1
View File
@@ -88,7 +88,7 @@ def test_list_deployments_sends_query_params():
c = HostBackendClient(
"https://api.example.com", "test-key", transport=httpx.MockTransport(handler)
)
assert c.list_deployments("my app") == []
assert c.list_deployments(name_contains="my app") == []
def _capturing_client(captured: dict) -> HostBackendClient:
@@ -612,3 +612,39 @@ def test_is_cloud_recognises_the_managed_control_plane(control_plane_url, expect
endpoints = ControlPlaneEndpoints.from_control_plane_url(control_plane_url)
assert endpoints.is_cloud is expected
@pytest.mark.parametrize(
("call", "expected_params"),
[
pytest.param(
lambda c: c.list_deployments(name="agent"),
{"name": "agent"},
id="exact_name_filters_server_side",
),
pytest.param(
lambda c: c.list_deployments(name_contains="age"),
{"name_contains": "age"},
id="substring_search_keeps_its_own_parameter",
),
pytest.param(
lambda c: c.list_deployments(),
{"name_contains": ""},
id="no_filter_lists_everything",
),
],
)
def test_list_deployments_sends_one_name_filter(call, expected_params):
seen: dict = {}
def handler(req: httpx.Request) -> httpx.Response:
seen.update(dict(req.url.params))
return httpx.Response(200, json={"resources": []})
call(
HostBackendClient(
"https://api.example.com", "key", transport=httpx.MockTransport(handler)
)
)
assert seen == expected_params