diff --git a/README.md b/README.md index c759290..8f3205c 100644 --- a/README.md +++ b/README.md @@ -142,6 +142,7 @@ cld logout # Revokes and removes a saved OAuth login. 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. ``` ## Docker Usage @@ -315,6 +316,28 @@ cld [cli options] migrate [command options] upload_mapping file For details, see the [Cloudinary CLI documentation](https://cloudinary.com/documentation/cloudinary_cli#migrate). +### `agent signup` + +**For AI agents only.** Creates a Free-plan Cloudinary account on behalf of a human. No existing configuration is required to run it. A verification email is sent to the address, and the returned credentials are **inert until the human completes that verification**. The new product environment is saved as a named configuration (named after the cloud) so it is ready to use once activated. + +``` +cld agent signup [command options] +``` + +Example: + +``` +cld agent signup you@example.com claude-code claude-fable-5 "test the agent account flow" +``` + +Options: + +* `--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. +* `--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. + ## 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/auth/__init__.py b/cloudinary_cli/auth/__init__.py index c0d6fc6..d959d83 100644 --- a/cloudinary_cli/auth/__init__.py +++ b/cloudinary_cli/auth/__init__.py @@ -23,13 +23,9 @@ from cloudinary_cli.defaults import logger, normalize_region, DEFAULT_REGION, CLOUDINARY_REGION from cloudinary_cli.utils.config_utils import ( load_config, - update_config, remove_config_keys, - user_config_names, - get_default_config_name, - set_default_config, + save_named_config, is_reserved_config_name, - is_env_configured, ) from cloudinary_cli.utils.utils import log_exception, is_interactive @@ -62,32 +58,8 @@ def login(region=None, name=None, set_default=False): raise RuntimeError("Login token did not include a cloud name; cannot save this login.") config_name = name or _derive_config_name(session.cloud_name, region) - was_default = get_default_config_name() == config_name # before we touch the config - update_config({config_name: to_cloudinary_url(session)}) - - if was_default: - return config_name, "already" - if set_default or _should_auto_default(config_name): - set_default_config(config_name) - return config_name, "made" - return config_name, "no" - - -def _should_auto_default(name): - """ - True when the just-saved login should become the default without an explicit flag: it is the - only saved config, the environment configures nothing, and no default is already stored. - - A stored default outranks the environment, so auto-defaulting is suppressed when an env config - is present: a single `cld login` must not silently override a user's CLOUDINARY_URL. They can - still opt in with `--set-default`. - """ - cfg = load_config() - return ( - user_config_names(cfg) == [name] - and not is_env_configured() - and not get_default_config_name() - ) + default_status = save_named_config(config_name, to_cloudinary_url(session), set_default=set_default) + return config_name, default_status def logout(name): diff --git a/cloudinary_cli/auth/session.py b/cloudinary_cli/auth/session.py index 46b6fe3..0d79644 100644 --- a/cloudinary_cli/auth/session.py +++ b/cloudinary_cli/auth/session.py @@ -4,13 +4,14 @@ import base64 import json import time -import urllib.parse from dataclasses import dataclass from cloudinary_cli.defaults import ( OAUTH_EXPIRY_SKEW_SECONDS, api_host_for_region, ) +from cloudinary_cli.utils.url_utils import url_params, url_host +from cloudinary_cli.utils.config_utils import build_config_url # Query-string keys that carry the OAuth session inside a cloudinary:// URL. _OAUTH_MARKER = "oauth_token" @@ -71,15 +72,14 @@ def to_cloudinary_url(session): "issuer": session.issuer or "", "upload_prefix": api_host_for_region(session.region), } - return f"cloudinary://{session.cloud_name}?{urllib.parse.urlencode(params)}" + return build_config_url(session.cloud_name, params) def from_cloudinary_url(url): """Parse an OAuth cloudinary:// URL back into a Session.""" - parsed = urllib.parse.urlparse(url) - q = {k: v[0] for k, v in urllib.parse.parse_qs(parsed.query).items()} + q = url_params(url) return Session( - cloud_name=parsed.hostname, + cloud_name=url_host(url), access_token=q.get("oauth_token"), refresh_token=q.get("refresh_token") or None, issued_at=int(q.get("issued_at", 0) or 0), @@ -92,8 +92,7 @@ def from_cloudinary_url(url): def is_oauth_url(url): if not isinstance(url, str): return False - query = urllib.parse.urlparse(url).query - return _OAUTH_MARKER in urllib.parse.parse_qs(query) + return _OAUTH_MARKER in url_params(url) def _decode_jwt_payload(access_token): diff --git a/cloudinary_cli/cli_group.py b/cloudinary_cli/cli_group.py index a1b5301..541dfae 100644 --- a/cloudinary_cli/cli_group.py +++ b/cloudinary_cli/cli_group.py @@ -28,7 +28,9 @@ @click_log.simple_verbosity_option(logger) @click.pass_context def cli(ctx, config, config_saved): - resolve_cli_config(config, config_saved) + subcommand = cli.get_command(ctx, ctx.invoked_subcommand) if ctx.invoked_subcommand else None + warn_if_unconfigured = not getattr(subcommand, "config_optional", False) + resolve_cli_config(config, config_saved, warn_if_unconfigured=warn_if_unconfigured) if ctx.invoked_subcommand is None: click.echo(ctx.get_help()) diff --git a/cloudinary_cli/core/__init__.py b/cloudinary_cli/core/__init__.py index 8e2edeb..d2d5df3 100644 --- a/cloudinary_cli/core/__init__.py +++ b/cloudinary_cli/core/__init__.py @@ -1,6 +1,7 @@ import click from cloudinary_cli.core.admin import admin +from cloudinary_cli.core.agent import agent_group from cloudinary_cli.core.auth import login, logout from cloudinary_cli.core.config import config_command from cloudinary_cli.core.search import search, search_folders @@ -13,6 +14,7 @@ commands = [ config_command, + agent_group, login, logout, search, diff --git a/cloudinary_cli/core/agent.py b/cloudinary_cli/core/agent.py new file mode 100644 index 0000000..b61c1ba --- /dev/null +++ b/cloudinary_cli/core/agent.py @@ -0,0 +1,202 @@ +import json +import re + +import cloudinary.provisioning +from click import group, argument, option, echo, style, BadParameter, ClickException +from cloudinary.exceptions import Error as CloudinaryError, RateLimited + +from cloudinary_cli.defaults import logger, ACCOUNT_EMAIL_PARAM +from cloudinary_cli.utils.api_utils import call_api +from cloudinary_cli.utils.json_utils import print_json +from cloudinary_cli.utils.config_utils import ( + save_named_config, + is_reserved_config_name, + config_name_for_email, + build_config_url, + user_config_names, + config_optional, +) + + +@config_optional +@group("agent", help="Commands for AI agents acting on behalf of a human.") +def agent_group(): + pass + + +@agent_group.command("signup", + short_help="Create a Cloudinary account on behalf of a human (for AI agents only).", + help="""\b +For AI agents only: create a Free-plan Cloudinary account on behalf of a human. +A verification email is sent to the address; the credentials are inert until the human verifies it. +The returned product environment is saved as a named configuration (use --no-save to skip). +Format: cld agent signup +\te.g. cld agent signup you@example.com claude-code claude-fable-5 "test the agent account flow" +""") +@argument("email") +@argument("agent_framework") +@argument("agent_llm_model") +@argument("agent_goal") +@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("--json", "as_json", is_flag=True, + help="Output the full raw JSON response (agent contract) instead of the human summary.") +def signup(email, agent_framework, agent_llm_model, agent_goal, sdk_framework, name, set_default, no_save, + as_json): + if name and is_reserved_config_name(name): + raise BadParameter(f"'{name}' is a reserved configuration name.") + if not email or not email.strip(): + raise BadParameter("email must not be empty.") + + existing = config_name_for_email(email) + if existing: + raise ClickException(_already_have_config_message(email, existing)) + + try: + result = call_api(cloudinary.provisioning.create_agent_account, email, agent_framework, + agent_llm_model, agent_goal, sdk_framework=sdk_framework) + except RateLimited as e: + raise ClickException( + f"Rate limited while creating the account: {e}. This endpoint is limited per IP address; " + f"wait a bit and try again.") + except CloudinaryError as e: + raise ClickException(_signup_error_message(email, e)) + + # Show the freshly-minted credentials BEFORE saving, so a save failure can never lose them. + if as_json: + print_json(result) + else: + _print_signup_summary(result) + + if not no_save: + save_agent_config(result, email, name=name, set_default=set_default) + + logger.info("Note: the account's credentials are inert until the emailed verification is completed.") + + +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} `. " + f"If it's not activated yet, complete the verification email; or run `cld login` if you use OAuth.") + + +def _account_exists_message(email): + return (f"An account already exists for {email}, but no configuration is saved on this machine. " + f"If it's already verified, add its CLOUDINARY_URL with `cld config -n ` " + f"(or `cld login` if you use OAuth). If you just created it, complete the verification email first.") + + +def _signup_error_message(email, error): + text = str(error) + if "has already been taken" in text or "409" in text: + return _account_exists_message(email) + + detail = _parse_error_detail(text) + return f"Signup failed: {detail}." if detail else f"Signup failed: {text}." + + +def _parse_error_detail(text): + """Best-effort human message from a provisioning error string like + 'Error 400 - {"email":["is invalid"]}' or '... {"error":{"message":"boom"}}'. Returns None on any + failure so the caller falls back to the raw text.""" + match = re.search(r"\{.*\}", text) + if not match: + return None + try: + payload = json.loads(match.group(0)) + except ValueError: + return None + if not isinstance(payload, dict) or not payload: + return None + + error = payload.get("error") + if isinstance(error, dict) and error.get("message"): + return str(error["message"]) + + parts = [] + for field, msgs in payload.items(): + msgs = msgs if isinstance(msgs, list) else [msgs] + parts.append(f"{field} {', '.join(str(m) for m in msgs)}") + return "; ".join(parts) or None + + +# Top-level and product-environment response keys the summary renders explicitly (or deliberately +# omits, e.g. secrets folded into CLOUDINARY_URL). Any key NOT listed here is surfaced generically +# by _extra_rows so future fields the server adds are never silently dropped. +_KNOWN_TOP_LEVEL_KEYS = {"email", "plan_name", "product_environments", "guidance"} +_KNOWN_ENV_KEYS = {"cloud_name", "api_key", "api_secret", "api_environment_variable", "external_id"} + + +def _print_signup_summary(result): + environment = (result.get("product_environments") or [{}])[0] + rows = [ + ("Email", result.get("email", "")), + ("Plan", result.get("plan_name", "")), + ("Cloud name", environment.get("cloud_name", "")), + ("API key", environment.get("api_key", "")), + ("CLOUDINARY_URL", _config_url_from_environment(environment)), + ] + rows += _extra_rows(result, _KNOWN_TOP_LEVEL_KEYS) + rows += _extra_rows(environment, _KNOWN_ENV_KEYS) + rows = [(label, value) for label, value in rows if value] + + echo(style("Cloudinary account created.", 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)) + + guidance = result.get("guidance") + if guidance: + echo(f"\n{guidance}") + + +def _extra_rows(data, known_keys): + """(label, value) rows for scalar keys not already rendered, so response fields the server adds + in the future surface instead of being dropped. Skips nested dict/list values (shown elsewhere + or via --json) and empties.""" + rows = [] + for key, value in data.items(): + if key in known_keys or isinstance(value, (dict, list)) or value in (None, ""): + continue + rows.append((key.replace("_", " ").capitalize(), value)) + 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) + 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 + + if name and name in user_config_names(): + logger.warning(f"Overwriting existing config '{name}'.") + + try: + 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 + + 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.") + + +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 + try: + return build_config_url(environment["cloud_name"], params=params, + api_key=environment["api_key"], api_secret=environment["api_secret"]) + except (KeyError, ValueError): + return "" diff --git a/cloudinary_cli/core/auth.py b/cloudinary_cli/core/auth.py index 6fb4402..d40b2cb 100644 --- a/cloudinary_cli/core/auth.py +++ b/cloudinary_cli/core/auth.py @@ -2,9 +2,11 @@ from cloudinary_cli.auth import login as run_login, logout as run_logout, list_oauth_login_names from cloudinary_cli.defaults import logger +from cloudinary_cli.utils.config_utils import config_optional from cloudinary_cli.utils.utils import log_exception, prompt_user +@config_optional @command("login", help="Log in to Cloudinary via OAuth (opens a browser). The session is saved " "as a named configuration you can select with `-C`.") @argument("name", required=False) @@ -34,6 +36,7 @@ def login(name, region, set_default): return True +@config_optional @command("logout", help="Log out: revoke a saved OAuth login's token and remove its configuration. " "Run without a name to choose from the saved logins.") @argument("name", required=False) diff --git a/cloudinary_cli/core/config.py b/cloudinary_cli/core/config.py index 0c9897d..7e1b75e 100644 --- a/cloudinary_cli/core/config.py +++ b/cloudinary_cli/core/config.py @@ -1,11 +1,11 @@ import cloudinary from click import command, option, echo, BadParameter, UsageError -from cloudinary_cli.defaults import logger, DEFAULT_CONFIG_KEY +from cloudinary_cli.defaults import logger, DEFAULT_CONFIG_KEY, NO_CONFIG_MESSAGE from cloudinary_cli.utils.config_utils import ( load_config, verify_cloudinary_url, - update_config, + save_named_config, remove_config_keys, show_cloudinary_config, is_valid_cloudinary_config, @@ -15,6 +15,7 @@ clear_default_config, is_reserved_config_name, config_type, + config_optional, ) from cloudinary_cli.utils.utils import ConfigurationError from cloudinary_cli.utils.json_utils import print_json @@ -30,6 +31,7 @@ ) +@config_optional @command("config", help="Display the current configuration, and manage additional configurations.") @option("-n", "--new", help="""\b Create and name a configuration from a Cloudinary account environment variable. e.g. cld config -n """, nargs=2) @@ -80,13 +82,12 @@ def config_command(new, ls, as_json, show, rm, from_url, default, set_default, u config_name = config_name or cloudinary.config().cloud_name - update_config({config_name: cloudinary_url}) + default_status = save_named_config(config_name, cloudinary_url, set_default=set_default) logger.info("Config '{}' saved!".format(config_name)) logger.info("Example usage: cld -C {} ".format(config_name)) - if set_default: - set_default_config(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.") elif default: @@ -110,6 +111,8 @@ def config_command(new, ls, as_json, show, rm, from_url, default, set_default, u rows = list_configs() if as_json: print_json(rows) + elif not rows: + echo(NO_CONFIG_MESSAGE) else: echo(render_config_table(rows)) elif show: diff --git a/cloudinary_cli/defaults.py b/cloudinary_cli/defaults.py index cba172d..1132ad9 100644 --- a/cloudinary_cli/defaults.py +++ b/cloudinary_cli/defaults.py @@ -28,6 +28,40 @@ # names are rejected as user config names, so this can't collide with a saved config. 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. +ACCOUNT_EMAIL_PARAM = "account_email" + +# 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. +NO_CONFIG_MESSAGE = ( + "No Cloudinary configuration found.\n" + " - Log in with OAuth: cld login\n" + " - Add an API-key config: cld config -n " + "cloudinary://:@ --set-default\n" + " - Set an existing config\n" + " as the default: cld config -d \n" + " - AI agents only, create\n" + " an account for a human: cld agent signup " +) + +# Shown when saved configs exist but none is active (no default set, no environment config, and no +# -c/-C on the command line). The account is there; the CLI just doesn't know which one to use. +NO_DEFAULT_CONFIG_MESSAGE = ( + "No default Cloudinary configuration is set. Select one per command with `-C `, " + "or set a default with `cld config -d `.\n" + "List your saved configurations with `cld config -ls`." +) + +# Shown when an explicitly selected config (-c URL or -C saved name) has a cloud name but no +# credentials (api_key/api_secret or an OAuth token). The user picked a config on purpose, so the +# generic "no config found" guidance would be misleading; the config is just incomplete. +INCOMPLETE_CONFIG_MESSAGE = ( + "The selected configuration is incomplete: it has a cloud name but no credentials " + "(api_key/api_secret or an OAuth token). Operations that need authentication will fail." +) + # OAuth configuration for `cld login`. The region string derives both the API and # OAuth hosts; an unknown region simply fails to resolve. DEFAULT_REGION = 'api' @@ -82,7 +116,7 @@ def oauth_revoke_url_for_region(region): OAUTH_CALLBACK_PATH = '/callback' OAUTH_CALLBACK_TIMEOUT_SECONDS = 300 -OAUTH_EXPIRY_SKEW_SECONDS = 280 +OAUTH_EXPIRY_SKEW_SECONDS = 30 OAUTH_HTTP_TIMEOUT_SECONDS = 30 TEMPLATE_FOLDER_NAME = 'templates' diff --git a/cloudinary_cli/utils/api_utils.py b/cloudinary_cli/utils/api_utils.py index 3dbd6ca..0ff9c03 100644 --- a/cloudinary_cli/utils/api_utils.py +++ b/cloudinary_cli/utils/api_utils.py @@ -9,7 +9,7 @@ from cloudinary.utils import cloudinary_url from cloudinary_cli.defaults import logger -from cloudinary_cli.utils.config_utils import is_valid_cloudinary_config +from cloudinary_cli.utils.config_utils import is_valid_cloudinary_config, user_config_names from cloudinary_cli.utils.file_utils import (normalize_file_extension, posix_rel_path, get_destination_folder, populate_duplicate_name) from cloudinary_cli.utils.json_utils import print_json, write_json_to_file @@ -35,6 +35,15 @@ # (NConfig.max_resource_count_for_delete). Used purely for clearer prompt wording. MAX_DESTRUCTIVE_BULK_PER_CALL = 1000 +# Public, unauthenticated API methods that must run without a Cloudinary configuration. +PUBLIC_API_METHODS = { + "provisioning": {"create_agent_account"}, +} + + +def is_public_api_method(api_name, method_name): + return method_name in PUBLIC_API_METHODS.get(api_name, set()) + def is_destructive_bulk_api_method(method_name): return method_name in DESTRUCTIVE_BULK_API_METHODS @@ -368,7 +377,10 @@ def handle_api_command( if not confirm_destructive_bulk_api_method(api_name, func.__name__, force): return False - if not is_valid_cloudinary_config(): + if not is_public_api_method(api_name, func.__name__) and not is_valid_cloudinary_config(): + if user_config_names(): + raise ConfigurationError("No default Cloudinary configuration is set. " + "Select one with `-C ` or set a default with `cld config -d `.") raise ConfigurationError("No Cloudinary configuration found.") try: diff --git a/cloudinary_cli/utils/config_listing.py b/cloudinary_cli/utils/config_listing.py index 1c138e7..d12a59b 100644 --- a/cloudinary_cli/utils/config_listing.py +++ b/cloudinary_cli/utils/config_listing.py @@ -12,6 +12,7 @@ config_type, cloudinary_config_details, is_env_configured, + email_from_url, ) from cloudinary_cli.utils.config_resolver import ( active_config_name, @@ -31,6 +32,8 @@ 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_COLUMN = ("email", "EMAIL") def list_configs(): @@ -45,8 +48,8 @@ def list_configs(): rows.append(_url_row()) # an inline -c URL: not a saved config, but it is what's active now if is_env_configured(): rows.append(_env_row(env_active=active_config_is_env())) - rows += [ - { + for name in user_config_names(cfg): + row = { "name": name, "cloud_name": cloud_name_from_url(cfg[name]), "type": config_type(cfg[name]), @@ -54,8 +57,10 @@ def list_configs(): "default": name == default, "active": name == active_name, } - for name in user_config_names(cfg) - ] + 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 + rows.append(row) return rows @@ -88,8 +93,11 @@ def active_config_meta(config_obj): def render_config_table(rows): - headers = [title for _, title in _TABLE_COLUMNS] - cells = [[_cell(row, key) for key, _ in _TABLE_COLUMNS] for row in 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) + 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() diff --git a/cloudinary_cli/utils/config_resolver.py b/cloudinary_cli/utils/config_resolver.py index 24fc308..38b35ad 100644 --- a/cloudinary_cli/utils/config_resolver.py +++ b/cloudinary_cli/utils/config_resolver.py @@ -1,10 +1,16 @@ #!/usr/bin/env python3 import cloudinary -from click import UsageError +from click import UsageError, echo from cloudinary_cli.auth import refresh_url_if_stale from cloudinary_cli.auth.session import strip_oauth_internal_keys -from cloudinary_cli.defaults import logger, DEFAULT_CONFIG_KEY +from cloudinary_cli.defaults import ( + logger, + DEFAULT_CONFIG_KEY, + NO_CONFIG_MESSAGE, + NO_DEFAULT_CONFIG_MESSAGE, + INCOMPLETE_CONFIG_MESSAGE, +) from cloudinary_cli.utils.config_utils import ( load_config, config_to_dict, @@ -13,15 +19,7 @@ is_valid_cloudinary_config, is_env_configured, user_config_names, -) - -_UNCONFIGURED_MESSAGE = ( - "No Cloudinary configuration found.\n" - " - Log in with OAuth: cld login\n" - " - Add an API-key config: cld config -n " - "cloudinary://:@ --set-default\n" - " - Set an existing config\n" - " as the default: cld config -d " + validate_config_url, ) # What the last resolve_cli_config selected, by precedence. One of: @@ -36,7 +34,7 @@ _active_source = None -def resolve_cli_config(config=None, config_saved=None): +def resolve_cli_config(config=None, config_saved=None, warn_if_unconfigured=True): """Select a config by precedence and load it into the SDK global. No network I/O.""" global _active_name, _active_source _active_name = None @@ -45,12 +43,15 @@ def resolve_cli_config(config=None, config_saved=None): if config and config_saved: raise UsageError("-c/--config and -C/--config_saved are mutually exclusive; pass only one.") + cfg = load_config() + + # -c/-C explicitly select a config; if it is shape-invalid it is incomplete (missing + # credentials), not absent, so the generic "no config found" guidance would mislead. if config: + _validate_inline_config(config, cfg) _active_source = "url" refresh_cloudinary_config(config) - return _format_ok() - - cfg = load_config() + return _format_ok(warn_if_unconfigured, INCOMPLETE_CONFIG_MESSAGE) if config_saved: if config_saved not in user_config_names(cfg): @@ -58,14 +59,14 @@ def resolve_cli_config(config=None, config_saved=None): _active_name = config_saved _active_source = "saved" refresh_cloudinary_config(cfg[config_saved], saved_name=config_saved) - return _format_ok() + return _format_ok(warn_if_unconfigured, INCOMPLETE_CONFIG_MESSAGE) default = cfg.get(DEFAULT_CONFIG_KEY) if default and default in cfg: _active_name = default _active_source = "saved" refresh_cloudinary_config(cfg[default], saved_name=default) - return _format_ok() + return _format_ok(warn_if_unconfigured) # No stored default: fall back to the environment. Install it as an OAuthConfig (static, no # saved name -> never refreshes) so the active global is always an OAuthConfig and exposes @@ -74,7 +75,12 @@ def resolve_cli_config(config=None, config_saved=None): _active_source = "env" from cloudinary_cli.auth.oauth_config import install_env_config install_env_config() - return _format_ok() + return _format_ok(warn_if_unconfigured) + + # Nothing resolved. If saved configs exist, the account is there but no default is set, so guide + # the user to pick one rather than claiming there is no configuration at all. + message = NO_DEFAULT_CONFIG_MESSAGE if user_config_names(cfg) else NO_CONFIG_MESSAGE + return _format_ok(warn_if_unconfigured, message) def active_config_name(): @@ -92,10 +98,26 @@ def active_config_is_url(): return _active_source == "url" -def _format_ok(): +def _validate_inline_config(config, cfg): + """-c/--config takes a CLOUDINARY_URL, not a saved config name. Fail early with a clear message + (and point at -C when the value matches a saved config) instead of letting a malformed value + surface as a raw SDK error deep inside command execution.""" + try: + validate_config_url(config) + except ValueError as e: + if config in user_config_names(cfg): + raise UsageError(f"-c/--config expects a CLOUDINARY_URL, but '{config}' is a saved " + f"configuration name. Select it with -C/--config_saved instead: " + f"cld -C {config} .") + raise UsageError(f"-c/--config expects a CLOUDINARY_URL " + f"(cloudinary://:@): {e}") + + +def _format_ok(warn=True, message=NO_CONFIG_MESSAGE): """Format-only check: is a usable-SHAPED config loaded? Does NOT contact the network.""" if not is_valid_cloudinary_config(): - logger.warning(_UNCONFIGURED_MESSAGE) + if warn: + echo(message, err=True) return False return True diff --git a/cloudinary_cli/utils/config_utils.py b/cloudinary_cli/utils/config_utils.py index e5a5926..a2a4c77 100644 --- a/cloudinary_cli/utils/config_utils.py +++ b/cloudinary_cli/utils/config_utils.py @@ -13,10 +13,19 @@ CLOUDINARY_CLI_CONFIG_FILE, OLD_CLOUDINARY_CLI_CONFIG_FILE, DEFAULT_CONFIG_KEY, + ACCOUNT_EMAIL_PARAM, logger, ) from cloudinary_cli.utils.json_utils import write_json_to_file, read_json_from_file -from cloudinary_cli.utils.utils import log_exception +from cloudinary_cli.utils.url_utils import set_url_params, url_param + +def config_optional(cmd): + """Mark a Click command/group as not requiring a resolved Cloudinary config, so the top-level + group callback skips the 'No configuration found' banner for it. The command owns its own config + handling (or needs none).""" + cmd.config_optional = True + return cmd + # Cross-process lock guarding read-modify-write of the config file. Reentrant within a process, # so callers may hold it across a multi-step update (e.g. token refresh) without deadlocking. @@ -109,6 +118,37 @@ def clear_default_config(): remove_config_keys(DEFAULT_CONFIG_KEY) +def save_named_config(name, cloudinary_url, set_default=False): + """ + Persist a named configuration and apply the auto-default rule, returning one of: + "made" - this save became the default (explicit set_default, or auto-defaulted as the sole config), + "already" - the name was already the stored default (left unchanged), + "no" - saved but not the default. + + Auto-default (without set_default) applies only when this is the first usable config: it is the + only saved config, the environment configures nothing, and no default is already stored. A stored + default outranks the environment, so a single save must not silently override a user's CLOUDINARY_URL. + """ + was_default = get_default_config_name() == name # before we touch the config + update_config({name: cloudinary_url}) + + if was_default: + return "already" + if set_default or _should_auto_default(name): + set_default_config(name) + return "made" + return "no" + + +def _should_auto_default(name): + cfg = load_config() + return ( + user_config_names(cfg) == [name] + and not is_env_configured() + and not get_default_config_name() + ) + + def user_config_names(cfg=None): """Saved config names with the reserved default key filtered out.""" cfg = cfg if cfg is not None else load_config() @@ -120,6 +160,55 @@ def is_reserved_config_name(name): return name.startswith("__") and name.endswith("__") +def _normalize_email(email): + return (email or "").strip().lower() + + +def validate_config_url(cloudinary_url): + """Return cloudinary_url if it is a well-formed cloudinary:// config URL, else raise ValueError. + Reuses the SDK's own scheme validation (Config._config_from_parsed_url) and additionally requires + a cloud name, which the SDK does not enforce.""" + config = cloudinary.Config() + # noinspection PyProtectedMember + parsed = config._parse_cloudinary_url(cloudinary_url) + # noinspection PyProtectedMember + config._config_from_parsed_url(parsed) # raises ValueError on a non-cloudinary:// scheme + if not parsed.hostname: + raise ValueError("Invalid CLOUDINARY_URL: missing cloud name.") + return cloudinary_url + + +def build_config_url(cloud_name, params=None, api_key=None, api_secret=None): + """Build (and validate) a cloudinary://[key:secret@]cloud_name?params config URL. The single place + that constructs a config URL from parts. URL -> Config parsing stays with the SDK; the SDK has no + Config -> URL serializer, so building one lives here. Raises ValueError on an invalid result. + + api_key/api_secret are placed in the userinfo verbatim: the SDK reads userinfo raw (it does not + percent-decode it), and Cloudinary keys/secrets are alphanumeric, so they need no escaping.""" + userinfo = f"{api_key}:{api_secret or ''}@" if api_key else "" + url = set_url_params(f"cloudinary://{userinfo}{cloud_name}", **(params or {})) + return validate_config_url(url) + + +def email_from_url(cloudinary_url): + """The normalized ACCOUNT_EMAIL_PARAM value stored in a saved URL, or None.""" + value = url_param(cloudinary_url, ACCOUNT_EMAIL_PARAM) + return _normalize_email(value) if value else None + + +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.""" + email = _normalize_email(email) + if not email: + return None + cfg = load_config() + for name in user_config_names(cfg): + if email_from_url(cfg[name]) == email: + return name + return None + + def refresh_cloudinary_config(cloudinary_url, saved_name=None): """Install cloudinary_url as the active config. OAuth URLs install a self-refreshing config bound to saved_name (so token rotations persist); other URLs use the plain SDK config.""" diff --git a/cloudinary_cli/utils/url_utils.py b/cloudinary_cli/utils/url_utils.py new file mode 100644 index 0000000..6e683b0 --- /dev/null +++ b/cloudinary_cli/utils/url_utils.py @@ -0,0 +1,29 @@ +"""Generic URL query-string helpers — the single place that touches urllib for reading/writing query +params and hosts. Knows nothing about Cloudinary; the cloudinary:// config codec lives in config_utils.""" +from urllib.parse import urlsplit, urlunsplit, urlencode, parse_qs + + +def url_params(url): + """The URL's query string as a flat {key: value} dict (first value per key), or {} if none.""" + query = urlsplit(url or "").query + return {k: v[0] for k, v in parse_qs(query, keep_blank_values=True).items()} + + +def url_param(url, key): + """A single query param value from the URL, or None if absent.""" + return url_params(url).get(key) + + +def url_host(url): + """The host (netloc without userinfo/port) of a URL.""" + return urlsplit(url or "").hostname + + +def set_url_params(url, **params): + """Return url with the given query params added or overridden. Values are urlencoded, so an '@' + in a value never collides with the userinfo '@'. Existing params are preserved.""" + parts = urlsplit(url) + query = parse_qs(parts.query, keep_blank_values=True) + for key, value in params.items(): + query[key] = [value] + return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query, doseq=True), parts.fragment)) diff --git a/requirements.txt b/requirements.txt index 856c21d..efe7965 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -cloudinary>=1.44.4 +cloudinary>=1.45.0 pygments jinja2 click diff --git a/test/test_auth_session.py b/test/test_auth_session.py index bdef819..432623b 100644 --- a/test/test_auth_session.py +++ b/test/test_auth_session.py @@ -337,10 +337,10 @@ class TestLoginGuards(unittest.TestCase): def test_missing_cloud_name_raises_and_saves_nothing(self): session = _session(cloud_name=None) with patch("cloudinary_cli.auth._run_browser_flow", return_value=session), \ - patch("cloudinary_cli.auth.update_config") as update_config: + patch("cloudinary_cli.auth.save_named_config") as save_named_config: with self.assertRaises(RuntimeError): login(region="api-eu") - update_config.assert_not_called() + save_named_config.assert_not_called() class TestBrowserFlowNonInteractive(unittest.TestCase): diff --git a/test/test_cli.py b/test/test_cli.py index 87e1208..f833cc3 100644 --- a/test/test_cli.py +++ b/test/test_cli.py @@ -10,6 +10,7 @@ class TestCLI(unittest.TestCase): COMMANDS = [ 'admin', + 'agent', 'config', 'login', 'logout', diff --git a/test/test_cli_agent.py b/test/test_cli_agent.py new file mode 100644 index 0000000..8dbe701 --- /dev/null +++ b/test/test_cli_agent.py @@ -0,0 +1,402 @@ +import json +import os +import tempfile +import unittest +from contextlib import contextmanager +from unittest.mock import patch + +from click.testing import CliRunner +from cloudinary.exceptions import BadRequest, RateLimited, AlreadyExists, GeneralError +from filelock import FileLock + +from cloudinary_cli.cli import cli +from cloudinary_cli.utils import config_utils + +AGENT_RESPONSE = { + "external_id": "acct-123", + "email": "you@example.com", + "plan_name": "free", + "product_environments": [ + { + "external_id": "env-456", + "cloud_name": "testcloud", + "api_key": "111", + "api_secret": "secret", + "api_environment_variable": "CLOUDINARY_URL=cloudinary://111:secret@testcloud", + } + ], + "guidance": "A verification email has been sent.", +} + +SIGNUP_ARGS = [ + "agent", "signup", + "you@example.com", "claude-code", "claude-opus-4-8", "test the agent account flow", +] + + +class TestCLIAgentSignup(unittest.TestCase): + runner = CliRunner() + + def _invoke(self, extra_args=None, default_status="no", existing=None, save_side_effect=None): + with patch("cloudinary.provisioning.create_agent_account", + return_value=AGENT_RESPONSE) as create, \ + patch("cloudinary_cli.core.agent.config_name_for_email", + return_value=existing) as lookup, \ + patch("cloudinary_cli.core.agent.user_config_names", return_value=[]), \ + patch("cloudinary_cli.core.agent.save_named_config", + return_value=default_status, side_effect=save_side_effect) as save: + result = self.runner.invoke(cli, SIGNUP_ARGS + (extra_args or [])) + return result, create, save, lookup + + def test_signup_pretty_output_default(self): + result, create, save, _ = self._invoke() + + self.assertEqual(0, result.exit_code) + create.assert_called_once_with( + "you@example.com", "claude-code", "claude-opus-4-8", "test the agent account flow", + sdk_framework=None) + self.assertIn("Cloudinary account created.", result.output) + self.assertIn("Cloud name:", result.output) + self.assertIn("testcloud", result.output) + self.assertIn("CLOUDINARY_URL:", result.output) + self.assertIn("cloudinary://111:secret@testcloud", result.output) + self.assertIn("Config 'testcloud' saved!", result.output) + self.assertIn("inert until the emailed verification", result.output) + # guidance from the response is shown in the human path too + self.assertIn("A verification email has been sent.", result.output) + # the saved URL carries the account email + name_arg, url_arg = save.call_args.args + self.assertEqual("testcloud", name_arg) + self.assertIn("cloudinary://111:secret@testcloud", url_arg) + self.assertIn("account_email=you%40example.com", url_arg) + self.assertEqual({"set_default": False}, save.call_args.kwargs) + + def test_signup_no_save_persists_nothing(self): + result, _, save, lookup = self._invoke(["--no-save"]) + + self.assertEqual(0, result.exit_code) + save.assert_not_called() + lookup.assert_called_once() # pre-flight still runs + self.assertIn("Cloudinary account created.", result.output) + self.assertNotIn("Config 'testcloud' saved!", result.output) + + def test_signup_name_overrides_config_name(self): + result, _, save, _ = self._invoke(["--name", "myagent"]) + + self.assertEqual(0, result.exit_code) + self.assertEqual("myagent", save.call_args.args[0]) + + def test_signup_set_default_forwarded(self): + result, _, save, _ = self._invoke(["--set-default"], default_status="made") + + self.assertEqual(0, result.exit_code) + self.assertEqual({"set_default": True}, save.call_args.kwargs) + self.assertIn("Default set to 'testcloud'", result.output) + + def test_signup_auto_defaulted_message(self): + result, _, _, _ = self._invoke(default_status="made") + + self.assertEqual(0, result.exit_code) + self.assertIn("Default set to 'testcloud'", result.output) + + def test_signup_passes_sdk_framework(self): + result, create, _, _ = self._invoke(["--sdk-framework", "python"]) + + self.assertEqual(0, result.exit_code) + create.assert_called_once_with( + "you@example.com", "claude-code", "claude-opus-4-8", "test the agent account flow", + sdk_framework="python") + + def test_signup_surfaces_unknown_future_keys(self): + response = dict( + AGENT_RESPONSE, + trial_days=14, # novel top-level scalar + docs_url="https://example.com/start", # novel top-level scalar + metadata={"ignored": True}, # nested -> not dumped into the summary + ) + response["product_environments"] = [ + dict(AGENT_RESPONSE["product_environments"][0], region="us-east") # novel env scalar + ] + with patch("cloudinary.provisioning.create_agent_account", + return_value=response), \ + patch("cloudinary_cli.core.agent.config_name_for_email", return_value=None), \ + patch("cloudinary_cli.core.agent.user_config_names", return_value=[]), \ + patch("cloudinary_cli.core.agent.save_named_config", return_value="no"): + result = self.runner.invoke(cli, SIGNUP_ARGS) + + self.assertEqual(0, result.exit_code) + self.assertIn("Trial days:", result.output) + self.assertIn("14", result.output) + self.assertIn("Docs url:", result.output) + self.assertIn("https://example.com/start", result.output) + self.assertIn("Region:", result.output) # novel env key surfaced too + self.assertIn("us-east", result.output) + self.assertNotIn("ignored", result.output) # nested values not dumped + + def test_signup_json_flag_emits_raw_json(self): + result, _, save, _ = self._invoke(["--json"]) + + self.assertEqual(0, result.exit_code) + self.assertIn('"external_id": "acct-123"', result.output) + self.assertIn("A verification email has been sent.", result.output) # guidance present + self.assertNotIn("CLOUDINARY_URL:", result.output) # no pretty labels + save.assert_called_once() # saving is independent of output format + + def test_signup_json_no_save(self): + result, _, save, _ = self._invoke(["--json", "--no-save"]) + + self.assertEqual(0, result.exit_code) + self.assertIn('"external_id": "acct-123"', result.output) + save.assert_not_called() + + def test_signup_reserved_name_rejected(self): + result, create, save, _ = self._invoke(["--name", "__default__"]) + + self.assertEqual(2, result.exit_code) + self.assertIn("reserved configuration name", result.output) + create.assert_not_called() + save.assert_not_called() + + def test_signup_empty_email_rejected(self): + with patch("cloudinary.provisioning.create_agent_account") as create: + result = self.runner.invoke( + cli, ["agent", "signup", " ", "claude-code", "claude-opus-4-8", "goal"]) + + self.assertEqual(2, result.exit_code) + self.assertIn("email must not be empty", result.output) + create.assert_not_called() + + def test_signup_preflight_local_hit_skips_api(self): + result, create, save, _ = self._invoke(existing="testcloud") + + self.assertNotEqual(0, result.exit_code) + create.assert_not_called() + save.assert_not_called() + self.assertIn("you@example.com", result.output) + self.assertIn("saved as 'testcloud'", result.output) + self.assertIn("cld -C testcloud", result.output) + + def test_signup_preflight_blocks_even_with_no_save(self): + result, create, save, _ = self._invoke(["--no-save"], existing="testcloud") + + self.assertNotEqual(0, result.exit_code) + create.assert_not_called() + save.assert_not_called() + + def test_signup_name_collision_warns(self): + with patch("cloudinary.provisioning.create_agent_account", + return_value=AGENT_RESPONSE), \ + patch("cloudinary_cli.core.agent.config_name_for_email", return_value=None), \ + patch("cloudinary_cli.core.agent.user_config_names", return_value=["foo"]), \ + patch("cloudinary_cli.core.agent.save_named_config", return_value="no"): + result = self.runner.invoke(cli, SIGNUP_ARGS + ["--name", "foo"]) + + self.assertEqual(0, result.exit_code) + self.assertIn("Overwriting existing config 'foo'", result.output) + + def test_signup_save_failure_still_shows_credentials(self): + result, _, _, _ = self._invoke(save_side_effect=OSError("disk full")) + + self.assertIn("cloudinary://111:secret@testcloud", result.output) # creds shown first + self.assertIn("Could not save the configuration", result.output) + self.assertIn("cld config -n testcloud", result.output) # manual-add hint + + def test_signup_missing_product_env_surfaces_creds(self): + response = dict(AGENT_RESPONSE, product_environments=[]) + with patch("cloudinary.provisioning.create_agent_account", + return_value=response), \ + patch("cloudinary_cli.core.agent.config_name_for_email", return_value=None), \ + patch("cloudinary_cli.core.agent.user_config_names", return_value=[]), \ + patch("cloudinary_cli.core.agent.save_named_config") as save: + result = self.runner.invoke(cli, SIGNUP_ARGS) + + self.assertEqual(0, result.exit_code) + self.assertIn("Cloudinary account created.", result.output) + self.assertIn("you@example.com", result.output) # raw creds still surfaced + self.assertIn("Could not save the configuration automatically", result.output) + save.assert_not_called() + + def _invoke_failing(self, error, existing=None): + with patch("cloudinary.provisioning.create_agent_account", + side_effect=error) as create, \ + patch("cloudinary_cli.core.agent.config_name_for_email", return_value=existing), \ + patch("cloudinary_cli.core.agent.save_named_config") as save: + create.__name__ = "create_agent_account" # call_api reads func.__name__ when logging + result = self.runner.invoke(cli, SIGNUP_ARGS) + return result, save + + def test_signup_server_email_taken_no_local_config(self): + result, save = self._invoke_failing( + BadRequest('Error 400 - {"email":["has already been taken"]}')) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("An account already exists for you@example.com", result.output) + self.assertIn("no configuration is saved on this machine", result.output) + self.assertIn("cld config -n", result.output) + self.assertIn("verification email", result.output) + self.assertNotIn("cld config -ls", result.output) + save.assert_not_called() + + def test_signup_already_exists_409(self): + result, save = self._invoke_failing(AlreadyExists("Error 409 - account exists")) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("An account already exists for you@example.com", result.output) + save.assert_not_called() + + def test_signup_generic_500_clean_message(self): + result, save = self._invoke_failing( + GeneralError('Error 500 - {"error":{"message":"boom"}}')) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("Signup failed: boom.", result.output) + save.assert_not_called() + + def test_signup_other_bad_request_surfaces_message(self): + result, save = self._invoke_failing( + BadRequest('Error 400 - {"email":["is invalid"]}')) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("Signup failed: email is invalid.", result.output) + save.assert_not_called() + + def test_signup_bad_request_str_not_list(self): + result, _ = self._invoke_failing(BadRequest('Error 400 - {"email":"is invalid"}')) + + self.assertIn("Signup failed: email is invalid.", result.output) + + def test_signup_bad_request_empty_body(self): + result, _ = self._invoke_failing(BadRequest("Error 400 - {}")) + + self.assertNotEqual(0, result.exit_code) + self.assertNotIn("Signup failed: .", result.output) # no dangling empty detail + + def test_signup_bad_request_non_json(self): + result, _ = self._invoke_failing(BadRequest("Error 400 - service unavailable")) + + self.assertIn("Signup failed: Error 400 - service unavailable.", result.output) + + def test_signup_rate_limited_guides_to_retry(self): + result, save = self._invoke_failing(RateLimited("Error 420 - too many requests")) + + self.assertNotEqual(0, result.exit_code) + self.assertIn("Rate limited", result.output) + self.assertIn("per IP address", result.output) + save.assert_not_called() + + +_BANNER = "No Cloudinary configuration found" + + +class TestUnconfiguredBanner(unittest.TestCase): + """The top-level group prints the 'No configuration found' banner only for commands that + consume the resolved account; config-optional commands (login/logout/config/agent) stay silent.""" + + runner = CliRunner() + + @contextmanager + def _unconfigured(self, saved=None): + import cloudinary + with tempfile.TemporaryDirectory() as d: + config_file = os.path.join(d, "config.json") + if saved: # saved configs on disk but (deliberately) no __default__ set + with open(config_file, "w") as f: + json.dump(saved, f) + # Strip ambient CLOUDINARY_* (a developer's shell/IDE may export one) and rebuild the + # SDK config from the cleared env, so "unconfigured" is not polluted by a real account. + env = {k: v for k, v in os.environ.items() if not k.startswith("CLOUDINARY_")} + with patch.dict(os.environ, env, clear=True), \ + patch.object(config_utils, "CLOUDINARY_CLI_CONFIG_FILE", config_file), \ + patch.object(config_utils, "_config_lock", FileLock(config_file + ".lock")): + cloudinary.reset_config() + yield + + def _invoke(self, args, saved=None): + with self._unconfigured(saved=saved): + return self.runner.invoke(cli, args) + + def test_config_optional_commands_do_not_warn(self): + for args in (["login", "--help"], ["logout"], ["config", "-ls"], + ["config", "-n", "foo", "cloudinary://k:s@c"], ["agent", "--help"]): + result = self._invoke(args) + self.assertNotIn(_BANNER, result.stderr, args) + + def test_account_commands_still_warn(self): + for args in (["url", "sample"], ["utils"], ["admin"], ["uploader"]): + result = self._invoke(args) + self.assertIn(_BANNER, result.stderr, args) + + def test_banner_has_no_warning_prefix(self): + result = self._invoke(["url", "sample"]) + self.assertIn(_BANNER, result.stderr) + self.assertNotIn("warning:", result.stderr) # printed clean, not via logger.warning + self.assertIn("cld login", result.stderr) # guidance lines carry no prefix either + + def test_saved_configs_but_no_default_gives_distinct_banner(self): + saved = {"prod": "cloudinary://k:s@prodcloud"} # no __default__ + result = self._invoke(["admin", "usage"], saved=saved) + self.assertNotIn(_BANNER, result.stderr) # not the "found nothing" message + self.assertIn("No default Cloudinary configuration is set", result.stderr) + self.assertIn("-C ", result.stderr) + self.assertIn("cld config -d ", result.stderr) + + def test_no_default_api_error_is_accurate(self): + saved = {"prod": "cloudinary://k:s@prodcloud"} + result = self._invoke(["admin", "usage"], saved=saved) + self.assertNotEqual(0, result.exit_code) + self.assertIn("No default Cloudinary configuration is set", str(result.exception)) + + def test_inline_c_rejects_non_url(self): + result = self._invoke(["-c", "dummy", "url", "sample"]) + self.assertEqual(2, result.exit_code) + self.assertIn("-c/--config expects a CLOUDINARY_URL", result.output) + + def test_inline_c_with_saved_name_points_to_capital_c(self): + saved = {"prod": "cloudinary://k:s@prodcloud"} + result = self._invoke(["-c", "prod", "url", "sample"], saved=saved) + self.assertEqual(2, result.exit_code) + self.assertIn("'prod' is a saved configuration name", result.output) + self.assertIn("cld -C prod", result.output) + + def test_inline_c_accepts_valid_url(self): + result = self._invoke(["-c", "cloudinary://k:s@realcloud", "url", "sample"]) + self.assertEqual(0, result.exit_code) + self.assertIn("realcloud", result.output) + + def test_inline_c_keyless_url_reports_incomplete_not_missing(self): + # cloudinary:// is a valid config URL but has no credentials: URL building still + # works, so the banner must say "incomplete", not "no configuration found". + result = self._invoke(["-c", "cloudinary://dummy", "url", "sample"]) + self.assertEqual(0, result.exit_code) + self.assertNotIn(_BANNER, result.stderr) + self.assertIn("has a cloud name but no credentials", result.stderr) + self.assertIn("dummy", result.output) # URL still built + + def test_saved_keyless_config_reports_incomplete(self): + saved = {"keyless": "cloudinary://keylesscloud"} + result = self._invoke(["-C", "keyless", "admin", "usage"], saved=saved) + self.assertNotIn(_BANNER, result.stderr) + self.assertIn("has a cloud name but no credentials", result.stderr) + + def test_empty_config_ls_shows_guidance(self): + result = self._invoke(["config", "-ls"]) + self.assertEqual(0, result.exit_code) + self.assertIn(_BANNER, result.output) # printed to stdout, not the stderr banner + self.assertIn("cld login", result.output) + self.assertNotIn("NAME", result.output) # no empty header-only table + + def test_empty_config_ls_json_stays_empty_array(self): + result = self._invoke(["config", "-ls", "-j"]) + self.assertEqual(0, result.exit_code) + self.assertNotIn(_BANNER, result.output) + self.assertIn("[]", result.output) + + def test_bare_config_reports_no_config_via_own_error(self): + result = self._invoke(["config"]) + self.assertNotEqual(0, result.exit_code) + self.assertNotIn(_BANNER, result.stderr) # not the group banner + self.assertEqual("No Cloudinary configuration found.", str(result.exception)) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_cli_config.py b/test/test_cli_config.py index fac0964..d62f671 100644 --- a/test/test_cli_config.py +++ b/test/test_cli_config.py @@ -51,14 +51,16 @@ def test_cli_config_valid_config(self): def test_cli_config_invalid_config(self): result = self.runner.invoke(cli, ['--config', 'invalid', 'url', 'sample']) - self.assertEqual(1, result.exit_code) - self.assertIn('Invalid CLOUDINARY_URL scheme', str(result.exc_info[1])) + self.assertEqual(2, result.exit_code) + self.assertIn('-c/--config expects a CLOUDINARY_URL', result.output) + self.assertIn('Invalid CLOUDINARY_URL scheme', result.output) def test_cli_config_invalid_config_cloud_name(self): result = self.runner.invoke(cli, ['--config', self.INVALID_CLOUDINARY_URL, 'ping']) - self.assertEqual(1, result.exit_code) - self.assertIn('No Cloudinary configuration found.', str(result.exc_info[1])) + self.assertEqual(2, result.exit_code) + self.assertIn('-c/--config expects a CLOUDINARY_URL', result.output) + self.assertIn('missing cloud name', result.output) def test_cli_show_config(self): result = self.runner.invoke(cli, ['--config', self.TEST_CLOUDINARY_URL, 'config']) @@ -104,7 +106,9 @@ def test_cli_config_show_default_no_config(self): self.assertEqual(1, result.exit_code) - self.assertIn("No Cloudinary configuration found", result.output) + # Bare `config` raises its own ConfigurationError (not the group banner), so the message + # lives on the raised exception rather than stdout. + self.assertIn("No Cloudinary configuration found", str(result.exception)) def test_cli_config_show_non_existent(self): result = self.runner.invoke(cli, ['config', '--show', self.TEST_CLOUD_NAME]) diff --git a/test/test_cli_config_oauth.py b/test/test_cli_config_oauth.py index 5573c7f..54588d5 100644 --- a/test/test_cli_config_oauth.py +++ b/test/test_cli_config_oauth.py @@ -140,31 +140,30 @@ def test_noninteractive_stdin_errors_with_hint(self): class TestLoginSetDefault(unittest.TestCase): """`login` sets the default explicitly with --set-default and auto-defaults a sole login.""" - def _patches(self, saved): - session = Session(cloud_name="eu-cloud", access_token="a", refresh_token="r", - expires_at=int(time.time()) + 300, region="api-eu", - issuer="https://oauth.cloudinary.com/") + def _patches(self, saved, env_configured=False, stored_default=None): + # login now delegates the default decision to config_utils.save_named_config; patch that + # function's dependencies at their home module so the real auto-default logic runs. return patch.multiple( - "cloudinary_cli.auth", - _run_browser_flow=lambda region: session, + "cloudinary_cli.utils.config_utils", load_config=lambda: dict(saved), update_config=lambda *a, **k: None, - is_env_configured=lambda: False, + is_env_configured=lambda: env_configured, + get_default_config_name=lambda: stored_default, ) def test_set_default_flag_marks_default(self): from cloudinary_cli import auth with self._patches({"eu-cloud": _oauth_url(), "other": _oauth_url("other")}), \ - patch("cloudinary_cli.auth.set_default_config") as set_default, \ - patch("cloudinary_cli.auth.get_default_config_name", return_value=None): + patch("cloudinary_cli.auth._run_browser_flow", return_value=self._session()), \ + patch("cloudinary_cli.utils.config_utils.set_default_config") as set_default: auth.login(region="eu", name="eu-cloud", set_default=True) set_default.assert_called_once_with("eu-cloud") def test_auto_default_when_sole_config_no_env_no_default(self): from cloudinary_cli import auth with self._patches({"eu-cloud": _oauth_url()}), \ - patch("cloudinary_cli.auth.set_default_config") as set_default, \ - patch("cloudinary_cli.auth.get_default_config_name", return_value=None): + patch("cloudinary_cli.auth._run_browser_flow", return_value=self._session()), \ + patch("cloudinary_cli.utils.config_utils.set_default_config") as set_default: name, default_status = auth.login(region="eu", name="eu-cloud") set_default.assert_called_once_with("eu-cloud") self.assertEqual(("eu-cloud", "made"), (name, default_status)) @@ -172,8 +171,8 @@ def test_auto_default_when_sole_config_no_env_no_default(self): def test_returns_not_default_when_other_configs_exist(self): from cloudinary_cli import auth with self._patches({"eu-cloud": _oauth_url(), "other": _oauth_url("other")}), \ - patch("cloudinary_cli.auth.set_default_config"), \ - patch("cloudinary_cli.auth.get_default_config_name", return_value=None): + patch("cloudinary_cli.auth._run_browser_flow", return_value=self._session()), \ + patch("cloudinary_cli.utils.config_utils.set_default_config"): name, default_status = auth.login(region="eu", name="eu-cloud") self.assertEqual(("eu-cloud", "no"), (name, default_status)) @@ -181,9 +180,10 @@ def test_relogin_into_existing_default_reports_already(self): """Re-login into a config that is already the stored default must NOT set it again and must report "already" so the CLI doesn't tell the user to make it the default.""" from cloudinary_cli import auth - with self._patches({"eu-cloud": _oauth_url(), "other": _oauth_url("other")}), \ - patch("cloudinary_cli.auth.set_default_config") as set_default, \ - patch("cloudinary_cli.auth.get_default_config_name", return_value="eu-cloud"): + with self._patches({"eu-cloud": _oauth_url(), "other": _oauth_url("other")}, + stored_default="eu-cloud"), \ + patch("cloudinary_cli.auth._run_browser_flow", return_value=self._session()), \ + patch("cloudinary_cli.utils.config_utils.set_default_config") as set_default: name, default_status = auth.login(region="eu", name="eu-cloud") set_default.assert_not_called() self.assertEqual(("eu-cloud", "already"), (name, default_status)) @@ -191,28 +191,33 @@ def test_relogin_into_existing_default_reports_already(self): def test_no_auto_default_when_other_configs_exist(self): from cloudinary_cli import auth with self._patches({"eu-cloud": _oauth_url(), "other": _oauth_url("other")}), \ - patch("cloudinary_cli.auth.set_default_config") as set_default, \ - patch("cloudinary_cli.auth.get_default_config_name", return_value=None): + patch("cloudinary_cli.auth._run_browser_flow", return_value=self._session()), \ + patch("cloudinary_cli.utils.config_utils.set_default_config") as set_default: auth.login(region="eu", name="eu-cloud") set_default.assert_not_called() def test_no_auto_default_when_env_configured(self): from cloudinary_cli import auth - with self._patches({"eu-cloud": _oauth_url()}), \ - patch("cloudinary_cli.auth.is_env_configured", return_value=True), \ - patch("cloudinary_cli.auth.set_default_config") as set_default, \ - patch("cloudinary_cli.auth.get_default_config_name", return_value=None): + with self._patches({"eu-cloud": _oauth_url()}, env_configured=True), \ + patch("cloudinary_cli.auth._run_browser_flow", return_value=self._session()), \ + patch("cloudinary_cli.utils.config_utils.set_default_config") as set_default: auth.login(region="eu", name="eu-cloud") set_default.assert_not_called() def test_no_auto_default_when_default_already_stored(self): from cloudinary_cli import auth - with self._patches({"eu-cloud": _oauth_url()}), \ - patch("cloudinary_cli.auth.set_default_config") as set_default, \ - patch("cloudinary_cli.auth.get_default_config_name", return_value="something"): + with self._patches({"eu-cloud": _oauth_url()}, stored_default="something"), \ + patch("cloudinary_cli.auth._run_browser_flow", return_value=self._session()), \ + patch("cloudinary_cli.utils.config_utils.set_default_config") as set_default: auth.login(region="eu", name="eu-cloud") set_default.assert_not_called() + @staticmethod + def _session(): + return Session(cloud_name="eu-cloud", access_token="a", refresh_token="r", + expires_at=int(time.time()) + 300, region="api-eu", + issuer="https://oauth.cloudinary.com/") + def test_reserved_name_rejected(self): from cloudinary_cli import auth with patch("cloudinary_cli.auth._run_browser_flow"): @@ -526,10 +531,11 @@ def test_stored_default_applies_when_no_explicit_config(self): self.assertEqual("eu-cloud", cloudinary.config().cloud_name) def test_no_implicit_sole_login_without_default(self): - # A single saved login with no stored default no longer auto-applies. + # A single saved login with no stored default no longer auto-applies. The banner reflects + # that a config exists but no default is set (not that none was found). saved = {"eu-cloud": _oauth_url()} result = self._invoke(['url', 'sample'], saved=saved) - self.assertIn("No Cloudinary configuration found.", result.output) + self.assertIn("No default Cloudinary configuration is set", result.stderr) self.assertIsNone(cloudinary.config().cloud_name) def test_stored_default_beats_env(self): @@ -682,22 +688,19 @@ def test_set_default_without_create_flag_errors(self): def test_new_with_set_default(self): with patch("cloudinary_cli.core.config.verify_cloudinary_url", return_value=True), \ - patch("cloudinary_cli.core.config.update_config"), \ - patch("cloudinary_cli.core.config.set_default_config") as set_default: + patch("cloudinary_cli.core.config.save_named_config", return_value="made") as save: result = self.runner.invoke( cli, ['config', '-n', 'prod', 'cloudinary://k:s@prod', '--set-default']) self.assertEqual(0, result.exit_code, result.output) - set_default.assert_called_once_with("prod") + save.assert_called_once_with("prod", 'cloudinary://k:s@prod', set_default=True) self.assertIn("Default set to 'prod'", result.output) def test_set_default_on_failing_url_neither_saves_nor_defaults(self): with patch("cloudinary_cli.core.config.verify_cloudinary_url", return_value=False), \ - patch("cloudinary_cli.core.config.update_config") as update, \ - patch("cloudinary_cli.core.config.set_default_config") as set_default: + patch("cloudinary_cli.core.config.save_named_config") as save: self.runner.invoke( cli, ['config', '-n', 'prod', 'cloudinary://bad', '--set-default']) - update.assert_not_called() - set_default.assert_not_called() + save.assert_not_called() def test_unset_default(self): with patch("cloudinary_cli.core.config.load_config", return_value={}), \ diff --git a/test/test_default_config.py b/test/test_default_config.py index 5cafc89..6f0bbdc 100644 --- a/test/test_default_config.py +++ b/test/test_default_config.py @@ -3,6 +3,7 @@ from unittest.mock import patch import cloudinary_cli.utils.config_utils as config_utils +from cloudinary_cli.defaults import ACCOUNT_EMAIL_PARAM @contextmanager @@ -53,5 +54,160 @@ def test_is_reserved_config_name(self): self.assertFalse(config_utils.is_reserved_config_name("__prod")) +class TestSaveNamedConfig(unittest.TestCase): + def test_first_config_auto_defaults(self): + with _in_memory_config(), \ + patch("cloudinary_cli.utils.config_utils.is_env_configured", return_value=False): + status = config_utils.save_named_config("prod", "cloudinary://k:s@prod") + self.assertEqual("made", status) + self.assertEqual("prod", config_utils.get_default_config_name()) + + def test_second_config_does_not_auto_default(self): + with _in_memory_config({"prod": "cloudinary://k:s@prod", "__default__": "prod"}), \ + patch("cloudinary_cli.utils.config_utils.is_env_configured", return_value=False): + status = config_utils.save_named_config("staging", "cloudinary://k:s@staging") + self.assertEqual("no", status) + self.assertEqual("prod", config_utils.get_default_config_name()) + + def test_env_configured_suppresses_auto_default(self): + with _in_memory_config(), \ + patch("cloudinary_cli.utils.config_utils.is_env_configured", return_value=True): + status = config_utils.save_named_config("prod", "cloudinary://k:s@prod") + self.assertEqual("no", status) + self.assertIsNone(config_utils.get_default_config_name()) + + def test_set_default_forces_default(self): + with _in_memory_config({"prod": "cloudinary://k:s@prod", "__default__": "prod"}), \ + patch("cloudinary_cli.utils.config_utils.is_env_configured", return_value=False): + status = config_utils.save_named_config("staging", "cloudinary://k:s@staging", set_default=True) + self.assertEqual("made", status) + self.assertEqual("staging", config_utils.get_default_config_name()) + + def test_resaving_current_default_reports_already(self): + with _in_memory_config({"prod": "cloudinary://k:s@prod", "__default__": "prod"}), \ + patch("cloudinary_cli.utils.config_utils.is_env_configured", return_value=False): + status = config_utils.save_named_config("prod", "cloudinary://k:s@prod-new") + self.assertEqual("already", status) + self.assertEqual("prod", config_utils.get_default_config_name()) + self.assertEqual("cloudinary://k:s@prod-new", config_utils.load_config()["prod"]) + + +def _agent_url(cloud, email, key="k", secret="s"): + """A saved agent config URL carrying an account email, built the way production builds it.""" + return config_utils.build_config_url(cloud, params={ACCOUNT_EMAIL_PARAM: email}, + api_key=key, api_secret=secret) + + +class TestAccountEmailInUrl(unittest.TestCase): + def test_email_encoded_and_read_back(self): + url = _agent_url("cloud", "you@example.com") + self.assertIn("account_email=you%40example.com", url) # percent-encoded + self.assertEqual("you@example.com", config_utils.email_from_url(url)) + + def test_plus_addressing_roundtrips(self): + url = _agent_url("cloud", "someone+agent@example.com") + self.assertIn("account_email=someone%2Bagent%40example.com", url) + self.assertEqual("someone+agent@example.com", config_utils.email_from_url(url)) + + def test_email_from_url_normalizes(self): + # email_from_url lower-cases/strips what it reads back + url = config_utils.build_config_url("cloud", params={ACCOUNT_EMAIL_PARAM: "You@Example.com"}, + api_key="k", api_secret="s") + self.assertEqual("you@example.com", config_utils.email_from_url(url)) + + def test_email_from_url_none_when_absent(self): + self.assertIsNone(config_utils.email_from_url("cloudinary://k:s@cloud")) + + def test_config_name_for_email_finds_match(self): + with _in_memory_config({ + "agent1": _agent_url("c1", "you@example.com"), + "plain": "cloudinary://k:s@c2", + }): + self.assertEqual("agent1", config_utils.config_name_for_email("YOU@example.com ")) + + def test_config_name_for_email_unknown_returns_none(self): + with _in_memory_config({"plain": "cloudinary://k:s@c2"}): + self.assertIsNone(config_utils.config_name_for_email("you@example.com")) + + def test_config_name_for_email_self_heals_after_removal(self): + with _in_memory_config({"agent1": _agent_url("c1", "you@example.com")}): + self.assertEqual("agent1", config_utils.config_name_for_email("you@example.com")) + config_utils.remove_config_keys("agent1") + self.assertIsNone(config_utils.config_name_for_email("you@example.com")) + + def test_config_to_dict_surfaces_account_email(self): + import cloudinary + cfg = cloudinary.Config() + # noinspection PyProtectedMember + cfg._setup_from_parsed_url(cfg._parse_cloudinary_url(_agent_url("cloud", "you@example.com"))) + self.assertEqual("you@example.com", config_utils.config_to_dict(cfg).get("account_email")) + + +class TestBuildConfigUrl(unittest.TestCase): + def _parsed(self, url): + import cloudinary + cfg = cloudinary.Config() + # noinspection PyProtectedMember + cfg._setup_from_parsed_url(cfg._parse_cloudinary_url(url)) + return cfg + + def test_api_key_url_roundtrips_through_sdk(self): + url = config_utils.build_config_url("mycloud", api_key="111", api_secret="sek") + cfg = self._parsed(url) + self.assertEqual(("111", "sek", "mycloud"), (cfg.api_key, cfg.api_secret, cfg.cloud_name)) + + def test_api_key_url_with_email_param(self): + url = config_utils.build_config_url( + "mycloud", params={ACCOUNT_EMAIL_PARAM: "you@example.com"}, api_key="111", api_secret="sek") + cfg = self._parsed(url) + self.assertEqual("111", cfg.api_key) + self.assertEqual("you@example.com", cfg.__dict__.get(ACCOUNT_EMAIL_PARAM)) + + def test_keyless_oauth_url(self): + url = config_utils.build_config_url("mycloud", {"oauth_token": "abc", "region": "api"}) + self.assertTrue(url.startswith("cloudinary://mycloud?")) + cfg = self._parsed(url) + self.assertIsNone(cfg.api_key) + self.assertEqual("abc", cfg.__dict__.get("oauth_token")) + + def test_build_rejects_missing_cloud_name(self): + with self.assertRaises(ValueError): + config_utils.build_config_url("", api_key="111", api_secret="sek") + + +class TestValidateConfigUrl(unittest.TestCase): + def test_accepts_valid(self): + self.assertEqual("cloudinary://k:s@cloud", + config_utils.validate_config_url("cloudinary://k:s@cloud")) + + def test_rejects_bad_scheme(self): + with self.assertRaises(ValueError): + config_utils.validate_config_url("http://k:s@cloud") + + def test_rejects_missing_cloud_name(self): + with self.assertRaises(ValueError): + config_utils.validate_config_url("cloudinary://") + + +class TestConfigTableEmailColumn(unittest.TestCase): + def _row(self, name, cloud, **extra): + return dict({"name": name, "cloud_name": cloud, "type": "api_key", + "default": False, "active": False}, **extra) + + def test_email_column_hidden_when_no_emails(self): + from cloudinary_cli.utils.config_listing import render_config_table + table = render_config_table([self._row("a", "c1")]) + self.assertNotIn("EMAIL", table) + + def test_email_column_shown_when_any_email(self): + from cloudinary_cli.utils.config_listing import render_config_table + table = render_config_table([ + self._row("a", "c1"), + self._row("b", "c2", email="you@example.com"), + ]) + self.assertIn("EMAIL", table) + self.assertIn("you@example.com", table) + + if __name__ == "__main__": unittest.main()