diff --git a/libs/cli/langgraph_cli/cli.py b/libs/cli/langgraph_cli/cli.py index f487070ca..21715d50c 100644 --- a/libs/cli/langgraph_cli/cli.py +++ b/libs/cli/langgraph_cli/cli.py @@ -29,7 +29,7 @@ from langgraph_cli.exec import Runner, subp_exec from langgraph_cli.host_backend import HostBackendClient, HostBackendError from langgraph_cli.progress import Progress from langgraph_cli.templates import TEMPLATE_HELP_STRING, create_new -from langgraph_cli.util import warn_non_wolfi_distro +from langgraph_cli.util import format_deployments_table, warn_non_wolfi_distro from langgraph_cli.version import __version__ RESERVED_ENV_VARS = frozenset( @@ -287,6 +287,22 @@ OPT_API_VERSION = click.option( help="API server version to use for the base image. If unspecified, the latest version will be used.", ) +OPT_HOST_API_KEY = click.option( + "--api-key", + envvar="LANGGRAPH_HOST_API_KEY", + help=( + "API key. Can also be set via LANGGRAPH_HOST_API_KEY, " + "LANGSMITH_API_KEY, or LANGCHAIN_API_KEY environment variable or .env file." + ), +) + +OPT_HOST_URL = click.option( + "--host-url", + envvar="LANGGRAPH_HOST_URL", + default="https://api.host.langchain.com", + hidden=True, +) + OPT_ENGINE_RUNTIME_MODE = click.option( "--engine-runtime-mode", type=click.Choice(["combined_queue_worker", "distributed"]), @@ -295,7 +311,67 @@ OPT_ENGINE_RUNTIME_MODE = click.option( ) -@click.group() +class NestedHelpGroup(click.Group): + """Click group that shows one level of nested subcommands in top-level help.""" + + def format_commands( + self, ctx: click.Context, formatter: click.HelpFormatter + ) -> None: + command_entries: list[tuple[str, click.Command]] = [] + # Collect the top-level commands first, then append one level of nested + # subcommands using names like "deploy list" so they show up in the + # top-level help output. + for command_name in self.list_commands(ctx): + command = self.get_command(ctx, command_name) + if command is None or command.hidden: + continue + command_entries.append((command_name, command)) + if isinstance(command, click.Group): + # Build a child context so Click resolves the subcommands the same + # way it would for the nested group itself. + sub_ctx = click.Context(command, info_name=command_name, parent=ctx) + for subcommand_name in command.list_commands(sub_ctx): + subcommand = command.get_command(sub_ctx, subcommand_name) + if subcommand is None or subcommand.hidden: + continue + command_entries.append( + (f"{command_name} {subcommand_name}", subcommand) + ) + + # Compute the available width for help text up front so we can truncate + # descriptions before handing them to Click. That keeps each command on + # a single line instead of allowing wrapped descriptions. + command_width = max((len(name) for name, _ in command_entries), default=0) + help_width = max(formatter.width - command_width - 6, 10) + rows = [ + (name, command.get_short_help_str(help_width)) + for name, command in command_entries + ] + + if rows: + # Render the flattened command list using Click's standard + # definition-list formatter so alignment stays consistent with the + # rest of the CLI help output. + with formatter.section("Commands"): + formatter.write_dl(rows) + + +class DeployGroup(NestedHelpGroup): + """Group that treats leading '-' args as passthrough docker flags.""" + + def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]: + result = super().parse_args(ctx, args) + if ctx._protected_args and ctx._protected_args[0].startswith("-"): + # Click stores the would-be subcommand in _protected_args; if it looks + # like an option (e.g. --build-arg) treat it as passthrough docker + # args instead of insisting on a nested command. + ctx.args = [*ctx._protected_args, *ctx.args] + ctx._protected_args = [] + return ctx.args + return result + + +@click.group(cls=NestedHelpGroup) @click.version_option(version=__version__, prog_name="LangGraph CLI") def cli(): pass @@ -593,72 +669,89 @@ def build( ) -@click.option( - "--api-key", - envvar="LANGGRAPH_HOST_API_KEY", - help=( - "API key. Can also be set via LANGGRAPH_HOST_API_KEY, " - "LANGSMITH_API_KEY, or LANGCHAIN_API_KEY environment variable or .env file." - ), -) -@click.option( - "--name", - envvar="LANGSMITH_DEPLOYMENT_NAME", - help=( - "Deployment name. Can also be set via LANGSMITH_DEPLOYMENT_NAME " - "environment variable or .env file. Defaults to current directory name " - "if --deployment-id is not provided." - ), -) -@click.option( - "--deployment-id", - help=( - "ID of an existing deployment to update. If omitted, " - "--name is used to find or create the deployment." - ), -) -@click.option( - "--deployment-type", - type=click.Choice(["dev", "prod"]), - default="dev", - show_default=True, - help="Deployment type (used when creating a new deployment).", -) -@click.option( - "--no-wait", - is_flag=True, - default=False, - help="Skip waiting for deployment status.", -) -@OPT_VERBOSE -@click.option( - "--host-url", - envvar="LANGGRAPH_HOST_URL", - default="https://api.host.langchain.com", - hidden=True, -) -@click.option("--image-name", hidden=True) -@click.option("--image-tag", default="latest", hidden=True) -@click.option( - "--config", - "-c", - default=DEFAULT_CONFIG, - hidden=True, - type=click.Path( - exists=True, - file_okay=True, - dir_okay=False, - resolve_path=True, - path_type=pathlib.Path, - ), -) -@click.option("--pull/--no-pull", default=True, hidden=True) -@click.option("--base-image", hidden=True) -@click.option("--install-command", hidden=True) -@click.option("--build-command", hidden=True) -@click.option("--api-version", type=str, hidden=True) -@click.argument("docker_build_args", nargs=-1, type=click.UNPROCESSED) -@cli.command( +def _deploy_base_options( + func: Callable | None = None, + *, + include_docker_args: bool = True, + validate_config_path: bool = True, +): + """Apply shared deploy flags. + + The group shares most options but should not consume subcommands, so the + docker build args are only attached when requested. + """ + + def _apply(target: Callable) -> Callable: + decorators = [ + OPT_HOST_API_KEY, + click.option( + "--name", + envvar="LANGSMITH_DEPLOYMENT_NAME", + help=( + "Deployment name. Can also be set via LANGSMITH_DEPLOYMENT_NAME " + "environment variable or .env file. Defaults to current directory name " + "if --deployment-id is not provided." + ), + ), + click.option( + "--deployment-id", + help=( + "ID of an existing deployment to update. If omitted, " + "--name is used to find or create the deployment." + ), + ), + click.option( + "--deployment-type", + type=click.Choice(["dev", "prod"]), + default="dev", + show_default=True, + help="Deployment type (used when creating a new deployment).", + ), + click.option( + "--no-wait", + is_flag=True, + default=False, + help="Skip waiting for deployment status.", + ), + OPT_VERBOSE, + OPT_HOST_URL, + click.option("--image-name", hidden=True), + click.option("--image-tag", default="latest", hidden=True), + click.option( + "--config", + "-c", + default=DEFAULT_CONFIG, + hidden=True, + type=click.Path( + exists=validate_config_path, + file_okay=True, + dir_okay=False, + resolve_path=True, + path_type=pathlib.Path, + ), + ), + click.option("--pull/--no-pull", default=True, hidden=True), + click.option("--base-image", hidden=True), + click.option("--install-command", hidden=True), + click.option("--build-command", hidden=True), + click.option("--api-version", type=str, hidden=True), + ] + if include_docker_args: + # Only attach build args to the default command; on the group they + # would capture subcommand names like `list` before Click resolves + # them, making those subcommands unreachable. + decorators.append( + click.argument("docker_build_args", nargs=-1, type=click.UNPROCESSED) + ) + for decorator in reversed(decorators): + target = decorator(target) + return target + + return _apply(func) if func is not None else _apply + + +@cli.group( + cls=DeployGroup, help=( "[Beta] Build and deploy a LangGraph image to LangSmith Deployments.\n\n" "This command is in beta and under active development. " @@ -667,10 +760,26 @@ def build( "is located). This command also accepts build flags (--base-image, " "--pull, etc.). See 'langgraph build --help' for details." ), - context_settings=dict(ignore_unknown_options=True), + context_settings=dict(ignore_unknown_options=True, allow_extra_args=True), + invoke_without_command=True, # allow `deploy` click group to execute without command ) +@_deploy_base_options(include_docker_args=False, validate_config_path=False) +@click.pass_context @log_command -def deploy( +def deploy(ctx: click.Context, **_: object): + # We register deploy as both a group and a command here. + # if we detect no subcommand, we run _deploy (basically run langgraph deploy as a top level command) + # 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 + docker_build_args = tuple(ctx.args) + ctx.args = [] # Prevent Click from re-processing passthrough args later. + return ctx.forward(_deploy, docker_build_args=docker_build_args) + + +@_deploy_base_options() +@click.command(context_settings=dict(ignore_unknown_options=True)) +def _deploy( config: pathlib.Path, pull: bool, verbose: bool, @@ -1050,6 +1159,115 @@ def deploy( ) +def _create_host_backend_client( + host_url: str | None, + api_key: str | None, + env_vars: dict[str, str] | None = None, +) -> HostBackendClient: + if env_vars is None: + env_vars = _parse_env_from_config({}, pathlib.Path.cwd() / DEFAULT_CONFIG) + resolved_api_key = api_key + if not resolved_api_key: + for key_name in _API_KEY_ENV_NAMES: + val = env_vars.get(key_name) + if val: + resolved_api_key = val + break + val = os.environ.get(key_name) + if val: + resolved_api_key = val + break + if not resolved_api_key: + resolved_api_key = click.prompt("Host API key", hide_input=True) + return HostBackendClient(host_url, resolved_api_key) + + +def _call_host_backend_with_optional_tenant( + client: HostBackendClient, + operation: Callable[[HostBackendClient], object], +) -> object: + try: + return operation(client) + except HostBackendError as err: + if err.status_code == 403 and "requires workspace specification" in err.message: + click.secho( + "Your API key is org-scoped and requires a workspace ID.", + fg="yellow", + ) + click.secho( + "Find your workspace ID in LangSmith under Settings > Workspaces.", + fg="yellow", + ) + tenant_id = click.prompt("Workspace ID") + client = HostBackendClient( + client._base_url, client._api_key, tenant_id=tenant_id + ) + return operation(client) + raise + + +@OPT_HOST_API_KEY +@OPT_HOST_URL +@click.option( + "--name-contains", + default="", + help="Only show deployments whose names contain this value.", +) +@deploy.command("list", help="[Beta] List LangSmith Deployments.") +def deploy_list(api_key: str | None, host_url: str | None, name_contains: str) -> None: + client = _create_host_backend_client(host_url, api_key) + response = _call_host_backend_with_optional_tenant( + client, + lambda current_client: current_client.list_deployments( + name_contains=name_contains + ), + ) + resources = response.get("resources", []) if isinstance(response, dict) else [] + deployments = [item for item in resources if isinstance(item, dict)] + if not deployments: + click.echo("No deployments found.") + return + click.echo(format_deployments_table(deployments)) + + +@OPT_HOST_API_KEY +@OPT_HOST_URL +@click.option( + "--force", + is_flag=True, + default=False, + help="Delete without prompting for confirmation.", +) +@click.argument("deployment_id") +@deploy.command( + "delete", + help=( + "[Beta] Delete a LangSmith Deployment.\n\n" + "Use the `deploy list` command to list deployment IDs." + ), +) +def deploy_delete( + api_key: str | None, host_url: str | None, force: bool, deployment_id: str +) -> None: + if not force: + response = click.prompt( + click.style( + f"Are you sure you want to delete deployment ID {deployment_id}? (Y/n)", + fg="yellow", + ), + default="Y", + show_default=False, + ) + if response.strip().lower() not in {"y", "yes"}: + raise click.Abort() + client = _create_host_backend_client(host_url, api_key) + _call_host_backend_with_optional_tenant( + client, + lambda current_client: current_client.delete_deployment(deployment_id), + ) + click.secho(f"Deleted deployment {deployment_id}.", fg="green") + + def _normalize_image_name(value: str | None) -> str: """Sanitize a deployment/directory name into a valid Docker repository name. diff --git a/libs/cli/langgraph_cli/host_backend.py b/libs/cli/langgraph_cli/host_backend.py index f53055051..3e8be1a69 100644 --- a/libs/cli/langgraph_cli/host_backend.py +++ b/libs/cli/langgraph_cli/host_backend.py @@ -39,10 +39,14 @@ class HostBackendClient: ) def _request( - self, method: str, path: str, payload: dict[str, Any] | None = None + self, + method: str, + path: str, + payload: dict[str, Any] | None = None, + params: dict[str, Any] | None = None, ) -> Any: try: - resp = self._client.request(method, path, json=payload) + 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) @@ -65,12 +69,19 @@ class HostBackendClient: def create_deployment(self, payload: dict[str, Any]) -> dict[str, Any]: return self._request("POST", "/v2/deployments", payload) - def list_deployments(self, name_contains: str) -> dict[str, Any]: - return self._request("GET", f"/v2/deployments?name_contains={name_contains}") + def list_deployments(self, name_contains: str = "") -> dict[str, Any]: + return self._request( + "GET", + "/v2/deployments", + params={"name_contains": name_contains}, + ) def get_deployment(self, deployment_id: str) -> dict[str, Any]: return self._request("GET", f"/v2/deployments/{deployment_id}") + def delete_deployment(self, deployment_id: str) -> None: + return self._request("DELETE", f"/v2/deployments/{deployment_id}") + def request_push_token(self, deployment_id: str) -> dict[str, Any]: return self._request( "POST", diff --git a/libs/cli/langgraph_cli/util.py b/libs/cli/langgraph_cli/util.py index 683ed44bc..7e75e4020 100644 --- a/libs/cli/langgraph_cli/util.py +++ b/libs/cli/langgraph_cli/util.py @@ -1,3 +1,5 @@ +from collections.abc import Sequence + import click @@ -23,3 +25,35 @@ def warn_non_wolfi_distro(config_json: dict) -> None: fg="yellow", ) click.secho("") # Empty line for better readability + + +def _extract_deployment_url(deployment: dict[str, object]) -> str: + source_config = deployment.get("source_config") + if isinstance(source_config, dict): + custom_url = source_config.get("custom_url") + if isinstance(custom_url, str) and custom_url: + return custom_url + return "-" + + +def format_deployments_table(deployments: Sequence[dict[str, object]]) -> str: + headers = ("Deployment ID", "Deployment Name", "Deployment URL") + rows = [ + ( + str(deployment.get("id", "-") or "-"), + str(deployment.get("name", "-") or "-"), + _extract_deployment_url(deployment), + ) + for deployment in deployments + ] + widths = [ + max(len(headers[index]), *(len(row[index]) for row in rows)) + for index in range(len(headers)) + ] + + def format_row(row: Sequence[str]) -> str: + return " ".join(value.ljust(widths[index]) for index, value in enumerate(row)) + + lines = [format_row(headers), format_row(tuple("-" * width for width in widths))] + lines.extend(format_row(row) for row in rows) + return "\n".join(lines) diff --git a/libs/cli/tests/unit_tests/cli/test_cli.py b/libs/cli/tests/unit_tests/cli/test_cli.py index 801b83ab0..7768b9641 100644 --- a/libs/cli/tests/unit_tests/cli/test_cli.py +++ b/libs/cli/tests/unit_tests/cli/test_cli.py @@ -9,6 +9,7 @@ from pathlib import Path from click.testing import CliRunner +import langgraph_cli.cli as cli_module from langgraph_cli.cli import cli, prepare_args_and_stdin from langgraph_cli.config import Config, _get_pip_cleanup_lines, validate_config from langgraph_cli.docker import DEFAULT_POSTGRES_URI, DockerCapabilities, Version @@ -287,6 +288,238 @@ def test_version_option() -> None: ) +def test_top_level_help_shows_deploy_subcommands() -> None: + runner = CliRunner() + + result = runner.invoke(cli, ["--help"]) + + assert result.exit_code == 0, result.output + assert "deploy" in result.output + assert "deploy list" in result.output + assert "deploy delete" in result.output + assert "[Beta] List LangSmith Deployments." in result.output + + +def test_top_level_help_truncates_command_descriptions_to_single_line() -> None: + runner = CliRunner() + + result = runner.invoke(cli, ["--help"]) + + assert result.exit_code == 0, result.output + lines = result.output.splitlines() + deploy_line = next(line for line in lines if line.strip().startswith("deploy")) + deploy_list_line = next( + line for line in lines if line.strip().startswith("deploy list") + ) + + assert not lines[lines.index(deploy_line) + 1].startswith(" ") + assert "..." in deploy_line + assert "[Beta] List LangSmith Deployments." in deploy_list_line + + +def test_deploy_list_command(monkeypatch) -> None: + runner = CliRunner() + captured: dict[str, str] = {} + + class FakeClient: + def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None): + captured["host_url"] = host_url + captured["api_key"] = api_key + captured["tenant_id"] = tenant_id or "" + + 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"}, + }, + ] + } + + monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient) + + result = runner.invoke( + cli, + [ + "deploy", + "list", + "--api-key", + "test-key", + "--host-url", + "https://api.example.com", + "--name-contains", + "alp", + ], + ) + + assert result.exit_code == 0, result.output + assert captured == { + "host_url": "https://api.example.com", + "api_key": "test-key", + "tenant_id": "", + "name_contains": "alp", + } + assert "Deployment ID" in result.output + assert "Deployment Name" in result.output + assert "Deployment URL" in result.output + assert "dep-123" in result.output + assert "https://beta.example.com" in result.output + + +def test_deploy_list_command_no_results(monkeypatch) -> None: + runner = CliRunner() + + class FakeClient: + def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None): + pass + + def list_deployments(self, name_contains: str = ""): + return {"resources": []} + + monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient) + + result = runner.invoke( + cli, + [ + "deploy", + "list", + "--api-key", + "test-key", + "--host-url", + "https://api.example.com", + ], + ) + + assert result.exit_code == 0, result.output + assert result.output.strip() == "No deployments found." + + +def test_deploy_delete_command(monkeypatch) -> None: + runner = CliRunner() + captured: dict[str, str] = {} + + class FakeClient: + def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None): + captured["host_url"] = host_url + captured["api_key"] = api_key + captured["tenant_id"] = tenant_id or "" + + def delete_deployment(self, deployment_id: str): + captured["deployment_id"] = deployment_id + return None + + monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient) + + result = runner.invoke( + cli, + [ + "deploy", + "delete", + "--api-key", + "test-key", + "--host-url", + "https://api.example.com", + "dep-123", + ], + input="y\n", + ) + + assert result.exit_code == 0, result.output + assert captured == { + "host_url": "https://api.example.com", + "api_key": "test-key", + "tenant_id": "", + "deployment_id": "dep-123", + } + assert ( + "Are you sure you want to delete deployment ID dep-123? (Y/n):" in result.output + ) + assert result.output.strip().endswith("Deleted deployment dep-123.") + + +def test_deploy_delete_command_cancelled(monkeypatch) -> None: + runner = CliRunner() + deleted = False + + class FakeClient: + def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None): + pass + + def delete_deployment(self, deployment_id: str): + nonlocal deleted + deleted = True + return None + + monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient) + + result = runner.invoke( + cli, + [ + "deploy", + "delete", + "--api-key", + "test-key", + "--host-url", + "https://api.example.com", + "dep-123", + ], + input="n\n", + ) + + assert result.exit_code == 1, result.output + assert not deleted + assert "Aborted!" in result.output + + +def test_deploy_delete_command_force(monkeypatch) -> None: + runner = CliRunner() + captured: dict[str, str] = {} + + class FakeClient: + def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None): + captured["host_url"] = host_url + captured["api_key"] = api_key + captured["tenant_id"] = tenant_id or "" + + def delete_deployment(self, deployment_id: str): + captured["deployment_id"] = deployment_id + return None + + monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient) + + result = runner.invoke( + cli, + [ + "deploy", + "delete", + "--force", + "--api-key", + "test-key", + "--host-url", + "https://api.example.com", + "dep-123", + ], + ) + + assert result.exit_code == 0, result.output + assert "Are you sure you want to delete deployment ID dep-123?" not in result.output + assert captured == { + "host_url": "https://api.example.com", + "api_key": "test-key", + "tenant_id": "", + "deployment_id": "dep-123", + } + assert result.output.strip() == "Deleted deployment dep-123." + + def test_dockerfile_command_basic() -> None: """Test the 'dockerfile' command with basic configuration.""" runner = CliRunner() diff --git a/libs/cli/tests/unit_tests/test_config.py b/libs/cli/tests/unit_tests/test_config.py index b71dd76c8..050284905 100644 --- a/libs/cli/tests/unit_tests/test_config.py +++ b/libs/cli/tests/unit_tests/test_config.py @@ -1784,7 +1784,9 @@ def test_config_to_compose_distributed_mode(): # Executor service is present with correct base image assert "langgraph-executor:" in actual_compose_stdin assert "FROM langchain/langgraph-executor:3.11" in actual_compose_stdin - assert 'entrypoint: ["sh", "/storage/executor_entrypoint.sh"]' in actual_compose_stdin + assert ( + 'entrypoint: ["sh", "/storage/executor_entrypoint.sh"]' in actual_compose_stdin + ) # Executor has required environment variables assert "EXECUTOR_GRPC_PORT:" in actual_compose_stdin diff --git a/libs/cli/tests/unit_tests/test_host_backend.py b/libs/cli/tests/unit_tests/test_host_backend.py index 3e91d4553..e4d41f4c8 100644 --- a/libs/cli/tests/unit_tests/test_host_backend.py +++ b/libs/cli/tests/unit_tests/test_host_backend.py @@ -135,6 +135,28 @@ def test_list_deployments(client): 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} diff --git a/libs/cli/tests/unit_tests/test_util.py b/libs/cli/tests/unit_tests/test_util.py index f4b439959..3e05becdd 100644 --- a/libs/cli/tests/unit_tests/test_util.py +++ b/libs/cli/tests/unit_tests/test_util.py @@ -1,6 +1,11 @@ from unittest.mock import patch -from langgraph_cli.util import clean_empty_lines, warn_non_wolfi_distro +from langgraph_cli.util import ( + _extract_deployment_url, + clean_empty_lines, + format_deployments_table, + warn_non_wolfi_distro, +) def test_clean_empty_lines(): @@ -186,3 +191,36 @@ def test_warn_non_wolfi_distro_does_not_modify_config(): warn_non_wolfi_distro(config_copy) assert config_copy == original_config # Config should remain unchanged + + +def test_extract_deployment_url_uses_custom_url(): + deployment = {"source_config": {"custom_url": "https://example.com/custom"}} + assert _extract_deployment_url(deployment) == "https://example.com/custom" + + +def test_extract_deployment_url_defaults_to_dash(): + assert _extract_deployment_url({"id": "dep-123"}) == "-" + + +def test_format_deployments_table(): + output = format_deployments_table( + [ + { + "id": "dep-123", + "name": "alpha", + "source_config": {"custom_url": "https://alpha.example.com"}, + }, + { + "id": "dep-456", + "name": "beta", + "url": "https://beta.example.com", + }, + ] + ) + assert "Deployment ID" in output + assert "Deployment Name" in output + assert "Deployment URL" in output + assert "dep-123" in output + assert "alpha" in output + assert "https://alpha.example.com" in output + assert "dep-456" in output