Fix customer-registry deployment creation and image handling

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

libs/cli/langgraph_cli/deploy.py

@@ -639,15 +639,23 @@ def _create_deployment(
     source: str,
     config_rel: str | None = None,
     secrets: list[dict[str, str]] | None = None,
+    image_uri: str | None = None,
+    tracked_packages: list[str] | None = None,
 ) -> tuple[str, int]:
     """Create a deployment and return its ID and next step number."""
     _log_deploy_step(step, f"Creating deployment '{name}'")
+    extra = (
+        {"image_uri": image_uri, "tracked_packages": tracked_packages}
+        if image_uri
+        else {}
+    )
     created = client.create_deployment(
         name=name,
         deployment_type=deployment_type,
         source=source,
         config_path=config_rel,
         secrets=secrets,
+        **extra,
     )
     created_id = created.get("id") if isinstance(created, dict) else None
     if not isinstance(created_id, str) or not created_id:
@@ -666,7 +674,6 @@ def _smith_dashboard_base_url(host_url: str | None) -> str:
         return "https://smith.langchain.com"
     parsed = urlparse(host_url)
     hostname = parsed.hostname or ""
-    # Self-hosted: host_url is <scheme>://<host>/api-host — return just the root
     path = parsed.path.rstrip("/")
     if path == "/api-host" or path.endswith("/api-host"):
         return f"{parsed.scheme}://{parsed.netloc}"
@@ -946,43 +953,29 @@ def _build_image_tagged(
     verbose: bool,
 ) -> None:
     """Build a Docker image to *tag*, using buildx on non-x86_64 hosts to target linux/amd64."""
-    if platform.machine() != "x86_64":
-        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,
-                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,
-                tag,
-                docker_build_args,
-                install_command,
-                build_command,
-                verbose=verbose,
-            )
+    needs_buildx = platform.machine() != "x86_64"
+    build_flags = ["--platform", "linux/amd64", "--load"] if needs_buildx else []
+    if needs_buildx and 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,
+            tag,
+            docker_build_args,
+            install_command,
+            build_command,
+            docker_command=("docker", "buildx", "build")
+            if needs_buildx
+            else ("docker", "build"),
+            extra_flags=build_flags,
+            verbose=verbose,
+        )
 
 
 def _run_local_build(
@@ -1152,14 +1145,13 @@ def _run_local_build(
     )
 
 
-def _run_external_deploy(
+def _push_external_image(
     *,
-    client: HostBackendClient,
-    deployment_id: str,
     step: int,
     config: pathlib.Path,
     config_json: dict,
-    image_uri: str,
+    push_to: str,
+    prebuilt_image: str | None,
     verbose: bool,
     pull: bool,
     api_version: str | None,
@@ -1167,55 +1159,40 @@ def _run_external_deploy(
     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 image, push using existing Docker credentials, and update the deployment."""
+) -> str:
+    """Build or retag an image and push using existing Docker credentials."""
     with Runner() as runner:
-        _log_deploy_step(step, f"Building image {image_uri}")
-        _build_image_tagged(
-            runner,
-            config,
-            config_json,
-            base_image,
-            api_version,
-            pull,
-            image_uri,
-            docker_build_args,
-            install_command,
-            build_command,
-            verbose=verbose,
-        )
-        step += 1
-
-        _log_deploy_step(step, f"Pushing image {image_uri}")
+        if prebuilt_image:
+            _log_deploy_step(step, f"Validating image {prebuilt_image}")
+            _validate_prebuilt_image(runner, prebuilt_image, verbose=verbose)
+            runner.run(
+                subp_exec("docker", "tag", prebuilt_image, push_to, verbose=verbose)
+            )
+        else:
+            _log_deploy_step(step, f"Building image {push_to}")
+            _build_image_tagged(
+                runner,
+                config,
+                config_json,
+                base_image,
+                api_version,
+                pull,
+                push_to,
+                docker_build_args,
+                install_command,
+                build_command,
+                verbose=verbose,
+            )
+        _log_deploy_step(step + 1, f"Pushing image {push_to}")
         with Progress(message="Pushing...", elapsed=not verbose):
-            runner.run(subp_exec("docker", "push", image_uri, verbose=verbose))
-        step += 1
-
-        resolved_image = _resolve_pushed_image_digest(
+            runner.run(subp_exec("docker", "push", push_to, verbose=verbose))
+        return _resolve_pushed_image_digest(
             runner,
-            remote_image=image_uri,
+            remote_image=push_to,
             docker_config_dir=None,
             verbose=verbose,
         )
 
-        _log_deploy_step(step, f"Updating deployment {deployment_id}")
-        updated = client.update_deployment_external(
-            deployment_id,
-            resolved_image,
-            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",
-    )
-
 
 def _run_remote_build(
     *,
@@ -1352,30 +1329,20 @@ def _create_host_backend_client(
     tenant_id = env_vars.get("LANGSMITH_TENANT_ID") or os.environ.get(
         "LANGSMITH_TENANT_ID"
     )
-    # If no explicit host URL was provided, check LANGSMITH_ENDPOINT as a
-    # fallback so self-hosted customers don't need to know about LANGGRAPH_HOST_URL.
-    # Self-hosted control plane always lives at <langsmith_endpoint>/api-host.
-    _cloud_default = "https://api.host.langchain.com"
-    _cloud_endpoints = {
-        "https://api.smith.langchain.com",
-        "https://api.langchain.com",
-    }
-    resolved_host = host_url
-    if not resolved_host or resolved_host == _cloud_default:
-        langsmith_endpoint = env_vars.get("LANGSMITH_ENDPOINT") or os.environ.get(
+    if not host_url:
+        from urllib.parse import urlparse
+
+        endpoint = env_vars.get("LANGSMITH_ENDPOINT") or os.environ.get(
             "LANGSMITH_ENDPOINT"
         )
-        if (
-            langsmith_endpoint
-            and langsmith_endpoint.rstrip("/") not in _cloud_endpoints
-        ):
-            from urllib.parse import urlparse as _urlparse
-
-            _p = _urlparse(langsmith_endpoint)
-            resolved_host = f"{_p.scheme}://{_p.netloc}/api-host"
+        parsed = urlparse(endpoint or "https://api.smith.langchain.com")
+        if parsed.hostname in ("api.smith.langchain.com", "api.langchain.com"):
+            host_url = "https://api.host.langchain.com"
+        elif parsed.hostname == "eu.api.smith.langchain.com":
+            host_url = "https://eu.api.host.langchain.com"
         else:
-            resolved_host = _cloud_default
-    return HostBackendClient(resolved_host, resolved_api_key, tenant_id=tenant_id)
+            host_url = f"{parsed.scheme}://{parsed.netloc}/api-host"
+    return HostBackendClient(host_url, resolved_api_key, tenant_id=tenant_id)
 
 
 def _call_host_backend_with_optional_tenant(
@@ -1454,7 +1421,7 @@ 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,
 )
 
@@ -1594,13 +1561,11 @@ def _deploy_base_options(
                 ),
             ),
             click.option(
-                "--image-uri",
+                "--push-to",
                 help=(
-                    "Image URI to build, push, and deploy "
-                    "(e.g. 123456789.dkr.ecr.us-east-1.amazonaws.com/repo:tag). "
-                    "Builds the project, pushes using existing Docker credentials, "
-                    "and triggers a deployment revision. "
-                    "Required for self-hosted deployments."
+                    "Push to this customer-managed registry URI for self-hosted/hybrid "
+                    "deployments using existing Docker credentials. Builds the project "
+                    "or retags the local image supplied with --image."
                 ),
             ),
             click.option(
@@ -1704,7 +1669,7 @@ def _deploy_cmd(
     name: str | None,
     image_name: str | None,
     image: str | None,
-    image_uri: str | None,
+    push_to: str | None,
     tag: str,
     base_image: str | None,
     install_command: str | None,
@@ -1751,12 +1716,10 @@ def _deploy_cmd(
 
     secrets = _secrets_from_env(_env_without_deployment_name(env_vars))
 
-    if image_uri and remote_build_flag is True:
-        raise click.UsageError("--image-uri cannot be combined with --remote.")
-    if image_uri and image:
-        raise click.UsageError("--image-uri cannot be combined with --image.")
+    if push_to and remote_build_flag is True:
+        raise click.UsageError("--push-to cannot be combined with --remote.")
 
-    use_external_docker = image_uri is not None
+    use_external_docker = push_to is not None
 
     if use_external_docker:
         use_remote_build = False
@@ -1782,45 +1745,35 @@ def _deploy_cmd(
         name,
         not_found_message=(
             "No deployment found. Will create."
-            if (use_remote_build or use_external_docker)
+            if use_remote_build
             else "No deployment found. Will create after build."
         ),
     )
 
-    if needs_creation:
-        if use_external_docker:
-            source = "external_docker"
-        elif use_remote_build:
-            source = "internal_source"
-        else:
-            source = "internal_docker"
+    if needs_creation and not use_external_docker:
         deployment_id, step = _create_deployment(
             client,
             step,
             name=name,
             deployment_type=deployment_type,
-            source=source,
+            source="internal_source" if use_remote_build else "internal_docker",
             secrets=secrets,
         )
 
-    if not deployment_id:
+    if not deployment_id and not use_external_docker:
         raise click.ClickException("Failed to determine deployment ID")
 
-    # Validate that an existing deployment is compatible with --image-uri.
-    # update_deployment_external sends an external_docker revision; applying
-    # it to a deployment created with a different source mode may be rejected
-    # by the backend or silently produce an inconsistent revision.
     if use_external_docker and not needs_creation:
         existing = _call_host_backend_with_optional_tenant(
             client, lambda c: c.get_deployment(deployment_id)
         )
         existing_source = existing.get("source") if isinstance(existing, dict) else None
-        if existing_source and existing_source != "external_docker":
+        if existing_source != "external_docker":
             raise click.UsageError(
                 f"Deployment {deployment_id} uses a different build mode and "
-                f"cannot be updated with --image-uri. To use --image-uri, omit "
+                f"cannot be updated with --push-to. To use --push-to, omit "
                 f"--deployment-id to create a new deployment, or remove "
-                f"--image-uri to continue using the current build mode."
+                f"--push-to to continue using the current build mode."
             )
 
     # Scan local sources for tracked packages so the new revision carries
@@ -1834,13 +1787,12 @@ def _deploy_cmd(
 
     # -- 3. Build (divergent path) --
     if use_external_docker:
-        build_result = _run_external_deploy(
-            client=client,
-            deployment_id=deployment_id,
+        resolved_image = _push_external_image(
             step=step,
             config=config,
             config_json=config_json,
-            image_uri=image_uri,
+            push_to=push_to,
+            prebuilt_image=image,
             verbose=verbose,
             pull=pull,
             api_version=api_version,
@@ -1848,8 +1800,32 @@ def _deploy_cmd(
             install_command=install_command,
             build_command=build_command,
             docker_build_args=docker_build_args,
-            secrets=secrets,
-            tracked_packages=tracked_packages,
+        )
+        if needs_creation:
+            deployment_id, step = _create_deployment(
+                client,
+                step + 2,
+                name=name,
+                deployment_type=deployment_type,
+                source="external_docker",
+                secrets=secrets,
+                image_uri=resolved_image,
+                tracked_packages=tracked_packages,
+            )
+            updated = client.get_deployment(deployment_id)
+        else:
+            updated = client.update_deployment_external(
+                deployment_id,
+                resolved_image,
+                secrets=secrets,
+                tracked_packages=tracked_packages,
+            )
+        build_result = BuildResult(
+            updated=updated,
+            progress_message="Deploying...",
+            timeout_seconds=300,
+            poll_interval_seconds=1,
+            no_result_message="Deployment updated",
         )
     elif use_remote_build:
         build_result = _run_remote_build(

libs/cli/langgraph_cli/host_backend.py

@@ -81,6 +81,8 @@ class HostBackendClient:
         source: str,
         config_path: str | None = None,
         secrets: list[dict[str, str]] | None = None,
+        image_uri: str | None = None,
+        tracked_packages: list[str] | None = None,
     ) -> dict[str, Any]:
         """Create a deployment."""
         payload: dict[str, Any] = {
@@ -89,6 +91,11 @@ class HostBackendClient:
             "source_config": {"deployment_type": deployment_type},
             "source_revision_config": {},
         }
+        if source == "external_docker":
+            payload["source_config"] = {}
+            payload["source_revision_config"] = {"image_uri": image_uri}
+        if tracked_packages:
+            payload["tracked_packages"] = tracked_packages
         if source == "internal_source" and config_path:
             payload["source_revision_config"]["langgraph_config_path"] = config_path
         if secrets is not None:

libs/cli/tests/unit_tests/cli/test_cli.py

@@ -6,8 +6,10 @@ import tempfile
 import textwrap
 from contextlib import contextmanager
 from pathlib import Path
+from unittest.mock import MagicMock
 
 import click
+import pytest
 from click.testing import CliRunner
 
 import langgraph_cli.deploy as deploy_module
@@ -1356,47 +1358,89 @@ def test_prepare_args_and_stdin_distributed_mode() -> None:
     assert "executor_entrypoint.sh" in actual_stdin
 
 
-def test_deploy_image_uri_rejects_incompatible_source(monkeypatch, tmp_path) -> None:
-    """--image-uri raises UsageError when applied to a non-external_docker deployment."""
-    # --no-input sets the module-level _no_input global; ensure it's restored.
+@pytest.mark.parametrize(
+    "source,image",
+    [
+        (None, None),
+        (None, "local:latest"),
+        ("external_docker", None),
+        ("external_docker", "local:latest"),
+        ("internal_docker", None),
+    ],
+)
+def test_deploy_push_to(monkeypatch, tmp_path, source, image):
     monkeypatch.setattr(deploy_module, "_no_input", False)
-
+    monkeypatch.setattr(deploy_module, "_emitter", None)
     config = tmp_path / "langgraph.json"
     config.write_text('{"graphs": {"agent": "agent.py:graph"}, "dependencies": ["."]}')
-
-    class FakeClient:
-        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
-            self.base_url = host_url
-
-        def get_deployment(self, deployment_id: str):
-            return {
-                "id": deployment_id,
-                "name": "test-deploy",
-                "source": "internal_docker",
-                "tenant_id": "tenant-1",
-            }
-
-    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
-
-    runner = CliRunner()
-    result = runner.invoke(
-        cli,
-        [
-            "deploy",
-            "--api-key",
-            "test-key",
-            "--host-url",
-            "https://api.example.com",
-            "--deployment-id",
-            "dep-123",
-            "--image-uri",
-            "registry.example.com/app:latest",
-            "--config",
-            str(config),
-            "--no-input",
-        ],
+    events = []
+    client = MagicMock(base_url="https://smith.example.com/api-host")
+    client.get_deployment.return_value = {"id": "dep-123", "source": source}
+    client.list_deployments.return_value = {"deployments": []}
+    client.create_deployment.side_effect = lambda **kw: (
+        events.append(("create", kw)) or {"id": "dep-123"}
     )
-
-    assert result.exit_code != 0
-    assert "different build mode" in result.output
-    assert "cannot be updated with --image-uri" in result.output
+    client.update_deployment_external.side_effect = lambda *a, **kw: (
+        events.append(("update", a)) or {}
+    )
+    monkeypatch.setattr(deploy_module, "HostBackendClient", lambda *a, **kw: client)
+    monkeypatch.setattr(
+        deploy_module,
+        "_build_image_tagged",
+        lambda *a, **kw: events.append(("build", a[6])),
+    )
+    monkeypatch.setattr(
+        deploy_module,
+        "_validate_prebuilt_image",
+        lambda *a, **kw: events.append(("validate", a[1])),
+    )
+    monkeypatch.setattr(deploy_module, "subp_exec", lambda *a, **kw: a)
+    runner = MagicMock()
+    runner.__enter__.return_value = runner
+    runner.run.side_effect = lambda args: events.append(args)
+    monkeypatch.setattr(deploy_module, "Runner", lambda: runner)
+    digest = "registry.example.com/app@sha256:abc123"
+    monkeypatch.setattr(
+        deploy_module,
+        "_resolve_pushed_image_digest",
+        lambda *a, **kw: events.append(("digest",)) or digest,
+    )
+    args = [
+        "deploy",
+        "--api-key",
+        "key",
+        "--push-to",
+        "registry.example.com/app:latest",
+        "--config",
+        str(config),
+        "--no-input",
+        "--no-wait",
+    ]
+    args += ["--deployment-id", "dep-123"] if source else ["--name", "app"]
+    if image:
+        args += ["--image", image]
+    result = CliRunner().invoke(cli, args)
+    if source == "internal_docker":
+        assert result.exit_code != 0
+        assert "cannot be updated with --push-to" in result.output
+        assert events == []
+        return
+    assert result.exit_code == 0, result.output
+    destination = "registry.example.com/app:latest"
+    expected = (
+        [("validate", image), ("docker", "tag", image, destination)]
+        if image
+        else [("build", destination)]
+    )
+    assert events[: len(expected) + 2] == expected + [
+        ("docker", "push", destination),
+        ("digest",),
+    ]
+    if source:
+        assert events[-1] == ("update", ("dep-123", digest))
+        client.create_deployment.assert_not_called()
+    else:
+        assert events[-1][0] == "create"
+        assert events[-1][1]["image_uri"] == digest
+        assert events[-1][1]["source"] == "external_docker"
+        client.update_deployment_external.assert_not_called()

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -543,53 +543,47 @@ class TestCreateHostBackendClientNoInput:
         assert client is not None
 
 
-class TestCreateHostBackendClientEndpointFallback:
-    def test_langsmith_endpoint_env_var_used_as_fallback(self, monkeypatch):
-        monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test")
-        monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://smith.example.com/api/v1")
-        monkeypatch.delenv("LANGGRAPH_HOST_URL", raising=False)
-        client = _create_host_backend_client(host_url=None, api_key=None, env_vars={})
-        assert client.base_url == "https://smith.example.com/api-host"
-
-    def test_langsmith_endpoint_from_env_vars_dict(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"},
-        )
-        assert client.base_url == "https://smith.example.com/api-host"
-
-    def test_cloud_langsmith_endpoint_not_used_as_self_hosted(self, monkeypatch):
-        monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test")
-        monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://api.smith.langchain.com")
-        client = _create_host_backend_client(host_url=None, api_key=None, env_vars={})
-        assert client.base_url == "https://api.host.langchain.com"
-
-    def test_langchain_api_endpoint_not_used_as_self_hosted(self, monkeypatch):
-        monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test")
-        monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://api.langchain.com")
-        client = _create_host_backend_client(host_url=None, api_key=None, env_vars={})
-        assert client.base_url == "https://api.host.langchain.com"
-
-    def test_explicit_host_url_takes_precedence_over_langsmith_endpoint(
-        self, monkeypatch
-    ):
-        monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test")
-        monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://smith.example.com/api/v1")
-        client = _create_host_backend_client(
-            host_url="https://custom.host.com",
-            api_key=None,
-            env_vars={},
-        )
-        assert client.base_url == "https://custom.host.com"
-
-    def test_no_endpoint_falls_back_to_cloud_default(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={})
-        assert client.base_url == "https://api.host.langchain.com"
+@pytest.mark.parametrize(
+    "endpoint,host,expected",
+    [
+        (None, None, "https://api.host.langchain.com"),
+        ("https://api.smith.langchain.com/", None, "https://api.host.langchain.com"),
+        ("https://api.langchain.com", None, "https://api.host.langchain.com"),
+        (
+            "https://eu.api.smith.langchain.com",
+            None,
+            "https://eu.api.host.langchain.com",
+        ),
+        (
+            "https://smith.example.com/api/v1",
+            None,
+            "https://smith.example.com/api-host",
+        ),
+        (
+            "https://smith.example.com",
+            "https://api.host.langchain.com",
+            "https://api.host.langchain.com",
+        ),
+        (
+            "https://smith.example.com",
+            "https://custom.host.com",
+            "https://custom.host.com",
+        ),
+    ],
+)
+@pytest.mark.parametrize("from_env", [True, False])
+def test_endpoint_fallback(monkeypatch, endpoint, host, expected, from_env):
+    monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
+    env_vars = {}
+    if endpoint:
+        if from_env:
+            monkeypatch.setenv("LANGSMITH_ENDPOINT", endpoint)
+        else:
+            env_vars["LANGSMITH_ENDPOINT"] = endpoint
+    client = _create_host_backend_client(
+        host_url=host, api_key="key", env_vars=env_vars
+    )
+    assert client.base_url == expected
 
 
 class TestSmithDashboardBaseUrl:
@@ -647,23 +641,16 @@ class TestSmithDashboardBaseUrl:
             == "https://smith.langchain.com"
         )
 
-    def test_self_hosted_api_host_suffix(self):
-        assert (
-            _smith_dashboard_base_url("https://smith.example.com/api-host")
-            == "https://smith.example.com"
-        )
-
-    def test_self_hosted_api_host_trailing_slash(self):
-        assert (
-            _smith_dashboard_base_url("https://smith.example.com/api-host/")
-            == "https://smith.example.com"
-        )
-
-    def test_self_hosted_localhost_api_host(self):
-        assert (
-            _smith_dashboard_base_url("http://localhost:8080/api-host")
-            == "http://localhost:8080"
-        )
+    @pytest.mark.parametrize(
+        "url,expected",
+        [
+            ("https://smith.example.com/api-host", "https://smith.example.com"),
+            ("https://smith.example.com/api-host/", "https://smith.example.com"),
+            ("http://localhost:8080/api-host", "http://localhost:8080"),
+        ],
+    )
+    def test_self_hosted(self, url, expected):
+        assert _smith_dashboard_base_url(url) == expected
 
 
 class TestResolvePushedImageDigest:

libs/cli/tests/unit_tests/test_host_backend.py

@@ -178,45 +178,32 @@ def test_update_deployment_no_secrets(client):
     assert result == {"ok": True}
 
 
-def test_update_deployment_external():
+@pytest.mark.parametrize("create", [True, False])
+def test_external_deployment_payload(create):
     captured: dict = {}
     c = _capturing_client(captured)
-    result = c.update_deployment_external(
-        "dep-123", "registry.example.com/app@sha256:abc123"
-    )
-    assert result == {"ok": True}
+    image = "registry.example.com/app@sha256:abc123"
+    kwargs = {
+        "secrets": [{"name": "KEY", "value": "value"}],
+        "tracked_packages": ["google-adk:1.0.0"],
+    }
+    if create:
+        c.create_deployment("app", "dev", "external_docker", image_uri=image, **kwargs)
+    else:
+        c.update_deployment_external("dep-123", image, **kwargs)
     body = json.loads(captured["body"])
-    assert "revision_source" not in body
-    assert body["source_revision_config"]["image_uri"] == (
-        "registry.example.com/app@sha256:abc123"
-    )
-
-
-def test_update_deployment_external_forwards_tracked_packages():
-    captured: dict = {}
-    c = _capturing_client(captured)
-    c.update_deployment_external(
-        "dep-123",
-        "registry.example.com/app:latest",
-        tracked_packages=["google-adk:1.0.0"],
-    )
-    body = json.loads(captured["body"])
-    assert body["tracked_packages"] == ["google-adk:1.0.0"]
-    assert "revision_source" not in body
-
-
-def test_update_deployment_external_omits_tracked_packages_when_absent():
-    captured: dict = {}
-    c = _capturing_client(captured)
-    c.update_deployment_external("dep-123", "registry.example.com/app:latest")
-    body = json.loads(captured["body"])
-    assert "tracked_packages" not in body
-    assert "revision_source" not in body
+    assert body == {
+        **(
+            {"name": "app", "source": "external_docker", "source_config": {}}
+            if create
+            else {}
+        ),
+        "source_revision_config": {"image_uri": image},
+        **kwargs,
+    }
 
 
 def test_request_builds_full_url_with_path_prefix():
-    """base_url with a path prefix (/api-host) must not be dropped when paths start with /."""
-
     def handler(req: httpx.Request) -> httpx.Response:
         assert str(req.url) == "https://smith.example.com/api-host/v2/deployments"
         return httpx.Response(200, json={"ok": True})