Compare commits

...
Author SHA1 Message Date
gdibben 6b84ae69a0 Prevent --json from prompting interactively on a 403
Per review: --json still went through _create_host_backend_client
and _call_host_backend_with_optional_tenant, both of which prompt
interactively (writing to stdout, reading stdin) when the module-
level _no_input flag is left at its default False. deploy_listeners_
list never set it, unlike the main deploy command, so a missing API
key or an org-scoped key without LANGSMITH_TENANT_ID would corrupt
the JSON output with warning text and could hang waiting for input
that a scripted invocation never provides.

Sets _no_input = json_output at the top of the command, same pattern
_deploy_cmd already uses, so both paths raise a ClickException to
stderr instead. Added a regression test that fails without the fix
(click.prompt is genuinely invoked) and passes with it.
2026-10-08 08:26:40 +01:00
gdibben 48b2346e33 feat(cli): add --json to deploy listeners list
The table output isn't scriptable, so a deploy automation pipeline
has no reliable way to extract a listener's id to pass to
--listener-id. --json prints the listeners as a JSON array instead
(an empty array when there are none, so callers never need to
special-case the human-readable "no listeners" message).
2026-10-08 07:57:05 +01:00
2 changed files with 141 additions and 1 deletions
+15 -1
View File
@@ -2518,6 +2518,13 @@ def deploy_listeners() -> None:
@OPT_HOST_API_KEY
@OPT_HOST_URL
@click.option(
"--json",
"json_output",
is_flag=True,
default=False,
help="Emit the listeners as a JSON array instead of a table.",
)
@deploy_listeners.command(
"list",
help=(
@@ -2526,11 +2533,18 @@ def deploy_listeners() -> None:
"--listener-id <id>` to deploy through it."
),
)
def deploy_listeners_list(api_key: str | None, host_url: str | None) -> None:
def deploy_listeners_list(
api_key: str | None, host_url: str | None, json_output: bool
) -> None:
global _no_input
_no_input = json_output
client = _create_host_backend_client(host_url, api_key)
listeners = _call_host_backend_with_optional_tenant(
client, lambda c: c.list_listeners()
)
if json_output:
click.echo(json_mod.dumps(listeners))
return
if not listeners:
click.echo("No listeners found for this workspace.")
return
+126
View File
@@ -14,6 +14,7 @@ import langgraph_cli.deploy as deploy_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
from langgraph_cli.host_backend import HostBackendError
from langgraph_cli.util import clean_empty_lines
FORMATTED_CLEANUP_LINES = _get_pip_cleanup_lines(
@@ -535,6 +536,131 @@ def test_deploy_listeners_list_command_no_results(monkeypatch) -> None:
assert result.output.strip() == "No listeners found for this workspace."
def test_deploy_listeners_list_command_json(monkeypatch) -> None:
# --json sets the module-level _no_input global as a side effect; pin
# the "before" value so monkeypatch restores it after this test.
monkeypatch.setattr(deploy_module, "_no_input", False)
runner = CliRunner()
listeners = [
{
"id": "listener-1",
"compute_id": "prod-cluster",
"compute_config": {"k8s_namespaces": ["agents"]},
},
{
"id": "listener-2",
"compute_id": "multi-cluster",
"compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
},
]
class FakeClient:
def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
pass
def list_listeners(self):
return listeners
monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
result = runner.invoke(
cli,
[
"deploy",
"listeners",
"list",
"--json",
"--api-key",
"test-key",
"--host-url",
"https://api.example.com",
],
)
assert result.exit_code == 0, result.output
assert json.loads(result.output) == listeners
def test_deploy_listeners_list_command_json_no_results(monkeypatch) -> None:
# --json sets the module-level _no_input global as a side effect; pin
# the "before" value so monkeypatch restores it after this test.
monkeypatch.setattr(deploy_module, "_no_input", False)
runner = CliRunner()
class FakeClient:
def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
pass
def list_listeners(self):
return []
monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
result = runner.invoke(
cli,
[
"deploy",
"listeners",
"list",
"--json",
"--api-key",
"test-key",
"--host-url",
"https://api.example.com",
],
)
assert result.exit_code == 0, result.output
assert json.loads(result.output) == []
def test_deploy_listeners_list_command_json_org_scoped_403(monkeypatch) -> None:
"""With --json, an org-scoped 403 must raise cleanly instead of
prompting for a workspace ID, which would write to stdout and hang
waiting for input that a scripted invocation never provides."""
# deploy_listeners_list sets the module-level _no_input global as a
# side effect; pin its "before" value so monkeypatch restores it after
# this test instead of leaking True into whichever test runs next.
monkeypatch.setattr(deploy_module, "_no_input", False)
runner = CliRunner()
class FakeClient:
def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
pass
def list_listeners(self):
raise HostBackendError("requires workspace specification", status_code=403)
def set_tenant(self, tenant_id: str) -> None:
raise AssertionError("must not retry with a prompted tenant with --json")
monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
prompt_calls = []
monkeypatch.setattr(
click, "prompt", lambda *a, **k: prompt_calls.append((a, k)) or "unreachable"
)
result = runner.invoke(
cli,
[
"deploy",
"listeners",
"list",
"--json",
"--api-key",
"test-key",
"--host-url",
"https://api.example.com",
],
)
assert prompt_calls == [], "must not prompt interactively with --json"
assert result.exit_code != 0
assert "LANGSMITH_TENANT_ID" in result.output
def test_deploy_revisions_list_command(monkeypatch) -> None:
runner = CliRunner()
captured: dict[str, str] = {}