diff --git a/cloudinary_cli/__init__.py b/cloudinary_cli/__init__.py index 3d1dcb0..b426e9f 100644 --- a/cloudinary_cli/__init__.py +++ b/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 diff --git a/cloudinary_cli/core/provisioning.py b/cloudinary_cli/core/provisioning.py index 92451d3..add5c8d 100644 --- a/cloudinary_cli/core/provisioning.py +++ b/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, diff --git a/cloudinary_cli/utils/config_resolver.py b/cloudinary_cli/utils/config_resolver.py index 38b35ad..d8cbdd7 100644 --- a/cloudinary_cli/utils/config_resolver.py +++ b/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 diff --git a/cloudinary_cli/utils/env_config.py b/cloudinary_cli/utils/env_config.py new file mode 100644 index 0000000..d21ee26 --- /dev/null +++ b/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 diff --git a/test/test_cli.py b/test/test_cli.py index 40116bb..48e7ac3 100644 --- a/test/test_cli.py +++ b/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'])