From c67bcef812fcf9049e91225bf62c0c53eee61bbf Mon Sep 17 00:00:00 2001 From: Constantine Nathanson Date: Tue, 11 Aug 2026 18:33:20 +0300 Subject: [PATCH] Fix stale default configuration after removing a saved config `__default__` could name a config that no longer exists, which suppressed the auto-default rule on the next save. - `load_config` drops a `__default__` naming no saved config; stale keys clear on the next write, since every mutation rebuilds the file from a load_config result. - Add `remove_named_config`, counterpart to `save_named_config`: one atomic write that deletes the config and re-points the default. Used by `logout` and `config -rm`. - Promote a lone survivor to default when the removed config held it, unless the environment configures Cloudinary - a stored default outranks the environment, so promoting would take precedence from CLOUDINARY_URL. `_is_sole_usable_config` holds that rule for the save and remove paths. - `cld logout` with a single saved login confirms it instead of showing a one-item menu. Co-Authored-By: Claude Opus 5 --- cloudinary_cli/auth/__init__.py | 4 +- cloudinary_cli/core/auth.py | 16 +++++ cloudinary_cli/core/config.py | 11 ++-- cloudinary_cli/defaults.py | 11 ++-- cloudinary_cli/utils/config_utils.py | 46 +++++++++++++-- test/test_cli_config_oauth.py | 76 +++++++++++++++++++----- test/test_default_config.py | 87 +++++++++++++++++++++++++++- 7 files changed, 212 insertions(+), 39 deletions(-) diff --git a/cloudinary_cli/auth/__init__.py b/cloudinary_cli/auth/__init__.py index d959d83..3f0e258 100644 --- a/cloudinary_cli/auth/__init__.py +++ b/cloudinary_cli/auth/__init__.py @@ -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, ) @@ -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" diff --git a/cloudinary_cli/core/auth.py b/cloudinary_cli/core/auth.py index d40b2cb..3e3ff0f 100644 --- a/cloudinary_cli/core/auth.py +++ b/cloudinary_cli/core/auth.py @@ -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}") @@ -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 `.") + if choice is None: + return "invalid", None + if choice.strip().lower() not in ("y", "yes"): + return "cancelled", None + return "selected", name diff --git a/cloudinary_cli/core/config.py b/cloudinary_cli/core/config.py index 7e1b75e..a054859 100644 --- a/cloudinary_cli/core/config.py +++ b/cloudinary_cli/core/config.py @@ -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, @@ -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: diff --git a/cloudinary_cli/defaults.py b/cloudinary_cli/defaults.py index 1132ad9..34af062 100644 --- a/cloudinary_cli/defaults.py +++ b/cloudinary_cli/defaults.py @@ -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 " - "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 " + " - Log in with OAuth: cld login\n" + " - Add an API-key config: cld config -n cloudinary://:@ --set-default\n" + " - Set an existing config as the default: cld config -d \n" + " - AI agents - provision an environment to be claimed by a human: cld agent signup " ) # Shown when saved configs exist but none is active (no default set, no environment config, and no diff --git a/cloudinary_cli/utils/config_utils.py b/cloudinary_cli/utils/config_utils.py index a2a4c77..24e9028 100644 --- a/cloudinary_cli/utils/config_utils.py +++ b/cloudinary_cli/utils/config_utils.py @@ -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 @@ -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 ` 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): diff --git a/test/test_cli_config_oauth.py b/test/test_cli_config_oauth.py index 54588d5..255ac15 100644 --- a/test/test_cli_config_oauth.py +++ b/test/test_cli_config_oauth.py @@ -2,6 +2,7 @@ import os import time import unittest +from contextlib import contextmanager from unittest.mock import patch import cloudinary @@ -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 @@ -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) @@ -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") @@ -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 @@ -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() @@ -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) @@ -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 `) 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 ", result.output) @@ -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( diff --git a/test/test_default_config.py b/test/test_default_config.py index 6f0bbdc..718799b 100644 --- a/test/test_default_config.py +++ b/test/test_default_config.py @@ -12,7 +12,9 @@ def _in_memory_config(initial=None): store = {"cfg": dict(initial or {})} def _load(): - return dict(store["cfg"]) + cfg = dict(store["cfg"]) + config_utils._rectify_config(cfg) # the real load_config rectifies; mirror it here + return cfg def _save(cfg): store["cfg"] = dict(cfg) @@ -143,6 +145,89 @@ def test_config_to_dict_surfaces_account_email(self): self.assertEqual("you@example.com", config_utils.config_to_dict(cfg).get("account_email")) +class TestDanglingDefault(unittest.TestCase): + """A __default__ naming no saved config is invalid state; reads must not observe it.""" + + def test_dangling_default_is_ignored(self): + with _in_memory_config({"__default__": "gone"}): + self.assertIsNone(config_utils.get_default_config_name()) + + def test_dangling_default_does_not_block_auto_default(self): + with _in_memory_config({"__default__": "gone"}), \ + 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_saving_under_dangling_default_name_is_not_already(self): + with _in_memory_config({"__default__": "gone"}), \ + patch("cloudinary_cli.utils.config_utils.is_env_configured", return_value=False): + status = config_utils.save_named_config("gone", "cloudinary://k:s@gone") + self.assertEqual("made", status) + self.assertEqual("gone", config_utils.get_default_config_name()) + + def test_dangling_default_dropped_from_file_on_next_write(self): + with _in_memory_config({"prod": "cloudinary://k:s@prod", "__default__": "gone"}) as store: + config_utils.update_config({"staging": "cloudinary://k:s@staging"}) + self.assertNotIn("__default__", store["cfg"]) + + +class TestRemoveNamedConfig(unittest.TestCase): + def test_keeps_default_pointing_at_another_config(self): + with _in_memory_config({"prod": "cloudinary://k:s@prod", + "staging": "cloudinary://k:s@staging", + "__default__": "prod"}) as store, \ + patch("cloudinary_cli.utils.config_utils.is_env_configured", return_value=False): + self.assertTrue(config_utils.remove_named_config("staging")) + self.assertEqual("prod", store["cfg"]["__default__"]) + + def test_promotes_lone_survivor_to_default(self): + with _in_memory_config({"prod": "cloudinary://k:s@prod", + "staging": "cloudinary://k:s@staging", + "__default__": "prod"}) as store, \ + patch("cloudinary_cli.utils.config_utils.is_env_configured", return_value=False): + self.assertTrue(config_utils.remove_named_config("prod")) + self.assertEqual("staging", store["cfg"]["__default__"]) + + def test_no_promotion_when_env_configured(self): + with _in_memory_config({"prod": "cloudinary://k:s@prod", + "staging": "cloudinary://k:s@staging", + "__default__": "prod"}) as store, \ + patch("cloudinary_cli.utils.config_utils.is_env_configured", return_value=True): + self.assertTrue(config_utils.remove_named_config("prod")) + self.assertNotIn("__default__", store["cfg"]) # env keeps precedence + + def test_no_promotion_when_several_survive(self): + with _in_memory_config({"a": "cloudinary://k:s@a", "b": "cloudinary://k:s@b", + "c": "cloudinary://k:s@c", "__default__": "a"}) as store, \ + patch("cloudinary_cli.utils.config_utils.is_env_configured", return_value=False): + self.assertTrue(config_utils.remove_named_config("a")) + self.assertNotIn("__default__", store["cfg"]) # ambiguous, so the user picks + + def test_no_promotion_when_removed_was_not_default(self): + with _in_memory_config({"prod": "cloudinary://k:s@prod", + "staging": "cloudinary://k:s@staging"}) as store, \ + patch("cloudinary_cli.utils.config_utils.is_env_configured", return_value=False): + self.assertTrue(config_utils.remove_named_config("staging")) + self.assertNotIn("__default__", store["cfg"]) # there was no default to re-point + + def test_removing_last_config_leaves_no_default(self): + with _in_memory_config({"prod": "cloudinary://k:s@prod", "__default__": "prod"}) as store, \ + patch("cloudinary_cli.utils.config_utils.is_env_configured", return_value=False): + self.assertTrue(config_utils.remove_named_config("prod")) + self.assertEqual({}, store["cfg"]) + + def test_missing_name_reports_false(self): + with _in_memory_config({"prod": "cloudinary://k:s@prod"}) as store: + self.assertFalse(config_utils.remove_named_config("nope")) + self.assertEqual({"prod": "cloudinary://k:s@prod"}, store["cfg"]) + + def test_refuses_reserved_key(self): + with _in_memory_config({"prod": "cloudinary://k:s@prod", "__default__": "prod"}) as store: + self.assertFalse(config_utils.remove_named_config("__default__")) + self.assertEqual("prod", store["cfg"]["__default__"]) + + class TestBuildConfigUrl(unittest.TestCase): def _parsed(self, url): import cloudinary