Show a one-line error for an invalid CLOUDINARY_URL

cloudinary_cli/__init__.py

@@ -1,4 +1,8 @@
 from cloudinary_cli.version import __version__
+from cloudinary_cli.utils.env_config import import_sdk
+
+# Must run before any other `import cloudinary`, so that an invalid CLOUDINARY_URL does not stop the CLI.
+import_sdk()
 
 import cloudinary
 

cloudinary_cli/core/provisioning.py

@@ -1,7 +1,10 @@
+import sys
+
 from click import command, argument, option
 import cloudinary.provisioning
 
 from cloudinary_cli.utils.api_utils import handle_api_command
+from cloudinary_cli.utils.env_config import env_config_error
 
 
 @command("provisioning",
@@ -19,6 +22,11 @@
 @option("--save", nargs=1, help="Save output to a file.")
 @option("-d", "--doc", is_flag=True, help="Open the Provisioning API reference in a browser.")
 def provisioning(params, optional_parameter, optional_parameter_parsed, ls, save, doc):
+    # -c, -C and the saved default do not replace CLOUDINARY_ACCOUNT_URL, so an invalid one is always an error here.
+    account_url_error = env_config_error("CLOUDINARY_ACCOUNT_URL")
+    if account_url_error:
+        sys.exit(account_url_error)
+
     return handle_api_command(params, optional_parameter, optional_parameter_parsed, ls, save, doc,
                               doc_url="https://cloudinary.com/documentation/provisioning_api",
                               api_instance=cloudinary.provisioning,

cloudinary_cli/utils/config_resolver.py

@@ -1,4 +1,6 @@
 #!/usr/bin/env python3
+import sys
+
 import cloudinary
 from click import UsageError, echo
 
@@ -21,6 +23,7 @@
     user_config_names,
     validate_config_url,
 )
+from cloudinary_cli.utils.env_config import env_config_error
 
 # What the last resolve_cli_config selected, by precedence. One of:
 #   "url"   -> an inline -c CLOUDINARY_URL
@@ -70,7 +73,12 @@ def resolve_cli_config(config=None, config_saved=None, warn_if_unconfigured=True
 
     # 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
-    # has_oauth uniformly; if nothing is configured, _format_ok warns.
+    # has_oauth uniformly; if nothing is configured, _format_ok warns. An invalid CLOUDINARY_URL is an
+    # error only here, and not for the commands that work without a config (such as `config -n`).
+    url_error = env_config_error("CLOUDINARY_URL")
+    if url_error and warn_if_unconfigured:
+        sys.exit(url_error)
+
     if is_env_configured():
         _active_source = "env"
         from cloudinary_cli.auth.oauth_config import install_env_config

cloudinary_cli/utils/env_config.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+# Do not import the SDK at the top: import_sdk must run before the first `import cloudinary`.
+import os
+
+_ENV_VARS = ("CLOUDINARY_URL", "CLOUDINARY_ACCOUNT_URL")
+_errors = {}
+
+
+def import_sdk():
+    """
+    Import the SDK so that an invalid CLOUDINARY_URL or CLOUDINARY_ACCOUNT_URL does not stop the CLI.
+
+    The SDK loads these variables on import and raises when one is invalid, so they are held back during
+    the import. Then each config class loads the environment through a safe wrapper: a failed load leaves
+    an empty config and records the error, which env_config_error returns when the variable is needed.
+    """
+    held = {name: os.environ.pop(name) for name in _ENV_VARS if name in os.environ}
+    try:
+        import cloudinary
+        import cloudinary.provisioning
+    finally:
+        os.environ.update(held)
+
+    _wrap_load_from_env(cloudinary.Config, "CLOUDINARY_URL")
+    _wrap_load_from_env(cloudinary.provisioning.AccountConfig, "CLOUDINARY_ACCOUNT_URL")
+    cloudinary.reset_config()
+    cloudinary.provisioning.reset_config()
+
+
+def env_config_error(name):
+    """The error message for an invalid environment variable `name`, or None if it loaded."""
+    return _errors.get(name)
+
+
+def _wrap_load_from_env(config_class, name):
+    load_from_env = config_class._load_config_from_env
+
+    def safe_load_from_env(self):
+        before = dict(self.__dict__)
+        try:
+            load_from_env(self)
+        except (ValueError, TypeError) as e:
+            # Do not keep the values that loaded before the error.
+            self.__dict__.clear()
+            self.__dict__.update(before)
+            _errors[name] = f"error: {e}. Fix or unset the {name} environment variable."
+
+    config_class._load_config_from_env = safe_load_from_env

test/test_cli.py

@@ -1,3 +1,8 @@
+import json
+import os
+import subprocess
+import sys
+import tempfile
 import unittest
 
 from click.testing import CliRunner
@@ -48,6 +53,73 @@ def test_cli_version(self):
         self.assertIn('Cloudinary SDK', result.output)
         self.assertIn('Python', result.output)
 
+    def _run_cli(self, *args, saved=None, **env_vars):
+        """Run the CLI in a subprocess with only env_vars set, and a temp CLOUDINARY_HOME with the saved configs."""
+        with tempfile.TemporaryDirectory() as home:
+            if saved:
+                with open(os.path.join(home, "config.json"), "w") as f:
+                    json.dump(saved, f)
+            env = {k: v for k, v in os.environ.items() if not k.startswith("CLOUDINARY_")}
+            env.update(CLOUDINARY_HOME=home, **env_vars)
+            return subprocess.run([sys.executable, "-m", "cloudinary_cli.cli", *args],
+                                  env=env, capture_output=True, text=True)
+
+    def assertUrlOutput(self, result):
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertIn("res.cloudinary.com/demo/image/upload/sample", result.stdout)
+
+    def assertEnvError(self, result, name):
+        self.assertEqual(1, result.returncode)
+        self.assertEqual(1, len(result.stderr.strip().splitlines()), result.stderr)
+        self.assertTrue(result.stderr.strip().endswith(f"Fix or unset the {name} environment variable."),
+                        result.stderr)
+
+    def test_invalid_cloudinary_url_env(self):
+        self.assertEnvError(self._run_cli("url", "sample", CLOUDINARY_URL="garbage"), "CLOUDINARY_URL")
+
+    def test_invalid_cloudinary_url_env_query_keys(self):
+        # The SDK raises TypeError, not ValueError, on this URL.
+        result = self._run_cli("url", "sample", CLOUDINARY_URL="cloudinary://1:s@demo?a=1&a[b]=2")
+        self.assertEnvError(result, "CLOUDINARY_URL")
+
+    def test_invalid_cloudinary_url_env_with_config_override(self):
+        self.assertUrlOutput(self._run_cli("-c", "cloudinary://123:abc@demo", "url", "sample",
+                                           CLOUDINARY_URL="garbage"))
+
+    def test_invalid_cloudinary_url_env_with_config_saved(self):
+        self.assertUrlOutput(self._run_cli("-C", "demo", "url", "sample", saved={"demo": "cloudinary://123:abc@demo"},
+                                           CLOUDINARY_URL="garbage"))
+
+    def test_invalid_cloudinary_url_env_with_saved_default(self):
+        saved = {"demo": "cloudinary://123:abc@demo", "__default__": "demo"}
+        self.assertUrlOutput(self._run_cli("url", "sample", saved=saved, CLOUDINARY_URL="garbage"))
+
+    def test_invalid_cloudinary_url_env_with_cloud_name(self):
+        # With CLOUDINARY_CLOUD_NAME set, the SDK ignores CLOUDINARY_URL.
+        self.assertUrlOutput(self._run_cli("url", "sample", CLOUDINARY_URL="garbage", CLOUDINARY_CLOUD_NAME="demo",
+                                           CLOUDINARY_API_KEY="123", CLOUDINARY_API_SECRET="abc"))
+
+    def test_invalid_cloudinary_url_env_config_new(self):
+        # `config -n` works without a config, so the invalid variable does not block it.
+        # The ping of the new config fails (fake credentials), which is not the error under test.
+        result = self._run_cli("config", "-n", "demo", "cloudinary://123:abc@demo", CLOUDINARY_URL="garbage")
+        self.assertNotIn("CLOUDINARY_URL", result.stderr)
+        self.assertNotIn("Traceback", result.stderr)
+
+    def test_invalid_cloudinary_account_url_env(self):
+        # Only `provisioning` uses CLOUDINARY_ACCOUNT_URL. Other commands continue to the API call,
+        # which fails with the fake credentials.
+        result = self._run_cli("admin", "ping", CLOUDINARY_URL="cloudinary://123:abc@demo",
+                               CLOUDINARY_ACCOUNT_URL="garbage")
+        self.assertNotIn("CLOUDINARY_ACCOUNT_URL", result.stderr)
+        self.assertNotIn("Traceback", result.stderr)
+
+    def test_invalid_cloudinary_account_url_env_provisioning(self):
+        # -c does not replace CLOUDINARY_ACCOUNT_URL.
+        result = self._run_cli("-c", "cloudinary://123:abc@demo", "provisioning", "users",
+                               CLOUDINARY_ACCOUNT_URL="garbage")
+        self.assertEnvError(result, "CLOUDINARY_ACCOUNT_URL")
+
     def test_unknown_command_suggests_similar(self):
         result = self.runner.invoke(cli, ['serach', 'cat'])