diff --git a/libs/cli/langgraph_cli/deploy.py b/libs/cli/langgraph_cli/deploy.py index a2ed5c74d..8da036ebb 100644 --- a/libs/cli/langgraph_cli/deploy.py +++ b/libs/cli/langgraph_cli/deploy.py @@ -26,6 +26,7 @@ from langgraph_cli.dependency_tracking import find_tracked_packages from langgraph_cli.docker import build_docker_image, can_build_locally from langgraph_cli.exec import CommandRunner, Runner, subp_exec from langgraph_cli.host_backend import ( + MAX_PAGE_SIZE, ControlPlaneEndpoints, HostBackendClient, HostBackendError, @@ -101,15 +102,15 @@ _NATIVE_AMD64_MACHINE = "x86_64" _PUSH_ATTEMPTS = 3 _LOCAL_BUILD_TAG_PREFIX = "langgraph-deploy-tmp" _OPERATOR_DEFAULT_RESOURCE_SPEC: Mapping[str, object] = {} -_LISTENER_REQUIRED_MARKER = "listener_id' is required" -_HYBRID_LISTENER_GUIDANCE = ( - "This workspace deploys through a listener in your own cluster, and the " - "control plane needs a listener ID to create a deployment. Create the " - "deployment once in the LangSmith UI, choosing the listener and namespace, " - "then re-run with --deployment-id ." -) - _CUSTOMER_REGISTRY_SOURCE: SourceName = "external_docker" +_LISTENER_REQUIRED_MARKER = "listener_id' is required" +_LISTENERS_SHOWN = 10 +_LISTENER_NOT_FOUND_STATUSES = frozenset({404, 422}) +_LISTENERS_DOCS_URL = "https://docs.langchain.com/langsmith/control-plane#listeners" +_NO_LISTENERS = ( + "This workspace has no listeners, so --listener-id and --k8s-namespace " + "do not apply." +) _TERMINAL_STATUSES = frozenset( @@ -161,6 +162,134 @@ class ByAgent: DeploymentSelector = ById | ByName | ByAgent +@dataclass(frozen=True, slots=True) +class Listener: + id: str + compute_id: str + namespaces: tuple[str, ...] + + @classmethod + def from_resource(cls, resource: Mapping[str, object]) -> "Listener": + identifier = str(resource.get("id") or "") + if not identifier: + raise HostBackendError( + "The control plane returned a listener without an id." + ) + compute_config = resource.get("compute_config") + namespaces = ( + compute_config.get("k8s_namespaces") + if isinstance(compute_config, Mapping) + else None + ) + return cls( + identifier, + str(resource.get("compute_id", "")), + tuple(str(namespace) for namespace in namespaces) + if isinstance(namespaces, list) + else (), + ) + + +@dataclass(frozen=True, slots=True) +class Unplaced: + @property + def summary(self) -> str: + return "" + + def source_config(self) -> dict[str, object]: + return {} + + +@dataclass(frozen=True, slots=True) +class OnListener: + listener_id: str + k8s_namespace: str + + @property + def summary(self) -> str: + return ( + f"Deploying through listener {self.listener_id} " + f"in namespace {self.k8s_namespace}" + ) + + 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 ensure_not_requested(self, deployment_id: str) -> None: + if self.requested: + raise click.UsageError( + "Listener and namespace are fixed when a deployment is created. " + f"Deployment {deployment_id} already exists, so drop --listener-id " + "and --k8s-namespace, or create a new deployment with a different " + "--name." + ) + + def on(self, listener: Listener) -> Placement: + return OnListener(listener.id, self._namespace(listener)) + + def among(self, listeners: Sequence[Listener]) -> Placement: + if not listeners: + if self.requested: + raise click.UsageError(_NO_LISTENERS) + return Unplaced() + if len(listeners) > 1: + raise click.UsageError( + "This workspace has several listeners. Choose one with " + f"--listener-id:\n{_describe_listeners(listeners)}" + ) + return self.on(listeners[0]) + + 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(listeners: Sequence[Listener]) -> str: + shown = listeners[:_LISTENERS_SHOWN] + lines = [ + f" {listener.id} cluster {listener.compute_id} " + f"namespaces: {', '.join(listener.namespaces)}" + for listener in shown + ] + if len(listeners) > len(shown): + lines.append(f" ... and {len(listeners) - len(shown)} more") + if len(listeners) == MAX_PAGE_SIZE: + lines.append(f" (only the first {MAX_PAGE_SIZE} listeners were read)") + return "\n".join(lines) + + @dataclass(frozen=True, slots=True) class ExistingDeployment: id: str @@ -379,15 +508,16 @@ def _source_of(resource: object) -> str | None: def find_deployment_by_name( client: HostBackendClient, name: str ) -> ExistingDeployment | None: - listed = client.list_deployments(name_contains=name) - resources = listed.get("resources", []) if isinstance(listed, dict) else [] - for resource in resources: - if ( - isinstance(resource, dict) - and resource.get("name") == name - and resource.get("id") - ): + listed = client.list_deployments(name=name, name_contains=name, limit=MAX_PAGE_SIZE) + for resource in listed: + if resource.get("name") == name and resource.get("id"): return ExistingDeployment(str(resource["id"]), _source_of(resource)) + if len(listed) >= MAX_PAGE_SIZE: + raise click.ClickException( + "This workspace has more deployments than the CLI can search, so it " + f"cannot tell whether '{name}' already exists. Pass --deployment-id to " + "update an existing deployment." + ) return None @@ -683,14 +813,22 @@ def _find_deployment( existing = _call_host_backend_with_optional_tenant( client, lambda c: c.list_deployments( - agent_id=selector.agent_id, agent_environment=selector.environment + agent_id=selector.agent_id, + agent_environment=selector.environment, + limit=MAX_PAGE_SIZE, ), ) + if len(existing) > 1: + raise click.ClickException( + "This control plane does not filter deployments by agent, so the " + f"CLI cannot tell which one belongs to '{selector.agent_id}' in " + f"{selector.environment}. Deploy by --name instead." + ) found = next( ( ExistingDeployment(str(dep["id"]), _source_of(dep)) - for dep in existing.get("resources", []) - if not dep.get("is_preview") + for dep in existing + if dep.get("id") and not dep.get("is_preview") ), None, ) @@ -758,21 +896,18 @@ def _create_deployment( def _get_deployment_status_url( - updated: object, deployment_id: str, host_url: str + updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints ) -> str | None: - """Compute the LangSmith dashboard URL for a deployment, if possible.""" tenant_id = updated.get("tenant_id") if isinstance(updated, dict) else None if not tenant_id: return None - base = ControlPlaneEndpoints.from_control_plane_url(host_url).dashboard_url - return f"{base}/o/{tenant_id}/host/deployments/{deployment_id}" + return f"{endpoints.dashboard_url}/o/{tenant_id}/host/deployments/{deployment_id}" def _emit_deployment_status_url( - updated: object, deployment_id: str, host_url: str + updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints ) -> str | None: - """Emit the deployment status URL and return it.""" - url = _get_deployment_status_url(updated, deployment_id, host_url) + url = _get_deployment_status_url(updated, deployment_id, endpoints) if url: _get_emitter().status_url(url) return url @@ -790,14 +925,11 @@ def _poll_revision_status( ) -> tuple[str, str | None]: """Poll latest revision status until terminal status or timeout.""" em = _get_emitter() - revisions_resp = client.list_revisions(deployment_id, limit=1) - resources = ( - revisions_resp.get("resources", []) if isinstance(revisions_resp, dict) else [] - ) - if not resources: + revisions = client.list_revisions(deployment_id, limit=1) + if not revisions: return "", None - revision_id = str(resources[0]["id"]) + revision_id = str(revisions[0]["id"]) last_status = "" deadline = time.time() + timeout_seconds start_time = time.monotonic() @@ -1318,6 +1450,7 @@ def _run_remote_build( @dataclass(frozen=True, slots=True) class DeployContext: client: HostBackendClient + endpoints: ControlPlaneEndpoints spec: BuildSpec verbose: bool selector: DeploymentSelector @@ -1347,19 +1480,66 @@ def _resolve_or_create( ) if found is not None: return found.id, step - created, step = _create_deployment( - ctx.client, - step, - name=ctx.selector.name if isinstance(ctx.selector, ByName) else None, - agent=asdict(ctx.selector) if isinstance(ctx.selector, ByAgent) else None, - source=source, - source_config={"deployment_type": ctx.deployment_type}, - source_revision_config={}, - secrets=ctx.secrets, - ) + try: + created, step = _create_deployment( + ctx.client, + step, + name=ctx.selector.name if isinstance(ctx.selector, ByName) else None, + agent=asdict(ctx.selector) if isinstance(ctx.selector, ByAgent) else None, + source=source, + source_config={"deployment_type": ctx.deployment_type}, + source_revision_config={}, + secrets=ctx.secrets, + ) + except HostBackendError as err: + if _needs_a_listener(err): + raise ListenerRequiredError( + "The image has to come from a registry you manage, so re-run with " + "--push-to /." + ) from None + raise return created.id, step +class ListenerRequiredError(click.UsageError): + def __init__(self, remedy: str) -> None: + super().__init__( + "This workspace deploys through a listener in your own cluster. " + f"{remedy}\nLearn about listeners: {_LISTENERS_DOCS_URL}" + ) + + +def _needs_a_listener(err: HostBackendError) -> bool: + return err.status_code == 400 and _LISTENER_REQUIRED_MARKER in ( + err.detail or err.message + ) + + +def _requested_listener(client: HostBackendClient, listener_id: str) -> Listener: + try: + resource = _call_host_backend_with_optional_tenant( + client, lambda c: c.get_listener(listener_id) + ) + except HostBackendError as err: + if err.status_code not in _LISTENER_NOT_FOUND_STATUSES: + raise + available = _available_listeners(client) + if not available: + raise click.UsageError(_NO_LISTENERS) from None + raise click.UsageError( + f"Listener {listener_id} was not found in this workspace. " + f"Available listeners:\n{_describe_listeners(available)}" + ) from None + return Listener.from_resource(resource) + + +def _available_listeners(client: HostBackendClient) -> tuple[Listener, ...]: + resources = _call_host_backend_with_optional_tenant( + client, lambda c: c.list_listeners() + ) + return tuple(Listener.from_resource(resource) for resource in resources) + + def _ensure_customer_registry_source(existing: ExistingDeployment) -> None: if existing.source != _CUSTOMER_REGISTRY_SOURCE: raise click.UsageError( @@ -1422,6 +1602,7 @@ class RemoteBuildSource: class CustomerRegistrySource: reference: ImageReference prebuilt_image: str | None + requested_placement: RequestedPlacement def run(self, ctx: DeployContext) -> DeployOutcome: if isinstance(ctx.selector, ById): @@ -1443,6 +1624,7 @@ class CustomerRegistrySource: self, ctx: DeployContext, existing: ExistingDeployment, step: int ) -> DeployOutcome: _ensure_customer_registry_source(existing) + self.requested_placement.ensure_not_requested(existing.id) image_uri, step = self._publish(ctx, step) _log_deploy_step(step, f"Updating deployment {existing.id}") updated = ctx.client.update_deployment( @@ -1456,7 +1638,25 @@ class CustomerRegistrySource: existing.id, _image_revision_result(updated, "Deployment updated") ) + def _resolve_placement(self, ctx: DeployContext) -> Placement: + requested = self.requested_placement + if requested.listener_id is not None: + return requested.on(_requested_listener(ctx.client, requested.listener_id)) + if not (ctx.endpoints.is_cloud or requested.requested): + return Unplaced() + return requested.among(_available_listeners(ctx.client)) + + def _announce(self, placement: Placement) -> None: + if isinstance(placement, OnListener): + _get_emitter().info( + placement.summary, + listener_id=placement.listener_id, + k8s_namespace=placement.k8s_namespace, + ) + def _create(self, ctx: DeployContext, name: str | None, step: int) -> DeployOutcome: + placement = self._resolve_placement(ctx) + self._announce(placement) image_uri, step = self._publish(ctx, step) try: created, _ = _create_deployment( @@ -1467,13 +1667,19 @@ class CustomerRegistrySource: if isinstance(ctx.selector, ByAgent) else None, source=_CUSTOMER_REGISTRY_SOURCE, - source_config={"resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC}, + source_config={ + "resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC, + **placement.source_config(), + }, source_revision_config={"image_uri": image_uri}, secrets=ctx.secrets, ) except HostBackendError as err: - if err.status_code == 400 and _LISTENER_REQUIRED_MARKER in err.message: - raise click.ClickException(_HYBRID_LISTENER_GUIDANCE) from None + if _needs_a_listener(err): + raise ListenerRequiredError( + "Re-run with --listener-id and --k8s-namespace.\n" + f"{err.detail or err.message}" + ) from None raise return DeployOutcome( created.id, _image_revision_result(created.resource, "Deployment created") @@ -1534,14 +1740,31 @@ def _select_source( image_name: str | None, tag: str | None, remote_build_flag: bool | None, + placement: RequestedPlacement, + selector: DeploymentSelector, ) -> DeploymentSource: + if push_to is None and placement.requested: + raise click.UsageError( + "--listener-id and --k8s-namespace only apply when creating a " + "deployment with --push-to." + ) + if placement.requested and isinstance(selector, ById): + raise click.UsageError( + "Listener and namespace are fixed when a deployment is created, so " + "they cannot be set for an existing --deployment-id. Drop them, or " + "create a new deployment with --name." + ) if push_to is not None: if remote_build_flag is True: raise click.UsageError("--push-to cannot be combined with --remote.") reference = _push_reference(push_to, tag) if image is None: _require_local_docker() - return CustomerRegistrySource(reference, prebuilt_image=image) + return CustomerRegistrySource( + reference=reference, + prebuilt_image=image, + requested_placement=placement, + ) if image and remote_build_flag is True: raise click.UsageError("--image cannot be combined with --remote builds.") use_remote_build, local_build_error = _resolve_build_mode( @@ -1647,9 +1870,7 @@ def _call_host_backend_with_optional_tenant( prompted_for_tenant = True continue if err.status_code == 403 and "not enabled" in err.message.lower(): - smith_base = ControlPlaneEndpoints.from_control_plane_url( - client.base_url - ).dashboard_url + smith_base = client.endpoints.dashboard_url raise HostBackendError( "LangSmith Deployment is not enabled for this organization. " f"Enable it at {smith_base}/host/deployments" @@ -1854,6 +2075,21 @@ def _deploy_base_options( "Give the tag here or with --tag (default: latest)." ), ), + click.option( + "--listener-id", + help=( + "Listener that will run the deployment, for workspaces that " + "deploy through a listener in your own cluster. Only used when " + "creating a deployment with --push-to." + ), + ), + click.option( + "--k8s-namespace", + help=( + "Kubernetes namespace the listener deploys into. Only used when " + "creating a deployment with --push-to." + ), + ), click.option( "--config", "-c", @@ -1964,6 +2200,8 @@ def _deploy_cmd( image_name: str | None, image: str | None, push_to: str | None, + listener_id: str | None, + k8s_namespace: str | None, tag: str | None, base_image: str | None, install_command: str | None, @@ -2031,12 +2269,15 @@ def _deploy_cmd( secrets = _secrets_from_env(_env_without_deployment_name(env_vars)) + selector = ByAgent(**agent) if agent else deployment_selector(deployment_id, name) source = _select_source( push_to=push_to, image=image, image_name=image_name, tag=tag, remote_build_flag=remote_build_flag, + placement=RequestedPlacement(listener_id, k8s_namespace), + selector=selector, ) client = _create_host_backend_client(host_url, api_key, env_vars=env_vars) @@ -2049,6 +2290,7 @@ def _deploy_cmd( outcome = source.run( DeployContext( client=client, + endpoints=client.endpoints, spec=BuildSpec( config=config, config_json=config_json, @@ -2060,9 +2302,7 @@ def _deploy_cmd( build_command=build_command, ), verbose=verbose, - selector=ByAgent(**agent) - if agent - else deployment_selector(deployment_id, name), + selector=selector, deployment_type=deployment_type, secrets=secrets, tracked_packages=tracked_packages, @@ -2071,7 +2311,7 @@ def _deploy_cmd( dep_status_url = _emit_deployment_status_url( outcome.build_result.updated, outcome.deployment_id, - client.base_url, + client.endpoints, ) if no_wait: @@ -2158,16 +2398,10 @@ def deploy_list( if environment is not None: filters["agent_environment"] = environment client = _create_host_backend_client(host_url, api_key) - response = _call_host_backend_with_optional_tenant( + deployments = _call_host_backend_with_optional_tenant( client, lambda c: c.list_deployments(name_contains=name_contains, **filters), ) - resources = response.get("resources") if isinstance(response, dict) else None - deployments = ( - [item for item in resources if isinstance(item, dict)] - if isinstance(resources, list) - else [] - ) if not deployments: click.echo("No deployments found.") return @@ -2207,16 +2441,10 @@ def deploy_revisions_list( api_key: str | None, host_url: str | None, limit: int, deployment_id: str ) -> None: client = _create_host_backend_client(host_url, api_key) - response = _call_host_backend_with_optional_tenant( + revisions = _call_host_backend_with_optional_tenant( client, lambda c: c.list_revisions(deployment_id, limit=limit), ) - resources = response.get("resources") if isinstance(response, dict) else None - revisions = ( - [item for item in resources if isinstance(item, dict)] - if isinstance(resources, list) - else [] - ) if not revisions: click.echo(f"No revisions found for deployment {deployment_id}.") return @@ -2366,17 +2594,12 @@ def deploy_logs( dep_id = found.id if log_type == "build" and not revision_id: - revisions_resp = client.list_revisions(dep_id, limit=1) - resources = ( - revisions_resp.get("resources", []) - if isinstance(revisions_resp, dict) - else [] - ) - if not resources: + revisions = client.list_revisions(dep_id, limit=1) + if not revisions: raise click.ClickException( "No revisions found for this deployment. Cannot fetch build logs." ) - revision_id = str(resources[0]["id"]) + revision_id = str(revisions[0]["id"]) click.secho(f"Using latest revision: {revision_id}", fg="cyan") payload: dict = {"limit": limit, "order": "desc"} diff --git a/libs/cli/langgraph_cli/host_backend.py b/libs/cli/langgraph_cli/host_backend.py index 81af22d83..4f2afe42a 100644 --- a/libs/cli/langgraph_cli/host_backend.py +++ b/libs/cli/langgraph_cli/host_backend.py @@ -18,6 +18,7 @@ CLOUD_DASHBOARD_HOST = "smith.langchain.com" CONTROL_PLANE_PATH = "/api-host" LANGSMITH_API_PATHS = ("/api/v1", "/api") LOCAL_HOSTNAMES = ("localhost", "127.0.0.1") +MAX_PAGE_SIZE = 100 SourceName = Literal["internal_docker", "internal_source", "external_docker"] @@ -36,6 +37,13 @@ class ControlPlaneEndpoints: return cls.from_langsmith_endpoint(langsmith_endpoint) return cls(CLOUD_CONTROL_PLANE_URL, CLOUD_DASHBOARD_URL) + @property + def is_cloud(self) -> bool: + hostname = urlparse(self.control_plane_url).hostname or "" + return hostname == CLOUD_CONTROL_PLANE_HOST or hostname.endswith( + f".{CLOUD_CONTROL_PLANE_HOST}" + ) + @classmethod def from_control_plane_url(cls, url: str) -> ControlPlaneEndpoints: control_plane_url = url.rstrip("/") @@ -83,12 +91,36 @@ def _without_api_path(path: str) -> str: return path +def _resources(payload: object) -> list[dict[str, Any]]: + if not isinstance(payload, dict): + return [] + resources = payload.get("resources") + if not isinstance(resources, list): + return [] + return [item for item in resources if isinstance(item, dict)] + + class HostBackendError(click.ClickException): """Raised when the host backend returns an error response.""" - def __init__(self, message: str, status_code: int | None = None): + def __init__( + self, + message: str, + status_code: int | None = None, + detail: str | None = None, + ): super().__init__(message) self.status_code = status_code + self.detail = detail + + +def _error_detail(response: httpx.Response) -> str | None: + try: + body = response.json() + except ValueError: + return None + detail = body.get("detail") if isinstance(body, dict) else None + return detail if isinstance(detail, str) else None class HostBackendClient: @@ -110,7 +142,8 @@ class HostBackendClient: } if tenant_id: headers["X-Tenant-ID"] = tenant_id - self._base_url = base_url.rstrip("/") + self._endpoints = ControlPlaneEndpoints.from_control_plane_url(base_url) + self._base_url = self._endpoints.control_plane_url self._client = httpx.Client( base_url=self._base_url, headers=headers, @@ -122,6 +155,10 @@ class HostBackendClient: def base_url(self) -> str: return self._base_url + @property + def endpoints(self) -> ControlPlaneEndpoints: + return self._endpoints + def set_tenant(self, tenant_id: str) -> None: self._client.headers["X-Tenant-ID"] = tenant_id @@ -136,10 +173,12 @@ class HostBackendClient: resp = self._client.request(method, path, json=payload, params=params) resp.raise_for_status() except httpx.HTTPStatusError as err: - detail = err.response.text or str(err.response.status_code) + detail = _error_detail(err.response) + reason = detail or err.response.text or str(err.response.status_code) raise HostBackendError( - f"{method} {path} failed with status {err.response.status_code}: {detail}", + f"{method} {path} failed with status {err.response.status_code}: {reason}", status_code=err.response.status_code, + detail=detail, ) from None except httpx.TransportError as err: raise HostBackendError(str(err)) from None @@ -178,20 +217,29 @@ class HostBackendClient: def list_deployments( self, - name_contains: str = "", *, + name: str | None = None, + name_contains: str | None = None, + limit: int | None = None, agent_id: str | None = None, agent_environment: str | None = None, - ) -> dict[str, Any]: - params = {"name_contains": name_contains} - if agent_id is not None: - params["agent_id"] = agent_id - if agent_environment is not None: - params["agent_environment"] = agent_environment - return self._request( - "GET", - "/v2/deployments", - params=params, + ) -> list[dict[str, Any]]: + given = ( + ("name", name), + ("name_contains", name_contains), + ("limit", limit), + ("agent_id", agent_id), + ("agent_environment", agent_environment), + ) + params = {key: value for key, value in given if value is not None} + return _resources(self._request("GET", "/v2/deployments", params=params)) + + def get_listener(self, listener_id: str) -> dict[str, Any]: + return self._request("GET", f"/v2/listeners/{listener_id}") + + def list_listeners(self) -> list[dict[str, Any]]: + return _resources( + self._request("GET", "/v2/listeners", params={"limit": MAX_PAGE_SIZE}) ) def get_deployment(self, deployment_id: str) -> dict[str, Any]: @@ -266,10 +314,15 @@ class HostBackendClient: payload["secrets"] = secrets return self._request("PATCH", f"/v2/deployments/{deployment_id}", payload) - def list_revisions(self, deployment_id: str, limit: int = 1) -> dict[str, Any]: - return self._request( - "GET", - f"/v2/deployments/{deployment_id}/revisions?limit={limit}", + def list_revisions( + self, deployment_id: str, limit: int = 1 + ) -> list[dict[str, Any]]: + return _resources( + self._request( + "GET", + f"/v2/deployments/{deployment_id}/revisions", + params={"limit": limit}, + ) ) def get_revision(self, deployment_id: str, revision_id: str) -> dict[str, Any]: diff --git a/libs/cli/tests/unit_tests/cli/test_cli.py b/libs/cli/tests/unit_tests/cli/test_cli.py index ca0ec7a04..8cce48db8 100644 --- a/libs/cli/tests/unit_tests/cli/test_cli.py +++ b/libs/cli/tests/unit_tests/cli/test_cli.py @@ -382,20 +382,18 @@ def test_deploy_list_command(monkeypatch) -> None: def list_deployments(self, name_contains: str = ""): captured["name_contains"] = name_contains - return { - "resources": [ - { - "id": "dep-123", - "name": "alpha", - "source_config": {"custom_url": "https://alpha.example.com"}, - }, - { - "id": "dep-456", - "name": "beta", - "source_config": {"custom_url": "https://beta.example.com"}, - }, - ] - } + return [ + { + "id": "dep-123", + "name": "alpha", + "source_config": {"custom_url": "https://alpha.example.com"}, + }, + { + "id": "dep-456", + "name": "beta", + "source_config": {"custom_url": "https://beta.example.com"}, + }, + ] monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient) @@ -435,7 +433,7 @@ def test_deploy_list_command_no_results(monkeypatch) -> None: pass def list_deployments(self, name_contains: str = ""): - return {"resources": []} + return [] monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient) @@ -468,20 +466,18 @@ def test_deploy_revisions_list_command(monkeypatch) -> None: def list_revisions(self, deployment_id: str, limit: int = 1): captured["deployment_id"] = deployment_id captured["limit"] = str(limit) - return { - "resources": [ - { - "id": "rev-123", - "status": "CREATING", - "created_at": "2023-11-07T05:31:56Z", - }, - { - "id": "rev-456", - "status": "DEPLOYED", - "created_at": "2023-11-08T10:00:00Z", - }, - ] - } + return [ + { + "id": "rev-123", + "status": "CREATING", + "created_at": "2023-11-07T05:31:56Z", + }, + { + "id": "rev-456", + "status": "DEPLOYED", + "created_at": "2023-11-08T10:00:00Z", + }, + ] monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient) @@ -522,7 +518,7 @@ def test_deploy_revisions_list_command_no_results(monkeypatch) -> None: pass def list_revisions(self, deployment_id: str, limit: int = 1): - return {"resources": []} + return [] monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient) @@ -555,7 +551,7 @@ def test_deploy_revisions_list_command_with_explicit_limit(monkeypatch) -> None: def list_revisions(self, deployment_id: str, limit: int = 1): captured["deployment_id"] = deployment_id captured["limit"] = str(limit) - return {"resources": []} + return [] monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient) diff --git a/libs/cli/tests/unit_tests/cli/test_deploy_command.py b/libs/cli/tests/unit_tests/cli/test_deploy_command.py index fd9817633..457725e37 100644 --- a/libs/cli/tests/unit_tests/cli/test_deploy_command.py +++ b/libs/cli/tests/unit_tests/cli/test_deploy_command.py @@ -1,5 +1,6 @@ import asyncio import json +import uuid from collections.abc import Callable, Iterator from contextlib import contextmanager from dataclasses import dataclass, field @@ -17,6 +18,7 @@ from langgraph_cli.host_backend import HostBackendClient from langgraph_cli.image_reference import ImageReference CONTROL_PLANE_URL = "https://control-plane.example.com" +CLOUD_CONTROL_PLANE_URL = "https://api.host.langchain.com" REGISTRY_URL = "https://registry.example.com/team" PUSH_TOKEN = "push-token" PUSHED_IMAGE = "registry.example.com/team/my-app:latest" @@ -24,10 +26,25 @@ PUSHED_DIGEST = "registry.example.com/team/my-app@sha256:abc123" PUSH_REPOSITORY = "registry.example.com/team/agent" EXTERNAL_IMAGE = f"{PUSH_REPOSITORY}:latest" EXTERNAL_DIGEST = f"{PUSH_REPOSITORY}@sha256:abc123" -LISTENER_REQUIRED = ( - "Source configuration error: 'source_config.listener_id' is required for " - "workspace with available listener IDs: ['listener-1']" -) +LISTENER_ID = "11111111-1111-4111-8111-111111111111" +OTHER_LISTENER_ID = "22222222-2222-4222-8222-222222222222" +PAGE_TWO_LISTENER_ID = "33333333-3333-4333-8333-333333333333" +UNKNOWN_LISTENER_ID = "99999999-9999-4999-8999-999999999999" +LISTENER = { + "id": LISTENER_ID, + "compute_id": "prod-cluster", + "compute_config": {"k8s_namespaces": ["agents"]}, +} +OTHER_LISTENER = { + "id": OTHER_LISTENER_ID, + "compute_id": "other-cluster", + "compute_config": {"k8s_namespaces": ["agents"]}, +} +TWO_NAMESPACE_LISTENER = { + "id": LISTENER_ID, + "compute_id": "prod-cluster", + "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]}, +} CREATED_ID = "dep-created" TRACKED_PACKAGES = ["langgraph:1.0.0"] SIGNED_UPLOAD_URL = "https://storage.example.com/signed" @@ -38,7 +55,12 @@ DIGESTS_FORMAT = "{{json .RepoDigests}}" NOT_A_CLI_DEPLOYMENT = ( "push token is only available for 'internal_docker' source deployments" ) +LISTENER_REQUIRED = ( + "Source configuration error: 'source_config.listener_id' is required " + f"for workspace with available listener IDs: ['{LISTENER_ID}']" +) LIST_DEPLOYMENTS = "GET /v2/deployments" +LIST_LISTENERS = "GET /v2/listeners" CREATE_DEPLOYMENT = "POST /v2/deployments" @@ -58,12 +80,22 @@ def _get(deployment_id: str) -> str: return f"GET /v2/deployments/{deployment_id}" +def _looks_like_a_uuid(value: str) -> bool: + try: + uuid.UUID(value) + except ValueError: + return False + return True + + @dataclass class ControlPlaneDouble: timeline: list[str] existing_deployments: list[dict] = field(default_factory=list) push_token_status: int = 200 create_error: str | None = None + listeners: list[dict] = field(default_factory=list) + listeners_by_id: dict[str, dict] = field(default_factory=dict) bodies: dict[str, dict] = field(default_factory=dict) def handle(self, request: httpx.Request) -> httpx.Response: @@ -71,14 +103,45 @@ 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 == "GET" and path.startswith("/v2/listeners/"): + listener_id = path.rsplit("/", 1)[-1] + if not _looks_like_a_uuid(listener_id): + return httpx.Response( + 422, + json={ + "detail": [ + {"type": "uuid_parsing", "loc": ["path", "listener_id"]} + ] + }, + ) + known = {listener["id"]: listener for listener in self.listeners} + known.update(self.listeners_by_id) + if listener_id not in known: + return httpx.Response( + 404, json={"detail": f"Listener ID {listener_id} not found."} + ) + return httpx.Response(200, json=known[listener_id]) 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) + return httpx.Response(400, json={"detail": self.create_error}) return httpx.Response(201, json={"id": CREATED_ID, "tenant_id": "tenant-1"}) if path.endswith("/push-token"): if self.push_token_status != 200: @@ -199,7 +262,7 @@ class DeployProject: timeline: list[str] uploads: list[tuple[str, str, int]] - def run(self, *args: str) -> Result: + def run(self, *args: str, host_url: str = CONTROL_PLANE_URL) -> Result: return CliRunner().invoke( cli, [ @@ -207,7 +270,7 @@ class DeployProject: "--api-key", "test-key", "--host-url", - CONTROL_PLANE_URL, + host_url, "--name", "my-app", "--no-input", @@ -611,18 +674,6 @@ def test_push_to_rejects_a_non_external_deployment_before_any_docker_work( assert deploy_project.docker.verbs() == [] -def test_push_to_explains_the_listener_requirement_of_hybrid_workspaces( - deploy_project: DeployProject, -) -> None: - deploy_project.control_plane.create_error = LISTENER_REQUIRED - - result = deploy_project.run("--push-to", PUSH_REPOSITORY) - - assert result.exit_code != 0 - assert "listener" in result.output - assert "--deployment-id" in result.output - - def test_push_to_with_deployment_id_fetches_the_deployment_once( deploy_project: DeployProject, ) -> None: @@ -652,3 +703,414 @@ def test_invalid_tag_fails_before_any_control_plane_call( assert result.exit_code != 0 assert "Image tag may only contain" in result.output assert deploy_project.timeline == [] + + +def test_push_to_places_a_new_deployment_on_the_only_listener( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER] + + result = deploy_project.run( + "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL + ) + + assert result.exit_code == 0, result.output + assert deploy_project.timeline == [ + LIST_DEPLOYMENTS, + LIST_LISTENERS, + "docker build", + "docker push", + "docker inspect-digest", + CREATE_DEPLOYMENT, + ] + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == { + "resource_spec": {}, + "listener_id": LISTENER_ID, + "listener_config": {"k8s_namespace": "agents"}, + } + assert f"Deploying through listener {LISTENER_ID} in namespace agents" in ( + result.output + ) + + +def test_push_to_places_a_new_deployment_on_the_chosen_listener( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER, OTHER_LISTENER] + + result = deploy_project.run( + "--push-to", + PUSH_REPOSITORY, + "--listener-id", + OTHER_LISTENER_ID, + "--k8s-namespace", + "agents", + host_url=CLOUD_CONTROL_PLANE_URL, + ) + + assert result.exit_code == 0, result.output + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == { + "resource_spec": {}, + "listener_id": OTHER_LISTENER_ID, + "listener_config": {"k8s_namespace": "agents"}, + } + + +@pytest.mark.parametrize( + ("listeners", "args", "message"), + [ + pytest.param( + [LISTENER, OTHER_LISTENER], (), "--listener-id", id="two_listeners" + ), + pytest.param( + [TWO_NAMESPACE_LISTENER], (), "--k8s-namespace", id="two_namespaces" + ), + pytest.param( + [LISTENER], + ("--k8s-namespace", "nope"), + "does not serve namespace", + id="unknown_namespace", + ), + ], +) +def test_push_to_refuses_an_unresolved_placement_before_any_docker_work( + deploy_project: DeployProject, listeners, args, message +) -> None: + deploy_project.control_plane.listeners = listeners + + result = deploy_project.run( + "--push-to", PUSH_REPOSITORY, *args, host_url=CLOUD_CONTROL_PLANE_URL + ) + + assert result.exit_code != 0 + assert message in result.output + assert deploy_project.docker.verbs() == [] + assert CREATE_DEPLOYMENT not in deploy_project.timeline + + +def test_self_hosted_control_plane_keeps_its_default_placement( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER] + + result = deploy_project.run("--push-to", PUSH_REPOSITORY) + + assert result.exit_code == 0, result.output + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == { + "resource_spec": {} + } + + +def test_self_hosted_control_plane_places_when_asked( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER] + + result = deploy_project.run( + "--push-to", PUSH_REPOSITORY, "--listener-id", LISTENER_ID + ) + + assert result.exit_code == 0, result.output + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == { + "resource_spec": {}, + "listener_id": LISTENER_ID, + "listener_config": {"k8s_namespace": "agents"}, + } + + +def test_updating_a_deployment_never_looks_up_listeners( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER] + deploy_project.control_plane.existing_deployments = [ + {"id": "dep-ext", "name": "my-app", "source": "external_docker"} + ] + + result = deploy_project.run( + "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL + ) + + assert result.exit_code == 0, result.output + assert LIST_LISTENERS not in deploy_project.timeline + + +def test_listener_flags_are_refused_for_a_deployment_id_without_any_call( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run( + "--push-to", + PUSH_REPOSITORY, + "--deployment-id", + "dep-ext", + "--k8s-namespace", + "agents", + host_url=CLOUD_CONTROL_PLANE_URL, + ) + + assert result.exit_code != 0 + assert "fixed when a deployment is created" in result.output + assert deploy_project.timeline == [] + + +def test_listener_flags_are_refused_on_an_existing_deployment( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER] + deploy_project.control_plane.existing_deployments = [ + {"id": "dep-ext", "name": "my-app", "source": "external_docker"} + ] + + result = deploy_project.run( + "--push-to", + PUSH_REPOSITORY, + "--listener-id", + LISTENER_ID, + host_url=CLOUD_CONTROL_PLANE_URL, + ) + + assert result.exit_code != 0 + assert "fixed when a deployment is created" in result.output + assert deploy_project.docker.verbs() == [] + + +def test_a_deployment_without_a_listener_announces_nothing( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run("--push-to", PUSH_REPOSITORY) + + assert result.exit_code == 0, result.output + assert "listener" not in result.output + + +def test_a_self_hosted_create_without_flags_never_looks_up_listeners( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER] + + result = deploy_project.run("--push-to", PUSH_REPOSITORY) + + assert result.exit_code == 0, result.output + assert LIST_LISTENERS not in deploy_project.timeline + + +def test_a_control_plane_that_demands_a_listener_names_the_flags( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.create_error = LISTENER_REQUIRED + + result = deploy_project.run("--push-to", PUSH_REPOSITORY) + + assert result.exit_code != 0 + assert "--listener-id" in result.output + assert "--k8s-namespace" in result.output + assert LISTENER_ID in result.output + assert "{" not in result.output + assert "POST /v2/deployments failed" not in result.output + + +def test_listener_flags_without_push_to_make_no_call_at_all( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run("--listener-id", LISTENER_ID) + + assert result.exit_code != 0 + assert "--push-to" in result.output + assert deploy_project.timeline == [] + + +def test_a_truncated_listener_page_says_so(deploy_project: DeployProject) -> None: + deploy_project.control_plane.listeners = [ + { + "id": str(uuid.UUID(int=index)), + "compute_id": "cluster", + "compute_config": {"k8s_namespaces": ["agents"]}, + } + for index in range(100) + ] + + result = deploy_project.run( + "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL + ) + + assert result.exit_code != 0 + assert "first 100" in result.output + + +def test_a_managed_build_in_a_listener_workspace_points_at_push_to( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.create_error = LISTENER_REQUIRED + + result = deploy_project.run("--no-remote") + + assert result.exit_code != 0 + assert "--push-to" in result.output + assert deploy_project.docker.verbs() == [] + + +@pytest.mark.parametrize( + "args", + [ + pytest.param(("--no-remote",), id="managed_build"), + pytest.param(("--push-to", PUSH_REPOSITORY), id="push_to"), + ], +) +def test_a_listener_requirement_links_the_listener_docs( + deploy_project: DeployProject, args: tuple[str, ...] +) -> None: + deploy_project.control_plane.create_error = LISTENER_REQUIRED + + result = deploy_project.run(*args) + + assert result.exit_code != 0 + assert "https://docs.langchain.com/langsmith/control-plane#listeners" in ( + result.output + ) + + +def test_a_managed_control_plane_without_listeners_creates_as_before( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run( + "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL + ) + + assert result.exit_code == 0, result.output + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == { + "resource_spec": {} + } + assert deploy_project.timeline.count(LIST_LISTENERS) == 1 + + +def test_a_listener_without_an_id_is_reported_rather_than_ignored( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [ + {"compute_id": "broken", "compute_config": {"k8s_namespaces": ["agents"]}}, + LISTENER, + ] + + result = deploy_project.run( + "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL + ) + + assert result.exit_code != 0 + assert "without an id" in result.output + assert deploy_project.docker.verbs() == [] + + +def _listener_route(listener_id: str) -> str: + return f"GET /v2/listeners/{listener_id}" + + +def test_an_explicit_listener_is_fetched_by_id_not_searched( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER, OTHER_LISTENER] + + result = deploy_project.run( + "--push-to", + PUSH_REPOSITORY, + "--listener-id", + OTHER_LISTENER_ID, + host_url=CLOUD_CONTROL_PLANE_URL, + ) + + assert result.exit_code == 0, result.output + assert _listener_route(OTHER_LISTENER_ID) in deploy_project.timeline + assert LIST_LISTENERS not in deploy_project.timeline + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == { + "resource_spec": {}, + "listener_id": OTHER_LISTENER_ID, + "listener_config": {"k8s_namespace": "agents"}, + } + + +def test_an_explicit_listener_beyond_the_first_page_still_works( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [ + { + "id": str(uuid.UUID(int=index)), + "compute_id": "cluster", + "compute_config": {"k8s_namespaces": ["agents"]}, + } + for index in range(100) + ] + deploy_project.control_plane.listeners_by_id = { + PAGE_TWO_LISTENER_ID: { + "id": PAGE_TWO_LISTENER_ID, + "compute_id": "far-cluster", + "compute_config": {"k8s_namespaces": ["agents"]}, + } + } + + result = deploy_project.run( + "--push-to", + PUSH_REPOSITORY, + "--listener-id", + PAGE_TWO_LISTENER_ID, + host_url=CLOUD_CONTROL_PLANE_URL, + ) + + assert result.exit_code == 0, result.output + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == { + "resource_spec": {}, + "listener_id": PAGE_TWO_LISTENER_ID, + "listener_config": {"k8s_namespace": "agents"}, + } + + +def test_an_unknown_listener_names_the_ones_that_exist( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER] + + result = deploy_project.run( + "--push-to", + PUSH_REPOSITORY, + "--listener-id", + UNKNOWN_LISTENER_ID, + host_url=CLOUD_CONTROL_PLANE_URL, + ) + + assert result.exit_code != 0 + assert "was not found" in result.output + assert LISTENER_ID in result.output + assert "prod-cluster" in result.output + assert deploy_project.docker.verbs() == [] + + +def test_an_explicit_listener_in_a_workspace_without_any_is_refused( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run( + "--push-to", + PUSH_REPOSITORY, + "--listener-id", + LISTENER_ID, + host_url=CLOUD_CONTROL_PLANE_URL, + ) + + assert result.exit_code != 0 + assert "no listeners" in result.output + assert deploy_project.docker.verbs() == [] + + +def test_a_listener_id_that_is_not_an_identifier_still_names_the_real_ones( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER] + + result = deploy_project.run( + "--push-to", + PUSH_REPOSITORY, + "--listener-id", + "not-a-listener", + host_url=CLOUD_CONTROL_PLANE_URL, + ) + + assert result.exit_code != 0 + assert "was not found" in result.output + assert LISTENER_ID in result.output + assert "uuid_parsing" not in result.output diff --git a/libs/cli/tests/unit_tests/test_deploy_agents.py b/libs/cli/tests/unit_tests/test_deploy_agents.py index 2f37627cc..334c852b1 100644 --- a/libs/cli/tests/unit_tests/test_deploy_agents.py +++ b/libs/cli/tests/unit_tests/test_deploy_agents.py @@ -72,9 +72,9 @@ def test_agent_create(deployment_api, tmp_path, monkeypatch): result = CliRunner().invoke(cli, AGENT_ARGS) assert result.exit_code == 0, result.output assert dict(requests[0].url.params) == { - "name_contains": "", "agent_id": "customer-support", "agent_environment": "staging", + "limit": "100", } payload = json.loads(requests[1].content) assert payload["agent"] == { @@ -103,3 +103,17 @@ def test_agent_rejects_explicit_name(deployment_api, monkeypatch): assert result.exit_code == 2 assert "cannot be combined" in result.output assert not requests + + +def test_agent_lookup_refuses_a_control_plane_that_ignores_the_filter(deployment_api): + state, requests, _ = deployment_api + state["resources"] = [ + {"id": "someone-elses", "is_preview": False}, + {"id": "another", "is_preview": False}, + ] + + result = CliRunner().invoke(cli, AGENT_ARGS) + + assert result.exit_code != 0 + assert "does not filter deployments by agent" in result.output + assert len(requests) == 1 diff --git a/libs/cli/tests/unit_tests/test_deploy_helpers.py b/libs/cli/tests/unit_tests/test_deploy_helpers.py index 62c34625b..02dce8679 100644 --- a/libs/cli/tests/unit_tests/test_deploy_helpers.py +++ b/libs/cli/tests/unit_tests/test_deploy_helpers.py @@ -13,10 +13,17 @@ import pytest import langgraph_cli.deploy as deploy_mod from langgraph_cli.deploy import ( + ById, + ByName, CustomerRegistrySource, DockerBuildCommand, + ExistingDeployment, + Listener, ManagedRegistrySource, + OnListener, RemoteBuildSource, + RequestedPlacement, + Unplaced, _call_host_backend_with_optional_tenant, _create_host_backend_client, _docker_config_for_token, @@ -27,6 +34,7 @@ from langgraph_cli.deploy import ( _resolve_pushed_image_digest, _select_source, _validate_prebuilt_image, + find_deployment_by_name, normalize_image_tag, normalize_name, ) @@ -280,11 +288,13 @@ class TestCallHostBackendWithOptionalTenant: return c def test_success_passes_through(self): - client = self._make_client(lambda req: httpx.Response(200, json={"ok": True})) + client = self._make_client( + lambda req: httpx.Response(200, json={"resources": [{"id": "dep-1"}]}) + ) result = _call_host_backend_with_optional_tenant( client, lambda c: c.list_deployments() ) - assert result == {"ok": True} + assert result == [{"id": "dep-1"}] def test_403_not_enabled_gives_actionable_error(self): detail = ( @@ -607,6 +617,8 @@ class TestSelectSource: "image_name": None, "tag": None, "remote_build_flag": None, + "placement": RequestedPlacement(), + "selector": ByName("my-app"), } REPOSITORY = "registry.example.com/app" @@ -617,7 +629,9 @@ class TestSelectSource: {"push_to": REPOSITORY}, True, CustomerRegistrySource( - ImageReference(REPOSITORY, "latest"), prebuilt_image=None + reference=ImageReference(REPOSITORY, "latest"), + prebuilt_image=None, + requested_placement=RequestedPlacement(), ), id="push_to_selects_the_external_source_with_the_default_tag", ), @@ -625,7 +639,9 @@ class TestSelectSource: {"push_to": f"{REPOSITORY}:v2"}, True, CustomerRegistrySource( - ImageReference(REPOSITORY, "v2"), prebuilt_image=None + reference=ImageReference(REPOSITORY, "v2"), + prebuilt_image=None, + requested_placement=RequestedPlacement(), ), id="push_to_keeps_a_tag_given_in_the_reference", ), @@ -633,7 +649,9 @@ class TestSelectSource: {"push_to": REPOSITORY, "tag": "v3"}, True, CustomerRegistrySource( - ImageReference(REPOSITORY, "v3"), prebuilt_image=None + reference=ImageReference(REPOSITORY, "v3"), + prebuilt_image=None, + requested_placement=RequestedPlacement(), ), id="tag_flag_composes_with_push_to", ), @@ -641,10 +659,25 @@ class TestSelectSource: {"push_to": REPOSITORY, "image": "app:dev"}, False, CustomerRegistrySource( - ImageReference(REPOSITORY, "latest"), prebuilt_image="app:dev" + reference=ImageReference(REPOSITORY, "latest"), + prebuilt_image="app:dev", + requested_placement=RequestedPlacement(), ), id="prebuilt_image_is_retagged_for_push_to_without_docker_checks", ), + pytest.param( + { + "push_to": REPOSITORY, + "placement": RequestedPlacement("listener-1", "agents"), + }, + True, + CustomerRegistrySource( + reference=ImageReference(REPOSITORY, "latest"), + prebuilt_image=None, + requested_placement=RequestedPlacement("listener-1", "agents"), + ), + id="push_to_carries_the_requested_placement", + ), pytest.param( {"remote_build_flag": True}, True, @@ -720,6 +753,16 @@ class TestSelectSource: "--image cannot be combined with --remote builds.", id="image_with_remote", ), + pytest.param( + {"placement": RequestedPlacement(listener_id="listener-1")}, + "only apply when creating a deployment with --push-to", + id="listener_without_push_to", + ), + pytest.param( + {"placement": RequestedPlacement(k8s_namespace="agents")}, + "only apply when creating a deployment with --push-to", + id="namespace_without_push_to", + ), ], ) def test_conflicting_flags_are_rejected(self, monkeypatch, flags, message): @@ -890,3 +933,289 @@ class TestResolvePushedImageDigest: frame_locals = captured["coro"].cr_frame.f_locals assert "--config" not in frame_locals["args"] captured["coro"].close() + + +class TestListener: + @pytest.mark.parametrize( + ("resource", "expected"), + [ + pytest.param( + { + "id": "listener-1", + "compute_id": "prod-cluster", + "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]}, + }, + Listener("listener-1", "prod-cluster", ("agents", "agents-staging")), + id="reads_id_cluster_and_namespaces", + ), + pytest.param( + {"id": "listener-1", "compute_id": "c", "compute_config": {}}, + Listener("listener-1", "c", ()), + id="missing_namespaces", + ), + pytest.param( + {"id": "listener-1", "compute_id": "c", "compute_config": None}, + Listener("listener-1", "c", ()), + id="null_compute_config", + ), + pytest.param( + {"id": "listener-1"}, + Listener("listener-1", "", ()), + id="only_an_id", + ), + ], + ) + 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", "expected"), + [ + pytest.param( + RequestedPlacement(), (), Unplaced(), id="no_listeners_no_request" + ), + pytest.param( + RequestedPlacement(), + (ONE_NAMESPACE,), + OnListener("listener-1", "agents"), + id="uses_the_only_possible_answer", + ), + pytest.param( + RequestedPlacement(k8s_namespace="agents-staging"), + (TWO_NAMESPACES,), + OnListener("listener-2", "agents-staging"), + id="namespace_alone_picks_the_only_listener", + ), + ], + ) + def test_resolves_to_a_placement(self, request_, listeners, expected): + assert request_.among(listeners) == 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(k8s_namespace="agents"), + (), + "no listeners", + id="namespace_without_any_listener", + ), + pytest.param( + RequestedPlacement(), + (TWO_NAMESPACES,), + "--k8s-namespace", + id="several_namespaces_need_a_choice", + ), + ], + ) + def test_refuses_and_names_the_choices(self, request_, listeners, message): + with pytest.raises(click.UsageError, match=message): + request_.among(listeners) + + def test_the_error_lists_every_listener_with_its_cluster_and_namespaces(self): + with pytest.raises(click.UsageError) as error: + RequestedPlacement().among((ONE_NAMESPACE, TWO_NAMESPACES)) + + 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 + + +def test_finding_a_deployment_by_name_narrows_the_search_for_every_server_version(): + 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", + "name_contains": "agent", + "limit": "100", + } + assert found == ExistingDeployment("dep-1", "github") + + +def test_a_server_that_ignores_the_exact_name_filter_never_matches_another_deployment(): + client = HostBackendClient( + "https://api.example.com", + "key", + transport=httpx.MockTransport( + lambda req: httpx.Response( + 200, + json={ + "resources": [ + { + "id": "dep-other", + "name": "another-teams-agent", + "source": "external_docker", + } + ] + }, + ) + ), + ) + + assert find_deployment_by_name(client, "brand-new-agent") is None + + +def test_a_full_page_without_a_match_refuses_to_claim_the_name_is_free(): + page = [ + {"id": f"dep-{index}", "name": f"other-agent-{index}"} for index in range(100) + ] + client = HostBackendClient( + "https://api.example.com", + "key", + transport=httpx.MockTransport( + lambda req: httpx.Response(200, json={"resources": page}) + ), + ) + + with pytest.raises(click.ClickException, match="--deployment-id"): + find_deployment_by_name(client, "brand-new-agent") + + +def test_a_partial_page_without_a_match_means_the_name_is_free(): + client = HostBackendClient( + "https://api.example.com", + "key", + transport=httpx.MockTransport( + lambda req: httpx.Response( + 200, json={"resources": [{"id": "dep-1", "name": "other"}]} + ) + ), + ) + + assert find_deployment_by_name(client, "brand-new-agent") is None + + +@pytest.mark.parametrize( + "resource", + [ + pytest.param({"compute_id": "c"}, id="no_id"), + pytest.param({"id": ""}, id="empty_id"), + ], +) +def test_a_listener_without_an_id_is_refused(resource): + with pytest.raises(HostBackendError, match="without an id"): + Listener.from_resource(resource) + + +def test_a_deployment_id_with_listener_flags_is_refused_without_probing_docker( + monkeypatch, +): + def explode() -> tuple[bool, str | None]: + raise AssertionError("docker must not be probed for an argv-only conflict") + + monkeypatch.setattr(deploy_mod, "can_build_locally", explode) + + with pytest.raises(click.UsageError, match="--deployment-id"): + _select_source( + push_to="registry.example.com/app", + image=None, + image_name=None, + tag=None, + remote_build_flag=None, + placement=RequestedPlacement(listener_id="listener-1"), + selector=ById("dep-1"), + ) + + +class TestPlacementOnAKnownListener: + @pytest.mark.parametrize( + ("request_", "listener", "expected"), + [ + pytest.param( + RequestedPlacement(listener_id="listener-1"), + ONE_NAMESPACE, + OnListener("listener-1", "agents"), + id="the_only_namespace_is_used", + ), + pytest.param( + RequestedPlacement(listener_id="listener-2", k8s_namespace="agents"), + TWO_NAMESPACES, + OnListener("listener-2", "agents"), + id="the_chosen_namespace_is_used", + ), + ], + ) + def test_places_on_the_listener(self, request_, listener, expected): + assert request_.on(listener) == expected + + @pytest.mark.parametrize( + ("request_", "listener", "message"), + [ + pytest.param( + RequestedPlacement(listener_id="listener-2"), + 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_namespaces(self, request_, listener, message): + with pytest.raises(click.UsageError, match=message): + request_.on(listener) diff --git a/libs/cli/tests/unit_tests/test_host_backend.py b/libs/cli/tests/unit_tests/test_host_backend.py index 686b516c3..bf9f922f7 100644 --- a/libs/cli/tests/unit_tests/test_host_backend.py +++ b/libs/cli/tests/unit_tests/test_host_backend.py @@ -79,19 +79,6 @@ def test_request_transport_error_raises(): c._request("GET", "/test") -def test_list_deployments_sends_query_params(): - def handler(req: httpx.Request) -> httpx.Response: - assert req.url.path == "/v2/deployments" - assert req.url.params["name_contains"] == "my app" - return httpx.Response(200, json={"ok": True}) - - c = HostBackendClient( - "https://api.example.com", "test-key", transport=httpx.MockTransport(handler) - ) - result = c.list_deployments("my app") - assert result == {"ok": True} - - def _capturing_client(captured: dict) -> HostBackendClient: def handler(req: httpx.Request) -> httpx.Response: captured["body"] = req.read() @@ -421,7 +408,7 @@ def test_injected_transport_receives_requests_under_the_prefixed_base_url(): transport=httpx.MockTransport(handler), ) - assert c.list_revisions("dep-1", limit=2) == {"ok": True} + assert c.list_revisions("dep-1", limit=2) == [] assert seen == { "url": "https://smith.example.com/api-host/v2/deployments/dep-1/revisions?limit=2", "api_key": "key", @@ -546,3 +533,144 @@ def test_control_plane_endpoints_resolve(host_url, langsmith_endpoint, expected) endpoints = ControlPlaneEndpoints.resolve(host_url, langsmith_endpoint) assert (endpoints.control_plane_url, endpoints.dashboard_url) == expected + + +@pytest.mark.parametrize( + ("payload", "expected"), + [ + pytest.param( + {"resources": [{"id": "a"}, {"id": "b"}]}, + [{"id": "a"}, {"id": "b"}], + id="list_returns_the_resources", + ), + pytest.param({"resources": []}, [], id="empty_list"), + pytest.param({}, [], id="missing_key"), + pytest.param({"resources": None}, [], id="null_resources"), + pytest.param( + {"resources": ["nope", {"id": "a"}]}, [{"id": "a"}], id="skips_non_objects" + ), + pytest.param([], [], id="unexpected_envelope"), + ], +) +def test_list_endpoints_return_resource_objects(payload, expected): + def handler(req: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=payload) + + c = HostBackendClient( + "https://api.example.com", "key", transport=httpx.MockTransport(handler) + ) + + assert c.list_deployments() == expected + + +def test_list_listeners_asks_for_a_full_page(): + seen: dict = {} + + def handler(req: httpx.Request) -> httpx.Response: + seen["url"] = str(req.url) + return httpx.Response(200, json={"resources": [{"id": "listener-1"}]}) + + c = HostBackendClient( + "https://api.example.com", "key", transport=httpx.MockTransport(handler) + ) + + assert c.list_listeners() == [{"id": "listener-1"}] + assert seen["url"] == "https://api.example.com/v2/listeners?limit=100" + + +@pytest.mark.parametrize( + ("control_plane_url", "expected"), + [ + pytest.param("https://api.host.langchain.com", True, id="cloud"), + pytest.param("https://eu.api.host.langchain.com", True, id="cloud_region"), + pytest.param("https://dev.api.host.langchain.com", True, id="cloud_dev"), + pytest.param("https://smith.example.com/api-host", False, id="self_hosted"), + pytest.param( + "https://corp.example.com/langsmith/api-host", + False, + id="self_hosted_prefix", + ), + pytest.param("http://localhost:8080/api-host", False, id="local"), + pytest.param( + "https://evil-api.host.langchain.com", False, id="lookalike_needs_a_dot" + ), + ], +) +def test_is_cloud_recognises_the_managed_control_plane(control_plane_url, expected): + 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(), + {}, + id="no_filter_sends_no_parameters", + ), + pytest.param( + lambda c: c.list_deployments( + name="agent", name_contains="agent", limit=100 + ), + {"name": "agent", "name_contains": "agent", "limit": "100"}, + id="both_filters_travel_together_for_older_servers", + ), + ], +) +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 + + +@pytest.mark.parametrize( + ("body", "expected"), + [ + pytest.param( + {"detail": "Source configuration error: bad listener"}, + "Source configuration error: bad listener", + id="fastapi_detail_is_unwrapped", + ), + pytest.param( + {"detail": {"loc": ["body"], "msg": "nope"}}, + None, + id="a_structured_detail_is_left_alone", + ), + pytest.param({"other": "shape"}, None, id="an_unknown_shape_is_left_alone"), + ], +) +def test_error_detail_is_readable(body, expected): + c = HostBackendClient( + "https://api.example.com", + "key", + transport=httpx.MockTransport(lambda req: httpx.Response(400, json=body)), + ) + + with pytest.raises(HostBackendError) as error: + c.get_deployment("dep-1") + + assert error.value.detail == expected + if expected is not None: + assert error.value.message.endswith(expected)