From edf73e97e95ba8d0a637785a32406f217cdede58 Mon Sep 17 00:00:00 2001 From: Constantine Nathanson Date: Wed, 19 Aug 2026 20:16:24 +0300 Subject: [PATCH 1/2] Add support for `agent cloud create` and `agent cloud claim` commands Claimable Clouds are temporary clouds whose credentials work immediately, with media delivery restricted to an IP allow-list, disabled after ~24h unless a human claims them. `agent cloud create` provisions one and saves it as a named configuration along with the claim URL, which the server returns exactly once and cannot be looked up again. Delivery IPs are forwarded verbatim: the server owns the count, routability and CIDR rules, and rejects bad input with a 400 that does not consume creation quota. `agent cloud claim` opens (or prints) the stored claim URL. `--print` and `--json` never launch a browser, so headless and agent callers get the URL on stdout. Config metadata is carried as query params on the saved cloudinary:// URL: claim_url, expires_at, delivery_ips, account_id and cloud_email. `config -ls` gains a STATUS column and `config -s` a header flag, both showing an expiry countdown; JSON output carries raw timestamps plus claimable_cloud and expired flags. Extract parse_expiry, is_expired and expires_in_hint into utils, shared by the OAuth epoch expiry and the ISO-8601 cloud expiry. Document both commands in the README, and split Requirements into an agent path and a human path so Claimable Clouds are the documented way for an agent to get an account. Co-Authored-By: Claude Opus 5 --- README.md | 80 ++- cloudinary_cli/core/agent.py | 321 ++++++++++- cloudinary_cli/core/config.py | 16 +- cloudinary_cli/defaults.py | 15 +- cloudinary_cli/utils/api_utils.py | 2 +- cloudinary_cli/utils/config_listing.py | 40 +- cloudinary_cli/utils/config_utils.py | 60 +- cloudinary_cli/utils/utils.py | 44 ++ requirements.txt | 2 +- test/test_cli_agent_cloud.py | 740 +++++++++++++++++++++++++ test/test_utils.py | 55 +- 11 files changed, 1347 insertions(+), 28 deletions(-) create mode 100644 test/test_cli_agent_cloud.py diff --git a/README.md b/README.md index 8f3205c..48d5ef2 100644 --- a/README.md +++ b/README.md @@ -9,11 +9,24 @@ The Cloudinary CLI (Command Line Interface) enables you to interact with Cloudin It is fully documented at [https://cloudinary.com/documentation/cloudinary_cli](https://cloudinary.com/documentation/cloudinary_cli). +**Built for AI agents too.** [`cld agent cloud create`](#agent-cloud) provisions a working Cloudinary cloud in a single command — no signup, no verification email, no credentials to ask a human for — so an agent can start uploading and transforming right away, then hand the human a claim URL to keep it. + ## Requirements -Your own Cloudinary account. If you don't already have one, sign up at [https://cloudinary.com/users/register/free](https://cloudinary.com/users/register/free). Python 3.8 or later. You can install Python from [https://www.python.org/](https://www.python.org/). Note that the Python Package Installer (pip) is installed with it. +A Cloudinary account. There are two ways to get one: + +**Are you an AI agent?** Run [`cld agent cloud create`](#agent-cloud) — no signup, no verification email, no human in the loop. You get a working cloud name, API key and secret in one command, and can upload and transform immediately: + +``` +cld agent cloud create +``` + +The cloud is temporary: media delivery is limited to the IP that created it, and it expires in about 24 hours unless a human claims it. Hand them the claim URL from the output — or run [`cld agent cloud claim`](#agent-cloud) — and it becomes a permanent account, keeping the same credentials and everything already uploaded. + +**Are you a human?** Sign up at [https://cloudinary.com/users/register/free](https://cloudinary.com/users/register/free), then see [Configuration](#configuration) below. If you'd rather an agent set it up for you, [`cld agent signup`](#agent-signup) creates a Free-plan account in your name and emails you to verify it. + ## Installation The CLI is published on PyPI as [`cloudinary-cli`](https://pypi.org/project/cloudinary-cli/). The package name (`cloudinary-cli`) is what you install; the command it provides is **`cld`** (it also installs a `cloudinary` alias). Pick the method that fits your setup. If you just want a working `cld` command and aren't sure, use **pipx** or **uv** — they install the CLI in its own isolated environment, so it won't conflict with other Python packages and you don't need to manage a virtual environment yourself. @@ -143,6 +156,7 @@ cld search --help # Shows usage for the Search API. cld admin # Lists Admin API methods. cld uploader # Lists Upload API methods. cld agent signup # For AI agents: creates a Cloudinary account on behalf of a human. +cld agent cloud # For AI agents: creates a temporary cloud that works immediately. ``` ## Docker Usage @@ -338,6 +352,70 @@ Options: * `--sdk-framework ` — the Cloudinary SDK framework the agent intends to use. * `--json` — output the full raw JSON response (the agent contract) instead of the human-readable summary. +### `agent cloud` + +**For AI agents acting on behalf of a human.** Creates and claims *Claimable Clouds*: temporary clouds whose credentials work immediately, with no signup and no verification email. No existing configuration is required. + +Media delivery is restricted to an IP allow-list, and the cloud is **disabled after about 24 hours** — along with everything uploaded to it — unless a human claims it. Claiming makes it permanent, keeps the credentials and assets, and lifts the IP restriction. + +#### `agent cloud create` + +``` +cld agent cloud create [command options] [email] +``` + +Example: + +``` +cld agent cloud create you@example.com --claim +``` + +The cloud is saved as a named configuration along with its claim URL, which **the server returns exactly once and cannot be looked up again**. A cloud saved without it — or created outside the CLI — cannot be claimed via `agent cloud claim`. + +The optional `email` only pre-fills the claim page. It is never verified and no mail is sent to it at creation, but it must be a real unused address: the server rejects addresses already taken and disposable domains. + +Options: + +* `--ip
` — an additional IP permitted to deliver media, repeatable. Omit to let the server use the address the request comes from; the literal `requester_ip` means that same address. +* `--name ` — name for the saved configuration (default: the returned cloud name). +* `--set-default` — set the saved configuration as the default. +* `--no-save` — show the credentials but do not save them as a configuration. The claim URL is then only in the output, so store it yourself. +* `--claim` — open the claim page as soon as the cloud is created. +* `--agent-framework `, `--agent-llm-model `, `--agent-goal `, `--sdk-framework ` — attribution for the agent creating the cloud. +* `--json` — output the full raw JSON response (the agent contract) instead of the human-readable summary. + +Delivery IPs are sent to the server as given; it validates them and returns the allow-list it actually stored. Read `delivery_ips` back from the response rather than assuming the list you sent was kept — behind a proxy or VPN your own address may be dropped, and the server refuses to create a cloud with no publicly routable address in the list. + +Because the allow-list covers media delivery only, the Upload and Admin APIs are unaffected. Uploads succeeding while a delivery URL returns `x-cld-error: ACL deny` is the expected symptom of the restriction, not a broken cloud or bad credentials. It is therefore not a confidentiality control. + +#### `agent cloud claim` + +``` +cld agent cloud claim [command options] [name] +``` + +Example: + +``` +cld agent cloud claim mycloud --print +``` + +Opens the claim page for a saved Claimable Cloud. Run without a name to pick from the saved clouds that have not expired. + +Claiming is a **human action completed in a browser**: they enter an email address there, then click the link sent to it. This command only opens or prints the page — it cannot claim anything itself, and nothing reports whether a claim succeeded. + +Options: + +* `--print` (or `--no-open`) — print the claim URL instead of opening a browser. Required for headless and agent use; also implied by `--json`. The CLI prints rather than opens whenever stdout is not a terminal. +* `--json` — output `{cloud_name, claim_url, expires_at}` as JSON. + +Saved Claimable Clouds are flagged in `cld config -ls` and `cld config -s ` with an expiry countdown, so you can see what is still claimable: + +``` +NAME CLOUD TYPE DEFAULT ACTIVE STATUS +mycloud abcd1234 api_key unclaimed, expires in 8h +``` + ## Additional configurations A configuration is a reference to a specified Cloudinary account or cloud name via its environment variable. You set the default configuration during setup and installation. Using different configurations allows you to access different Cloudinary cloud names, such as sub-accounts of your main Cloudinary account, or any additional Cloudinary accounts you may have. diff --git a/cloudinary_cli/core/agent.py b/cloudinary_cli/core/agent.py index d7dce16..de5b840 100644 --- a/cloudinary_cli/core/agent.py +++ b/cloudinary_cli/core/agent.py @@ -1,18 +1,32 @@ import json import re - +import sys import cloudinary.provisioning -from click import group, argument, option, echo, style, BadParameter, ClickException +from click import group, argument, option, echo, style, launch, BadParameter, ClickException from cloudinary.exceptions import Error as CloudinaryError, RateLimited -from cloudinary_cli.defaults import logger, ACCOUNT_EMAIL_PARAM +from cloudinary_cli.defaults import ( + logger, + ACCOUNT_EMAIL_PARAM, + CLOUD_CLAIM_URL_PARAM, + CLOUD_EXPIRES_AT_PARAM, + CLOUD_DELIVERY_IPS_PARAM, + CLOUD_ACCOUNT_ID_PARAM, + CLOUD_EMAIL_PARAM, + CLOUD_PLACEHOLDER_EMAIL_DOMAIN, +) from cloudinary_cli.utils.api_utils import call_api from cloudinary_cli.utils.json_utils import print_json +from cloudinary_cli.utils.utils import is_expired, prompt_user from cloudinary_cli.utils.config_utils import ( save_named_config, is_reserved_config_name, config_name_for_email, build_config_url, + claim_url_from_url, + claimable_cloud_names, + expires_at_from_url, + load_config, user_config_names, config_optional, ) @@ -72,11 +86,17 @@ def signup(email, agent_framework, agent_llm_model, agent_goal, sdk_framework, n _print_signup_summary(result) if not no_save: - save_agent_config(result, email, name=name, set_default=set_default) + save_agent_config(_signup_environment(result), name=name, set_default=set_default, + params=_account_email_params(email)) logger.info("Note: the account's credentials are inert until the emailed verification is completed.") +def _signup_environment(result): + """The credentials block of a signup response, nested under product_environments[].""" + return (result.get("product_environments") or [{}])[0] + + def _already_have_config_message(email, name): return (f"You already signed up with {email} (saved as '{name}'). " f"Use it with `cld -C {name} `. " @@ -131,7 +151,7 @@ def _parse_error_detail(text): def _print_signup_summary(result): - environment = (result.get("product_environments") or [{}])[0] + environment = _signup_environment(result) rows = [ ("Email", result.get("email", "")), ("Plan", result.get("plan_name", "")), @@ -166,14 +186,17 @@ def _extra_rows(data, known_keys): return rows -def save_agent_config(result, email, name=None, set_default=False): - environment = (result.get("product_environments") or [{}])[0] - config_name = name or environment.get("cloud_name") - stored_url = _config_url_from_environment(environment, email=email) +def save_agent_config(credentials, name=None, set_default=False, params=None): + """Persist a flat credentials dict ({cloud_name, api_key, api_secret}) as a named configuration, + optionally carrying CLI-local metadata query params. On any failure it warns with the exact + `cld config -n …` command to run by hand rather than raising. Returns the saved config name, or + None when nothing was saved.""" + config_name = name or credentials.get("cloud_name") + stored_url = _config_url_from_environment(credentials, params=params) if not stored_url or not config_name: logger.warning("Could not save the configuration automatically (missing credentials in the response). " "Add it manually with `cld config -n `.") - return + return None if name and name in user_config_names(): logger.warning(f"Overwriting existing config '{name}'.") @@ -182,22 +205,284 @@ def save_agent_config(result, email, name=None, set_default=False): default_status = save_named_config(config_name, stored_url, set_default=set_default) except Exception as e: logger.warning(f"Could not save the configuration '{config_name}': {e}. " - f"Add it manually with `cld config -n {config_name} {_config_url_from_environment(environment)}`.") - return + f"Add it manually with `cld config -n {config_name} {_config_url_from_environment(credentials, params=params)}`.") + return None logger.info(f"Config '{config_name}' saved!") logger.info(f"Example usage: cld -C {config_name} ") if default_status == "made": logger.info(f"Default set to '{config_name}'. Run `cld ` to use it, " f"or `cld -C {config_name} ` to select it explicitly.") + return config_name -def _config_url_from_environment(environment, email=None): - """Build a validated cloudinary:// config URL from a product-environment's credential fields, - optionally carrying the account email. Returns "" when the response lacks the credentials.""" - params = {ACCOUNT_EMAIL_PARAM: email.strip().lower()} if email and email.strip() else None +def _config_url_from_environment(credentials, params=None): + """Build a validated cloudinary:// config URL from a flat credentials dict, optionally carrying + CLI-local metadata query params. Returns "" when the response lacks the credentials.""" try: - return build_config_url(environment["cloud_name"], params=params, - api_key=environment["api_key"], api_secret=environment["api_secret"]) + return build_config_url(credentials["cloud_name"], params=params or None, + api_key=credentials["api_key"], api_secret=credentials["api_secret"]) except (KeyError, ValueError): return "" + + +def _account_email_params(email): + """The saved-URL params recording a verified signup email.""" + return {ACCOUNT_EMAIL_PARAM: email.strip().lower()} if email and email.strip() else {} + + +@agent_group.group("cloud", help="Create and claim Claimable Clouds (for AI agents acting on behalf of a human).") +def cloud_group(): + pass + + +@cloud_group.command("create", + short_help="Provision a Claimable Cloud with immediately working credentials.", + help="""\b +Create a Claimable Cloud: a temporary cloud whose credentials work immediately, with no signup and +no verification email. Media delivery is locked to an IP allow-list (by default the address this +request comes from), and the cloud is disabled in about 24 hours unless a human claims it via +`cld agent cloud claim`, which makes it permanent and lifts the IP lock. +The cloud is saved as a named configuration (use --no-save to skip), along with the claim URL, +which the server returns exactly once and cannot be looked up again. +Note the IP allow-list restricts media delivery only - the Upload and Admin APIs authenticate by +signature and ignore it, so it is not a confidentiality control. +The optional EMAIL only pre-fills the claim page: it is never verified, no mail is sent to it at +creation, and it must be a real unused address - the server rejects taken and disposable domains. +Format: cld agent cloud create [EMAIL] +\te.g. cld agent cloud create you@example.com --claim +""") +@option("--ip", "ips", multiple=True, + help="Additional IP permitted to deliver media, repeatable. Omit to let the server detect this " + "request's own address. The literal 'requester_ip' means that address.") +@argument("email", required=False) +@option("--agent-framework", "agent_framework", help="The agent framework creating the cloud.") +@option("--agent-llm-model", "agent_llm_model", help="The LLM model powering the agent.") +@option("--agent-goal", "agent_goal", help="A short description of what the agent is trying to achieve.") +@option("--sdk-framework", "sdk_framework", help="The Cloudinary SDK framework the agent intends to use.") +@option("--name", help="Name for the saved configuration (default: the returned cloud name).") +@option("--set-default", "set_default", is_flag=True, help="Set the saved configuration as the default.") +@option("--no-save", "no_save", is_flag=True, help="Do not save the returned credentials as a configuration.") +@option("--claim", "claim_now", is_flag=True, help="Open the claim page as soon as the cloud is created.") +@option("--json", "as_json", is_flag=True, + help="Output the full raw JSON response (agent contract) instead of the human summary.") +def cloud_create(ips, email, agent_framework, agent_llm_model, agent_goal, sdk_framework, name, set_default, + no_save, claim_now, as_json): + ips = list(ips) + if name and is_reserved_config_name(name): + raise BadParameter(f"'{name}' is a reserved configuration name.") + + if not no_save: + _refuse_config_collision(name) + + try: + result = call_api(cloudinary.provisioning.create_cloud, delivery_ips=ips or None, email=email, + agent_framework=agent_framework, agent_llm_model=agent_llm_model, + agent_goal=agent_goal, sdk_framework=sdk_framework) + except RateLimited as e: + raise ClickException(f"Rate limited while creating the cloud: {e}. This endpoint is limited per IP " + f"address (about 10 clouds per 24 hours); wait a bit and try again.") + except CloudinaryError as e: + raise ClickException(_cloud_error_message(e)) + + if as_json: + print_json(result) + else: + _print_cloud_summary(result) + + saved_name = None + if no_save: + logger.warning("Nothing was saved (--no-save), so `cld agent cloud claim` cannot open the claim page " + "later. The claim URL above is the only way to keep this cloud - store it now.") + else: + saved_name = save_agent_config(result, name=name, set_default=set_default, params=_cloud_params(result)) + + if claim_now: + _open_claim_url(result.get("claim_url"), result.get("cloud_name"), print_only=False) + elif saved_name: + logger.info(f"Claim it with `cld agent cloud claim {saved_name}` - a human must finish in a browser.") + + +@cloud_group.command("claim", + short_help="Open the claim page for a saved Claimable Cloud.", + help="""\b +Open the claim page of a Claimable Cloud saved by `cld agent cloud create`, using the claim URL +stored with the configuration - the server returns it exactly once and there is no way to look it +up again. Run without a name to choose from the saved Claimable Clouds. +Claiming is a human action completed in a browser: they enter an email address, then click a link +sent to it. This command only opens (or prints) the page - it cannot claim anything itself, and +nothing reports whether a claim succeeded. +\te.g. cld agent cloud claim mycloud --print +""") +@argument("name", required=False) +@option("--print", "--no-open", "print_only", is_flag=True, + help="Print the claim URL instead of opening a browser. Required for headless and agent use.") +@option("--json", "as_json", is_flag=True, help="Output {cloud_name, claim_url, expires_at} as JSON.") +def cloud_claim(name, print_only, as_json): + if not name: + name = _select_claimable_cloud() + if name is None: + return + + config_url = _saved_config_url(name) + claim_url = claim_url_from_url(config_url) + if not claim_url: + raise ClickException( + f"No claim URL is stored with the configuration '{name}'. It is returned only when the cloud is " + f"created and cannot be re-fetched, so a cloud saved before this feature - or created outside " + f"the CLI - cannot be claimed from here. Create a new one with `cld agent cloud create`.") + + expires_at = expires_at_from_url(config_url) + if as_json: + print_json({"cloud_name": _cloud_name_for_config(config_url, name), + "claim_url": claim_url, "expires_at": expires_at}) + _warn_if_expired(expires_at) + # --json is the machine-readable contract: it already carries the claim URL, so never also + # spawn a browser on the caller's behalf. + _open_claim_url(claim_url, name, print_only=print_only or as_json, quiet=as_json) + + +def _refuse_config_collision(name): + """Refuse a create that would overwrite an existing config, before any network call.""" + if name and name in user_config_names(): + raise ClickException(f"A configuration named '{name}' already exists. Choose another name with --name, " + f"or use --no-save to print the credentials without saving them.") + + +def _cloud_params(result): + """The CLI-local metadata stored on a Claimable Cloud's saved config URL.""" + params = {key: result[key] for key in (CLOUD_CLAIM_URL_PARAM, CLOUD_EXPIRES_AT_PARAM, + CLOUD_ACCOUNT_ID_PARAM) if result.get(key)} + + delivery_ips = result.get("delivery_ips") + if delivery_ips: + params[CLOUD_DELIVERY_IPS_PARAM] = ",".join(delivery_ips) + + email = (result.get("email") or "").strip() + if email and not email.endswith(CLOUD_PLACEHOLDER_EMAIL_DOMAIN): + params[CLOUD_EMAIL_PARAM] = email + + return params + + +# Response keys the Claimable Cloud summary renders explicitly (or omits, e.g. secrets folded into +# CLOUDINARY_URL). Anything else is surfaced generically by _extra_rows. +_KNOWN_CLOUD_KEYS = {"cloud_name", "api_key", "api_secret", "api_environment_variable", "delivery_ips", + "expires_at", "claim_url", "guidance"} + + +def _print_cloud_summary(result): + rows = [ + ("Cloud name", result.get("cloud_name", "")), + ("API key", result.get("api_key", "")), + ("CLOUDINARY_URL", _config_url_from_environment(result)), + ("Delivery IPs", ", ".join(result.get("delivery_ips") or [])), + ] + rows += _extra_rows(result, _KNOWN_CLOUD_KEYS) + rows = [(label, value) for label, value in rows if value] + + echo(style("Claimable Cloud created. The credentials below work immediately.", fg="green")) + if rows: + width = max(len(label) for label, _ in rows) + 1 + template = "{0:" + str(width) + "} {1}" + echo("\n".join(template.format(f"{label}:", value) for label, value in rows)) + + expires_at = result.get("expires_at") + claim_url = result.get("claim_url") + if expires_at: + echo(style(f"\nExpires at: {expires_at} (about 24h) - unclaimed clouds are disabled and their assets " + f"removed.", fg="yellow")) + if claim_url: + echo(style(f"Claim URL: {claim_url}", fg="yellow")) + echo("Opening it is the only way to keep this cloud. It is returned once and cannot be looked up " + "again. A human must enter an email there and then click the link sent to it.") + + guidance = result.get("guidance") + if guidance: + echo(f"\n{guidance}") + + +def _cloud_error_message(error): + """Actionable guidance for a create_cloud failure, recognised by the server's message text.""" + text = str(error) + lowered = text.lower() + + if "rate limit" in lowered: + return (f"Rate limited while creating the cloud: {text}. This endpoint is limited per IP address " + f"(about 10 clouds per 24 hours); wait a bit and try again.") + if "registration is currently unavailable" in lowered or "registration_disabled" in lowered: + return f"Cloud provisioning is currently disabled by Cloudinary: {text}." + if "location not permitted" in lowered: + return (f"Could not create the cloud: {text}. This applies to the address the request comes from and " + f"to every --ip supplied, so a delivery IP in a restricted country fails the whole call.") + + detail = _parse_error_detail(text) + return f"Could not create the cloud: {detail}." if detail else f"Could not create the cloud: {text}." + + +def _saved_config_url(name): + """The URL of the named saved config.""" + cfg = load_config() + if name not in user_config_names(cfg): + raise ClickException(f"Config {name} does not exist") + return cfg[name] + + +def _select_claimable_cloud(): + """Pick a saved Claimable Cloud to claim when no name was given, or None when there is nothing to + do.""" + names = claimable_cloud_names(include_expired=False) + if not names: + if claimable_cloud_names(): + logger.info("Every saved Claimable Cloud has expired, so there is nothing left to claim. " + "Create one with `cld agent cloud create`, or name one explicitly to open its " + "claim page anyway.") + else: + logger.info("No saved Claimable Clouds to claim. Create one with `cld agent cloud create`.") + return None + if len(names) == 1: + return names[0] + + echo("Saved Claimable Clouds:") + for i, name in enumerate(names, start=1): + echo(f" {i}) {name}") + + choice = prompt_user(f"Select a cloud to claim [1-{len(names)}] (or Enter to cancel): ", + noninteractive_hint="Pass the configuration name directly: `cld agent cloud claim `.") + if choice is None: + raise ClickException("No selection was made.") + choice = choice.strip() + if not choice: + return None + if not (choice.isdigit() and 1 <= int(choice) <= len(names)): + raise ClickException(f"Invalid selection '{choice}'. Expected a number between 1 and {len(names)}.") + return names[int(choice) - 1] + + +def _cloud_name_for_config(config_url, name): + from cloudinary_cli.utils.config_utils import cloud_name_from_url + return cloud_name_from_url(config_url) or name + + +def _warn_if_expired(expires_at): + """Warn when the stored expiry has passed, without refusing.""" + if is_expired(expires_at): + logger.warning(f"This cloud expired at {expires_at}, so it has most likely been disabled and its assets " + f"removed. Opening the claim page anyway; create a new cloud with " + f"`cld agent cloud create` if it no longer works.") + + +def _open_claim_url(claim_url, name, print_only, quiet=False): + """Open the claim page, or print the URL when asked to or when not attached to a TTY.""" + if not claim_url: + raise ClickException(f"No claim URL is available for '{name}'.") + + if print_only or not sys.stdout.isatty(): + if not quiet: + echo(claim_url) + return + + launch(claim_url) + logger.info(f"Opened the claim page for '{name}' in your browser. Claiming is not complete yet: a human " + f"must enter an email address there, then click the link sent to it.") + logger.info(f"If the browser did not open, visit this URL to claim the cloud:\n{claim_url}") diff --git a/cloudinary_cli/core/config.py b/cloudinary_cli/core/config.py index a054859..abf0a60 100644 --- a/cloudinary_cli/core/config.py +++ b/cloudinary_cli/core/config.py @@ -15,8 +15,11 @@ is_reserved_config_name, config_type, config_optional, + claim_url_from_url, + expires_at_from_url, + cloud_expiry_status, ) -from cloudinary_cli.utils.utils import ConfigurationError +from cloudinary_cli.utils.utils import ConfigurationError, is_expired from cloudinary_cli.utils.json_utils import print_json from cloudinary_cli.utils.config_resolver import active_config_name, active_config_is_url from cloudinary_cli.auth import refresh_config, refresh_configs, relogin_command @@ -180,8 +183,19 @@ def _show_config_header(name, cfg): flags.append("default") if active_config_name() == name: flags.append("active") + status = cloud_expiry_status(expires_at_from_url(cfg[name])) if claim_url_from_url(cfg[name]) else None + if status: + flags.append(status) suffix = f" [{', '.join(flags)}]" if flags else "" echo(f"name: {name} ({config_type(cfg[name])}){suffix}\n") + if not status: + return + if is_expired(expires_at_from_url(cfg[name])): + echo("Claimable Cloud, past its expiry: it has most likely been disabled, along with everything " + "uploaded to it.\nCreate another with `cld agent cloud create`.\n") + else: + echo("Claimable Cloud: disabled at expiry, along with everything uploaded to it.\n" + f"Claim with `cld agent cloud claim {name}` - completed in a browser.\n") def _show_active_header(): diff --git a/cloudinary_cli/defaults.py b/cloudinary_cli/defaults.py index 34af062..c529d85 100644 --- a/cloudinary_cli/defaults.py +++ b/cloudinary_cli/defaults.py @@ -29,9 +29,22 @@ DEFAULT_CONFIG_KEY = "__default__" # Query param carried inside a saved cloudinary:// URL recording the email the account was created -# for (via `cld agent signup`). Stripped before display and before reaching the SDK. +# for (via `cld agent signup`). The SDK absorbs unknown query params as config attributes, so this +# surfaces in `cld config -s` alongside the real config fields. ACCOUNT_EMAIL_PARAM = "account_email" +# Query params carried inside a saved cloudinary:// URL for a Claimable Cloud (`cld agent cloud +# create`). Like ACCOUNT_EMAIL_PARAM these reach the SDK config and are shown by `cld config -s`; +# the claim URL is displayed unmasked on purpose, since handing it to a human is the point. +CLOUD_CLAIM_URL_PARAM = "claim_url" +CLOUD_EXPIRES_AT_PARAM = "expires_at" +CLOUD_DELIVERY_IPS_PARAM = "delivery_ips" +CLOUD_ACCOUNT_ID_PARAM = "account_id" +# The claim-page email, stored only when the caller supplied a real one. +CLOUD_EMAIL_PARAM = "cloud_email" +# Domain of the synthetic placeholder address the server generates when no email is supplied. +CLOUD_PLACEHOLDER_EMAIL_DOMAIN = "@cloud.cloudinary.invalid" + # Guidance shown when no configuration is available (the group callback for account-consuming # commands, and the empty `config -ls`). Printed verbatim to stderr, without the logger's # "warning:" prefix, so the copy-pasteable command lines stay clean. diff --git a/cloudinary_cli/utils/api_utils.py b/cloudinary_cli/utils/api_utils.py index 0ff9c03..b095013 100644 --- a/cloudinary_cli/utils/api_utils.py +++ b/cloudinary_cli/utils/api_utils.py @@ -37,7 +37,7 @@ # Public, unauthenticated API methods that must run without a Cloudinary configuration. PUBLIC_API_METHODS = { - "provisioning": {"create_agent_account"}, + "provisioning": {"create_agent_account", "create_cloud"}, } diff --git a/cloudinary_cli/utils/config_listing.py b/cloudinary_cli/utils/config_listing.py index d12a59b..d8282cb 100644 --- a/cloudinary_cli/utils/config_listing.py +++ b/cloudinary_cli/utils/config_listing.py @@ -13,7 +13,11 @@ cloudinary_config_details, is_env_configured, email_from_url, + claim_url_from_url, + expires_at_from_url, + cloud_expiry_status, ) +from cloudinary_cli.utils.utils import is_expired from cloudinary_cli.utils.config_resolver import ( active_config_name, active_config_is_env, @@ -32,8 +36,10 @@ def config_type_label(config_obj): _TABLE_COLUMNS = [("name", "NAME"), ("cloud_name", "CLOUD"), ("type", "TYPE"), ("default", "DEFAULT"), ("active", "ACTIVE")] -# EMAIL is appended dynamically (see render_config_table) only when at least one row carries one. +# EMAIL and STATUS are appended dynamically (see render_config_table) only when at least one row +# carries one. _EMAIL_COLUMN = ("email", "EMAIL") +_STATUS_COLUMN = ("status", "STATUS") def list_configs(): @@ -60,6 +66,11 @@ def list_configs(): email = email_from_url(cfg[name]) if email: # only surfaced when the config records an account email (e.g. from `agent signup`) row["email"] = email + if claim_url_from_url(cfg[name]): # a Claimable Cloud saved by `agent cloud create` + expires_at = expires_at_from_url(cfg[name]) + row["status"] = cloud_expiry_status(expires_at) + row["expires_at"] = expires_at + row.update(_claimable_cloud_meta(cfg[name])) rows.append(row) return rows @@ -72,10 +83,19 @@ def config_meta(name, cfg, config_obj): "type": config_type(cfg[name]), "default": cfg.get(DEFAULT_CONFIG_KEY) == name, "active": active_config_name() == name, + **_claimable_cloud_meta(cfg[name]), **cloudinary_config_details(config_obj), } +def _claimable_cloud_meta(url): + """Machine-readable Claimable Cloud facts for JSON consumers, or {} for an ordinary config. + `expired` is null when the stored expiry cannot be read.""" + if not claim_url_from_url(url): + return {} + return {"claimable_cloud": True, "expired": is_expired(expires_at_from_url(url))} + + def active_config_meta(config_obj): """JSON view of the active config for bare `cld config` (saved name, -c URL, or env).""" name = active_config_name() @@ -96,12 +116,28 @@ def render_config_table(rows): columns = list(_TABLE_COLUMNS) if any(row.get("email") for row in rows): # add EMAIL only when some config records one columns.append(_EMAIL_COLUMN) + if any(row.get("status") for row in rows): # add STATUS only when some config is a Claimable Cloud + columns.append(_STATUS_COLUMN) headers = [title for _, title in columns] cells = [[_cell(row, key) for key, _ in columns] for row in rows] widths = [max(len(headers[i]), *(len(r[i]) for r in cells)) if cells else len(headers[i]) for i in range(len(headers))] line = lambda values: " ".join(v.ljust(widths[i]) for i, v in enumerate(values)).rstrip() - return "\n".join([line(headers)] + [line(r) for r in cells]) + table = "\n".join([line(headers)] + [line(r) for r in cells]) + hint = _claim_hint(rows) + return f"{table}\n\n{hint}" if hint else table + + +def _claim_hint(rows): + """A footer telling the user how to keep an unclaimed cloud, shown only when a still-live one is + listed.""" + live = sum(1 for row in rows if row.get("status") and not is_expired(row.get("expires_at"))) + if not live: + return "" + noun, pronoun = ("Clouds", "them") if live > 1 else ("Cloud", "it") + return (f"{live} unclaimed Claimable {noun}: disabled at expiry, along with " + f"everything uploaded to {pronoun}.\n" + f"Claim with `cld agent cloud claim` - completed in a browser.") def _url_row(): diff --git a/cloudinary_cli/utils/config_utils.py b/cloudinary_cli/utils/config_utils.py index 24e9028..4ab5b02 100644 --- a/cloudinary_cli/utils/config_utils.py +++ b/cloudinary_cli/utils/config_utils.py @@ -14,10 +14,16 @@ OLD_CLOUDINARY_CLI_CONFIG_FILE, DEFAULT_CONFIG_KEY, ACCOUNT_EMAIL_PARAM, + CLOUD_CLAIM_URL_PARAM, + CLOUD_EXPIRES_AT_PARAM, + CLOUD_DELIVERY_IPS_PARAM, + CLOUD_ACCOUNT_ID_PARAM, + CLOUD_EMAIL_PARAM, logger, ) from cloudinary_cli.utils.json_utils import write_json_to_file, read_json_from_file from cloudinary_cli.utils.url_utils import set_url_params, url_param +from cloudinary_cli.utils.utils import expires_in_hint, is_expired def config_optional(cmd): """Mark a Click command/group as not requiring a resolved Cloudinary config, so the top-level @@ -230,6 +236,54 @@ def email_from_url(cloudinary_url): return _normalize_email(value) if value else None +def claim_url_from_url(cloudinary_url): + """The Claimable Cloud claim URL stored in a saved config URL, or None.""" + return url_param(cloudinary_url, CLOUD_CLAIM_URL_PARAM) or None + + +def expires_at_from_url(cloudinary_url): + """The Claimable Cloud expiry timestamp stored in a saved config URL, or None.""" + return url_param(cloudinary_url, CLOUD_EXPIRES_AT_PARAM) or None + + +def delivery_ips_from_url(cloudinary_url): + """The Claimable Cloud's stored delivery allow-list as a list, or [] when none is recorded.""" + value = url_param(cloudinary_url, CLOUD_DELIVERY_IPS_PARAM) + return [ip for ip in (value or "").split(",") if ip] if value else [] + + +def account_id_from_url(cloudinary_url): + """The Claimable Cloud's account id stored in a saved config URL, or None.""" + return url_param(cloudinary_url, CLOUD_ACCOUNT_ID_PARAM) or None + + +def cloud_email_from_url(cloudinary_url): + """The claim-page email stored for a Claimable Cloud, or None.""" + return url_param(cloudinary_url, CLOUD_EMAIL_PARAM) or None + + +def cloud_expiry_status(expires_at): + """A Claimable Cloud's status label for `config -ls`, e.g. "unclaimed, expires in 8h". Advisory + only: no endpoint reports claim state, so a claimed cloud still reads as unclaimed.""" + countdown = expires_in_hint(expires_at) + if not countdown: + return "unclaimed" + if countdown == "expired": + return "unclaimed, expired" + return f"unclaimed, {countdown}" + + +def claimable_cloud_names(include_expired=True): + """The names of all saved configs carrying a stored claim URL. With include_expired=False, drops + the ones whose stored expiry has passed; an unreadable expiry is kept, since unknown is not + expired.""" + cfg = load_config() + names = [name for name in user_config_names(cfg) if claim_url_from_url(cfg[name])] + if include_expired: + return names + return [name for name in names if not is_expired(expires_at_from_url(cfg[name]))] + + def config_name_for_email(email): """The saved config whose URL records this account email, or None. Scans only saved configs, so a removed config drops out automatically (the URL is gone with it). Returns the first match.""" @@ -320,7 +374,7 @@ def _expires_at_fields(value): return { "epoch": epoch, "utc": datetime.fromtimestamp(epoch, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC"), - "expired": epoch <= int(time.time()), + "expired": bool(is_expired(epoch)), } @@ -355,7 +409,9 @@ def _format_epoch(value): def _format_expires_at(value): parts = _expires_at_fields(value) if parts is None: - return value + # Not an epoch: a Claimable Cloud's ISO-8601 expiry. + countdown = expires_in_hint(value) + return f"{value} ({countdown})" if countdown else value state = "expired" if parts["expired"] else "valid" return f"{parts['epoch']} ({parts['utc']}, {state})" diff --git a/cloudinary_cli/utils/utils.py b/cloudinary_cli/utils/utils.py index a4bd68c..bf50d2a 100644 --- a/cloudinary_cli/utils/utils.py +++ b/cloudinary_cli/utils/utils.py @@ -87,6 +87,50 @@ def expiry_hint(epoch): return str(epoch) +def parse_expiry(value): + """Any expiry the CLI stores as an aware UTC datetime, or None if it cannot be read. Accepts an + ISO-8601 string ('Z' or offset), a Unix epoch as int or numeric string, and a datetime. A naive + value is read as UTC.""" + if value is None or isinstance(value, bool): + return None + if isinstance(value, datetime): + return value if value.tzinfo else value.replace(tzinfo=timezone.utc) + try: + return datetime.fromtimestamp(int(value), tz=timezone.utc) + except (TypeError, ValueError, OverflowError, OSError): + pass + try: + parsed = datetime.fromisoformat(str(value).strip().replace("Z", "+00:00")) + except ValueError: + return None + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + + +def is_expired(value, now=None): + """Whether an expiry has passed. None for an unreadable expiry.""" + expiry = parse_expiry(value) + if expiry is None: + return None + return expiry <= (now or datetime.now(timezone.utc)) + + +def expires_in_hint(value, now=None, expired_label="expired", unknown_label=""): + """An expiry as a short countdown label: "expires in 8h", "expires in 45m", or expired_label once + it has passed. Returns unknown_label when the value cannot be read. Values are rounded.""" + expiry = parse_expiry(value) + if expiry is None: + return unknown_label + + seconds = (expiry - (now or datetime.now(timezone.utc))).total_seconds() + if seconds <= 0: + return expired_label + if seconds < 3600: + return f"expires in {max(1, round(seconds / 60))}m" + if seconds <= 86400: + return f"expires in {round(seconds / 3600)}h" + return f"expires in {round(seconds / 86400)}d" + + def log_exception(e, message=None, debug_message=None): message = f"{message}, error: {str(e)}" if message is not None else str(e) debug_message = debug_message or message diff --git a/requirements.txt b/requirements.txt index efe7965..d012f9d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -cloudinary>=1.45.0 +cloudinary>=1.46.0 pygments jinja2 click diff --git a/test/test_cli_agent_cloud.py b/test/test_cli_agent_cloud.py new file mode 100644 index 0000000..a521269 --- /dev/null +++ b/test/test_cli_agent_cloud.py @@ -0,0 +1,740 @@ +import json +import os +import tempfile +import unittest +from unittest.mock import patch +from urllib.parse import quote + +from click.testing import CliRunner +from cloudinary.exceptions import BadRequest, NotAllowed, RateLimited +from filelock import FileLock + +from cloudinary_cli.cli import cli +from cloudinary_cli.utils import config_utils +from cloudinary_cli.utils.config_utils import claim_url_from_url, expires_at_from_url, validate_config_url + +# Shaped like a real `POST /v1_1/provisioning/clouds` response, with every value obviously fake: +# the claim_url keeps its own ?token= query (and the -._ characters in it) because that is what the +# saved-URL encoding has to survive, and the delivery IP is from the TEST-NET-3 documentation range. +FAKE_CLOUD = "fakecloud" +FAKE_API_KEY = "100000000000001" +FAKE_API_SECRET = "fake-api-secret_NOT-REAL-000" +FAKE_CLAIM_TOKEN = "fake-claim-token.NOT-REAL_000" +FAKE_DELIVERY_IP = "203.0.113.10" +FAKE_CLOUDINARY_URL = f"cloudinary://{FAKE_API_KEY}:{FAKE_API_SECRET}@{FAKE_CLOUD}" + +CLOUD_RESPONSE = { + "account_id": "00000000-0000-4000-8000-000000000001", + "email": "cloud-fake0000000000@cloud.cloudinary.invalid", + "cloud_name": FAKE_CLOUD, + "api_key": FAKE_API_KEY, + "api_secret": FAKE_API_SECRET, + "api_environment_variable": f"CLOUDINARY_URL={FAKE_CLOUDINARY_URL}", + "claimed": False, + "expires_at": "2026-08-19T16:13:09Z", + "delivery_ips": [FAKE_DELIVERY_IP], + "claim_url": "https://console.example.invalid/users/agent_email_confirmation" + f"?token={FAKE_CLAIM_TOKEN}", + "guidance": "A Claimable Cloud is ready and the API key and secret below work immediately.", +} + +CREATE_ARGS = ["agent", "cloud", "create"] + + +class CloudTestCase(unittest.TestCase): + """Each test runs against a temporary config file, so nothing touches the developer's real one.""" + + runner = CliRunner() + + def setUp(self): + self._dir = tempfile.TemporaryDirectory() + self.addCleanup(self._dir.cleanup) + self.config_file = os.path.join(self._dir.name, "config.json") + for ctx in (patch.object(config_utils, "CLOUDINARY_CLI_CONFIG_FILE", self.config_file), + patch.object(config_utils, "_config_lock", FileLock(self.config_file + ".lock")), + patch.object(config_utils, "_config_cache", None), + patch.object(config_utils, "_config_cache_stat", None)): + ctx.start() + self.addCleanup(ctx.stop) + + def write_config(self, config): + with open(self.config_file, "w") as f: + json.dump(config, f) + + def read_config(self): + with open(self.config_file) as f: + return json.load(f) + + +class TestCloudCreate(CloudTestCase): + def _invoke(self, extra_args=None, response=None, error=None): + with patch("cloudinary.provisioning.create_cloud", + return_value=response or CLOUD_RESPONSE, side_effect=error) as create: + create.__name__ = "create_cloud" # call_api reads func.__name__ when logging + result = self.runner.invoke(cli, CREATE_ARGS + (extra_args or [])) + return result, create + + def test_create_prints_credentials_and_saves_config(self): + result, create = self._invoke() + + self.assertEqual(0, result.exit_code) + self.assertIn("Claimable Cloud created.", result.output) + self.assertIn(FAKE_CLOUD, result.output) + self.assertIn(FAKE_CLOUDINARY_URL, result.output) + self.assertIn(FAKE_DELIVERY_IP, result.output) + self.assertIn(CLOUD_RESPONSE["claim_url"], result.output) + self.assertIn("2026-08-19T16:13:09Z", result.output) + self.assertIn(CLOUD_RESPONSE["guidance"], result.output) # server guidance passed through verbatim + self.assertIn(f"Config '{FAKE_CLOUD}' saved!", result.output) + + def test_create_without_ip_omits_delivery_ips_entirely(self): + result, create = self._invoke() + + self.assertEqual(0, result.exit_code) + self.assertIsNone(create.call_args.kwargs["delivery_ips"]) + + def test_create_forwards_ips_and_attribution(self): + result, create = self._invoke([ + "you@cloudinary.com", + "--ip", "8.8.8.8", "--ip", "requester_ip", + "--agent-framework", "claude-code", "--agent-llm-model", "claude-opus-5", + "--agent-goal", "test", "--sdk-framework", "python", + ]) + + self.assertEqual(0, result.exit_code) + self.assertEqual({ + "delivery_ips": ["8.8.8.8", "requester_ip"], + "email": "you@cloudinary.com", + "agent_framework": "claude-code", + "agent_llm_model": "claude-opus-5", + "agent_goal": "test", + "sdk_framework": "python", + }, create.call_args.kwargs) + + def test_create_forwards_ips_verbatim_without_client_side_validation(self): + # the server owns the delivery-IP rules (count, routability, CIDR); the CLI must not + # duplicate them and reject something the server would have accepted + result, create = self._invoke( + ["--ip", "1.1.1.1", "--ip", "2.2.2.2", "--ip", "3.3.3.3", "--ip", "10.0.0.0/8"]) + + self.assertEqual(0, result.exit_code) + self.assertEqual(["1.1.1.1", "2.2.2.2", "3.3.3.3", "10.0.0.0/8"], + create.call_args.kwargs["delivery_ips"]) + + def test_create_accepts_ipv6(self): + result, create = self._invoke(["--ip", "2001:4860:4860::8888"]) + + self.assertEqual(0, result.exit_code) + self.assertEqual(["2001:4860:4860::8888"], create.call_args.kwargs["delivery_ips"]) + + def test_create_reserved_name_rejected(self): + result, create = self._invoke(["--name", "__default__"]) + + self.assertEqual(2, result.exit_code) + self.assertIn("reserved configuration name", result.output) + create.assert_not_called() + + def test_create_name_collision_makes_no_api_call(self): + self.write_config({"taken": "cloudinary://k:s@somecloud"}) + result, create = self._invoke(["--name", "taken"]) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("already exists", result.output) + create.assert_not_called() # never burn a rate-limited cloud we would refuse to store + + def test_create_name_collision_allowed_with_no_save(self): + self.write_config({"taken": "cloudinary://k:s@somecloud"}) + result, create = self._invoke(["--name", "taken", "--no-save"]) + + self.assertEqual(0, result.exit_code) + create.assert_called_once() + + def test_create_persists_claim_url_and_expiry_on_saved_config(self): + result, _ = self._invoke() + + self.assertEqual(0, result.exit_code) + stored = self.read_config()[FAKE_CLOUD] + validate_config_url(stored) + self.assertEqual(CLOUD_RESPONSE["claim_url"], claim_url_from_url(stored)) + self.assertEqual(CLOUD_RESPONSE["expires_at"], expires_at_from_url(stored)) + + def test_create_persists_delivery_ips_and_account_id(self): + result, _ = self._invoke() + + self.assertEqual(0, result.exit_code) + stored = self.read_config()[FAKE_CLOUD] + from cloudinary_cli.utils.config_utils import delivery_ips_from_url, account_id_from_url + # the server rewrites the list it was sent, and there is no endpoint to read it back + self.assertEqual(CLOUD_RESPONSE["delivery_ips"], delivery_ips_from_url(stored)) + self.assertEqual(CLOUD_RESPONSE["account_id"], account_id_from_url(stored)) + + def test_persisted_delivery_ips_survive_ipv6(self): + response = dict(CLOUD_RESPONSE, delivery_ips=["8.8.8.8", "2001:4860:4860::8888"]) + result, _ = self._invoke(response=response) + + from cloudinary_cli.utils.config_utils import delivery_ips_from_url + self.assertEqual(response["delivery_ips"], delivery_ips_from_url(self.read_config()[FAKE_CLOUD])) + + def test_create_stores_a_real_email_but_not_the_placeholder(self): + from cloudinary_cli.utils.config_utils import cloud_email_from_url + + result, _ = self._invoke() # response carries the synthetic @cloud.cloudinary.invalid + self.assertIsNone(cloud_email_from_url(self.read_config()[FAKE_CLOUD])) + + self.setUp() # fresh config file + result, _ = self._invoke(response=dict(CLOUD_RESPONSE, email="you@cloudinary.com")) + self.assertEqual("you@cloudinary.com", cloud_email_from_url(self.read_config()[FAKE_CLOUD])) + + def test_stored_cloud_email_is_not_the_signup_account_email(self): + # a real claim-page email must still not make config_name_for_email match this cloud + self._invoke(response=dict(CLOUD_RESPONSE, email="you@cloudinary.com")) + + stored = self.read_config()[FAKE_CLOUD] + self.assertNotIn("account_email", stored) + self.assertIsNone(config_utils.email_from_url(stored)) + self.assertIsNone(config_utils.config_name_for_email("you@cloudinary.com")) + + def test_noisy_response_fields_are_not_persisted(self): + result, _ = self._invoke() + + stored = self.read_config()[FAKE_CLOUD] + self.assertNotIn("guidance", stored) # paragraphs of prose + self.assertNotIn("claimed", stored) # always false, never updated + self.assertNotIn("api_environment_variable", stored) # duplicates the URL itself + + def test_create_does_not_write_account_email_param(self): + result, _ = self._invoke(["you@cloudinary.com"]) + + self.assertEqual(0, result.exit_code) + stored = self.read_config()[FAKE_CLOUD] + self.assertNotIn("account_email", stored) + # so a Claimable Cloud never masquerades as a signup account + self.assertIsNone(config_utils.config_name_for_email("you@cloudinary.com")) + + def test_create_no_save_persists_nothing_and_warns(self): + result, _ = self._invoke(["--no-save"]) + + self.assertEqual(0, result.exit_code) + self.assertFalse(os.path.exists(self.config_file)) + self.assertIn("Nothing was saved", result.output) + self.assertIn("claim", result.output) + self.assertIn(CLOUD_RESPONSE["claim_url"], result.output) + + def test_create_set_default(self): + result, _ = self._invoke(["--set-default"]) + + self.assertEqual(0, result.exit_code) + self.assertEqual(FAKE_CLOUD, self.read_config()["__default__"]) + + def test_create_custom_name(self): + result, _ = self._invoke(["--name", "mycloud"]) + + self.assertEqual(0, result.exit_code) + self.assertIn("mycloud", self.read_config()) + + def test_create_json_emits_raw_response(self): + result, _ = self._invoke(["--json"]) + + self.assertEqual(0, result.exit_code) + payload = json.loads(result.output[:result.output.rindex("}") + 1]) + self.assertEqual(CLOUD_RESPONSE, payload) # unmodified + self.assertNotIn("CLOUDINARY_URL:", result.output) # no pretty labels + + def test_create_shows_credentials_even_when_save_fails(self): + with patch("cloudinary.provisioning.create_cloud", return_value=CLOUD_RESPONSE), \ + patch("cloudinary_cli.core.agent.save_named_config", side_effect=OSError("disk full")): + result = self.runner.invoke(cli, CREATE_ARGS) + + self.assertEqual(0, result.exit_code) + self.assertIn(FAKE_CLOUDINARY_URL, result.output) + self.assertIn("Could not save the configuration", result.output) + self.assertIn(f"cld config -n {FAKE_CLOUD}", result.output) # manual-add hint keeps the claim URL + self.assertIn("claim_url", result.output) + + def test_create_surfaces_unknown_future_keys(self): + response = dict(CLOUD_RESPONSE, region="us-east", nested={"ignored": True}) + result, _ = self._invoke(response=response) + + self.assertEqual(0, result.exit_code) + self.assertIn("Region:", result.output) + self.assertIn("us-east", result.output) + self.assertNotIn("ignored", result.output) + + def test_create_makes_no_outbound_ip_lookup(self): + # the server derives the allow-list from the request's own source address; the CLI must not + # probe an echo service to second-guess it + with patch("cloudinary_cli.core.agent.call_api", return_value=CLOUD_RESPONSE), \ + patch("requests.get", side_effect=AssertionError("no echo-service call")): + result = self.runner.invoke(cli, CREATE_ARGS) + + self.assertEqual(0, result.exit_code) + + +class TestCloudCreateErrors(CloudTestCase): + """Server messages captured verbatim from staging — the SDK surfaces the message only, not the + error envelope's `code`, so the guidance is keyed on these strings.""" + + def _invoke_failing(self, error, extra_args=None): + with patch("cloudinary.provisioning.create_cloud", side_effect=error) as create: + create.__name__ = "create_cloud" + return self.runner.invoke(cli, CREATE_ARGS + (extra_args or [])) + + def test_public_ip_rejection_surfaces_the_server_message(self): + # the server explains the rejection; the CLI passes it through rather than guessing + # whether the private address came from --ip or from the request's own source + result = self._invoke_failing( + BadRequest("Error 400 - delivery_ips must contain at least one public IP address")) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("delivery_ips must contain at least one public IP address", result.output) + + def test_too_many_ips_server_message_surfaced(self): + result = self._invoke_failing(BadRequest("Error 400 - At most 3 delivery_ips are allowed")) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("At most 3 delivery_ips are allowed", result.output) + + def test_invalid_ip_server_message_surfaced(self): + result = self._invoke_failing(BadRequest("Error 400 - Invalid delivery IP: not-an-ip")) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("Invalid delivery IP: not-an-ip", result.output) + + def test_rate_limited(self): + result = self._invoke_failing(RateLimited("Error 429 - Rate limit exceeded")) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("Rate limited", result.output) + self.assertIn("10 clouds per 24 hours", result.output) + + def test_registration_disabled(self): + result = self._invoke_failing( + NotAllowed("Error 403 - Agent account registration is currently unavailable")) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("currently disabled by Cloudinary", result.output) + + def test_geo_location_not_permitted_mentions_supplied_ips(self): + result = self._invoke_failing(NotAllowed("Error 403 - Location not permitted")) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("Location not permitted", result.output) + self.assertIn("every --ip supplied", result.output) + + +def _hours_from_now(hours): + from datetime import datetime, timedelta, timezone + return (datetime.now(timezone.utc) + timedelta(hours=hours)).strftime("%Y-%m-%dT%H:%M:%SZ") + + +class TestCloudClaim(CloudTestCase): + # Expiry is relative to now: the claim picker skips expired clouds, so a hardcoded timestamp + # would silently change what these tests exercise once it passed. + CLOUD_URL = (FAKE_CLOUDINARY_URL + + "?claim_url=" + quote(CLOUD_RESPONSE["claim_url"], safe="") + + "&expires_at=" + quote(_hours_from_now(8), safe="")) + + def _invoke(self, args, is_tty=True): + # CliRunner swaps sys.stdout for a capture buffer, so the TTY check is steered by patching + # isatty on whatever object is installed at call time, not on the original stream. + with patch("cloudinary_cli.core.agent.launch") as launch, \ + patch("cloudinary_cli.core.agent.sys") as agent_sys: + agent_sys.stdout.isatty.return_value = is_tty + result = self.runner.invoke(cli, ["agent", "cloud", "claim"] + args) + return result, launch + + def test_claim_opens_stored_url_on_a_tty(self): + self.write_config({"mycloud": self.CLOUD_URL}) + result, launch = self._invoke(["mycloud"]) + + self.assertEqual(0, result.exit_code) + launch.assert_called_once_with(CLOUD_RESPONSE["claim_url"]) + self.assertIn("Opened the claim page", result.output) + self.assertIn("must enter an email address", result.output) + # printed as well as opened: launch() cannot confirm a browser came up, and the URL is + # issued once and cannot be re-fetched + self.assertIn(CLOUD_RESPONSE["claim_url"], result.output) + self.assertNotIn("claimed", result.output.lower().replace("unclaimed", "")) + + def test_claim_print_flag_does_not_open_browser(self): + self.write_config({"mycloud": self.CLOUD_URL}) + result, launch = self._invoke(["mycloud", "--print"]) + + self.assertEqual(0, result.exit_code) + launch.assert_not_called() + # the bare URL and nothing else, so `--print` stays pipeable + self.assertEqual(CLOUD_RESPONSE["claim_url"], result.output.strip()) + + def test_claim_prints_instead_of_opening_when_not_a_tty(self): + self.write_config({"mycloud": self.CLOUD_URL}) + result, launch = self._invoke(["mycloud"], is_tty=False) + + self.assertEqual(0, result.exit_code) + launch.assert_not_called() # an agent must hand the URL to its human, not spawn a browser + self.assertIn(CLOUD_RESPONSE["claim_url"], result.output) + + def test_claim_selects_sole_saved_cloud_when_name_omitted(self): + self.write_config({"mycloud": self.CLOUD_URL, "__default__": "mycloud"}) + result, launch = self._invoke(["--print"]) + + self.assertEqual(0, result.exit_code) + self.assertIn(CLOUD_RESPONSE["claim_url"], result.output) + + def test_claim_ignores_default_config_that_is_not_a_claimable_cloud(self): + # The default is an ordinary account with no claim URL; the sole Claimable Cloud is chosen + # instead of failing on the default the way a plain default-config lookup would. + self.write_config({"ordinary": "cloudinary://k:s@somecloud", + "mycloud": self.CLOUD_URL, + "__default__": "ordinary"}) + result, _ = self._invoke(["--print"]) + + self.assertEqual(0, result.exit_code) + self.assertIn(CLOUD_RESPONSE["claim_url"], result.output) + + def test_claim_without_any_saved_cloud_reports_nothing_to_do(self): + self.write_config({"ordinary": "cloudinary://k:s@somecloud", "__default__": "ordinary"}) + result, launch = self._invoke([]) + + self.assertEqual(0, result.exit_code) # nothing to do is not an error + launch.assert_not_called() + self.assertIn("No saved Claimable Clouds to claim", result.output) + + def test_claim_lists_choices_when_several_are_saved(self): + second = self.CLOUD_URL.replace(FAKE_CLOUD, "second").replace("fake-claim-token", "other-token") + self.write_config({"a": self.CLOUD_URL, "b": second}) + with patch("cloudinary_cli.core.agent.prompt_user", return_value="2"): + result, launch = self._invoke([]) + + self.assertEqual(0, result.exit_code) + self.assertIn("1) a", result.output) + self.assertIn("2) b", result.output) + self.assertIn("other-token", launch.call_args.args[0]) # the selected one, not the first + + def test_claim_cancelled_selection_does_nothing(self): + second = self.CLOUD_URL.replace(FAKE_CLOUD, "second") + self.write_config({"a": self.CLOUD_URL, "b": second}) + with patch("cloudinary_cli.core.agent.prompt_user", return_value=""): + result, launch = self._invoke([]) + + self.assertEqual(0, result.exit_code) + launch.assert_not_called() + + def test_claim_non_interactive_selection_errors_with_hint(self): + second = self.CLOUD_URL.replace(FAKE_CLOUD, "second") + self.write_config({"a": self.CLOUD_URL, "b": second}) + with patch("cloudinary_cli.core.agent.prompt_user", return_value=None): + result, launch = self._invoke([]) + + self.assertNotEqual(0, result.exit_code) + launch.assert_not_called() + + def test_claim_json_output(self): + self.write_config({"mycloud": self.CLOUD_URL}) + result, _ = self._invoke(["mycloud", "--json"]) + + self.assertEqual(0, result.exit_code) + payload = json.loads(result.output[:result.output.rindex("}") + 1]) + self.assertEqual({"cloud_name": FAKE_CLOUD, + "claim_url": CLOUD_RESPONSE["claim_url"], + "expires_at": expires_at_from_url(self.CLOUD_URL)}, payload) + + def test_claim_json_does_not_open_a_browser_even_on_a_tty(self): + # --json is the machine-readable contract; the payload already carries the claim URL + self.write_config({"mycloud": self.CLOUD_URL}) + result, launch = self._invoke(["mycloud", "--json"], is_tty=True) + + self.assertEqual(0, result.exit_code) + launch.assert_not_called() + + def test_claim_without_stored_claim_url_explains_why(self): + self.write_config({"plain": "cloudinary://k:s@somecloud"}) + result, launch = self._invoke(["plain"]) + + self.assertNotEqual(0, result.exit_code) + launch.assert_not_called() + self.assertIn("No claim URL is stored", result.output) + self.assertIn("cannot be re-fetched", result.output) + + def test_claim_unknown_config(self): + self.write_config({"mycloud": self.CLOUD_URL}) + result, _ = self._invoke(["nosuch"]) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("does not exist", result.output) + + EXPIRED_URL = CLOUD_URL.split("&expires_at=")[0] + "&expires_at=2020-01-01T00%3A00%3A00Z" + + def test_claim_warns_but_still_prints_when_expired(self): + self.write_config({"old": self.EXPIRED_URL}) + result, _ = self._invoke(["old", "--print"]) + + self.assertEqual(0, result.exit_code) + self.assertIn("expired", result.output) + self.assertIn(CLOUD_RESPONSE["claim_url"], result.output) # printed anyway: local clock isn't authoritative + + def test_expired_cloud_is_not_offered_in_the_picker(self): + self.write_config({"live": self.CLOUD_URL, "old": self.EXPIRED_URL}) + result, launch = self._invoke(["--print"]) + + self.assertEqual(0, result.exit_code) + self.assertNotIn("old", result.output) # sole live cloud is chosen outright, no menu + launch.assert_not_called() + self.assertIn(CLOUD_RESPONSE["claim_url"], result.output) + + def test_picker_says_so_when_every_cloud_has_expired(self): + self.write_config({"old": self.EXPIRED_URL}) + result, launch = self._invoke([]) + + self.assertEqual(0, result.exit_code) # nothing claimable is not an error + launch.assert_not_called() + self.assertIn("has expired", result.output) + + +class TestConfigListingStatus(CloudTestCase): + """`config -ls` gains a STATUS column for saved Claimable Clouds, the same way EMAIL appears + only when some config records one.""" + + @staticmethod + def _cloud_url(cloud, expires_at): + from urllib.parse import quote + return (f"cloudinary://k:s@{cloud}?claim_url={quote('https://console.cloudinary.com/x?token=ABC', safe='')}" + f"&expires_at={quote(expires_at, safe='')}") + + @staticmethod + def _in_hours(hours): + from datetime import datetime, timedelta, timezone + return (datetime.now(timezone.utc) + timedelta(hours=hours)).strftime("%Y-%m-%dT%H:%M:%SZ") + + def test_status_column_absent_without_any_claimable_cloud(self): + self.write_config({"ordinary": "cloudinary://k:s@plaincloud"}) + result = self.runner.invoke(cli, ["config", "-ls"]) + + self.assertEqual(0, result.exit_code) + self.assertNotIn("STATUS", result.output) + + def test_status_column_shows_remaining_time(self): + self.write_config({"ordinary": "cloudinary://k:s@plaincloud", + "mycloud": self._cloud_url(FAKE_CLOUD, self._in_hours(8))}) + result = self.runner.invoke(cli, ["config", "-ls"]) + + self.assertEqual(0, result.exit_code) + self.assertIn("STATUS", result.output) + self.assertIn("unclaimed, expires in 8h", result.output) + + def test_status_column_marks_expired_cloud(self): + self.write_config({"oldcloud": self._cloud_url("oldcloud", "2020-01-01T00:00:00Z")}) + result = self.runner.invoke(cli, ["config", "-ls"]) + + self.assertEqual(0, result.exit_code) + self.assertIn("unclaimed, expired", result.output) + + def test_status_never_claims_a_cloud_was_claimed(self): + # Nothing reports claim status, so the column must never say "claimed". + self.write_config({"mycloud": self._cloud_url(FAKE_CLOUD, self._in_hours(8))}) + result = self.runner.invoke(cli, ["config", "-ls"]) + + self.assertNotIn("claimed", result.output.replace("unclaimed", "")) + + def test_claim_hint_shown_when_a_live_cloud_is_listed(self): + self.write_config({"ordinary": "cloudinary://k:s@plaincloud", + "mycloud": self._cloud_url(FAKE_CLOUD, self._in_hours(8))}) + result = self.runner.invoke(cli, ["config", "-ls"]) + + self.assertIn("1 unclaimed Claimable Cloud:", result.output) + self.assertIn("uploaded to it.", result.output) # singular count, singular pronoun + self.assertIn("Claim with `cld agent cloud claim`", result.output) + + def test_claim_hint_counts_only_live_clouds(self): + self.write_config({"live": self._cloud_url("livecloud", self._in_hours(8)), + "alsolive": self._cloud_url("second", self._in_hours(24)), + "dead": self._cloud_url("deadcloud", "2020-01-01T00:00:00Z")}) + result = self.runner.invoke(cli, ["config", "-ls"]) + + self.assertIn("2 unclaimed Claimable Clouds:", result.output) # the expired one is not counted + self.assertIn("uploaded to them.", result.output) + + def test_no_claim_hint_when_every_cloud_has_expired(self): + self.write_config({"dead": self._cloud_url("deadcloud", "2020-01-01T00:00:00Z")}) + result = self.runner.invoke(cli, ["config", "-ls"]) + + self.assertIn("unclaimed, expired", result.output) # still listed + self.assertNotIn("cld agent cloud claim", result.output) # but not advertised as claimable + + def test_no_claim_hint_without_any_cloud(self): + self.write_config({"ordinary": "cloudinary://k:s@plaincloud"}) + result = self.runner.invoke(cli, ["config", "-ls"]) + + self.assertNotIn("unclaimed", result.output) + self.assertNotIn("cld agent cloud claim", result.output) + + def test_cloud_with_unreadable_expiry_still_gets_the_hint(self): + # unknown expiry is not the same as expired, so it stays claimable + self.write_config({"noexp": "cloudinary://k:s@noexp?claim_url=https%3A%2F%2Fc.io%2Fx%3Ftoken%3DD"}) + result = self.runner.invoke(cli, ["config", "-ls"]) + + self.assertIn("Claim with `cld agent cloud claim`", result.output) + + def test_json_listing_has_no_hint_text(self): + self.write_config({"mycloud": self._cloud_url(FAKE_CLOUD, self._in_hours(8))}) + result = self.runner.invoke(cli, ["config", "-ls", "-j"]) + + json.loads(result.output) # parses: no prose appended to the JSON payload + + def test_ordinary_config_row_has_no_status(self): + self.write_config({"ordinary": "cloudinary://k:s@plaincloud", + "mycloud": self._cloud_url(FAKE_CLOUD, self._in_hours(8))}) + result = self.runner.invoke(cli, ["config", "-ls", "-j"]) + + rows = {row["name"]: row for row in json.loads(result.output)} + self.assertNotIn("status", rows["ordinary"]) + self.assertIn("status", rows["mycloud"]) + + def test_json_carries_raw_expires_at_not_the_label(self): + expires_at = self._in_hours(8) + self.write_config({"mycloud": self._cloud_url(FAKE_CLOUD, expires_at)}) + result = self.runner.invoke(cli, ["config", "-ls", "-j"]) + + row = json.loads(result.output)[0] + self.assertEqual(expires_at, row["expires_at"]) # consumers get the timestamp, not "expires in 8h" + + def test_show_surfaces_claim_url_and_expiry(self): + expires_at = self._in_hours(8) + self.write_config({"mycloud": self._cloud_url(FAKE_CLOUD, expires_at)}) + result = self.runner.invoke(cli, ["config", "-s", "mycloud"]) + + self.assertEqual(0, result.exit_code) + self.assertIn("claim_url", result.output) + self.assertIn("https://console.cloudinary.com/x?token=ABC", result.output) + self.assertIn(expires_at, result.output) # ISO expiry not mangled by the OAuth epoch formatter + + def test_show_header_flags_the_cloud_and_its_deadline(self): + self.write_config({"mycloud": self._cloud_url(FAKE_CLOUD, self._in_hours(8))}) + result = self.runner.invoke(cli, ["config", "-s", "mycloud"]) + + self.assertIn("[unclaimed, expires in 8h]", result.output) # not just "(api_key)" + self.assertIn("Claimable Cloud: disabled at expiry", result.output) + self.assertIn("cld agent cloud claim mycloud", result.output) + + def test_show_annotates_the_iso_expiry_with_a_countdown(self): + expires_at = self._in_hours(8) + self.write_config({"mycloud": self._cloud_url(FAKE_CLOUD, expires_at)}) + result = self.runner.invoke(cli, ["config", "-s", "mycloud"]) + + self.assertIn(f"{expires_at} (expires in 8h)", result.output) + + def test_show_of_expired_cloud_does_not_advise_claiming(self): + self.write_config({"dead": self._cloud_url("deadcloud", "2020-01-01T00:00:00Z")}) + result = self.runner.invoke(cli, ["config", "-s", "dead"]) + + self.assertIn("past its expiry", result.output) + self.assertIn("cld agent cloud create", result.output) # make a new one instead + self.assertNotIn("cld agent cloud claim", result.output) + + def test_show_of_ordinary_config_is_unchanged(self): + self.write_config({"plain": "cloudinary://k:s@plaincloud"}) + result = self.runner.invoke(cli, ["config", "-s", "plain"]) + + self.assertIn("name: plain (api_key)", result.output) + self.assertNotIn("Claimable Cloud", result.output) + self.assertNotIn("unclaimed", result.output) + + def test_show_json_marks_the_config_as_a_claimable_cloud(self): + self.write_config({"mycloud": self._cloud_url(FAKE_CLOUD, self._in_hours(8))}) + result = self.runner.invoke(cli, ["config", "-s", "mycloud", "-j"]) + + payload = json.loads(result.output) + # a consumer should not have to infer the kind from the presence of claim_url, + # nor parse a countdown string to learn whether it is still live + self.assertIs(True, payload["claimable_cloud"]) + self.assertIs(False, payload["expired"]) + + def test_show_json_marks_an_expired_cloud(self): + self.write_config({"dead": self._cloud_url("deadcloud", "2020-01-01T00:00:00Z")}) + result = self.runner.invoke(cli, ["config", "-s", "dead", "-j"]) + + self.assertIs(True, json.loads(result.output)["expired"]) + + def test_show_json_reports_unknown_expiry_as_null_not_false(self): + self.write_config({"noexp": "cloudinary://k:s@noexp?claim_url=https%3A%2F%2Fc.io%2Fx%3Ftoken%3DD"}) + result = self.runner.invoke(cli, ["config", "-s", "noexp", "-j"]) + + payload = json.loads(result.output) + self.assertIs(True, payload["claimable_cloud"]) + self.assertIsNone(payload["expired"]) # unknown stays distinct from "still valid" + + def test_show_json_of_ordinary_config_omits_cloud_fields(self): + self.write_config({"plain": "cloudinary://k:s@plaincloud"}) + result = self.runner.invoke(cli, ["config", "-s", "plain", "-j"]) + + payload = json.loads(result.output) + self.assertNotIn("claimable_cloud", payload) + self.assertNotIn("expired", payload) + + def test_listing_json_carries_the_same_cloud_fields(self): + self.write_config({"mycloud": self._cloud_url(FAKE_CLOUD, self._in_hours(8)), + "plain": "cloudinary://k:s@plaincloud"}) + result = self.runner.invoke(cli, ["config", "-ls", "-j"]) + + rows = {row["name"]: row for row in json.loads(result.output)} + self.assertIs(True, rows["mycloud"]["claimable_cloud"]) + self.assertIs(False, rows["mycloud"]["expired"]) + self.assertNotIn("claimable_cloud", rows["plain"]) + + def test_show_json_keeps_the_raw_expiry(self): + expires_at = self._in_hours(8) + self.write_config({"mycloud": self._cloud_url(FAKE_CLOUD, expires_at)}) + result = self.runner.invoke(cli, ["config", "-s", "mycloud", "-j"]) + + # the countdown is a display concern; JSON consumers get the timestamp alone + self.assertEqual(expires_at, json.loads(result.output)["expires_at"]) + + +class TestCloudExpiryStatus(unittest.TestCase): + """The Claimable Cloud label layered over the generic countdown.""" + + def _in(self, **delta): + from datetime import datetime, timedelta, timezone + from cloudinary_cli.utils.config_utils import cloud_expiry_status + return cloud_expiry_status((datetime.now(timezone.utc) + timedelta(**delta)).strftime("%Y-%m-%dT%H:%M:%SZ")) + + def test_fresh_cloud_reads_in_hours_not_days(self): + self.assertEqual("unclaimed, expires in 24h", self._in(hours=24)) + + def test_past_expiry(self): + self.assertEqual("unclaimed, expired", self._in(hours=-1)) + + def test_always_leads_with_unclaimed(self): + # the state the user must act on, not just a countdown that reads like a refreshable token + self.assertTrue(self._in(hours=8).startswith("unclaimed")) + + def test_unreadable_expiry_still_reports_unclaimed(self): + from cloudinary_cli.utils.config_utils import cloud_expiry_status + self.assertEqual("unclaimed", cloud_expiry_status(None)) + self.assertEqual("unclaimed", cloud_expiry_status("not-a-date")) + + +class TestClaimUrlRoundTrip(unittest.TestCase): + def test_realistic_claim_url_survives_write_then_read(self): + from cloudinary_cli.core.agent import _cloud_params, _config_url_from_environment + + url = _config_url_from_environment(CLOUD_RESPONSE, params=_cloud_params(CLOUD_RESPONSE)) + + validate_config_url(url) + self.assertEqual(CLOUD_RESPONSE["claim_url"], claim_url_from_url(url)) + self.assertEqual(CLOUD_RESPONSE["expires_at"], expires_at_from_url(url)) + + def test_saved_url_parses_cleanly_as_an_sdk_config(self): + import cloudinary + from cloudinary_cli.core.agent import _cloud_params, _config_url_from_environment + + url = _config_url_from_environment(CLOUD_RESPONSE, params=_cloud_params(CLOUD_RESPONSE)) + config = cloudinary.Config() + config._setup_from_parsed_url(config._parse_cloudinary_url(url)) + + self.assertEqual(FAKE_CLOUD, config.cloud_name) + self.assertEqual(FAKE_API_KEY, config.api_key) + self.assertEqual(FAKE_API_SECRET, config.api_secret) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_utils.py b/test/test_utils.py index d186d2d..6fd3eb2 100644 --- a/test/test_utils.py +++ b/test/test_utils.py @@ -1,9 +1,62 @@ import builtins import unittest +from datetime import datetime, timedelta, timezone from unittest.mock import patch from cloudinary_cli.utils.utils import parse_option_value, parse_args_kwargs, whitelist_keys, merge_responses, \ - normalize_list_params, chunker, group_params, confirm_action, get_user_action, prompt_user, is_interactive + normalize_list_params, chunker, group_params, confirm_action, get_user_action, prompt_user, is_interactive, \ + parse_expiry, is_expired, expires_in_hint + + +NOW = datetime(2026, 8, 19, 12, 0, 0, tzinfo=timezone.utc) + + +class ExpiryHelpersTest(unittest.TestCase): + """Generic expiry helpers, shared by the OAuth token expiry (epoch) and the Claimable Cloud + expiry (ISO-8601 string).""" + + EPOCH = int(NOW.timestamp()) + + def test_parses_iso_with_z_offset_and_naive_alike(self): + for value in ("2026-08-19T12:00:00Z", "2026-08-19T12:00:00+00:00", "2026-08-19T12:00:00"): + self.assertEqual(NOW, parse_expiry(value), value) + + def test_parses_epoch_as_int_and_string(self): + self.assertEqual(NOW, parse_expiry(self.EPOCH)) + self.assertEqual(NOW, parse_expiry(str(self.EPOCH))) + + def test_non_utc_offset_is_honoured(self): + self.assertEqual(NOW, parse_expiry("2026-08-19T14:00:00+02:00")) + + def test_unreadable_values_parse_to_none(self): + for value in (None, "", "not-a-date", True, [], {}): + self.assertIsNone(parse_expiry(value), repr(value)) + + def test_is_expired_distinguishes_unknown_from_expired(self): + self.assertTrue(is_expired("2020-01-01T00:00:00Z", now=NOW)) + self.assertFalse(is_expired("2027-01-01T00:00:00Z", now=NOW)) + self.assertIsNone(is_expired("not-a-date")) # unknown, not expired + + def test_countdown_rounds_rather_than_truncates(self): + # 7h59m reads as 8h, not 7h + self.assertEqual("expires in 8h", expires_in_hint(NOW + timedelta(hours=7, minutes=59), now=NOW)) + + def test_countdown_units(self): + self.assertEqual("expires in 45m", expires_in_hint(NOW + timedelta(minutes=45), now=NOW)) + self.assertEqual("expires in 24h", expires_in_hint(NOW + timedelta(hours=24), now=NOW)) + self.assertEqual("expires in 3d", expires_in_hint(NOW + timedelta(days=3), now=NOW)) + + def test_countdown_never_reads_zero_minutes(self): + self.assertEqual("expires in 1m", expires_in_hint(NOW + timedelta(seconds=20), now=NOW)) + + def test_countdown_labels_are_overridable(self): + self.assertEqual("expired", expires_in_hint(NOW - timedelta(seconds=1), now=NOW)) + self.assertEqual("gone", expires_in_hint(NOW - timedelta(seconds=1), now=NOW, expired_label="gone")) + self.assertEqual("", expires_in_hint("not-a-date")) + self.assertEqual("unknown", expires_in_hint("not-a-date", unknown_label="unknown")) + + def test_accepts_a_datetime_directly(self): + self.assertEqual("expires in 2h", expires_in_hint(NOW + timedelta(hours=2), now=NOW)) class NonInteractiveInputTest(unittest.TestCase): From 59ac45e10a15faee0294d7253cd0ea3f8146942d Mon Sep 17 00:00:00 2001 From: Constantine Nathanson Date: Wed, 19 Aug 2026 20:27:40 +0300 Subject: [PATCH 2/2] Select the config row by name in the `-ls -j` expiry test The test read row [0] and assumed it was the saved cloud. An ambient CLOUDINARY_URL adds an "(environment)" row ahead of it, which carries no expires_at, so the assertion raised KeyError. Co-Authored-By: Claude Opus 5 --- test/test_cli_agent_cloud.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/test/test_cli_agent_cloud.py b/test/test_cli_agent_cloud.py index a521269..460042b 100644 --- a/test/test_cli_agent_cloud.py +++ b/test/test_cli_agent_cloud.py @@ -594,7 +594,8 @@ def test_json_carries_raw_expires_at_not_the_label(self): self.write_config({"mycloud": self._cloud_url(FAKE_CLOUD, expires_at)}) result = self.runner.invoke(cli, ["config", "-ls", "-j"]) - row = json.loads(result.output)[0] + # select by name, not position: an ambient CLOUDINARY_URL adds an "(environment)" row first + row = {r["name"]: r for r in json.loads(result.output)}["mycloud"] self.assertEqual(expires_at, row["expires_at"]) # consumers get the timestamp, not "expires in 8h" def test_show_surfaces_claim_url_and_expiry(self):