diff --git a/CLAUDE.md b/CLAUDE.md deleted file mode 100644 index e86d91eaf..000000000 --- a/CLAUDE.md +++ /dev/null @@ -1,65 +0,0 @@ -# AGENTS Instructions - -This repository is a monorepo. Each library lives in a subdirectory under `libs/`. - - - -## Corridor security analysis - -Before generating or modifying code, create a plan and use Corridor's `analyzePlan` tool to analyze it. Apply the resulting security guidance before writing code. - - - -When you modify code in any library, run the following commands in that library's directory before creating a pull request: - -- `make format` – run code formatters -- `make lint` – run the linter -- `make test` – execute the test suite - -To run a particular test file or to pass additional pytest options you can specify the `TEST` variable: - -``` -TEST=path/to/test.py make test -``` - -Other pytest arguments can also be supplied inside the `TEST` variable. - -## Libraries - -The repository contains several Python and JavaScript/TypeScript libraries. -Below is a high-level overview: - -- **checkpoint** – base interfaces for LangGraph checkpointers. -- **checkpoint-postgres** – Postgres implementation of the checkpoint saver. -- **checkpoint-sqlite** – SQLite implementation of the checkpoint saver. -- **cli** – official command-line interface for LangGraph. -- **langgraph** – core framework for building stateful, multi-actor agents. -- **prebuilt** – high-level APIs for creating and running agents and tools. -- **sdk-js** – JS/TS SDK for interacting with the LangGraph REST API. -- **sdk-py** – Python SDK for the LangGraph Server API. - -### Dependency map - -The diagram below lists downstream libraries for each production dependency as -declared in that library's `pyproject.toml` (or `package.json`). - -```text -checkpoint -├── checkpoint-postgres -├── checkpoint-sqlite -├── prebuilt -└── langgraph - -prebuilt -└── langgraph - -sdk-py -├── langgraph -└── cli - -sdk-js (standalone) -``` - -Changes to a library may impact all of its dependents shown above. - -- Do NOT use Sphinx-style double backtick formatting (` ``code`` `). Use single backticks (`` `code` ``) for inline code references in docstrings and comments. diff --git a/libs/cli/langgraph_cli/deploy.py b/libs/cli/langgraph_cli/deploy.py index 149f0ba7c..8da036ebb 100644 --- a/libs/cli/langgraph_cli/deploy.py +++ b/libs/cli/langgraph_cli/deploy.py @@ -8,10 +8,11 @@ import platform import re import tempfile import time -from collections.abc import Callable, Sequence +from collections.abc import Callable, Mapping, Sequence from contextlib import contextmanager -from dataclasses import dataclass, field +from dataclasses import asdict, dataclass, field from datetime import datetime, timezone +from typing import Protocol, TypeVar import click import click.exceptions @@ -19,11 +20,19 @@ from dotenv import dotenv_values, set_key import langgraph_cli.config from langgraph_cli.analytics import log_command +from langgraph_cli.config import Config from langgraph_cli.constants import DEFAULT_CONFIG 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 Runner, subp_exec -from langgraph_cli.host_backend import HostBackendClient, HostBackendError +from langgraph_cli.exec import CommandRunner, Runner, subp_exec +from langgraph_cli.host_backend import ( + MAX_PAGE_SIZE, + ControlPlaneEndpoints, + HostBackendClient, + HostBackendError, + SourceName, +) +from langgraph_cli.image_reference import ImageReference from langgraph_cli.progress import Progress from langgraph_cli.util import warn_non_wolfi_distro @@ -84,7 +93,25 @@ _API_KEY_ENV_NAMES = ( "LANGCHAIN_API_KEY", ) +_T = TypeVar("_T") + _DEPLOYMENT_NAME_ENV = "LANGSMITH_DEPLOYMENT_NAME" +_DEFAULT_IMAGE_TAG = "latest" +_DEPLOYMENT_PLATFORM = "linux/amd64" +_NATIVE_AMD64_MACHINE = "x86_64" +_PUSH_ATTEMPTS = 3 +_LOCAL_BUILD_TAG_PREFIX = "langgraph-deploy-tmp" +_OPERATOR_DEFAULT_RESOURCE_SPEC: Mapping[str, object] = {} +_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( [ @@ -116,6 +143,159 @@ class BuildResult: show_build_logs_on_failure: bool = False +@dataclass(frozen=True, slots=True) +class ById: + deployment_id: str + + +@dataclass(frozen=True, slots=True) +class ByName: + name: str + + +@dataclass(frozen=True, slots=True) +class ByAgent: + agent_id: str + environment: str + + +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 + source: str | None + + # --------------------------------------------------------------------------- # Structured output emitter # --------------------------------------------------------------------------- @@ -287,12 +467,12 @@ def _get_emitter() -> _Emitter: # --------------------------------------------------------------------------- -def validate_deployment_selector(deployment_id: str | None, name: str | None) -> None: - """Ensure either deployment_id or name is provided.""" +def deployment_selector(deployment_id: str | None, name: str | None) -> ById | ByName: if deployment_id: - return - if not name: - raise click.UsageError("Either --deployment-id or --name is required.") + return ById(deployment_id) + if name: + return ByName(name) + raise click.UsageError("Either --deployment-id or --name is required.") def validate_deploy_commands( @@ -318,19 +498,26 @@ def validate_deploy_commands( # --------------------------------------------------------------------------- -def find_deployment_id_by_name( - client: HostBackendClient, name: str | None -) -> str | None: - """Return deployment ID for an exact name match, or None if not found.""" - if not name: +def _source_of(resource: object) -> str | None: + if not isinstance(resource, dict): return None - existing = client.list_deployments(name_contains=name) - if isinstance(existing, dict): - for dep in existing.get("resources", []): - if isinstance(dep, dict) and dep.get("name") == name: - found_id = dep.get("id") - if found_id: - return str(found_id) + source = resource.get("source") + return source if isinstance(source, str) else None + + +def find_deployment_by_name( + client: HostBackendClient, name: str +) -> ExistingDeployment | None: + 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 @@ -358,7 +545,7 @@ def normalize_image_tag(value: str) -> str: Tags may only contain [A-Za-z0-9_.-]. Defaults to "latest" when empty. """ if not value: - value = "latest" + value = _DEFAULT_IMAGE_TAG if not re.fullmatch(r"[A-Za-z0-9_.-]+", value): raise click.UsageError( "Image tag may only contain characters A-Z, a-z, 0-9, '_', '-', '.'" @@ -366,7 +553,9 @@ def normalize_image_tag(value: str) -> str: return value -def _validate_prebuilt_image(runner, image: str, *, verbose: bool) -> None: +def _validate_prebuilt_image( + runner: CommandRunner, image: str, *, verbose: bool +) -> None: """Ensure a prebuilt image exists locally for linux/amd64.""" try: stdout, _ = runner.run( @@ -393,13 +582,14 @@ def _validate_prebuilt_image(runner, image: str, *, verbose: bool) -> None: ) from None image_platform = (stdout or "").strip() - if image_platform != "linux/amd64": + if image_platform != _DEPLOYMENT_PLATFORM: detected = image_platform or "unknown" raise click.ClickException( f"Docker image '{image}' targets {detected}, but LangSmith Deployment " - "requires linux/amd64. Rebuild or pull the image for linux/amd64 before " - "deploying with --image." + f"requires {_DEPLOYMENT_PLATFORM}. Rebuild or pull the image for " + f"{_DEPLOYMENT_PLATFORM} before deploying with --image." ) + _get_emitter().info(f"Image is available for {_DEPLOYMENT_PLATFORM}") def _extract_deployment_url(deployment: dict[str, object]) -> str: @@ -599,55 +789,66 @@ def _log_deploy_step(step: int, message: str, **extra: object) -> None: _get_emitter().step(step, message, **extra) -def _resolve_deployment( +def _fetch_deployment( + client: HostBackendClient, step: int, selector: ById +) -> tuple[ExistingDeployment, int]: + _log_deploy_step(step, f"Using deployment {selector.deployment_id}") + resource = _call_host_backend_with_optional_tenant( + client, lambda c: c.get_deployment(selector.deployment_id) + ) + return ExistingDeployment(selector.deployment_id, _source_of(resource)), step + 1 + + +def _find_deployment( client: HostBackendClient, step: int, - deployment_id: str | None, - name: str | None, + selector: ByName | ByAgent, *, not_found_message: str, - agent: dict[str, str] | None = None, -) -> tuple[str | None, bool, int]: - """Resolve an existing deployment by ID or exact name match.""" - needs_creation = False - if deployment_id: - _log_deploy_step(step, f"Using deployment {deployment_id}") - _call_host_backend_with_optional_tenant( - client, lambda c: c.get_deployment(deployment_id) - ) - return deployment_id, needs_creation, step + 1 - - if agent is not None: +) -> tuple[ExistingDeployment | None, int]: + if isinstance(selector, ByAgent): _log_deploy_step( - step, f"Looking up agent '{agent['agent_id']}' in {agent['environment']}" + step, f"Looking up agent '{selector.agent_id}' in {selector.environment}" ) existing = _call_host_backend_with_optional_tenant( client, lambda c: c.list_deployments( - agent_id=agent["agent_id"], agent_environment=agent["environment"] + agent_id=selector.agent_id, + agent_environment=selector.environment, + limit=MAX_PAGE_SIZE, ), ) - found_id = next( + 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( ( - dep["id"] - for dep in existing.get("resources", []) - if not dep.get("is_preview") + ExistingDeployment(str(dep["id"]), _source_of(dep)) + for dep in existing + if dep.get("id") and not dep.get("is_preview") ), None, ) else: - _log_deploy_step(step, f"Looking up deployment '{name}'") - found_id = _call_host_backend_with_optional_tenant( - client, lambda c: find_deployment_id_by_name(c, name) + _log_deploy_step(step, f"Looking up deployment '{selector.name}'") + found = _call_host_backend_with_optional_tenant( + client, lambda c: find_deployment_by_name(c, selector.name) ) em = _get_emitter() - if found_id: - deployment_id = str(found_id) - em.info(f"Found existing deployment (ID: {deployment_id})") - else: - needs_creation = True + if found is None: em.warn(not_found_message) - return deployment_id, needs_creation, step + 1 + else: + em.info(f"Found existing deployment (ID: {found.id})") + return found, step + 1 + + +@dataclass(frozen=True, slots=True) +class CreatedDeployment: + id: str + resource: dict[str, object] def _create_deployment( @@ -655,13 +856,12 @@ def _create_deployment( step: int, *, name: str | None, - deployment_type: str, source: str, - config_rel: str | None = None, - secrets: list[dict[str, str]] | None = None, + source_config: dict[str, object], + source_revision_config: dict[str, object], + secrets: list[dict[str, str]], agent: dict[str, str] | None = None, -) -> tuple[str, int]: - """Create a deployment and return its ID and next step number.""" +) -> tuple[CreatedDeployment, int]: _log_deploy_step( step, f"Creating deployment for agent '{agent['agent_id']}' in {agent['environment']}" @@ -671,9 +871,9 @@ def _create_deployment( try: created = client.create_deployment( name=name, - deployment_type=deployment_type, source=source, - config_path=config_rel, + source_config=source_config, + source_revision_config=source_revision_config, secrets=secrets, agent=agent, ) @@ -690,48 +890,24 @@ def _create_deployment( "POST /v2/deployments succeeded but response missing a valid 'id'" ) if agent is not None: - _get_emitter().info(f"Deployment name: {created['name']}") + _get_emitter().info(f"Deployment name: {created.get('name')}") _get_emitter().info(f"Deployment ID: {created_id}", deployment_id=created_id) - return created_id, step + 1 - - -def _smith_dashboard_base_url(host_url: str | None) -> str: - """Derive the LangSmith dashboard base URL from the API host URL.""" - from urllib.parse import urlparse - - if not host_url: - return "https://smith.langchain.com" - parsed = urlparse(host_url) - hostname = parsed.hostname or "" - if hostname in ("localhost", "127.0.0.1"): - return host_url.rstrip("/") - - for api_host_suffix in ("api.host.langchain.com", "api.smith.langchain.com"): - if hostname == api_host_suffix: - return "https://smith.langchain.com" - if hostname.endswith(f".{api_host_suffix}"): - prefix = hostname[: -(len(api_host_suffix) + 1)] - return f"https://{prefix}.smith.langchain.com" - - return "https://smith.langchain.com" + return CreatedDeployment(created_id, created), step + 1 def _get_deployment_status_url( - updated: object, deployment_id: str, host_url: str | None = None + 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 = _smith_dashboard_base_url(host_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 | None = None + 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 @@ -749,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() @@ -930,7 +1103,7 @@ def _upload_to_gcs(signed_url: str, file_path: str, file_size: int) -> None: def _resolve_pushed_image_digest( - runner, + runner: CommandRunner, *, remote_image: str, docker_config_dir: str | None, @@ -943,19 +1116,25 @@ def _resolve_pushed_image_digest( Falls back to ``remote_image`` with a warning if no matching digest is found, rather than failing the deploy. """ - # rsplit preserves ``:port`` in the registry host. - repo_no_tag = remote_image.rsplit(":", 1)[0] - args: list[str] = ["docker"] - if docker_config_dir: - args += ["--config", docker_config_dir] - args += ["image", "inspect", "--format", "{{json .RepoDigests}}", remote_image] - stdout, _ = runner.run(subp_exec(*args, collect=True, verbose=verbose)) + reference = ImageReference.parse(remote_image) + stdout, _ = runner.run( + subp_exec( + *_docker_argv(docker_config_dir), + "image", + "inspect", + "--format", + "{{json .RepoDigests}}", + remote_image, + collect=True, + verbose=verbose, + ) + ) try: digests = json_mod.loads(stdout or "[]") or [] except json_mod.JSONDecodeError: digests = [] for d in digests: - if isinstance(d, str) and d.startswith(f"{repo_no_tag}@sha256:"): + if isinstance(d, str) and reference.matches_digest(d): return d _get_emitter().warn( f"Could not resolve image digest for {remote_image}; " @@ -964,86 +1143,123 @@ def _resolve_pushed_image_digest( return remote_image +@dataclass(frozen=True, slots=True) +class BuildSpec: + config: pathlib.Path + config_json: Config + base_image: str | None + api_version: str | None + pull: bool + docker_build_args: Sequence[str] + install_command: str | None + build_command: str | None + + +@dataclass(frozen=True, slots=True) +class DockerBuildCommand: + command: tuple[str, ...] + flags: tuple[str, ...] + + @classmethod + def for_host(cls, machine: str, *, verbose: bool) -> "DockerBuildCommand": + if machine == _NATIVE_AMD64_MACHINE: + return cls(("docker", "build"), ()) + flags: tuple[str, ...] = ("--platform", _DEPLOYMENT_PLATFORM, "--load") + if not verbose: + flags += ("--progress=quiet",) + return cls(("docker", "buildx", "build"), flags) + + +def _docker_argv(docker_config_dir: str | None) -> tuple[str, ...]: + if docker_config_dir is None: + return ("docker",) + return ("docker", "--config", docker_config_dir) + + +def _build_image( + runner: CommandRunner, spec: BuildSpec, tag: str, *, verbose: bool +) -> None: + build = DockerBuildCommand.for_host(platform.machine(), verbose=verbose) + with Progress(message="Building...", elapsed=not verbose): + build_docker_image( + runner, + lambda _msg: None, + spec.config, + spec.config_json, + spec.base_image, + spec.api_version, + spec.pull, + tag, + spec.docker_build_args, + spec.install_command, + spec.build_command, + docker_command=build.command, + extra_flags=build.flags, + verbose=verbose, + ) + + +def _push_image( + runner: CommandRunner, + image: str, + *, + docker_config_dir: str | None, + verbose: bool, +) -> None: + for attempt in range(1, _PUSH_ATTEMPTS + 1): + try: + with Progress(message="Pushing...", elapsed=not verbose): + runner.run( + subp_exec( + *_docker_argv(docker_config_dir), "push", image, verbose=verbose + ) + ) + return + except click.exceptions.Exit: + if attempt == _PUSH_ATTEMPTS: + raise + _get_emitter().warn( + f" Push failed, retrying (attempt {attempt + 1} of {_PUSH_ATTEMPTS})..." + ) + + +def _image_revision_result(resource: object, no_result_message: str) -> BuildResult: + return BuildResult( + updated=resource if isinstance(resource, dict) else {}, + progress_message="Deploying...", + timeout_seconds=300, + poll_interval_seconds=1, + no_result_message=no_result_message, + ) + + def _run_local_build( *, client: HostBackendClient, deployment_id: str, step: int, - config: pathlib.Path, - config_json: dict, + spec: BuildSpec, verbose: bool, - pull: bool, - api_version: str | None, - base_image: str | None, image_name: str | None, prebuilt_image: str | None, name: str | None, tag: str, - install_command: str | None, - build_command: str | None, - docker_build_args: Sequence[str], secrets: list[dict[str, str]], tracked_packages: list[str] | None, ) -> BuildResult: """Build locally with Docker, push to registry, update deployment.""" - # Use buildx to cross-compile for amd64 when running on a non-x86_64 host - # (e.g. Apple Silicon). On amd64 hosts, plain docker build is sufficient. - needs_buildx = platform.machine() != "x86_64" - local_tag = f"langgraph-deploy-tmp:{int(time.time())}" + local_tag = f"{_LOCAL_BUILD_TAG_PREFIX}:{int(time.time())}" image_to_push = prebuilt_image or local_tag with Runner() as runner: if prebuilt_image: _log_deploy_step(step, f"Validating image {prebuilt_image}") _validate_prebuilt_image(runner, prebuilt_image, verbose=verbose) - click.secho(" Image is available for linux/amd64", fg="green") else: - # -- Step: Build image -- _log_deploy_step(step, "Building image") - if needs_buildx: - build_flags: list[str] = [ - "--platform", - "linux/amd64", - "--load", - ] - if not verbose: - build_flags.append("--progress=quiet") - with Progress(message="Building...", elapsed=not verbose): - build_docker_image( - runner, - lambda _msg: None, - config, - config_json, - base_image, - api_version, - pull, - local_tag, - docker_build_args, - install_command, - build_command, - docker_command=("docker", "buildx", "build"), - extra_flags=build_flags, - verbose=verbose, - ) - else: - with Progress(message="Building...", elapsed=not verbose): - build_docker_image( - runner, - lambda _msg: None, - config, - config_json, - base_image, - api_version, - pull, - local_tag, - docker_build_args, - install_command, - build_command, - verbose=verbose, - ) + _build_image(runner, spec, local_tag, verbose=verbose) step += 1 - # -- Step: Get push token and authenticate -- _log_deploy_step(step, "Requesting push token") try: push_data = client.request_push_token(deployment_id) @@ -1071,15 +1287,15 @@ def _run_local_build( normalized_registry = registry_url.rstrip("/") if "://" in normalized_registry: normalized_registry = normalized_registry.split("//", 1)[1] - repo_seed = image_name or name or config.parent.name - repo_name = normalize_name(repo_seed) - tag_value = normalize_image_tag(tag) - remote_image = f"{normalized_registry}/{repo_name}:{tag_value}" - + repo_seed = image_name or name or spec.config.parent.name + remote_image = str( + ImageReference( + f"{normalized_registry}/{normalize_name(repo_seed)}", + tag, + ) + ) registry_host = normalized_registry.split("/")[0] - # Use a clean Docker config with only the push token so that - # system credential helpers (e.g. gcloud) don't interfere. with _docker_config_for_token(registry_host, deployment_token) as cfg: _log_deploy_step(step, f"Logging into {registry_host}") token_input = ( @@ -1089,53 +1305,23 @@ def _run_local_build( ) runner.run( subp_exec( - "docker", - "--config", - cfg, + *_docker_argv(cfg), "login", "-u", "oauth2accesstoken", "--password-stdin", registry_host, input=token_input, - verbose=verbose, + verbose=False, ) ) step += 1 - # -- Step: Tag and push -- _log_deploy_step(step, f"Pushing image {remote_image}") runner.run( - subp_exec( - "docker", - "tag", - image_to_push, - remote_image, - verbose=verbose, - ) + subp_exec("docker", "tag", image_to_push, remote_image, verbose=verbose) ) - max_push_retries = 3 - for attempt in range(max_push_retries): - try: - with Progress(message="Pushing...", elapsed=not verbose): - runner.run( - subp_exec( - "docker", - "--config", - cfg, - "push", - remote_image, - verbose=verbose, - ) - ) - break - except click.exceptions.Exit: - if attempt < max_push_retries - 1: - _get_emitter().warn( - f" Push failed, retrying (attempt {attempt + 2} of {max_push_retries})..." - ) - else: - raise + _push_image(runner, remote_image, docker_config_dir=cfg, verbose=verbose) step += 1 resolved_image = _resolve_pushed_image_digest( @@ -1145,22 +1331,16 @@ def _run_local_build( verbose=verbose, ) - # -- Step: Update deployment -- _log_deploy_step(step, f"Updating deployment {deployment_id}") updated = client.update_deployment( deployment_id, resolved_image, + revision_source="internal_docker", secrets=secrets, tracked_packages=tracked_packages, ) - return BuildResult( - updated=updated if isinstance(updated, dict) else {}, - progress_message="Deploying...", - timeout_seconds=300, - poll_interval_seconds=1, - no_result_message="Deployment updated", - ) + return _image_revision_result(updated, "Deployment updated") def _run_remote_build( @@ -1168,11 +1348,8 @@ def _run_remote_build( client: HostBackendClient, deployment_id: str, step: int, - config: pathlib.Path, - config_json: dict, + spec: BuildSpec, verbose: bool, - install_command: str | None, - build_command: str | None, secrets: list[dict[str, str]], tracked_packages: list[str] | None, ) -> BuildResult: @@ -1181,7 +1358,11 @@ def _run_remote_build( em = _get_emitter() _log_deploy_step(step, "Creating source archive") - with create_archive(config, config_json) as (archive_path, file_size, config_rel): + with create_archive(spec.config, spec.config_json) as ( + archive_path, + file_size, + config_rel, + ): em.info(f"Archive created ({file_size / _BYTES_PER_MIB:.1f} MB)") step += 1 @@ -1203,8 +1384,8 @@ def _run_remote_build( source_tarball_path=object_path, config_path=config_rel, secrets=secrets, - install_command=install_command, - build_command=build_command, + install_command=spec.install_command, + build_command=spec.build_command, tracked_packages=tracked_packages, ) @@ -1261,6 +1442,348 @@ def _run_remote_build( ) +# --------------------------------------------------------------------------- +# Deployment sources +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class DeployContext: + client: HostBackendClient + endpoints: ControlPlaneEndpoints + spec: BuildSpec + verbose: bool + selector: DeploymentSelector + deployment_type: str + secrets: list[dict[str, str]] + tracked_packages: list[str] | None + + +@dataclass(frozen=True, slots=True) +class DeployOutcome: + deployment_id: str + build_result: BuildResult + + +class DeploymentSource(Protocol): + def run(self, ctx: DeployContext) -> DeployOutcome: ... + + +def _resolve_or_create( + ctx: DeployContext, *, source: SourceName, not_found_message: str +) -> tuple[str, int]: + if isinstance(ctx.selector, ById): + existing, step = _fetch_deployment(ctx.client, 1, ctx.selector) + return existing.id, step + found, step = _find_deployment( + ctx.client, 1, ctx.selector, not_found_message=not_found_message + ) + if found is not None: + return found.id, step + 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( + f"Deployment {existing.id} was not created from an external image " + "and cannot be updated with --push-to. Run without --push-to to keep " + "its current build mode, or use a different --name to create a new " + "deployment." + ) + + +@dataclass(frozen=True, slots=True) +class ManagedRegistrySource: + prebuilt_image: str | None + image_name: str | None + tag: str + + def run(self, ctx: DeployContext) -> DeployOutcome: + deployment_id, step = _resolve_or_create( + ctx, + source="internal_docker", + not_found_message="No deployment found. Will create after build.", + ) + build_result = _run_local_build( + client=ctx.client, + deployment_id=deployment_id, + step=step, + spec=ctx.spec, + verbose=ctx.verbose, + image_name=self.image_name, + prebuilt_image=self.prebuilt_image, + name=ctx.selector.name if isinstance(ctx.selector, ByName) else None, + tag=self.tag, + secrets=ctx.secrets, + tracked_packages=ctx.tracked_packages, + ) + return DeployOutcome(deployment_id, build_result) + + +@dataclass(frozen=True, slots=True) +class RemoteBuildSource: + def run(self, ctx: DeployContext) -> DeployOutcome: + deployment_id, step = _resolve_or_create( + ctx, + source="internal_source", + not_found_message="No deployment found. Will create.", + ) + build_result = _run_remote_build( + client=ctx.client, + deployment_id=deployment_id, + step=step, + spec=ctx.spec, + verbose=ctx.verbose, + secrets=ctx.secrets, + tracked_packages=ctx.tracked_packages, + ) + return DeployOutcome(deployment_id, build_result) + + +@dataclass(frozen=True, slots=True) +class CustomerRegistrySource: + reference: ImageReference + prebuilt_image: str | None + requested_placement: RequestedPlacement + + def run(self, ctx: DeployContext) -> DeployOutcome: + if isinstance(ctx.selector, ById): + existing, step = _fetch_deployment(ctx.client, 1, ctx.selector) + return self._update(ctx, existing, step) + found, step = _find_deployment( + ctx.client, + 1, + ctx.selector, + not_found_message="No deployment found. Will create after push.", + ) + if found is not None: + return self._update(ctx, found, step) + return self._create( + ctx, ctx.selector.name if isinstance(ctx.selector, ByName) else None, step + ) + + def _update( + 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( + existing.id, + image_uri, + revision_source=None, + secrets=ctx.secrets, + tracked_packages=ctx.tracked_packages, + ) + return DeployOutcome( + 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( + ctx.client, + step, + name=name, + agent=asdict(ctx.selector) + if isinstance(ctx.selector, ByAgent) + else None, + source=_CUSTOMER_REGISTRY_SOURCE, + 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 _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") + ) + + def _publish(self, ctx: DeployContext, step: int) -> tuple[str, int]: + image = str(self.reference) + with Runner() as runner: + if self.prebuilt_image: + _log_deploy_step(step, f"Validating image {self.prebuilt_image}") + _validate_prebuilt_image( + runner, self.prebuilt_image, verbose=ctx.verbose + ) + runner.run( + subp_exec( + "docker", "tag", self.prebuilt_image, image, verbose=ctx.verbose + ) + ) + else: + _log_deploy_step(step, f"Building image {image}") + _build_image(runner, ctx.spec, image, verbose=ctx.verbose) + step += 1 + _log_deploy_step(step, f"Pushing image {image}") + _push_image(runner, image, docker_config_dir=None, verbose=ctx.verbose) + step += 1 + digest = _resolve_pushed_image_digest( + runner, remote_image=image, docker_config_dir=None, verbose=ctx.verbose + ) + return digest, step + + +def _require_local_docker() -> None: + supported, error = can_build_locally() + if not supported: + raise click.UsageError(error or "Unable to build locally.") + + +def _push_reference(push_to: str, tag: str | None) -> ImageReference: + try: + reference = ImageReference.parse(push_to) + except ValueError: + raise click.UsageError( + "--push-to takes a repository with an optional tag, not a digest." + ) from None + if reference.tag is not None and tag is not None: + raise click.UsageError( + "--push-to already includes a tag; do not combine it with --tag." + ) + if reference.tag is not None: + return reference + return reference.with_tag(normalize_image_tag(tag or _DEFAULT_IMAGE_TAG)) + + +def _select_source( + *, + push_to: str | None, + image: str | None, + 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=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( + remote_build_flag, force_local=image is not None + ) + if not use_remote_build: + return ManagedRegistrySource( + prebuilt_image=image, + image_name=image_name, + tag=normalize_image_tag(tag or _DEFAULT_IMAGE_TAG), + ) + if remote_build_flag is None and local_build_error: + em = _get_emitter() + em.note(f"{local_build_error}\nUsing remote build instead.") + if not em.json_mode: + click.echo() + return RemoteBuildSource() + + # --------------------------------------------------------------------------- # Host backend client factory # --------------------------------------------------------------------------- @@ -1298,13 +1821,19 @@ def _create_host_backend_client( tenant_id = env_vars.get("LANGSMITH_TENANT_ID") or os.environ.get( "LANGSMITH_TENANT_ID" ) - return HostBackendClient(host_url, resolved_api_key, tenant_id=tenant_id) + langsmith_endpoint = env_vars.get("LANGSMITH_ENDPOINT") or os.environ.get( + "LANGSMITH_ENDPOINT" + ) + endpoints = ControlPlaneEndpoints.resolve(host_url, langsmith_endpoint) + return HostBackendClient( + endpoints.control_plane_url, resolved_api_key, tenant_id=tenant_id + ) def _call_host_backend_with_optional_tenant( client: HostBackendClient, - operation: Callable[[HostBackendClient], object], -) -> object: + operation: Callable[[HostBackendClient], _T], +) -> _T: """Run *operation*, prompting for a workspace ID on org-scoped 403s. On success the original *client* is returned as-is. If the user is @@ -1337,11 +1866,11 @@ def _call_host_backend_with_optional_tenant( "Find your workspace ID in LangSmith under Settings > Workspaces.", fg="yellow", ) - client._client.headers["X-Tenant-ID"] = click.prompt("Workspace ID") + client.set_tenant(click.prompt("Workspace ID")) prompted_for_tenant = True continue if err.status_code == 403 and "not enabled" in err.message.lower(): - smith_base = _smith_dashboard_base_url(client._base_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" @@ -1377,16 +1906,22 @@ OPT_HOST_DEPLOYMENT_NAME = click.option( OPT_HOST_URL = click.option( "--host-url", envvar="LANGGRAPH_HOST_URL", - default="https://api.host.langchain.com", + default=None, hidden=True, ) OPT_AGENT_ID = click.option( - "--agent-id", help="Logical agent ID (requires agent mode enabled for the tenant)." + "--agent-id", + envvar="LANGSMITH_AGENT_ID", + show_envvar=True, + help="Logical agent ID (requires agent mode enabled for the tenant).", ) OPT_AGENT_ENVIRONMENT = click.option( - "--environment", + "--agent-environment", + "environment", + envvar="LANGSMITH_AGENT_ENVIRONMENT", + show_envvar=True, type=click.Choice(["development", "staging", "production"]), help="Agent environment (requires agent mode enabled for the tenant).", ) @@ -1503,7 +2038,10 @@ def _deploy_base_options( type=click.Choice(["dev", "prod"]), default="dev", show_default=True, - help="Deployment type (used when creating a new deployment).", + help=( + "Deployment type (used when creating a new deployment). " + "Ignored with --push-to." + ), ), click.option( "--no-wait", @@ -1517,9 +2055,8 @@ def _deploy_base_options( click.option( "--tag", "-t", - default="latest", - show_default=True, - help="Tag to use for the pushed deployment image.", + default=None, + help="Tag to use for the pushed deployment image. [default: latest]", ), click.option( "--image", @@ -1528,6 +2065,31 @@ def _deploy_base_options( "skip building. The image must target linux/amd64." ), ), + click.option( + "--push-to", + help=( + "Push the image to this repository in a registry you manage, " + "then deploy it from there. For self-hosted and hybrid " + "LangSmith. Uses your existing Docker credentials. Builds the " + "project, or retags the local image given with --image. " + "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", @@ -1610,6 +2172,12 @@ def deploy(ctx: click.Context, **_: object): # otherwise, we return None here and click will proceed to actually run the subcommand (list or delete) if ctx.invoked_subcommand is not None: return + if ( + ctx.params.get("agent_id") is not None + or ctx.params.get("environment") is not None + ) and ctx.get_parameter_source("name") == click.core.ParameterSource.ENVIRONMENT: + # Ignore the inherited name default so it does not conflict with agent mode. + ctx.params["name"] = None docker_build_args = tuple(ctx.args) ctx.args = [] # Prevent Click from re-processing passthrough args later. return ctx.forward(_deploy_cmd, docker_build_args=docker_build_args) @@ -1631,7 +2199,10 @@ def _deploy_cmd( environment: str | None, image_name: str | None, image: str | None, - tag: str, + push_to: str | None, + listener_id: str | None, + k8s_namespace: str | None, + tag: str | None, base_image: str | None, install_command: str | None, build_command: str | None, @@ -1652,17 +2223,17 @@ def _deploy_cmd( if not json_output: click.echo() - # -- 1. Preflight -- validate_deploy_commands(install_command, build_command) agent = None if agent_id is not None or environment is not None: + em.note("Note: --agent-id and --agent-environment flags are in private beta") if not agent_id or not agent_id.strip() or not environment: raise click.UsageError( - "--agent-id and --environment are required together." + "--agent-id and --agent-environment are required together." ) if name is not None or deployment_id is not None: raise click.UsageError( - "--agent-id and --environment cannot be combined with --name or --deployment-id." + "--agent-id and --agent-environment cannot be combined with --name or --deployment-id." ) agent = {"agent_id": agent_id, "environment": environment} if not config.exists(): @@ -1698,119 +2269,70 @@ def _deploy_cmd( secrets = _secrets_from_env(_env_without_deployment_name(env_vars)) - 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( - remote_build_flag, force_local=image is not None + 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, ) - if use_remote_build and remote_build_flag is None and local_build_error: - em.note(f"{local_build_error}\nUsing remote build instead.") - if not json_output: - click.echo() - # -- 2. Resolve / create deployment -- client = _create_host_backend_client(host_url, api_key, env_vars=env_vars) - step = 1 - - deployment_id, needs_creation, step = _resolve_deployment( - client, - step, - deployment_id, - name, - not_found_message=( - "No deployment found. Will create." - if use_remote_build - else "No deployment found. Will create after build." - ), - agent=agent, - ) - - if needs_creation: - deployment_id, step = _create_deployment( - client, - step, - name=name, - deployment_type=deployment_type, - source="internal_source" if use_remote_build else "internal_docker", - secrets=secrets, - agent=agent, - ) - - if not deployment_id: - raise click.ClickException("Failed to determine deployment ID") - - # Scan local sources for tracked packages so the new revision carries - # the same metadata GitHub-backed deploys produce. Failures must never - # block a deploy. try: tracked_packages = find_tracked_packages(config, config_json) or None except Exception as exc: em.warn(f"Skipped tracked-package scan: {exc}") tracked_packages = None - # -- 3. Build (divergent path) -- - if use_remote_build: - build_result = _run_remote_build( + outcome = source.run( + DeployContext( client=client, - deployment_id=deployment_id, - step=step, - config=config, - config_json=config_json, + endpoints=client.endpoints, + spec=BuildSpec( + config=config, + config_json=config_json, + base_image=base_image, + api_version=api_version, + pull=pull, + docker_build_args=docker_build_args, + install_command=install_command, + build_command=build_command, + ), verbose=verbose, - install_command=install_command, - build_command=build_command, + selector=selector, + deployment_type=deployment_type, secrets=secrets, tracked_packages=tracked_packages, ) - else: - build_result = _run_local_build( - client=client, - deployment_id=deployment_id, - step=step, - config=config, - config_json=config_json, - verbose=verbose, - pull=pull, - api_version=api_version, - base_image=base_image, - image_name=image_name, - prebuilt_image=image, - name=name, - tag=tag, - install_command=install_command, - build_command=build_command, - docker_build_args=docker_build_args, - secrets=secrets, - tracked_packages=tracked_packages, - ) - - # -- 4. Shared wait + result -- + ) dep_status_url = _emit_deployment_status_url( - build_result.updated, - deployment_id, - host_url, + outcome.build_result.updated, + outcome.deployment_id, + client.endpoints, ) if no_wait: - em.info(build_result.no_result_message) + em.info(outcome.build_result.no_result_message) return last_status, revision_id = _poll_revision_status( client, - deployment_id, - progress_message=build_result.progress_message, - timeout_seconds=build_result.timeout_seconds, - poll_interval_seconds=build_result.poll_interval_seconds, - on_poll=build_result.on_poll, - on_interrupt=build_result.on_interrupt, + outcome.deployment_id, + progress_message=outcome.build_result.progress_message, + timeout_seconds=outcome.build_result.timeout_seconds, + poll_interval_seconds=outcome.build_result.poll_interval_seconds, + on_poll=outcome.build_result.on_poll, + on_interrupt=outcome.build_result.on_interrupt, ) if not last_status: - em.info(build_result.no_result_message) + em.info(outcome.build_result.no_result_message) return if ( - build_result.show_build_logs_on_failure + outcome.build_result.show_build_logs_on_failure and last_status == "BUILD_FAILED" and not verbose and revision_id is not None @@ -1818,7 +2340,7 @@ def _deploy_cmd( em.error("Last build log lines:") try: logs_resp = client.get_build_logs( - deployment_id, + outcome.deployment_id, revision_id, {"order": "desc", "limit": 30}, ) @@ -1834,7 +2356,7 @@ def _deploy_cmd( _print_deployment_result( client, - deployment_id, + outcome.deployment_id, last_status, dashboard_label="Deployment dashboard", status_url=dep_status_url, @@ -1863,6 +2385,11 @@ def deploy_list( agent_id: str | None, environment: str | None, ) -> None: + if agent_id is not None or environment is not None: + click.secho( + "Note: --agent-id and --agent-environment flags are in private beta", + fg="yellow", + ) if agent_id is not None and not agent_id.strip(): raise click.UsageError("--agent-id must not be empty.") filters = {} @@ -1871,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 @@ -1920,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 @@ -2060,35 +2575,31 @@ def deploy_logs( start_time: str | None, end_time: str | None, follow: bool, - host_url: str, + host_url: str | None, ): env_vars = _parse_env_from_config({}, pathlib.Path.cwd() / DEFAULT_CONFIG) client = _create_host_backend_client(host_url, api_key, env_vars=env_vars) if not deployment_id and not name: name = env_vars.get(_DEPLOYMENT_NAME_ENV) - validate_deployment_selector(deployment_id, name) - if deployment_id: - dep_id = deployment_id + selector = deployment_selector(deployment_id, name) + if isinstance(selector, ById): + dep_id = selector.deployment_id else: + name_to_find = selector.name found = _call_host_backend_with_optional_tenant( - client, lambda c: find_deployment_id_by_name(c, name) + client, lambda c: find_deployment_by_name(c, name_to_find) ) - if not found: - raise click.ClickException(f"Deployment '{name}' not found.") - dep_id = str(found) + if found is None: + raise click.ClickException(f"Deployment '{name_to_find}' not found.") + 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/exec.py b/libs/cli/langgraph_cli/exec.py index 974fc75f3..32c3d3470 100644 --- a/libs/cli/langgraph_cli/exec.py +++ b/libs/cli/langgraph_cli/exec.py @@ -1,12 +1,18 @@ import asyncio import signal import sys -from collections.abc import Callable +from collections.abc import Callable, Coroutine from contextlib import contextmanager -from typing import cast +from typing import Any, Protocol, TypeVar, cast import click.exceptions +_T = TypeVar("_T") + + +class CommandRunner(Protocol): + def run(self, coro: Coroutine[Any, Any, _T]) -> _T: ... + @contextmanager def Runner(): diff --git a/libs/cli/langgraph_cli/host_backend.py b/libs/cli/langgraph_cli/host_backend.py index f5f928e89..1aea6fa00 100644 --- a/libs/cli/langgraph_cli/host_backend.py +++ b/libs/cli/langgraph_cli/host_backend.py @@ -2,18 +2,126 @@ from __future__ import annotations -from typing import Any +from dataclasses import dataclass +from typing import Any, Literal +from urllib.parse import urlparse import click import httpx +CLOUD_CONTROL_PLANE_URL = "https://api.host.langchain.com" +CLOUD_DASHBOARD_URL = "https://smith.langchain.com" +CLOUD_DOMAIN = "langchain.com" +CLOUD_API_HOST = "api.smith.langchain.com" +CLOUD_CONTROL_PLANE_HOST = "api.host.langchain.com" +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"] + + +@dataclass(frozen=True, slots=True) +class ControlPlaneEndpoints: + control_plane_url: str + dashboard_url: str + + @classmethod + def resolve( + cls, host_url: str | None, langsmith_endpoint: str | None + ) -> ControlPlaneEndpoints: + if host_url: + return cls.from_control_plane_url(host_url) + if langsmith_endpoint: + 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("/") + hostname = urlparse(control_plane_url).hostname or "" + if control_plane_url.endswith(CONTROL_PLANE_PATH): + return cls(control_plane_url, control_plane_url[: -len(CONTROL_PLANE_PATH)]) + if hostname in LOCAL_HOSTNAMES: + return cls(control_plane_url, control_plane_url) + return cls(control_plane_url, _cloud_dashboard_for(hostname)) + + @classmethod + def from_langsmith_endpoint(cls, endpoint: str) -> ControlPlaneEndpoints: + parsed = urlparse(endpoint.rstrip("/")) + hostname = parsed.hostname or "" + if _is_cloud_host(hostname): + return cls.from_control_plane_url( + f"https://{_cloud_control_plane_host_for(hostname)}" + ) + root = f"{parsed.scheme}://{parsed.netloc}{_without_api_path(parsed.path)}" + return cls(f"{root}{CONTROL_PLANE_PATH}", root) + + +def _is_cloud_host(hostname: str) -> bool: + return hostname == CLOUD_DOMAIN or hostname.endswith(f".{CLOUD_DOMAIN}") + + +def _cloud_control_plane_host_for(langsmith_api_host: str) -> str: + if langsmith_api_host.endswith(f".{CLOUD_API_HOST}"): + region = langsmith_api_host[: -len(CLOUD_API_HOST)] + return f"{region}{CLOUD_CONTROL_PLANE_HOST}" + return CLOUD_CONTROL_PLANE_HOST + + +def _cloud_dashboard_for(control_plane_host: str) -> str: + for api_host in (CLOUD_CONTROL_PLANE_HOST, CLOUD_API_HOST): + if control_plane_host.endswith(f".{api_host}"): + region = control_plane_host[: -len(api_host) - 1] + return f"https://{region}.{CLOUD_DASHBOARD_HOST}" + return CLOUD_DASHBOARD_URL + + +def _without_api_path(path: str) -> str: + for api_path in LANGSMITH_API_PATHS: + if path.endswith(api_path): + return path[: -len(api_path)] + 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: @@ -24,24 +132,37 @@ class HostBackendClient: base_url: str, api_key: str, tenant_id: str | None = None, + *, + transport: httpx.BaseTransport | None = None, ): if not base_url: raise click.UsageError("Host backend URL is required") - transport = httpx.HTTPTransport(retries=3) headers: dict[str, str] = { "X-Api-Key": api_key, "Accept": "application/json", } 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, - transport=transport, + transport=transport or httpx.HTTPTransport(retries=3), timeout=30, ) + @property + 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 + def _request( self, method: str, @@ -53,10 +174,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 @@ -72,45 +195,52 @@ class HostBackendClient: def create_deployment( self, + *, name: str | None, - deployment_type: str, - source: str, - config_path: str | None = None, + source: SourceName, + source_config: dict[str, object], + source_revision_config: dict[str, object], secrets: list[dict[str, str]] | None = None, agent: dict[str, str] | None = None, ) -> dict[str, Any]: - """Create a deployment.""" payload: dict[str, Any] = { "source": source, - "source_config": {"deployment_type": deployment_type}, - "source_revision_config": {}, + "source_config": source_config, + "source_revision_config": source_revision_config, } if agent is not None: payload["agent"] = agent else: payload["name"] = name - if source == "internal_source" and config_path: - payload["source_revision_config"]["langgraph_config_path"] = config_path if secrets is not None: payload["secrets"] = secrets return self._request("POST", "/v2/deployments", payload) 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]: @@ -136,22 +266,21 @@ class HostBackendClient: self, deployment_id: str, image_uri: str, + *, + revision_source: SourceName | None, secrets: list[dict[str, str]] | None = None, tracked_packages: list[str] | None = None, ) -> dict[str, Any]: payload: dict[str, Any] = { - "revision_source": "internal_docker", "source_revision_config": {"image_uri": image_uri}, } + if revision_source is not None: + payload["revision_source"] = revision_source if tracked_packages: payload["tracked_packages"] = tracked_packages if secrets is not None: payload["secrets"] = secrets - return self._request( - "PATCH", - f"/v2/deployments/{deployment_id}", - payload, - ) + return self._request("PATCH", f"/v2/deployments/{deployment_id}", payload) def update_deployment_internal_source( self, @@ -186,10 +315,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/langgraph_cli/image_reference.py b/libs/cli/langgraph_cli/image_reference.py new file mode 100644 index 000000000..803ea3d4c --- /dev/null +++ b/libs/cli/langgraph_cli/image_reference.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +from dataclasses import dataclass, replace + +DIGEST_SEPARATOR = "@sha256:" +DIGEST_MARKER = "@" +TAG_SEPARATOR = ":" +PATH_SEPARATOR = "/" + + +@dataclass(frozen=True, slots=True) +class ImageReference: + repository: str + tag: str | None = None + + @classmethod + def parse(cls, reference: str) -> ImageReference: + if DIGEST_MARKER in reference: + raise ValueError(f"{reference!r} carries a digest and cannot be tagged") + path_start = reference.rfind(PATH_SEPARATOR) + 1 + name, separator, tag = reference[path_start:].partition(TAG_SEPARATOR) + if not separator: + return cls(reference) + return cls(reference[:path_start] + name, tag) + + def with_tag(self, tag: str) -> ImageReference: + return replace(self, tag=tag) + + def matches_digest(self, repo_digest: str) -> bool: + return repo_digest.startswith(f"{self.repository}{DIGEST_SEPARATOR}") + + def __str__(self) -> str: + if self.tag is None: + return self.repository + return f"{self.repository}{TAG_SEPARATOR}{self.tag}" 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 new file mode 100644 index 000000000..457725e37 --- /dev/null +++ b/libs/cli/tests/unit_tests/cli/test_deploy_command.py @@ -0,0 +1,1116 @@ +import asyncio +import json +import uuid +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from dataclasses import dataclass, field +from pathlib import Path + +import click.exceptions +import httpx +import pytest +from click.testing import CliRunner, Result + +import langgraph_cli.archive as archive_module +import langgraph_cli.deploy as deploy_module +from langgraph_cli.cli import cli +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" +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_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" +ARCHIVE = ("/tmp/src.tgz", 2048, "langgraph.json") +OBJECT_PATH = "tarballs/src.tgz" +PLATFORM_FORMAT = "{{.Os}}/{{.Architecture}}" +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" + + +def _push_token(deployment_id: str) -> str: + return f"POST /v2/deployments/{deployment_id}/push-token" + + +def _upload_url(deployment_id: str) -> str: + return f"POST /v2/deployments/{deployment_id}/upload-url" + + +def _patch(deployment_id: str) -> str: + return f"PATCH /v2/deployments/{deployment_id}" + + +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: + route = f"{request.method} {request.url.path}" + self.timeline.append(route) + if request.content: + self.bodies[route] = json.loads(request.content) + return self._respond(request) + + 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"): + 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, 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: + return httpx.Response(self.push_token_status, text=NOT_A_CLI_DEPLOYMENT) + return httpx.Response( + 200, json={"token": PUSH_TOKEN, "registry_url": REGISTRY_URL} + ) + if path.endswith("/upload-url"): + return httpx.Response( + 200, json={"upload_url": SIGNED_UPLOAD_URL, "object_path": OBJECT_PATH} + ) + if method == "PATCH": + return httpx.Response(200, json={"tenant_id": "tenant-1"}) + if method == "GET": + deployment_id = path.rsplit("/", 1)[-1] + return httpx.Response( + 200, + json=next( + d for d in self.existing_deployments if d["id"] == deployment_id + ), + ) + raise AssertionError(f"unexpected control plane call: {method} {path}") + + def client_factory(self) -> Callable[..., HostBackendClient]: + transport = httpx.MockTransport(self.handle) + + def make( + host_url: str, api_key: str, tenant_id: str | None = None + ) -> HostBackendClient: + return HostBackendClient(host_url, api_key, tenant_id, transport=transport) + + return make + + +@dataclass +class DockerCommand: + args: tuple[str, ...] + kwargs: dict + + +@dataclass +class DockerDouble: + timeline: list[str] + failing_pushes: int = 0 + builds: list[dict] = field(default_factory=list) + commands: list[DockerCommand] = field(default_factory=list) + + def verbs(self) -> list[str]: + return [event for event in self.timeline if event.startswith("docker ")] + + def command(self, verb: str) -> DockerCommand: + return next(c for c in self.commands if verb in c.args) + + def build_docker_image( + self, + runner: object, + set_message: Callable[[str], None], + config: Path, + config_json: dict, + base_image: str | None, + api_version: str | None, + pull: bool, + tag: str, + passthrough: tuple[str, ...] = (), + install_command: str | None = None, + build_command: str | None = None, + docker_command: tuple[str, ...] | None = None, + extra_flags: tuple[str, ...] = (), + verbose: bool = True, + ) -> None: + self.timeline.append("docker build") + self.builds.append( + { + "tag": tag, + "docker_command": tuple(docker_command or ("docker", "build")), + "extra_flags": tuple(extra_flags), + } + ) + + async def subp_exec( + self, *args: str, **kwargs: object + ) -> tuple[str | None, str | None]: + self.commands.append(DockerCommand(args=args, kwargs=kwargs)) + self.timeline.append(f"docker {self._verb(args)}") + if "push" in args and self.failing_pushes > 0: + self.failing_pushes -= 1 + raise click.exceptions.Exit(1) + if PLATFORM_FORMAT in args: + return "linux/amd64\n", None + if DIGESTS_FORMAT in args: + repository = ImageReference.parse(args[-1]).repository + return json.dumps([f"{repository}@sha256:abc123"]), None + return None, None + + @staticmethod + def _verb(args: tuple[str, ...]) -> str: + if PLATFORM_FORMAT in args: + return "inspect-platform" + if DIGESTS_FORMAT in args: + return "inspect-digest" + return next(verb for verb in ("login", "tag", "push", "pull") if verb in args) + + +class _AsyncioRunner: + def run(self, coro): + return asyncio.run(coro) + + +@contextmanager +def _fake_runner() -> Iterator[_AsyncioRunner]: + yield _AsyncioRunner() + + +@dataclass +class DeployProject: + control_plane: ControlPlaneDouble + docker: DockerDouble + timeline: list[str] + uploads: list[tuple[str, str, int]] + + def run(self, *args: str, host_url: str = CONTROL_PLANE_URL) -> Result: + return CliRunner().invoke( + cli, + [ + "deploy", + "--api-key", + "test-key", + "--host-url", + host_url, + "--name", + "my-app", + "--no-input", + "--no-wait", + *args, + ], + ) + + +@pytest.fixture +def deploy_project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> DeployProject: + (tmp_path / "langgraph.json").write_text( + json.dumps({"graphs": {"agent": "agent.py:graph"}, "dependencies": ["."]}) + ) + monkeypatch.chdir(tmp_path) + for name in ("LANGSMITH_TENANT_ID", "LANGSMITH_ENDPOINT", "LANGGRAPH_HOST_URL"): + monkeypatch.delenv(name, raising=False) + + timeline: list[str] = [] + control_plane = ControlPlaneDouble(timeline) + docker = DockerDouble(timeline) + uploads: list[tuple[str, str, int]] = [] + + @contextmanager + def fake_create_archive(config_path: Path, config: dict) -> Iterator[tuple]: + timeline.append("create_archive") + yield ARCHIVE + + def fake_upload(signed_url: str, file_path: str, file_size: int) -> None: + timeline.append("upload_archive") + uploads.append((signed_url, file_path, file_size)) + + monkeypatch.setattr(deploy_module, "_no_input", False) + monkeypatch.setattr(deploy_module, "_emitter", None) + monkeypatch.setattr( + deploy_module, "HostBackendClient", control_plane.client_factory() + ) + monkeypatch.setattr(deploy_module, "build_docker_image", docker.build_docker_image) + monkeypatch.setattr(deploy_module, "subp_exec", docker.subp_exec) + monkeypatch.setattr(deploy_module, "Runner", _fake_runner) + monkeypatch.setattr(deploy_module, "can_build_locally", lambda: (True, None)) + monkeypatch.setattr( + deploy_module, + "find_tracked_packages", + lambda config, config_json: TRACKED_PACKAGES, + ) + monkeypatch.setattr(deploy_module.platform, "machine", lambda: "x86_64") + monkeypatch.setattr(archive_module, "create_archive", fake_create_archive) + monkeypatch.setattr(deploy_module, "_upload_to_gcs", fake_upload) + return DeployProject(control_plane, docker, timeline, uploads) + + +def test_first_local_deploy_creates_then_builds_pushes_and_updates_in_order( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run("--no-remote") + + assert result.exit_code == 0, result.output + assert deploy_project.timeline == [ + LIST_DEPLOYMENTS, + CREATE_DEPLOYMENT, + "docker build", + _push_token(CREATED_ID), + "docker login", + "docker tag", + "docker push", + "docker inspect-digest", + _patch(CREATED_ID), + ] + assert "Deployment updated" in result.output + + +def test_first_local_deploy_creates_an_internal_docker_deployment( + deploy_project: DeployProject, +) -> None: + deploy_project.run("--no-remote") + + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT] == { + "name": "my-app", + "source": "internal_docker", + "source_config": {"deployment_type": "dev"}, + "source_revision_config": {}, + "secrets": [], + } + + +@pytest.mark.parametrize( + ("machine", "expected_command", "expected_flags"), + [ + pytest.param( + "arm64", + ("docker", "buildx", "build"), + ("--platform", "linux/amd64", "--load", "--progress=quiet"), + id="apple_silicon_cross_builds_for_linux_amd64", + ), + pytest.param( + "x86_64", + ("docker", "build"), + (), + id="amd64_host_uses_plain_docker_build", + ), + ], +) +def test_local_build_targets_linux_amd64( + deploy_project: DeployProject, + monkeypatch: pytest.MonkeyPatch, + machine: str, + expected_command: tuple[str, ...], + expected_flags: tuple[str, ...], +) -> None: + monkeypatch.setattr(deploy_module.platform, "machine", lambda: machine) + + deploy_project.run("--no-remote") + + build = deploy_project.docker.builds[0] + assert build["tag"].startswith("langgraph-deploy-tmp:") + assert (build["docker_command"], build["extra_flags"]) == ( + expected_command, + expected_flags, + ) + + +def test_local_deploy_logs_in_with_the_control_plane_push_token( + deploy_project: DeployProject, +) -> None: + deploy_project.run("--no-remote") + + login = deploy_project.docker.command("login") + assert login.args[:2] == ("docker", "--config") + assert login.args[3:] == ( + "login", + "-u", + "oauth2accesstoken", + "--password-stdin", + "registry.example.com", + ) + assert login.kwargs["input"] == f"{PUSH_TOKEN}\n" + + +def test_local_deploy_tags_the_build_into_the_token_registry( + deploy_project: DeployProject, +) -> None: + deploy_project.run("--no-remote") + + built_tag = deploy_project.docker.builds[0]["tag"] + assert deploy_project.docker.command("tag").args == ( + "docker", + "tag", + built_tag, + PUSHED_IMAGE, + ) + assert deploy_project.docker.command("push").args[-1] == PUSHED_IMAGE + + +def test_local_deploy_records_the_pushed_digest_and_tracked_packages( + deploy_project: DeployProject, +) -> None: + deploy_project.run("--no-remote") + + assert deploy_project.control_plane.bodies[_patch(CREATED_ID)] == { + "revision_source": "internal_docker", + "source_revision_config": {"image_uri": PUSHED_DIGEST}, + "secrets": [], + "tracked_packages": TRACKED_PACKAGES, + } + + +def test_status_link_points_at_the_langsmith_dashboard( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run("--no-remote") + + assert ( + "View status: https://smith.langchain.com/o/tenant-1/host/deployments/dep-created" + in result.output + ) + + +def test_prebuilt_image_is_validated_and_pushed_without_a_build( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run("--image", "local/app:dev") + + assert result.exit_code == 0, result.output + assert deploy_project.docker.builds == [] + assert deploy_project.docker.verbs() == [ + "docker inspect-platform", + "docker login", + "docker tag", + "docker push", + "docker inspect-digest", + ] + assert deploy_project.docker.command("tag").args[2:] == ( + "local/app:dev", + PUSHED_IMAGE, + ) + + +def test_push_is_retried_until_the_third_attempt( + deploy_project: DeployProject, +) -> None: + deploy_project.docker.failing_pushes = 2 + + result = deploy_project.run("--no-remote") + + assert result.exit_code == 0, result.output + assert deploy_project.docker.verbs().count("docker push") == 3 + + +def test_three_failed_pushes_abort_before_the_deployment_is_updated( + deploy_project: DeployProject, +) -> None: + deploy_project.docker.failing_pushes = 3 + + result = deploy_project.run("--no-remote") + + assert result.exit_code != 0 + assert _patch(CREATED_ID) not in deploy_project.timeline + + +def test_existing_deployment_matched_by_exact_name_is_updated_not_created( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.existing_deployments = [ + {"id": "dep-other", "name": "my-app-2"}, + {"id": "dep-existing", "name": "my-app"}, + ] + + deploy_project.run("--no-remote") + + assert CREATE_DEPLOYMENT not in deploy_project.timeline + assert _patch("dep-existing") in deploy_project.timeline + + +def test_deployment_not_created_by_the_cli_gets_an_actionable_error( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.existing_deployments = [ + {"id": "dep-ui", "name": "my-app"} + ] + deploy_project.control_plane.push_token_status = 400 + + result = deploy_project.run("--no-remote") + + assert result.exit_code != 0 + assert "was not created by 'langgraph deploy'" in result.output + assert "docker login" not in deploy_project.timeline + + +def test_remote_build_creates_an_internal_source_deployment_and_uploads_the_archive( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run("--remote", "--install-command", "yarn install") + + assert result.exit_code == 0, result.output + assert deploy_project.timeline == [ + LIST_DEPLOYMENTS, + CREATE_DEPLOYMENT, + "create_archive", + _upload_url(CREATED_ID), + "upload_archive", + _patch(CREATED_ID), + ] + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source"] == ( + "internal_source" + ) + assert deploy_project.uploads == [(SIGNED_UPLOAD_URL, ARCHIVE[0], ARCHIVE[1])] + assert deploy_project.control_plane.bodies[_patch(CREATED_ID)] == { + "revision_source": "internal_source", + "source_revision_config": { + "source_tarball_path": OBJECT_PATH, + "langgraph_config_path": ARCHIVE[2], + }, + "source_config": {"install_command": "yarn install"}, + "secrets": [], + "tracked_packages": TRACKED_PACKAGES, + } + assert "Build triggered" in result.output + + +def test_push_to_builds_pushes_then_creates_an_external_deployment( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run("--push-to", PUSH_REPOSITORY) + + assert result.exit_code == 0, result.output + assert deploy_project.timeline == [ + LIST_DEPLOYMENTS, + "docker build", + "docker push", + "docker inspect-digest", + CREATE_DEPLOYMENT, + ] + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT] == { + "name": "my-app", + "source": "external_docker", + "source_config": {"resource_spec": {}}, + "source_revision_config": {"image_uri": EXTERNAL_DIGEST}, + "secrets": [], + } + assert "Deployment created" in result.output + + +def test_push_to_builds_directly_with_the_push_reference( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run("--push-to", PUSH_REPOSITORY) + + assert result.exit_code == 0, result.output + assert deploy_project.docker.builds[0]["tag"] == EXTERNAL_IMAGE + assert deploy_project.docker.command("push").args == ( + "docker", + "push", + EXTERNAL_IMAGE, + ) + + +def test_push_to_composes_with_the_tag_flag(deploy_project: DeployProject) -> None: + result = deploy_project.run("--push-to", PUSH_REPOSITORY, "--tag", "v1") + + assert result.exit_code == 0, result.output + assert deploy_project.docker.command("push").args[-1] == f"{PUSH_REPOSITORY}:v1" + + +def test_push_to_with_a_failing_push_creates_no_deployment( + deploy_project: DeployProject, +) -> None: + deploy_project.docker.failing_pushes = 3 + + result = deploy_project.run("--push-to", PUSH_REPOSITORY) + + assert result.exit_code != 0 + assert CREATE_DEPLOYMENT not in deploy_project.timeline + + +def test_verbose_never_echoes_the_push_token(deploy_project: DeployProject) -> None: + result = deploy_project.run("--no-remote", "--verbose") + + assert result.exit_code == 0, result.output + assert deploy_project.docker.command("login").kwargs["verbose"] is False + + +def test_push_to_retags_a_prebuilt_image_instead_of_building( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run( + "--image", "local/app:dev", "--push-to", PUSH_REPOSITORY + ) + + assert result.exit_code == 0, result.output + assert deploy_project.docker.builds == [] + assert deploy_project.docker.verbs() == [ + "docker inspect-platform", + "docker tag", + "docker push", + "docker inspect-digest", + ] + assert deploy_project.docker.command("tag").args == ( + "docker", + "tag", + "local/app:dev", + EXTERNAL_IMAGE, + ) + + +def test_push_to_updates_an_existing_external_deployment_with_the_new_image( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.existing_deployments = [ + {"id": "dep-ext", "name": "my-app", "source": "external_docker"} + ] + + result = deploy_project.run("--push-to", PUSH_REPOSITORY) + + assert result.exit_code == 0, result.output + assert deploy_project.timeline == [ + LIST_DEPLOYMENTS, + "docker build", + "docker push", + "docker inspect-digest", + _patch("dep-ext"), + ] + assert deploy_project.control_plane.bodies[_patch("dep-ext")] == { + "source_revision_config": {"image_uri": EXTERNAL_DIGEST}, + "secrets": [], + "tracked_packages": TRACKED_PACKAGES, + } + + +def test_push_to_rejects_a_non_external_deployment_before_any_docker_work( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.existing_deployments = [ + {"id": "dep-cli", "name": "my-app", "source": "internal_docker"} + ] + + result = deploy_project.run("--push-to", PUSH_REPOSITORY) + + assert result.exit_code != 0 + assert "cannot be updated with --push-to" in result.output + assert deploy_project.docker.verbs() == [] + + +def test_push_to_with_deployment_id_fetches_the_deployment_once( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.existing_deployments = [ + {"id": "dep-ext", "name": "another-name", "source": "external_docker"} + ] + + result = deploy_project.run( + "--deployment-id", "dep-ext", "--push-to", PUSH_REPOSITORY + ) + + assert result.exit_code == 0, result.output + assert deploy_project.timeline == [ + _get("dep-ext"), + "docker build", + "docker push", + "docker inspect-digest", + _patch("dep-ext"), + ] + + +def test_invalid_tag_fails_before_any_control_plane_call( + deploy_project: DeployProject, +) -> None: + result = deploy_project.run("--no-remote", "--tag", "not a tag") + + 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 c437c4354..334c852b1 100644 --- a/libs/cli/tests/unit_tests/test_deploy_agents.py +++ b/libs/cli/tests/unit_tests/test_deploy_agents.py @@ -58,7 +58,7 @@ AGENT_ARGS = [ "deploy", "--agent-id", "customer-support", - "--environment", + "--agent-environment", "staging", "--remote", "--no-wait", @@ -66,14 +66,15 @@ AGENT_ARGS = [ ] -def test_agent_create(deployment_api, tmp_path): +def test_agent_create(deployment_api, tmp_path, monkeypatch): + monkeypatch.setenv("LANGSMITH_DEPLOYMENT_NAME", "legacy") _, requests, build = deployment_api 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"] == { @@ -93,3 +94,26 @@ def test_agent_update(deployment_api): assert result.exit_code == 0, result.output assert len(requests) == 1 assert build.call_args.kwargs["deployment_id"] == "existing-id" + + +def test_agent_rejects_explicit_name(deployment_api, monkeypatch): + monkeypatch.setenv("LANGSMITH_DEPLOYMENT_NAME", "legacy") + _, requests, _ = deployment_api + result = CliRunner().invoke(cli, [*AGENT_ARGS, "--name", "legacy"]) + 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 fa348d590..02dce8679 100644 --- a/libs/cli/tests/unit_tests/test_deploy_helpers.py +++ b/libs/cli/tests/unit_tests/test_deploy_helpers.py @@ -13,6 +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, @@ -21,12 +32,14 @@ from langgraph_cli.deploy import ( _parse_env_from_config, _resolve_env_path, _resolve_pushed_image_digest, - _smith_dashboard_base_url, + _select_source, _validate_prebuilt_image, + find_deployment_by_name, normalize_image_tag, normalize_name, ) from langgraph_cli.host_backend import HostBackendClient, HostBackendError +from langgraph_cli.image_reference import ImageReference class TestDockerConfigForToken: @@ -259,31 +272,29 @@ class TestEnvWithoutDeploymentName: class TestCallHostBackendWithOptionalTenant: def _make_client(self, handler): - c = HostBackendClient("https://api.example.com", "test-key") - c._client = httpx.Client( - base_url="https://api.example.com", + c = HostBackendClient( + "https://api.example.com", + "test-key", transport=httpx.MockTransport(handler), - headers={"X-Api-Key": "test-key", "Accept": "application/json"}, - timeout=30, ) return c def _make_eu_client(self, handler): - c = HostBackendClient("https://eu.api.host.langchain.com", "test-key") - c._client = httpx.Client( - base_url="https://eu.api.host.langchain.com", + c = HostBackendClient( + "https://eu.api.host.langchain.com", + "test-key", transport=httpx.MockTransport(handler), - headers={"X-Api-Key": "test-key", "Accept": "application/json"}, - timeout=30, ) 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 = ( @@ -334,7 +345,6 @@ class TestCallHostBackendWithOptionalTenant: assert exc_info.value.status_code == 403 assert "smith.langchain.com" in exc_info.value.message assert seen_tenant_ids == [None, "workspace-123"] - assert client._client.headers["X-Tenant-ID"] == "workspace-123" def test_other_403_re_raises_original(self): client = self._make_client( @@ -540,66 +550,226 @@ class TestCreateHostBackendClientNoInput: assert client is not None -class TestSmithDashboardBaseUrl: - def test_dev_smith_api_host_url(self): - assert ( - _smith_dashboard_base_url("https://dev.api.smith.langchain.com/") - == "https://dev.smith.langchain.com" +class TestCreateHostBackendClientEndpoint: + def test_langsmith_endpoint_from_project_env_selects_self_hosted_control_plane( + self, monkeypatch + ): + monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test") + monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False) + + client = _create_host_backend_client( + host_url=None, + api_key=None, + env_vars={"LANGSMITH_ENDPOINT": "https://smith.example.com/api/v1"}, ) - def test_none_returns_default(self): - assert _smith_dashboard_base_url(None) == "https://smith.langchain.com" + assert client.base_url == "https://smith.example.com/api-host" - def test_empty_returns_default(self): - assert _smith_dashboard_base_url("") == "https://smith.langchain.com" + def test_explicit_host_url_wins_over_langsmith_endpoint(self, monkeypatch): + monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test") + monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://smith.example.com/api/v1") - def test_prod_host_url(self): - assert ( - _smith_dashboard_base_url("https://api.host.langchain.com") - == "https://smith.langchain.com" + client = _create_host_backend_client( + host_url="https://custom.host.com", api_key=None, env_vars={} ) - def test_dev_host_url(self): - assert ( - _smith_dashboard_base_url("https://dev.api.host.langchain.com") - == "https://dev.smith.langchain.com" + assert client.base_url == "https://custom.host.com" + + +class TestDockerBuildCommand: + @pytest.mark.parametrize( + ("machine", "verbose", "expected"), + [ + pytest.param( + "x86_64", + False, + DockerBuildCommand(("docker", "build"), ()), + id="amd64_host_builds_natively", + ), + pytest.param( + "arm64", + False, + DockerBuildCommand( + ("docker", "buildx", "build"), + ("--platform", "linux/amd64", "--load", "--progress=quiet"), + ), + id="other_hosts_cross_build_quietly", + ), + pytest.param( + "arm64", + True, + DockerBuildCommand( + ("docker", "buildx", "build"), + ("--platform", "linux/amd64", "--load"), + ), + id="verbose_cross_build_keeps_progress_output", + ), + ], + ) + def test_for_host_targets_the_deployment_platform(self, machine, verbose, expected): + assert DockerBuildCommand.for_host(machine, verbose=verbose) == expected + + +class TestSelectSource: + OPTIONS = { + "push_to": None, + "image": None, + "image_name": None, + "tag": None, + "remote_build_flag": None, + "placement": RequestedPlacement(), + "selector": ByName("my-app"), + } + REPOSITORY = "registry.example.com/app" + + @pytest.mark.parametrize( + ("flags", "docker_available", "expected"), + [ + pytest.param( + {"push_to": REPOSITORY}, + True, + CustomerRegistrySource( + reference=ImageReference(REPOSITORY, "latest"), + prebuilt_image=None, + requested_placement=RequestedPlacement(), + ), + id="push_to_selects_the_external_source_with_the_default_tag", + ), + pytest.param( + {"push_to": f"{REPOSITORY}:v2"}, + True, + CustomerRegistrySource( + reference=ImageReference(REPOSITORY, "v2"), + prebuilt_image=None, + requested_placement=RequestedPlacement(), + ), + id="push_to_keeps_a_tag_given_in_the_reference", + ), + pytest.param( + {"push_to": REPOSITORY, "tag": "v3"}, + True, + CustomerRegistrySource( + reference=ImageReference(REPOSITORY, "v3"), + prebuilt_image=None, + requested_placement=RequestedPlacement(), + ), + id="tag_flag_composes_with_push_to", + ), + pytest.param( + {"push_to": REPOSITORY, "image": "app:dev"}, + False, + CustomerRegistrySource( + 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, + RemoteBuildSource(), + id="remote_flag_selects_the_source_upload", + ), + pytest.param( + {}, + False, + RemoteBuildSource(), + id="no_local_docker_falls_back_to_the_source_upload", + ), + pytest.param( + {}, + True, + ManagedRegistrySource( + prebuilt_image=None, image_name=None, tag="latest" + ), + id="local_docker_selects_the_internal_docker_source", + ), + pytest.param( + {"image": "app:dev", "tag": "v1"}, + False, + ManagedRegistrySource( + prebuilt_image="app:dev", image_name=None, tag="v1" + ), + id="prebuilt_image_forces_the_internal_docker_source", + ), + ], + ) + def test_flags_select_one_source( + self, monkeypatch, mocker, flags, docker_available, expected + ): + mocker.patch( + "langgraph_cli.deploy._get_emitter", return_value=mocker.MagicMock() + ) + monkeypatch.setattr( + deploy_mod, + "can_build_locally", + lambda: (True, None) if docker_available else (False, "Docker is required"), ) - def test_eu_host_url(self): - assert ( - _smith_dashboard_base_url("https://eu.api.host.langchain.com") - == "https://eu.smith.langchain.com" + assert _select_source(**{**self.OPTIONS, **flags}) == expected + + def test_push_to_build_requires_local_docker(self, monkeypatch): + monkeypatch.setattr( + deploy_mod, "can_build_locally", lambda: (False, "Docker is required") ) - def test_staging_host_url(self): - assert ( - _smith_dashboard_base_url("https://staging.api.host.langchain.com") - == "https://staging.smith.langchain.com" - ) + with pytest.raises(click.UsageError, match="Docker is required"): + _select_source(**{**self.OPTIONS, "push_to": self.REPOSITORY}) - def test_localhost(self): - assert ( - _smith_dashboard_base_url("http://localhost:8080") - == "http://localhost:8080" - ) + @pytest.mark.parametrize( + ("flags", "message"), + [ + pytest.param( + {"push_to": REPOSITORY, "remote_build_flag": True}, + "--push-to cannot be combined with --remote.", + id="push_to_with_remote", + ), + pytest.param( + {"push_to": f"{REPOSITORY}:v1", "tag": "v2"}, + "already includes a tag", + id="push_to_with_a_tag_and_the_tag_flag", + ), + pytest.param( + {"push_to": f"{REPOSITORY}@sha256:abc"}, + "not a digest", + id="push_to_with_a_digest", + ), + pytest.param( + {"image": "app:dev", "remote_build_flag": True}, + "--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): + monkeypatch.setattr(deploy_mod, "can_build_locally", lambda: (True, None)) - def test_localhost_trailing_slash(self): - assert ( - _smith_dashboard_base_url("http://localhost:8080/") - == "http://localhost:8080" - ) - - def test_127_0_0_1(self): - assert ( - _smith_dashboard_base_url("http://127.0.0.1:3000") - == "http://127.0.0.1:3000" - ) - - def test_unknown_domain_returns_default(self): - assert ( - _smith_dashboard_base_url("https://custom.example.com") - == "https://smith.langchain.com" - ) + with pytest.raises(click.UsageError, match=message): + _select_source(**{**self.OPTIONS, **flags}) class TestResolvePushedImageDigest: @@ -650,6 +820,16 @@ class TestResolvePushedImageDigest: ) assert out == "us-central1-docker.pkg.dev/proj/repo@sha256:abc123" + def test_registry_port_without_tag_still_resolves_the_digest(self): + runner = self._runner('["localhost:5000/repo@sha256:abc123"]') + out = _resolve_pushed_image_digest( + runner, + remote_image="localhost:5000/repo", + docker_config_dir=None, + verbose=False, + ) + assert out == "localhost:5000/repo@sha256:abc123" + def test_empty_repodigests_falls_back_with_warning(self, mocker): emitter = mocker.MagicMock() mocker.patch("langgraph_cli.deploy._get_emitter", return_value=emitter) @@ -753,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 85c2a93d5..d89468c41 100644 --- a/libs/cli/tests/unit_tests/test_host_backend.py +++ b/libs/cli/tests/unit_tests/test_host_backend.py @@ -3,29 +3,16 @@ import json import httpx import pytest -from langgraph_cli.host_backend import HostBackendClient, HostBackendError - - -@pytest.fixture -def mock_transport(): - return httpx.MockTransport(lambda req: httpx.Response(200, json={"ok": True})) - - -@pytest.fixture -def client(mock_transport): - c = HostBackendClient("https://api.example.com", "test-key") - c._client = httpx.Client( - base_url="https://api.example.com", - transport=mock_transport, - headers={"X-Api-Key": "test-key", "Accept": "application/json"}, - timeout=30, - ) - return c +from langgraph_cli.host_backend import ( + ControlPlaneEndpoints, + HostBackendClient, + HostBackendError, +) def test_constructor_strips_trailing_slash(): c = HostBackendClient("https://api.example.com/", "key") - assert str(c._client.base_url) == "https://api.example.com" + assert c.base_url == "https://api.example.com" def test_constructor_empty_url_raises(): @@ -39,12 +26,8 @@ def test_request_sends_headers(): assert req.headers["accept"] == "application/json" return httpx.Response(200, json={"ok": True}) - c = HostBackendClient("https://api.example.com", "test-key") - c._client = httpx.Client( - base_url="https://api.example.com", - transport=httpx.MockTransport(handler), - headers={"X-Api-Key": "test-key", "Accept": "application/json"}, - timeout=30, + c = HostBackendClient( + "https://api.example.com", "test-key", transport=httpx.MockTransport(handler) ) result = c._request("GET", "/test") assert result == {"ok": True} @@ -56,12 +39,8 @@ def test_request_sends_json_payload(): assert req.content == b'{"key":"value"}' return httpx.Response(200, json={"created": True}) - c = HostBackendClient("https://api.example.com", "test-key") - c._client = httpx.Client( - base_url="https://api.example.com", - transport=httpx.MockTransport(handler), - headers={"X-Api-Key": "test-key", "Accept": "application/json"}, - timeout=30, + c = HostBackendClient( + "https://api.example.com", "test-key", transport=httpx.MockTransport(handler) ) result = c._request("POST", "/test", {"key": "value"}) assert result == {"created": True} @@ -69,25 +48,13 @@ def test_request_sends_json_payload(): def test_request_empty_body_returns_none(): transport = httpx.MockTransport(lambda req: httpx.Response(200, content=b"")) - c = HostBackendClient("https://api.example.com", "test-key") - c._client = httpx.Client( - base_url="https://api.example.com", - transport=transport, - headers={"X-Api-Key": "test-key", "Accept": "application/json"}, - timeout=30, - ) + c = HostBackendClient("https://api.example.com", "test-key", transport=transport) assert c._request("DELETE", "/test") is None def test_request_http_error_raises(): transport = httpx.MockTransport(lambda req: httpx.Response(404, text="not found")) - c = HostBackendClient("https://api.example.com", "test-key") - c._client = httpx.Client( - base_url="https://api.example.com", - transport=transport, - headers={"X-Api-Key": "test-key", "Accept": "application/json"}, - timeout=30, - ) + c = HostBackendClient("https://api.example.com", "test-key", transport=transport) with pytest.raises(HostBackendError, match="404"): c._request("GET", "/missing") @@ -96,13 +63,7 @@ def test_request_invalid_json_raises(): transport = httpx.MockTransport( lambda req: httpx.Response(200, content=b"not json") ) - c = HostBackendClient("https://api.example.com", "test-key") - c._client = httpx.Client( - base_url="https://api.example.com", - transport=transport, - headers={"X-Api-Key": "test-key", "Accept": "application/json"}, - timeout=30, - ) + c = HostBackendClient("https://api.example.com", "test-key", transport=transport) with pytest.raises(HostBackendError, match="Failed to decode"): c._request("GET", "/bad-json") @@ -111,84 +72,20 @@ def test_request_transport_error_raises(): def handler(req: httpx.Request) -> httpx.Response: raise httpx.ConnectError("connection refused") - c = HostBackendClient("https://api.example.com", "test-key") - c._client = httpx.Client( - base_url="https://api.example.com", - transport=httpx.MockTransport(handler), - headers={"X-Api-Key": "test-key", "Accept": "application/json"}, - timeout=30, + c = HostBackendClient( + "https://api.example.com", "test-key", transport=httpx.MockTransport(handler) ) with pytest.raises(HostBackendError, match="connection refused"): c._request("GET", "/test") -def test_create_deployment(client): - result = client.create_deployment( - name="my-deploy", deployment_type="dev", source="internal_docker" - ) - assert result == {"ok": True} - - -def test_get_deployment(client): - result = client.get_deployment("dep-123") - assert result == {"ok": True} - - -def test_list_deployments(client): - result = client.list_deployments("my-app") - assert result == {"ok": True} - - -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") - c._client = httpx.Client( - base_url="https://api.example.com", - transport=httpx.MockTransport(handler), - headers={"X-Api-Key": "test-key", "Accept": "application/json"}, - timeout=30, - ) - result = c.list_deployments("my app") - assert result == {"ok": True} - - -def test_delete_deployment(client): - result = client.delete_deployment("dep-123") - assert result == {"ok": True} - - -def test_request_push_token(client): - result = client.request_push_token("dep-123") - assert result == {"ok": True} - - -def test_update_deployment(client): - result = client.update_deployment( - "dep-123", "image:latest", secrets=[{"name": "KEY", "value": "val"}] - ) - assert result == {"ok": True} - - -def test_update_deployment_no_secrets(client): - result = client.update_deployment("dep-123", "image:latest") - assert result == {"ok": True} - - def _capturing_client(captured: dict) -> HostBackendClient: def handler(req: httpx.Request) -> httpx.Response: captured["body"] = req.read() return httpx.Response(200, json={"ok": True}) - c = HostBackendClient("https://api.example.com", "key") - c._client = httpx.Client( - base_url="https://api.example.com", - transport=httpx.MockTransport(handler), - headers={"X-Api-Key": "key", "Accept": "application/json"}, - timeout=30, + c = HostBackendClient( + "https://api.example.com", "key", transport=httpx.MockTransport(handler) ) return c @@ -199,6 +96,7 @@ def test_update_deployment_forwards_tracked_packages(): c.update_deployment( "dep-123", "image:latest", + revision_source="internal_docker", tracked_packages=["google-adk:1.0.0"], ) body = json.loads(captured["body"]) @@ -209,7 +107,7 @@ def test_update_deployment_forwards_tracked_packages(): def test_update_deployment_omits_tracked_packages_when_absent(): captured: dict = {} c = _capturing_client(captured) - c.update_deployment("dep-123", "image:latest") + c.update_deployment("dep-123", "image:latest", revision_source="internal_docker") body = json.loads(captured["body"]) assert "tracked_packages" not in body @@ -241,33 +139,14 @@ def test_update_deployment_internal_source_omits_tracked_packages_when_absent(): assert "tracked_packages" not in body -def test_list_revisions(client): - result = client.list_revisions("dep-123", limit=5) - assert result == {"ok": True} - - -def test_get_revision(client): - result = client.get_revision("dep-123", "rev-456") - assert result == {"ok": True} - - -def test_get_build_logs(client): - result = client.get_build_logs("proj-1", "rev-1", {"limit": 10}) - assert result == {"ok": True} - - def test_get_deploy_logs_all_revisions(): def handler(req: httpx.Request) -> httpx.Response: assert "/v1/projects/proj-1/deploy_logs" in str(req.url) assert "/revisions/" not in str(req.url) return httpx.Response(200, json={"logs": [{"message": "running"}]}) - c = HostBackendClient("https://api.example.com", "key") - c._client = httpx.Client( - base_url="https://api.example.com", - transport=httpx.MockTransport(handler), - headers={"X-Api-Key": "key", "Accept": "application/json"}, - timeout=30, + c = HostBackendClient( + "https://api.example.com", "key", transport=httpx.MockTransport(handler) ) result = c.get_deploy_logs("proj-1", {"limit": 10}) assert result == {"logs": [{"message": "running"}]} @@ -278,12 +157,526 @@ def test_get_deploy_logs_specific_revision(): assert "/v1/projects/proj-1/revisions/rev-2/deploy_logs" in str(req.url) return httpx.Response(200, json={"logs": []}) - c = HostBackendClient("https://api.example.com", "key") - c._client = httpx.Client( - base_url="https://api.example.com", - transport=httpx.MockTransport(handler), - headers={"X-Api-Key": "key", "Accept": "application/json"}, - timeout=30, + c = HostBackendClient( + "https://api.example.com", "key", transport=httpx.MockTransport(handler) ) result = c.get_deploy_logs("proj-1", {"limit": 10}, revision_id="rev-2") assert result == {"logs": []} + + +def _routing_client(seen: dict) -> HostBackendClient: + def handler(req: httpx.Request) -> httpx.Response: + seen["method"] = req.method + seen["url"] = str(req.url) + return httpx.Response(200, json={"ok": True}) + + c = HostBackendClient( + "https://api.example.com/prefix", "key", transport=httpx.MockTransport(handler) + ) + return c + + +@pytest.mark.parametrize( + ("call", "expected_body"), + [ + pytest.param( + lambda c: c.create_deployment( + name="my-deploy", + source="internal_docker", + source_config={"deployment_type": "dev"}, + source_revision_config={}, + ), + { + "name": "my-deploy", + "source": "internal_docker", + "source_config": {"deployment_type": "dev"}, + "source_revision_config": {}, + }, + id="internal_docker_create_omits_secrets_key_when_not_given", + ), + pytest.param( + lambda c: c.create_deployment( + name="my-deploy", + source="internal_docker", + source_config={"deployment_type": "prod"}, + source_revision_config={}, + secrets=[{"name": "KEY", "value": "val"}], + ), + { + "name": "my-deploy", + "source": "internal_docker", + "source_config": {"deployment_type": "prod"}, + "source_revision_config": {}, + "secrets": [{"name": "KEY", "value": "val"}], + }, + id="internal_docker_create_forwards_secrets", + ), + pytest.param( + lambda c: c.update_deployment( + "dep-123", + "registry.example.com/app@sha256:abc", + revision_source="internal_docker", + secrets=[{"name": "KEY", "value": "val"}], + ), + { + "revision_source": "internal_docker", + "source_revision_config": { + "image_uri": "registry.example.com/app@sha256:abc" + }, + "secrets": [{"name": "KEY", "value": "val"}], + }, + id="internal_docker_revision_names_its_source", + ), + pytest.param( + lambda c: c.update_deployment_internal_source( + "dep-123", + source_tarball_path="tarballs/src.tgz", + config_path="langgraph.json", + secrets=[], + install_command="yarn install", + build_command="yarn build", + ), + { + "revision_source": "internal_source", + "source_revision_config": { + "source_tarball_path": "tarballs/src.tgz", + "langgraph_config_path": "langgraph.json", + }, + "source_config": { + "install_command": "yarn install", + "build_command": "yarn build", + }, + "secrets": [], + }, + id="internal_source_revision_sends_js_build_commands", + ), + pytest.param( + lambda c: c.update_deployment_internal_source( + "dep-123", + source_tarball_path="tarballs/src.tgz", + config_path="langgraph.json", + ), + { + "revision_source": "internal_source", + "source_revision_config": { + "source_tarball_path": "tarballs/src.tgz", + "langgraph_config_path": "langgraph.json", + }, + }, + id="internal_source_revision_omits_source_config_without_commands", + ), + pytest.param( + lambda c: c.create_deployment( + name="agent", + source="external_docker", + source_config={"resource_spec": {}}, + source_revision_config={ + "image_uri": "registry.example.com/agent@sha256:1" + }, + secrets=[], + ), + { + "name": "agent", + "source": "external_docker", + "source_config": {"resource_spec": {}}, + "source_revision_config": { + "image_uri": "registry.example.com/agent@sha256:1" + }, + "secrets": [], + }, + id="create_sends_the_source_configs_as_given", + ), + pytest.param( + lambda c: c.update_deployment( + "dep-1", "registry.example.com/agent@sha256:2", revision_source=None + ), + { + "source_revision_config": { + "image_uri": "registry.example.com/agent@sha256:2" + } + }, + id="revision_without_source_override_omits_revision_source", + ), + pytest.param( + lambda c: c.update_deployment( + "dep-1", + "registry.example.com/agent@sha256:2", + revision_source="internal_docker", + tracked_packages=["langgraph:1.0.0"], + ), + { + "revision_source": "internal_docker", + "source_revision_config": { + "image_uri": "registry.example.com/agent@sha256:2" + }, + "tracked_packages": ["langgraph:1.0.0"], + }, + id="revision_with_source_override_names_it", + ), + ], +) +def test_request_body_matches_control_plane_contract(call, expected_body): + captured: dict = {} + call(_capturing_client(captured)) + assert json.loads(captured["body"]) == expected_body + + +@pytest.mark.parametrize( + ("call", "method", "route"), + [ + pytest.param( + lambda c: c.create_deployment( + name="n", + source="internal_docker", + source_config={"deployment_type": "dev"}, + source_revision_config={}, + ), + "POST", + "/v2/deployments", + id="create_deployment", + ), + pytest.param( + lambda c: c.get_deployment("dep-1"), + "GET", + "/v2/deployments/dep-1", + id="get_deployment", + ), + pytest.param( + lambda c: c.delete_deployment("dep-1"), + "DELETE", + "/v2/deployments/dep-1", + id="delete_deployment", + ), + pytest.param( + lambda c: c.update_deployment("dep-1", "img", revision_source=None), + "PATCH", + "/v2/deployments/dep-1", + id="patch_deployment", + ), + pytest.param( + lambda c: c.request_push_token("dep-1"), + "POST", + "/v2/deployments/dep-1/push-token", + id="push_token", + ), + pytest.param( + lambda c: c.request_upload_url("dep-1"), + "POST", + "/v2/deployments/dep-1/upload-url", + id="upload_url", + ), + pytest.param( + lambda c: c.list_revisions("dep-1", limit=5), + "GET", + "/v2/deployments/dep-1/revisions?limit=5", + id="list_revisions_puts_limit_in_query", + ), + pytest.param( + lambda c: c.get_revision("dep-1", "rev-2"), + "GET", + "/v2/deployments/dep-1/revisions/rev-2", + id="get_revision", + ), + pytest.param( + lambda c: c.get_build_logs("dep-1", "rev-2", {"limit": 10}), + "POST", + "/v1/projects/dep-1/revisions/rev-2/build_logs", + id="build_logs", + ), + ], +) +def test_request_targets_control_plane_route_under_base_url(call, method, route): + seen: dict = {} + call(_routing_client(seen)) + assert (seen["method"], seen["url"]) == ( + method, + f"https://api.example.com/prefix{route}", + ) + + +def test_injected_transport_receives_requests_under_the_prefixed_base_url(): + seen: dict = {} + + def handler(req: httpx.Request) -> httpx.Response: + seen["url"] = str(req.url) + seen["api_key"] = req.headers["x-api-key"] + return httpx.Response(200, json={"ok": True}) + + c = HostBackendClient( + "https://smith.example.com/api-host", + "key", + transport=httpx.MockTransport(handler), + ) + + 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", + } + + +CLOUD = ("https://api.host.langchain.com", "https://smith.langchain.com") + + +@pytest.mark.parametrize( + ("host_url", "langsmith_endpoint", "expected"), + [ + pytest.param(None, None, CLOUD, id="nothing_configured_targets_cloud"), + pytest.param( + None, "https://api.smith.langchain.com", CLOUD, id="cloud_langsmith_api" + ), + pytest.param( + None, + "https://api.smith.langchain.com/api/v1", + CLOUD, + id="cloud_langsmith_api_with_versioned_path", + ), + pytest.param( + None, "https://api.langchain.com", CLOUD, id="cloud_langchain_api_alias" + ), + pytest.param( + None, + "https://xapi.smith.langchain.com", + CLOUD, + id="lookalike_cloud_host_is_not_rewritten_into_a_control_plane", + ), + pytest.param( + None, + "https://eu.api.smith.langchain.com", + ("https://eu.api.host.langchain.com", "https://eu.smith.langchain.com"), + id="eu_cloud_maps_to_eu_control_plane", + ), + pytest.param( + None, + "https://dev.api.smith.langchain.com", + ("https://dev.api.host.langchain.com", "https://dev.smith.langchain.com"), + id="dev_cloud_maps_to_dev_control_plane", + ), + pytest.param( + None, + "https://aks.smith.langchain.dev/api", + ( + "https://aks.smith.langchain.dev/api-host", + "https://aks.smith.langchain.dev", + ), + id="self_hosted_api_path_becomes_api_host", + ), + pytest.param( + None, + "https://smith.example.com/api/v1", + ("https://smith.example.com/api-host", "https://smith.example.com"), + id="self_hosted_versioned_api_path_becomes_api_host", + ), + pytest.param( + None, + "https://smith.example.com", + ("https://smith.example.com/api-host", "https://smith.example.com"), + id="self_hosted_origin_gets_api_host_appended", + ), + pytest.param( + None, + "https://corp.example.com/langsmith/api/v1", + ( + "https://corp.example.com/langsmith/api-host", + "https://corp.example.com/langsmith", + ), + id="self_hosted_path_prefix_is_kept", + ), + pytest.param( + "https://custom.host.example", + "https://aks.smith.langchain.dev/api", + ("https://custom.host.example", "https://smith.langchain.com"), + id="explicit_host_url_beats_langsmith_endpoint", + ), + pytest.param( + "https://api.host.langchain.com", + "https://aks.smith.langchain.dev/api", + CLOUD, + id="explicit_cloud_host_url_beats_self_hosted_endpoint", + ), + pytest.param( + "https://smith.example.com/api-host/", + None, + ("https://smith.example.com/api-host", "https://smith.example.com"), + id="explicit_api_host_url_derives_dashboard_root", + ), + pytest.param( + "https://corp.example.com/langsmith/api-host", + None, + ( + "https://corp.example.com/langsmith/api-host", + "https://corp.example.com/langsmith", + ), + id="explicit_api_host_url_keeps_path_prefix_in_dashboard", + ), + pytest.param( + "http://localhost:8080", + None, + ("http://localhost:8080", "http://localhost:8080"), + id="localhost_dashboard_is_the_same_origin", + ), + pytest.param( + "http://localhost:8080/api-host", + None, + ("http://localhost:8080/api-host", "http://localhost:8080"), + id="localhost_api_host_dashboard_is_the_origin", + ), + pytest.param( + "https://eu.api.host.langchain.com", + None, + ("https://eu.api.host.langchain.com", "https://eu.smith.langchain.com"), + id="regional_control_plane_maps_to_regional_dashboard", + ), + pytest.param( + "https://dev.api.smith.langchain.com/", + None, + ("https://dev.api.smith.langchain.com", "https://dev.smith.langchain.com"), + id="dev_smith_api_maps_to_dev_dashboard", + ), + ], +) +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) diff --git a/libs/cli/tests/unit_tests/test_image_reference.py b/libs/cli/tests/unit_tests/test_image_reference.py new file mode 100644 index 000000000..901555940 --- /dev/null +++ b/libs/cli/tests/unit_tests/test_image_reference.py @@ -0,0 +1,71 @@ +import pytest + +from langgraph_cli.image_reference import ImageReference + + +@pytest.mark.parametrize( + ("reference", "repository", "tag"), + [ + pytest.param( + "registry.example.com/team/app:v1", + "registry.example.com/team/app", + "v1", + id="tag_after_last_slash", + ), + pytest.param( + "registry.example.com/team/app", + "registry.example.com/team/app", + None, + id="no_tag", + ), + pytest.param( + "localhost:5000/app", + "localhost:5000/app", + None, + id="registry_port_is_not_a_tag", + ), + pytest.param( + "localhost:5000/app:latest", + "localhost:5000/app", + "latest", + id="registry_port_with_tag", + ), + pytest.param("app:dev", "app", "dev", id="bare_name_with_tag"), + ], +) +def test_parse_splits_repository_and_tag(reference, repository, tag): + assert ImageReference.parse(reference) == ImageReference(repository, tag) + + +def test_with_tag_replaces_the_tag(): + assert ImageReference("r/app", "v1").with_tag("v2") == ImageReference("r/app", "v2") + + +@pytest.mark.parametrize( + ("reference", "expected"), + [ + pytest.param(ImageReference("r/app", "v1"), "r/app:v1", id="tagged"), + pytest.param(ImageReference("r/app"), "r/app", id="untagged"), + ], +) +def test_str_renders_the_docker_reference(reference, expected): + assert str(reference) == expected + + +@pytest.mark.parametrize( + ("repo_digest", "expected"), + [ + pytest.param("localhost:5000/app@sha256:abc", True, id="same_repository"), + pytest.param("localhost:5000/app-2@sha256:abc", False, id="other_repository"), + pytest.param("mirror.example.com/app@sha256:abc", False, id="other_registry"), + ], +) +def test_matches_digest_only_for_the_same_repository(repo_digest, expected): + assert ImageReference("localhost:5000/app", "v1").matches_digest(repo_digest) is ( + expected + ) + + +def test_parse_rejects_a_digest_reference(): + with pytest.raises(ValueError, match="digest"): + ImageReference.parse("registry.example.com/app@sha256:abc")