Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions cloudinary_cli/auth/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
from cloudinary_cli.defaults import logger, normalize_region, DEFAULT_REGION, CLOUDINARY_REGION
from cloudinary_cli.utils.config_utils import (
load_config,
remove_config_keys,
remove_named_config,
save_named_config,
is_reserved_config_name,
)
Expand Down Expand Up @@ -78,7 +78,7 @@ def logout(name):
return "not_oauth"

revoked = _revoke_login(name, saved[name])
remove_config_keys(name)
remove_named_config(name)
return "removed" if revoked else "revoke_failed"


Expand Down
16 changes: 16 additions & 0 deletions cloudinary_cli/core/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,9 @@ def _select_oauth_login():
logger.info("No saved OAuth logins to log out of.")
return "none", None

if len(names) == 1:
return _confirm_sole_login(names[0])

echo("Saved OAuth logins:")
for i, name in enumerate(names, start=1):
echo(f" {i}) {name}")
Expand All @@ -92,3 +95,16 @@ def _select_oauth_login():
logger.error(f"Invalid selection '{choice}'. Expected a number between 1 and {len(names)}.")
return "invalid", None
return "selected", names[int(choice) - 1]


def _confirm_sole_login(name):
"""With a single saved login there is nothing to choose, so confirm that one rather than
presenting a one-item menu. Logging out still revokes a token, so it is not done unasked."""
choice = prompt_user(
f"Log out of '{name}'? [y/N]: ",
noninteractive_hint="Pass the configuration name directly: `cld logout <name>`.")
if choice is None:
return "invalid", None
if choice.strip().lower() not in ("y", "yes"):
return "cancelled", None
return "selected", name
11 changes: 4 additions & 7 deletions cloudinary_cli/core/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,10 @@
load_config,
verify_cloudinary_url,
save_named_config,
remove_config_keys,
remove_named_config,
show_cloudinary_config,
is_valid_cloudinary_config,
user_config_names,
get_default_config_name,
set_default_config,
clear_default_config,
is_reserved_config_name,
Expand Down Expand Up @@ -101,12 +100,10 @@ def config_command(new, ls, as_json, show, rm, from_url, default, set_default, u
clear_default_config()
logger.info("Default configuration cleared.")
elif rm:
if remove_config_keys(rm):
logger.warning(f"Configuration '{rm}' not found.")
else:
if get_default_config_name() == rm:
clear_default_config()
if remove_named_config(rm):
logger.info(f"Configuration '{rm}' deleted.")
else:
logger.warning(f"Configuration '{rm}' not found.")
elif ls:
rows = list_configs()
if as_json:
Expand Down
11 changes: 4 additions & 7 deletions cloudinary_cli/defaults.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,13 +37,10 @@
# "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 <name> "
"cloudinary://<api_key>:<api_secret>@<cloud_name> --set-default\n"
" - Set an existing config\n"
" as the default: cld config -d <name>\n"
" - AI agents only, create\n"
" an account for a human: cld agent signup <email> <framework> <model> <goal>"
" - Log in with OAuth: cld login\n"
" - Add an API-key config: cld config -n <name> cloudinary://<api_key>:<api_secret>@<cloud_name> --set-default\n"
" - Set an existing config as the default: cld config -d <name>\n"
" - AI agents - provision an environment to be claimed by a human: cld agent signup <email> <framework> <model> <goal>"
)

# Shown when saved configs exist but none is active (no default set, no environment config, and no
Expand Down
46 changes: 40 additions & 6 deletions cloudinary_cli/utils/config_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,10 +72,21 @@ def load_config():
if stat is not None and stat == _config_cache_stat and _config_cache is not None:
return dict(_config_cache) # copy: callers mutate the result in place (e.g. cfg.update(...))
cfg = read_json_from_file(CLOUDINARY_CLI_CONFIG_FILE, does_not_exist_ok=True)
_rectify_config(cfg)
_config_cache, _config_cache_stat = cfg, stat
return dict(cfg)


def _rectify_config(cfg):
"""Correct invalid state in a freshly-read config, in place, so no caller has to guard against
it. The repair reaches disk on the next write, since every mutation rebuilds the file from a
load_config result."""
default = cfg.get(DEFAULT_CONFIG_KEY)
if default is not None and default not in [k for k in cfg if k != DEFAULT_CONFIG_KEY]:
del cfg[DEFAULT_CONFIG_KEY]
logger.debug(f"Ignoring stored default '{default}': no such saved configuration.")


def save_config(config):
# 0600 from the start: the config file holds secrets (api_secret, account_url, OAuth tokens),
# and writing the temp file 0600 before the atomic replace means it is never momentarily
Expand Down Expand Up @@ -140,13 +151,36 @@ def save_named_config(name, cloudinary_url, set_default=False):
return "no"


def remove_named_config(name):
"""
Delete a named configuration, and re-point the stored default when it named that config: a lone
surviving config is promoted, otherwise the default is cleared. The counterpart to
save_named_config: the single way to remove a saved config. Returns True if the configuration
existed and was removed.
"""
with config_lock():
cfg = load_config()
if name not in user_config_names(cfg):
return False
del cfg[name]
if cfg.get(DEFAULT_CONFIG_KEY) == name:
del cfg[DEFAULT_CONFIG_KEY]
remaining = user_config_names(cfg)
if remaining and _is_sole_usable_config(remaining[0], cfg):
cfg[DEFAULT_CONFIG_KEY] = remaining[0]
save_config(cfg)
return True


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()
)
return _is_sole_usable_config(name) and not get_default_config_name()


def _is_sole_usable_config(name, cfg=None):
"""Whether name is the only config a bare `cld <command>` could use: the only saved config, with
nothing configured in the environment. A stored default outranks the environment, so defaulting
to a saved config while CLOUDINARY_URL is set would silently override the user's choice."""
return user_config_names(cfg) == [name] and not is_env_configured()


def user_config_names(cfg=None):
Expand Down
76 changes: 60 additions & 16 deletions test/test_cli_config_oauth.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import os
import time
import unittest
from contextlib import contextmanager
from unittest.mock import patch

import cloudinary
Expand All @@ -11,6 +12,7 @@
from test.oauth_helpers import jwt_access_token
from cloudinary_cli.cli import cli
from cloudinary_cli.utils.config_resolver import config_to_api_kwargs, get_cloudinary_config
from cloudinary_cli.utils import config_utils
from cloudinary_cli.utils.config_utils import config_to_dict, show_cloudinary_config


Expand All @@ -20,6 +22,27 @@ def _oauth_url(cloud="eu-cloud", region="api-eu"):
expires_at=int(time.time()) + 300, region=region, issuer="https://oauth.cloudinary.com/"))


@contextmanager
def _patched_config_store(initial=None):
"""Back load_config/save_config with an in-memory dict (no real config.json or lock)."""
store = {"cfg": dict(initial or {})}

@contextmanager
def _noop_lock():
yield

def _load():
cfg = dict(store["cfg"])
config_utils._rectify_config(cfg) # the real load_config rectifies; mirror it here
return cfg

with patch("cloudinary_cli.utils.config_utils.load_config", side_effect=_load), \
patch("cloudinary_cli.utils.config_utils.save_config",
side_effect=lambda cfg: store.__setitem__("cfg", dict(cfg))), \
patch("cloudinary_cli.utils.config_utils.config_lock", _noop_lock):
yield store


class _RestoresSdkConfig(unittest.TestCase):
def setUp(self):
self._env_snapshot = dict(os.environ)
Expand All @@ -43,7 +66,7 @@ def test_removes_oauth_login(self):
from cloudinary_cli.auth import logout
saved = {"eu-cloud": _oauth_url()}
with patch("cloudinary_cli.auth.load_config", return_value=saved), \
patch("cloudinary_cli.auth.remove_config_keys") as remove, \
patch("cloudinary_cli.auth.remove_named_config") as remove, \
patch("cloudinary_cli.auth.flow.revoke") as revoke:
self.assertEqual("removed", logout("eu-cloud"))
remove.assert_called_once_with("eu-cloud")
Expand All @@ -54,7 +77,7 @@ def test_revoke_failure_still_removes_locally(self):
from cloudinary_cli.auth import logout
saved = {"eu-cloud": _oauth_url()}
with patch("cloudinary_cli.auth.load_config", return_value=saved), \
patch("cloudinary_cli.auth.remove_config_keys") as remove, \
patch("cloudinary_cli.auth.remove_named_config") as remove, \
patch("cloudinary_cli.auth.flow.revoke", side_effect=requests.ConnectionError()):
self.assertEqual("revoke_failed", logout("eu-cloud"))
remove.assert_called_once_with("eu-cloud") # local entry removed despite revoke failure
Expand All @@ -63,14 +86,14 @@ def test_refuses_non_oauth_config(self):
from cloudinary_cli.auth import logout
saved = {"mykey": "cloudinary://key:secret@cloud"}
with patch("cloudinary_cli.auth.load_config", return_value=saved), \
patch("cloudinary_cli.auth.remove_config_keys") as remove:
patch("cloudinary_cli.auth.remove_named_config") as remove:
self.assertEqual("not_oauth", logout("mykey"))
remove.assert_not_called()

def test_missing_name(self):
from cloudinary_cli.auth import logout
with patch("cloudinary_cli.auth.load_config", return_value={}), \
patch("cloudinary_cli.auth.remove_config_keys") as remove:
patch("cloudinary_cli.auth.remove_named_config") as remove:
self.assertEqual("not_found", logout("nope"))
remove.assert_not_called()

Expand All @@ -79,13 +102,15 @@ class TestLogoutInteractiveSelect(unittest.TestCase):
"""`cld logout` with no name lists OAuth logins and removes the chosen one."""

runner = CliRunner()
# The numbered menu only appears with more than one login, so menu tests need two.
_TWO_LOGINS = {"cloud-a": _oauth_url("cloud-a"), "cloud-b": _oauth_url("cloud-b")}

def test_lists_only_oauth_and_removes_selected(self):
saved = {"mykey": "cloudinary://key:secret@cloud",
"cloud-a": _oauth_url("cloud-a"), "cloud-b": _oauth_url("cloud-b")}
with patch("cloudinary_cli.auth.load_config", return_value=saved), \
patch("cloudinary_cli.auth.refresh.load_config", return_value=saved), \
patch("cloudinary_cli.auth.remove_config_keys") as remove, \
patch("cloudinary_cli.auth.remove_named_config") as remove, \
patch("cloudinary_cli.auth.flow.revoke"):
result = self.runner.invoke(cli, ["logout"], input="2\n")
self.assertIn("cloud-a", result.output)
Expand All @@ -96,40 +121,60 @@ def test_lists_only_oauth_and_removes_selected(self):
def test_no_oauth_logins(self):
with patch("cloudinary_cli.auth.refresh.load_config",
return_value={"mykey": "cloudinary://key:secret@cloud"}), \
patch("cloudinary_cli.auth.remove_config_keys") as remove:
patch("cloudinary_cli.auth.remove_named_config") as remove:
result = self.runner.invoke(cli, ["logout"], input="\n")
self.assertIn("No saved OAuth logins", result.output)
remove.assert_not_called()

def test_cancel_on_empty_input(self):
with patch("cloudinary_cli.auth.refresh.load_config", return_value={"cloud-a": _oauth_url("cloud-a")}), \
patch("cloudinary_cli.auth.remove_config_keys") as remove:
patch("cloudinary_cli.auth.remove_named_config") as remove:
result = self.runner.invoke(cli, ["logout"], input="\n")
remove.assert_not_called()
self.assertEqual(0, result.exit_code)

def test_invalid_non_numeric_errors(self):
with patch("cloudinary_cli.auth.refresh.load_config", return_value={"cloud-a": _oauth_url("cloud-a")}), \
patch("cloudinary_cli.auth.remove_config_keys") as remove:
with patch("cloudinary_cli.auth.refresh.load_config", return_value=self._TWO_LOGINS), \
patch("cloudinary_cli.auth.remove_named_config") as remove:
result = self.runner.invoke(cli, ["logout"], input="sdfdsf\n", standalone_mode=False)
self.assertIn("Invalid selection", result.output)
self.assertFalse(result.return_value) # main() maps falsy -> exit 1
remove.assert_not_called()

def test_out_of_range_errors(self):
with patch("cloudinary_cli.auth.refresh.load_config", return_value={"cloud-a": _oauth_url("cloud-a")}), \
patch("cloudinary_cli.auth.remove_config_keys") as remove:
with patch("cloudinary_cli.auth.refresh.load_config", return_value=self._TWO_LOGINS), \
patch("cloudinary_cli.auth.remove_named_config") as remove:
result = self.runner.invoke(cli, ["logout"], input="5\n", standalone_mode=False)
self.assertIn("Invalid selection", result.output)
self.assertFalse(result.return_value)
remove.assert_not_called()

def test_sole_login_confirms_instead_of_listing(self):
saved = {"cloud-a": _oauth_url("cloud-a")}
with patch("cloudinary_cli.auth.load_config", return_value=saved), \
patch("cloudinary_cli.auth.refresh.load_config", return_value=saved), \
patch("cloudinary_cli.auth.remove_named_config") as remove, \
patch("cloudinary_cli.auth.flow.revoke"):
result = self.runner.invoke(cli, ["logout"], input="y\n")
self.assertNotIn("Saved OAuth logins:", result.output) # no one-item menu
self.assertIn("cloud-a", result.output)
remove.assert_called_once_with("cloud-a")

def test_sole_login_declined_removes_nothing(self):
saved = {"cloud-a": _oauth_url("cloud-a")}
with patch("cloudinary_cli.auth.load_config", return_value=saved), \
patch("cloudinary_cli.auth.refresh.load_config", return_value=saved), \
patch("cloudinary_cli.auth.remove_named_config") as remove:
result = self.runner.invoke(cli, ["logout"], input="n\n")
remove.assert_not_called()
self.assertEqual(0, result.exit_code)

def test_noninteractive_stdin_errors_with_hint(self):
# Closed stdin (no input at all): the selection cannot be made, so error with the
# non-interactive form (`cld logout <name>`) and exit non-zero, not a silent no-op.
import builtins
with patch("cloudinary_cli.auth.refresh.load_config", return_value={"cloud-a": _oauth_url("cloud-a")}), \
patch("cloudinary_cli.auth.remove_config_keys") as remove, \
patch("cloudinary_cli.auth.remove_named_config") as remove, \
patch.object(builtins, "input", side_effect=EOFError()):
result = self.runner.invoke(cli, ["logout"], standalone_mode=False)
self.assertIn("cld logout <name>", result.output)
Expand Down Expand Up @@ -811,12 +856,11 @@ def test_synthetic_row_name_parenthesized_in_table_and_json(self):
self.assertIn("(environment)", by_name)

def test_rm_of_default_clears_it(self):
with patch("cloudinary_cli.core.config.remove_config_keys", return_value=[]), \
patch("cloudinary_cli.core.config.get_default_config_name", return_value="prod"), \
patch("cloudinary_cli.core.config.clear_default_config") as clear:
stored = {"prod": "cloudinary://k:s@prod", "__default__": "prod"}
with _patched_config_store(stored) as store:
result = self.runner.invoke(cli, ['config', '-rm', 'prod'])
self.assertEqual(0, result.exit_code, result.output)
clear.assert_called_once()
self.assertEqual({}, store["cfg"]) # the name and the default it held both gone

def test_reserved_name_rejected_on_new(self):
result = self.runner.invoke(
Expand Down
Loading
Loading