From 43e46a0beac18f67b0fab465133437affbbea3aa Mon Sep 17 00:00:00 2001 From: Ian Duffy Date: Thu, 27 Aug 2026 06:06:06 +0100 Subject: [PATCH] feat(no-ticket): add update notifications and `cloudsmith upgrade` The release workflow publishes a per-target install manifest to the public Cloudsmith raw repository with a versions/latest alias. The CLI now polls that manifest to learn the latest released version. A background thread refreshes a cached latest-version record at most once a day and a notice is printed on stderr when a newer version exists. The check rides alongside the command and waits at most one second at exit, at most once a day. It stays silent in CI, without a TTY, in JSON output modes, under `cloudsmith mcp`, and when CLOUDSMITH_NO_UPDATE_CHECK is set. All HTTP goes through the shared session so proxy and custom-CA setups keep working. `cloudsmith upgrade` (alias: `update`) detects the install channel. A standalone binary downloads the archive for its own target, verifies the manifest checksum and swaps the install directory atomically. The pip, pipx, uv, Homebrew, Docker and aqua channels get the exact upgrade command for that channel printed instead. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 3 + cloudsmith_cli/cli/commands/main.py | 3 + cloudsmith_cli/cli/commands/registry.py | 2 + cloudsmith_cli/cli/commands/upgrade.py | 92 +++++ .../cli/tests/commands/test_upgrade.py | 127 +++++++ .../cli/tests/test_update_notice.py | 22 ++ cloudsmith_cli/core/installation.py | 122 +++++++ cloudsmith_cli/core/self_update.py | 114 ++++++ .../core/tests/test_installation.py | 136 +++++++ cloudsmith_cli/core/tests/test_self_update.py | 164 +++++++++ .../core/tests/test_update_check.py | 339 ++++++++++++++++++ cloudsmith_cli/core/update_check.py | 187 ++++++++++ 12 files changed, 1311 insertions(+) create mode 100644 cloudsmith_cli/cli/commands/upgrade.py create mode 100644 cloudsmith_cli/cli/tests/commands/test_upgrade.py create mode 100644 cloudsmith_cli/cli/tests/test_update_notice.py create mode 100644 cloudsmith_cli/core/installation.py create mode 100644 cloudsmith_cli/core/self_update.py create mode 100644 cloudsmith_cli/core/tests/test_installation.py create mode 100644 cloudsmith_cli/core/tests/test_self_update.py create mode 100644 cloudsmith_cli/core/tests/test_update_check.py create mode 100644 cloudsmith_cli/core/update_check.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4b87e660..5b61e0a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Added +- Added `cloudsmith upgrade` (alias: `update`). The command polls the release manifests the CLI already publishes to the public Cloudsmith raw repository, detects how the CLI was installed, and acts accordingly: a standalone binary downloads the new archive, verifies its checksum and replaces itself in place, while pip, pipx, uv, Homebrew, Docker and aqua installs get the exact upgrade command for that channel printed instead. +- The CLI now tells you when a newer version exists. At most once a day it fetches the latest released version in a background thread (waiting at most one second at exit for the result) and prints a short notice on stderr. The notice only appears on an interactive terminal with human-readable output, and never in CI, in `-F json` modes, or under `cloudsmith mcp`. Set `CLOUDSMITH_NO_UPDATE_CHECK=1` to turn it off entirely. + - Added `cloudsmith repos gpg` for managing the GPG key a repository signs its package indexes with. `get` shows the active key and its armored public block, `upload` installs a key you supply, and `regenerate` replaces the current key with a freshly generated Cloudsmith one. Key material and passphrases are only ever read from a file, stdin, or a hidden prompt, never from a command-line value, and `--debug` is refused on `upload` so the request body can't be logged. Both mutating subcommands accept `-n/--dry-run`, which checks the inputs and the key currently in place - naming the fingerprint that would be replaced - then stops before the request, so a mistyped repository or a stale credential fails there rather than on the real attempt. `regenerate` asks you to type `regenerate` to confirm - with no terminal attached it fails instead of blocking, so pass `-y/--yes` for unattended runs. There's no `delete` subcommand because the API has no way to remove a repository's key. - Added `cloudsmith repos privileges` for managing explicit repository access from the terminal. `list` shows the teams, users and service accounts that were granted access explicitly; `set` grants access to any number of them and leaves everyone else untouched, asking first if it would lower access someone already has; `revoke` takes access away from the ones named, skipping any that had none; and `replace` makes a JSON file (or stdin) the complete truth for the repository. `revoke` and `replace` ask for confirmation first unless `-y` is passed. diff --git a/cloudsmith_cli/cli/commands/main.py b/cloudsmith_cli/cli/commands/main.py index be7b3250..3db702a6 100644 --- a/cloudsmith_cli/cli/commands/main.py +++ b/cloudsmith_cli/cli/commands/main.py @@ -2,6 +2,7 @@ import click +from ...core import update_check from ...core.api.version import get_version as get_api_version from ...core.utils import get_github_website, get_help_website from ...core.version import get_version as get_cli_version @@ -63,6 +64,8 @@ def main(ctx, opts, version): """Handle entrypoint to CLI.""" # pylint: disable=unused-argument + update_check.arm(ctx, opts) + if version: print_version(opts) elif ctx.invoked_subcommand is None: diff --git a/cloudsmith_cli/cli/commands/registry.py b/cloudsmith_cli/cli/commands/registry.py index f1d28b9a..f03caaf5 100644 --- a/cloudsmith_cli/cli/commands/registry.py +++ b/cloudsmith_cli/cli/commands/registry.py @@ -37,6 +37,7 @@ "status": f"{_PACKAGE}.status", "tags": f"{_PACKAGE}.tags", "tokens": f"{_PACKAGE}.tokens", + "upgrade": f"{_PACKAGE}.upgrade", "upstream": f"{_PACKAGE}.upstream", "vulnerabilities": f"{_PACKAGE}.vulnerabilities", "whoami": f"{_PACKAGE}.whoami", @@ -56,4 +57,5 @@ "quarantine": ["block"], "repositories": ["repos"], "tags": ["tag"], + "upgrade": ["update"], } diff --git a/cloudsmith_cli/cli/commands/upgrade.py b/cloudsmith_cli/cli/commands/upgrade.py new file mode 100644 index 00000000..3402658f --- /dev/null +++ b/cloudsmith_cli/cli/commands/upgrade.py @@ -0,0 +1,92 @@ +"""CLI/Commands - Upgrade the CLI.""" + +import click +import requests + +from ...core import installation, self_update, update_check +from ...core.version import get_version, parse_version +from .. import decorators, utils +from ..utils import maybe_spinner +from .main import main + + +def _fetch_manifest(opts, target): + try: + with maybe_spinner(opts): + return update_check.fetch_latest_manifest(target=target) + except (requests.RequestException, ValueError) as exc: + raise click.ClickException(f"Failed to fetch the latest version: {exc}") + + +def _is_up_to_date(latest, current): + try: + return parse_version(latest) <= parse_version(current) + except ValueError as exc: + raise click.ClickException(f"Cannot compare versions: {exc}") + + +def _run_self_update(opts, manifest, data): + use_stderr = utils.should_use_stderr(opts) + click.echo( + f"Downloading and installing version {manifest['version']} ... ", + nl=False, + err=use_stderr, + ) + try: + with maybe_spinner(opts): + self_update.perform_self_update(manifest) + except (self_update.SelfUpdateError, requests.RequestException, OSError) as exc: + click.secho("ERROR", fg="red", err=use_stderr) + raise click.ClickException(str(exc)) + click.secho("OK", fg="green", err=use_stderr) + data["upgraded"] = True + if utils.maybe_print_as_json(opts, data): + return + click.echo(f"The Cloudsmith CLI is now at version {manifest['version']}.") + + +@main.command(aliases=["update"]) +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@click.pass_context +def upgrade(ctx, opts): + """Upgrade the Cloudsmith CLI to the latest released version. + + The command detects how the CLI was installed. A standalone binary + replaces itself with the latest release. Every other install channel + gets the correct upgrade command for that channel. + """ + current = get_version() + channel = installation.detect_channel() + target = installation.detect_target() + if channel == installation.CHANNEL_STANDALONE and target is None: + raise click.ClickException( + "Cannot detect a supported platform for the standalone binary." + ) + + manifest = _fetch_manifest(opts, target) + latest = manifest["version"] + up_to_date = _is_up_to_date(latest, current) + update_check.store_latest_version(latest) + + data = {"current_version": current, "latest_version": latest, "channel": channel} + + if up_to_date: + data["up_to_date"] = True + if not utils.maybe_print_as_json(opts, data): + click.echo(f"The Cloudsmith CLI is up to date (version {current}).") + return + + instruction = installation.upgrade_instruction(channel) + if instruction is not None: + data["upgrade_command"] = instruction + if utils.maybe_print_as_json(opts, data): + return + click.echo( + f"A new version of the Cloudsmith CLI is available: {current} → {latest}" + ) + click.echo(f"The CLI was installed via {channel}. To upgrade, run:") + click.echo(f" {instruction}") + return + + _run_self_update(opts, manifest, data) diff --git a/cloudsmith_cli/cli/tests/commands/test_upgrade.py b/cloudsmith_cli/cli/tests/commands/test_upgrade.py new file mode 100644 index 00000000..d8614aed --- /dev/null +++ b/cloudsmith_cli/cli/tests/commands/test_upgrade.py @@ -0,0 +1,127 @@ +"""CLI/Commands - Upgrade - Tests.""" + +import contextlib +import json +from unittest import mock + +import requests + +from ....core import installation, self_update, update_check +from ...commands import main as main_module + + +def make_manifest(version="9.9.9"): + return { + "version": version, + "url": "https://dl.cloudsmith.io/example.tar.gz", + "archive": "cloudsmith-9.9.9-macos-arm64.tar.gz", + "sha256": "abc123", + } + + +@contextlib.contextmanager +def upgrade_context(channel=installation.CHANNEL_PIP, manifest=None, target=None): + manifest = manifest or make_manifest() + with ( + mock.patch.object( + update_check, "fetch_latest_manifest", return_value=manifest + ) as fetch_mock, + mock.patch.object(update_check, "write_cached_state") as cache_mock, + mock.patch.object(installation, "detect_channel", return_value=channel), + mock.patch.object(installation, "detect_target", return_value=target), + mock.patch.object(self_update, "perform_self_update") as perform_mock, + ): + yield {"fetch": fetch_mock, "cache": cache_mock, "perform": perform_mock} + + +def invoke(runner, args=("upgrade",)): + return runner.invoke(main_module.main, list(args), catch_exceptions=False) + + +def test_reports_up_to_date(runner): + with upgrade_context(manifest=make_manifest(version="0.0.1")): + result = invoke(runner) + assert result.exit_code == 0 + assert "up to date" in result.output + + +def test_prints_instruction_for_managed_channel(runner): + with upgrade_context(): + result = invoke(runner) + assert result.exit_code == 0 + assert "9.9.9" in result.output + assert "pip install --upgrade cloudsmith-cli" in result.output + + +def test_self_updates_standalone(runner): + manifest = make_manifest() + with upgrade_context( + channel=installation.CHANNEL_STANDALONE, + manifest=manifest, + target="macos-arm64", + ) as mocks: + result = invoke(runner) + assert result.exit_code == 0 + mocks["perform"].assert_called_once_with(manifest) + mocks["fetch"].assert_called_once_with(target="macos-arm64") + assert "9.9.9" in result.output + + +def test_standalone_without_target_fails(runner): + with upgrade_context(channel=installation.CHANNEL_STANDALONE, target=None): + result = invoke(runner) + assert result.exit_code != 0 + + +def test_self_update_failure_reports_error(runner): + with upgrade_context( + channel=installation.CHANNEL_STANDALONE, target="macos-arm64" + ) as mocks: + mocks["perform"].side_effect = self_update.SelfUpdateError("disk full") + result = invoke(runner) + assert result.exit_code != 0 + assert "disk full" in result.output + + +def test_fetch_failure_reports_error(runner): + with upgrade_context() as mocks: + mocks["fetch"].side_effect = requests.ConnectionError("offline") + result = invoke(runner) + assert result.exit_code != 0 + + +def test_json_output(runner): + with upgrade_context(): + result = invoke(runner, ("upgrade", "-F", "json")) + assert result.exit_code == 0 + data = json.loads(result.stdout)["data"] + assert data["latest_version"] == "9.9.9" + assert data["channel"] == "pip" + + +def test_update_alias(runner): + with upgrade_context(manifest=make_manifest(version="0.0.1")): + result = invoke(runner, ("update",)) + assert result.exit_code == 0 + assert "up to date" in result.output + + +def test_refreshes_notifier_cache(runner): + with upgrade_context() as mocks: + invoke(runner) + mocks["cache"].assert_called_once_with("9.9.9") + + +def test_invalid_manifest_version_reports_error(runner): + with upgrade_context(manifest=make_manifest(version="not-semver")) as mocks: + result = invoke(runner) + assert result.exit_code != 0 + assert "Cannot compare versions" in result.output + mocks["cache"].assert_not_called() + + +def test_cache_write_failure_is_ignored(runner): + with upgrade_context() as mocks: + mocks["cache"].side_effect = OSError("read-only") + result = invoke(runner) + assert result.exit_code == 0 diff --git a/cloudsmith_cli/cli/tests/test_update_notice.py b/cloudsmith_cli/cli/tests/test_update_notice.py new file mode 100644 index 00000000..57acecc9 --- /dev/null +++ b/cloudsmith_cli/cli/tests/test_update_notice.py @@ -0,0 +1,22 @@ +"""Update notice wiring - Tests.""" + +from unittest import mock + +from ...core import update_check +from ..commands import main as main_module + + +def test_main_arms_update_check(runner): + with mock.patch.object(update_check, "arm") as arm_mock: + result = runner.invoke(main_module.main, ["--version"], catch_exceptions=False) + assert result.exit_code == 0 + arm_mock.assert_called_once() + + +def test_update_check_stays_silent_without_tty(runner): + with mock.patch.object( + update_check, "start_background_check_if_stale" + ) as start_mock: + result = runner.invoke(main_module.main, ["--version"], catch_exceptions=False) + assert result.exit_code == 0 + start_mock.assert_not_called() diff --git a/cloudsmith_cli/core/installation.py b/cloudsmith_cli/core/installation.py new file mode 100644 index 00000000..f90e6264 --- /dev/null +++ b/cloudsmith_cli/core/installation.py @@ -0,0 +1,122 @@ +"""Detect how the CLI was installed and how to upgrade it.""" + +import os +import platform +import sys +from importlib import metadata + +CHANNEL_STANDALONE = "standalone" +CHANNEL_HOMEBREW = "homebrew" +CHANNEL_DOCKER = "docker" +CHANNEL_AQUA = "aqua" +CHANNEL_PIP = "pip" +CHANNEL_PIPX = "pipx" +CHANNEL_UV_TOOL = "uv-tool" +CHANNEL_UV_PIP = "uv-pip" +CHANNEL_UNKNOWN = "unknown" + +_DISTRIBUTION_NAME = "cloudsmith-cli" + +_UPGRADE_INSTRUCTIONS = { + CHANNEL_PIP: "pip install --upgrade cloudsmith-cli", + CHANNEL_PIPX: "pipx upgrade cloudsmith-cli", + CHANNEL_UV_TOOL: "uv tool upgrade cloudsmith-cli", + CHANNEL_UV_PIP: "uv pip install --upgrade cloudsmith-cli", + CHANNEL_HOMEBREW: "brew update && brew upgrade cloudsmith-cli", + CHANNEL_DOCKER: "docker pull cloudsmith/cloudsmith-cli:latest", + CHANNEL_AQUA: ( + "update the cloudsmith-io/cloudsmith-cli version in your aqua " + "configuration, then run `aqua install`" + ), + CHANNEL_UNKNOWN: "pip install --upgrade cloudsmith-cli", +} + + +def _running_in_container(): + if os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv"): + return True + return bool( + os.environ.get("container") or os.environ.get("KUBERNETES_SERVICE_HOST") + ) + + +def _distribution_location(): + """Return the installed distribution's directory, or None.""" + try: + distribution = metadata.distribution(_DISTRIBUTION_NAME) + except metadata.PackageNotFoundError: + return None + return str(distribution.locate_file("")) + + +def _distribution_installer(): + """Return the INSTALLER record of the installed distribution, or None.""" + try: + distribution = metadata.distribution(_DISTRIBUTION_NAME) + except metadata.PackageNotFoundError: + return None + installer = distribution.read_text("INSTALLER") + return installer.strip() if installer else None + + +def _detect_frozen_channel(executable_path): + normalized = os.path.realpath(executable_path).replace("\\", "/") + if "aquaproj-aqua" in normalized: + return CHANNEL_AQUA + if "/Cellar/" in normalized or "/linuxbrew/" in normalized: + return CHANNEL_HOMEBREW + if normalized.startswith("/opt/cloudsmith") and _running_in_container(): + return CHANNEL_DOCKER + return CHANNEL_STANDALONE + + +def _detect_package_channel(): + location = _distribution_location() + if location is None: + return CHANNEL_UNKNOWN + normalized = location.replace("\\", "/") + if "/pipx/" in normalized: + return CHANNEL_PIPX + if "/uv/tools/" in normalized: + return CHANNEL_UV_TOOL + installer = _distribution_installer() + if installer == "uv": + return CHANNEL_UV_PIP + if installer == "pip": + return CHANNEL_PIP + return CHANNEL_UNKNOWN + + +def detect_channel(frozen=None, executable_path=None): + """Detect the install channel of the running CLI.""" + frozen = getattr(sys, "frozen", False) if frozen is None else frozen + if frozen: + return _detect_frozen_channel(executable_path or sys.executable) + return _detect_package_channel() + + +def detect_target(): + """Detect the standalone build target for this platform, or None.""" + system = platform.system() + machine = platform.machine().lower() + arch = {"amd64": "x86_64", "x86_64": "x86_64"}.get(machine) + arm = machine in ("arm64", "aarch64") + if system == "Darwin": + return "macos-arm64" if arm else ("macos-x86_64" if arch else None) + if system == "Windows": + return "windows-x86_64" if arch else None + if system == "Linux": + if arm: + arch = "aarch64" + if arch is None: + return None + libc = "gnu" if platform.libc_ver()[0] == "glibc" else "musl" + return f"linux-{arch}-{libc}" + return None + + +def upgrade_instruction(channel): + """Return the upgrade command for a channel, or None for self-update.""" + if channel == CHANNEL_STANDALONE: + return None + return _UPGRADE_INSTRUCTIONS[channel] diff --git a/cloudsmith_cli/core/self_update.py b/cloudsmith_cli/core/self_update.py new file mode 100644 index 00000000..8c560031 --- /dev/null +++ b/cloudsmith_cli/core/self_update.py @@ -0,0 +1,114 @@ +"""Self-update for the standalone CLI bundle.""" + +import hashlib +import os +import shutil +import sys +import tarfile +import tempfile +import zipfile + +DOWNLOAD_TIMEOUT_SECONDS = 120.0 +_READ_CHUNK_BYTES = 1 << 20 + + +class SelfUpdateError(Exception): + """A self-update step failed.""" + + +def download_archive(url, dest_path, timeout=DOWNLOAD_TIMEOUT_SECONDS): + """Stream a release archive to a local file.""" + from .session import create_requests_session + + with create_requests_session().get(url, stream=True, timeout=timeout) as response: + response.raise_for_status() + with open(dest_path, "wb") as dest: + dest.writelines(response.iter_content(chunk_size=_READ_CHUNK_BYTES)) + + +def verify_sha256(path, expected): + """Verify the SHA-256 digest of a file against the manifest value.""" + digest = hashlib.sha256() + with open(path, "rb") as handle: + for chunk in iter(lambda: handle.read(_READ_CHUNK_BYTES), b""): + digest.update(chunk) + if digest.hexdigest() != expected.strip().lower(): + raise SelfUpdateError( + f"checksum mismatch for {os.path.basename(path)}: " + f"expected {expected}, got {digest.hexdigest()}" + ) + + +def extract_archive(archive_path, dest_dir): + """Extract a release archive (tar.gz or zip) into a directory.""" + os.makedirs(dest_dir, exist_ok=True) + if archive_path.endswith(".zip"): + with zipfile.ZipFile(archive_path) as bundle: + bundle.extractall(dest_dir) + return + with tarfile.open(archive_path, "r:gz") as bundle: + bundle.extractall(dest_dir, filter="data") + + +def swap_install_dir(install_dir, staging_dir): + """Replace the install directory with the staged one; return the old one.""" + old_dir = install_dir + ".old" + if os.path.exists(old_dir): + shutil.rmtree(old_dir) + os.rename(install_dir, old_dir) + try: + os.rename(staging_dir, install_dir) + except OSError: + os.rename(old_dir, install_dir) + raise + return old_dir + + +def _check_replaceable(install_dir): + if os.name == "nt": + raise SelfUpdateError( + "self-update cannot replace a running executable on Windows; " + "download the new archive and replace the install directory" + ) + parent = os.path.dirname(install_dir) + if not (os.access(parent, os.W_OK) and os.access(install_dir, os.W_OK)): + raise SelfUpdateError( + f"the install directory {install_dir} is not writable; " + "run the upgrade with sufficient privileges" + ) + + +def perform_self_update(manifest, executable_path=None): + """Download, verify, and atomically install the bundle from a manifest.""" + missing = [key for key in ("url", "sha256") if not manifest.get(key)] + if missing: + raise SelfUpdateError( + f"the release manifest is missing fields: {', '.join(missing)}" + ) + executable_path = executable_path or sys.executable + install_dir = os.path.dirname(os.path.realpath(executable_path)) + _check_replaceable(install_dir) + + parent = os.path.dirname(install_dir) + staging_dir = install_dir + ".new" + if os.path.exists(staging_dir): + shutil.rmtree(staging_dir) + suffix = ".zip" if manifest["url"].endswith(".zip") else ".tar.gz" + archive_fd, archive_path = tempfile.mkstemp(dir=parent, suffix=suffix) + os.close(archive_fd) + try: + download_archive(manifest["url"], archive_path) + verify_sha256(archive_path, manifest["sha256"]) + extract_archive(archive_path, staging_dir) + executable_name = os.path.basename(executable_path) + if not os.path.isfile(os.path.join(staging_dir, executable_name)): + raise SelfUpdateError( + f"the downloaded bundle has no {executable_name} executable" + ) + old_dir = swap_install_dir(install_dir, staging_dir) + finally: + if os.path.exists(archive_path): + os.unlink(archive_path) + if os.path.exists(staging_dir): + shutil.rmtree(staging_dir, ignore_errors=True) + shutil.rmtree(old_dir, ignore_errors=True) diff --git a/cloudsmith_cli/core/tests/test_installation.py b/cloudsmith_cli/core/tests/test_installation.py new file mode 100644 index 00000000..3b42727e --- /dev/null +++ b/cloudsmith_cli/core/tests/test_installation.py @@ -0,0 +1,136 @@ +"""Install channel detection - Tests.""" + +import unittest +from unittest import mock + +from .. import installation + + +class TestDetectFrozenChannel(unittest.TestCase): + def _detect(self, path, in_container=False): + with mock.patch.object( + installation, "_running_in_container", return_value=in_container + ): + return installation.detect_channel(frozen=True, executable_path=path) + + def test_aqua(self): + path = "/home/user/.local/share/aquaproj-aqua/pkgs/x/cloudsmith" + self.assertEqual(installation.CHANNEL_AQUA, self._detect(path)) + + def test_homebrew_macos(self): + path = "/opt/homebrew/Cellar/cloudsmith-cli/1.25.0/libexec/cloudsmith" + self.assertEqual(installation.CHANNEL_HOMEBREW, self._detect(path)) + + def test_homebrew_linux(self): + path = ( + "/home/linuxbrew/.linuxbrew/Cellar/cloudsmith-cli/1.25.0/libexec/cloudsmith" + ) + self.assertEqual(installation.CHANNEL_HOMEBREW, self._detect(path)) + + def test_docker(self): + path = "/opt/cloudsmith/cloudsmith" + self.assertEqual( + installation.CHANNEL_DOCKER, self._detect(path, in_container=True) + ) + + def test_standalone(self): + path = "/usr/local/cloudsmith/cloudsmith" + self.assertEqual(installation.CHANNEL_STANDALONE, self._detect(path)) + + def test_standalone_windows_path(self): + path = "C:\\Tools\\cloudsmith\\cloudsmith.exe" + self.assertEqual(installation.CHANNEL_STANDALONE, self._detect(path)) + + +class TestDetectPackageChannel(unittest.TestCase): + def _detect(self, location, installer): + with ( + mock.patch.object( + installation, "_distribution_location", return_value=location + ), + mock.patch.object( + installation, "_distribution_installer", return_value=installer + ), + ): + return installation.detect_channel(frozen=False) + + def test_pipx(self): + location = ( + "/home/user/.local/pipx/venvs/cloudsmith-cli/lib/python3.12/site-packages" + ) + self.assertEqual(installation.CHANNEL_PIPX, self._detect(location, "pip")) + + def test_uv_tool(self): + location = "/home/user/.local/share/uv/tools/cloudsmith-cli/lib/site-packages" + self.assertEqual(installation.CHANNEL_UV_TOOL, self._detect(location, "uv")) + + def test_uv_pip(self): + location = "/home/user/project/.venv/lib/site-packages" + self.assertEqual(installation.CHANNEL_UV_PIP, self._detect(location, "uv")) + + def test_pip(self): + location = "/usr/lib/python3.12/site-packages" + self.assertEqual(installation.CHANNEL_PIP, self._detect(location, "pip")) + + def test_unknown_without_distribution(self): + self.assertEqual(installation.CHANNEL_UNKNOWN, self._detect(None, None)) + + +class TestDetectTarget(unittest.TestCase): + def _detect(self, system, machine, libc="glibc"): + with ( + mock.patch("platform.system", return_value=system), + mock.patch("platform.machine", return_value=machine), + mock.patch("platform.libc_ver", return_value=(libc, "")), + ): + return installation.detect_target() + + def test_macos_arm64(self): + self.assertEqual("macos-arm64", self._detect("Darwin", "arm64")) + + def test_macos_x86_64(self): + self.assertEqual("macos-x86_64", self._detect("Darwin", "x86_64")) + + def test_windows(self): + self.assertEqual("windows-x86_64", self._detect("Windows", "AMD64")) + + def test_linux_gnu(self): + self.assertEqual("linux-x86_64-gnu", self._detect("Linux", "x86_64")) + + def test_linux_musl(self): + self.assertEqual( + "linux-aarch64-musl", self._detect("Linux", "aarch64", libc="") + ) + + def test_linux_arm64_alias(self): + self.assertEqual("linux-aarch64-gnu", self._detect("Linux", "arm64")) + + def test_unsupported(self): + self.assertIsNone(self._detect("SunOS", "sparc")) + + +class TestUpgradeInstruction(unittest.TestCase): + def test_each_managed_channel_has_an_instruction(self): + managed = ( + installation.CHANNEL_PIP, + installation.CHANNEL_PIPX, + installation.CHANNEL_UV_TOOL, + installation.CHANNEL_UV_PIP, + installation.CHANNEL_HOMEBREW, + installation.CHANNEL_DOCKER, + installation.CHANNEL_AQUA, + installation.CHANNEL_UNKNOWN, + ) + for channel in managed: + self.assertIsInstance(installation.upgrade_instruction(channel), str) + + def test_standalone_self_updates(self): + self.assertIsNone( + installation.upgrade_instruction(installation.CHANNEL_STANDALONE) + ) + + def test_pip_instruction(self): + self.assertIn( + "pip install --upgrade cloudsmith-cli", + installation.upgrade_instruction(installation.CHANNEL_PIP), + ) diff --git a/cloudsmith_cli/core/tests/test_self_update.py b/cloudsmith_cli/core/tests/test_self_update.py new file mode 100644 index 00000000..6e503df8 --- /dev/null +++ b/cloudsmith_cli/core/tests/test_self_update.py @@ -0,0 +1,164 @@ +"""Standalone self-update - Tests.""" + +import hashlib +import io +import os +import tarfile +import tempfile +import unittest +import zipfile +from unittest import mock + +from .. import self_update + + +def build_bundle_tar(path, executable_name="cloudsmith", content=b"new-binary"): + """Build a small onedir-style tar.gz bundle archive.""" + with tarfile.open(path, "w:gz") as tar: + exe_info = tarfile.TarInfo(executable_name) + exe_info.size = len(content) + exe_info.mode = 0o755 + tar.addfile(exe_info, io.BytesIO(content)) + lib_info = tarfile.TarInfo("_internal/lib.so") + lib_data = b"lib" + lib_info.size = len(lib_data) + tar.addfile(lib_info, io.BytesIO(lib_data)) + + +def sha256_of(path): + with open(path, "rb") as handle: + return hashlib.sha256(handle.read()).hexdigest() + + +class TestVerifySha256(unittest.TestCase): + def test_accepts_matching_digest(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "file") + with open(path, "wb") as handle: + handle.write(b"data") + self_update.verify_sha256(path, hashlib.sha256(b"data").hexdigest()) + + def test_rejects_mismatched_digest(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "file") + with open(path, "wb") as handle: + handle.write(b"data") + with self.assertRaises(self_update.SelfUpdateError): + self_update.verify_sha256(path, "0" * 64) + + +class TestExtractArchive(unittest.TestCase): + def test_extracts_tar_gz(self): + with tempfile.TemporaryDirectory() as tmp: + archive = os.path.join(tmp, "bundle.tar.gz") + build_bundle_tar(archive) + staging = os.path.join(tmp, "staging") + self_update.extract_archive(archive, staging) + exe_path = os.path.join(staging, "cloudsmith") + self.assertTrue(os.path.isfile(exe_path)) + self.assertTrue(os.access(exe_path, os.X_OK)) + self.assertTrue(os.path.isfile(os.path.join(staging, "_internal/lib.so"))) + + def test_extracts_zip(self): + with tempfile.TemporaryDirectory() as tmp: + archive = os.path.join(tmp, "bundle.zip") + with zipfile.ZipFile(archive, "w") as bundle: + bundle.writestr("cloudsmith.exe", b"new-binary") + staging = os.path.join(tmp, "staging") + self_update.extract_archive(archive, staging) + self.assertTrue(os.path.isfile(os.path.join(staging, "cloudsmith.exe"))) + + +class TestSwapInstallDir(unittest.TestCase): + def test_swaps_directories(self): + with tempfile.TemporaryDirectory() as tmp: + install_dir = os.path.join(tmp, "cloudsmith") + staging_dir = os.path.join(tmp, "cloudsmith.new") + os.makedirs(install_dir) + os.makedirs(staging_dir) + with open(os.path.join(install_dir, "cloudsmith"), "w") as handle: + handle.write("old") + with open(os.path.join(staging_dir, "cloudsmith"), "w") as handle: + handle.write("new") + + old_dir = self_update.swap_install_dir(install_dir, staging_dir) + + with open(os.path.join(install_dir, "cloudsmith")) as handle: + self.assertEqual("new", handle.read()) + with open(os.path.join(old_dir, "cloudsmith")) as handle: + self.assertEqual("old", handle.read()) + + +class TestPerformSelfUpdate(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.install_dir = os.path.join(self.tmp.name, "cloudsmith") + os.makedirs(os.path.join(self.install_dir, "_internal")) + self.executable = os.path.join(self.install_dir, "cloudsmith") + with open(self.executable, "w") as handle: + handle.write("old-binary") + self.archive = os.path.join(self.tmp.name, "fixture.tar.gz") + build_bundle_tar(self.archive) + + def _manifest(self, sha256=None): + return { + "version": "1.26.0", + "url": "https://dl.cloudsmith.io/example.tar.gz", + "archive": "cloudsmith-1.26.0-macos-arm64.tar.gz", + "sha256": sha256 or sha256_of(self.archive), + } + + def _fake_download(self, url, dest_path, timeout=None): + with open(self.archive, "rb") as src, open(dest_path, "wb") as dest: + dest.write(src.read()) + + def test_replaces_install_dir(self): + with mock.patch.object( + self_update, "download_archive", side_effect=self._fake_download + ): + self_update.perform_self_update( + self._manifest(), executable_path=self.executable + ) + with open(self.executable, "rb") as handle: + self.assertEqual(b"new-binary", handle.read()) + self.assertFalse(os.path.exists(self.install_dir + ".old")) + self.assertFalse(os.path.exists(self.install_dir + ".new")) + + def test_sha_mismatch_leaves_install_untouched(self): + with mock.patch.object( + self_update, "download_archive", side_effect=self._fake_download + ): + with self.assertRaises(self_update.SelfUpdateError): + self_update.perform_self_update( + self._manifest(sha256="0" * 64), + executable_path=self.executable, + ) + with open(self.executable) as handle: + self.assertEqual("old-binary", handle.read()) + + def test_rejects_manifest_without_url_or_sha256(self): + for missing_key in ("url", "sha256"): + manifest = self._manifest() + del manifest[missing_key] + with self.assertRaises(self_update.SelfUpdateError): + self_update.perform_self_update( + manifest, executable_path=self.executable + ) + + def test_refuses_windows(self): + with mock.patch.object(self_update.os, "name", "nt"): + with self.assertRaises(self_update.SelfUpdateError): + self_update.perform_self_update( + self._manifest(), executable_path=self.executable + ) + + def test_refuses_unwritable_parent(self): + if os.name == "nt" or os.geteuid() == 0: + self.skipTest("chmod-based write denial needs a non-root POSIX user") + os.chmod(self.tmp.name, 0o500) + self.addCleanup(os.chmod, self.tmp.name, 0o700) + with self.assertRaises(self_update.SelfUpdateError): + self_update.perform_self_update( + self._manifest(), executable_path=self.executable + ) diff --git a/cloudsmith_cli/core/tests/test_update_check.py b/cloudsmith_cli/core/tests/test_update_check.py new file mode 100644 index 00000000..50d84557 --- /dev/null +++ b/cloudsmith_cli/core/tests/test_update_check.py @@ -0,0 +1,339 @@ +"""Update check - Tests.""" + +import json +import os +import tempfile +import unittest +from unittest import mock + +from .. import update_check, version + +MANIFEST_TEXT = ( + "# Generated by the standalone CLI release workflow. Do not edit manually.\n" + "schema=1\n" + "version=1.26.0\n" + "target=macos-arm64\n" + "archive=cloudsmith-1.26.0-macos-arm64.tar.gz\n" + "url=https://dl.cloudsmith.io/example.tar.gz\n" + "sha256=abc123\n" +) + + +class TestParseManifest(unittest.TestCase): + def test_parses_key_value_lines(self): + manifest = update_check.parse_manifest(MANIFEST_TEXT) + self.assertEqual("1.26.0", manifest["version"]) + self.assertEqual("abc123", manifest["sha256"]) + + def test_skips_comments_and_blank_lines(self): + manifest = update_check.parse_manifest("# comment\n\nversion=1.0.0\n") + self.assertEqual({"version": "1.0.0"}, manifest) + + +class TestFetchLatestManifest(unittest.TestCase): + def setUp(self): + patcher = mock.patch("cloudsmith_cli.core.session.create_requests_session") + self.addCleanup(patcher.stop) + self.get_mock = patcher.start().return_value.get + + def test_fetches_probe_target_manifest(self): + self.get_mock.return_value.text = MANIFEST_TEXT + manifest = update_check.fetch_latest_manifest() + expected_url = update_check.MANIFEST_URL_TEMPLATE.format( + target=update_check.VERSION_PROBE_TARGET + ) + self.get_mock.assert_called_once_with(expected_url, timeout=5.0) + self.get_mock.return_value.raise_for_status.assert_called_once_with() + self.assertEqual("1.26.0", manifest["version"]) + + def test_fetches_explicit_target_manifest(self): + self.get_mock.return_value.text = MANIFEST_TEXT + update_check.fetch_latest_manifest(target="macos-arm64", timeout=2.0) + expected_url = update_check.MANIFEST_URL_TEMPLATE.format(target="macos-arm64") + self.get_mock.assert_called_once_with(expected_url, timeout=2.0) + + def test_rejects_manifest_without_version(self): + self.get_mock.return_value.text = "schema=1\n" + with self.assertRaises(ValueError): + update_check.fetch_latest_manifest() + + +class TestCachedState(unittest.TestCase): + def setUp(self): + self.tmp_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp_dir.cleanup) + self.cache_file = os.path.join(self.tmp_dir.name, "update_check.json") + patcher = mock.patch.object( + update_check, "get_cache_file_path", return_value=self.cache_file + ) + self.addCleanup(patcher.stop) + patcher.start() + + def test_round_trip(self): + with mock.patch("time.time", return_value=1000.0): + update_check.write_cached_state("1.26.0") + state = update_check.read_cached_state() + self.assertEqual("1.26.0", state["latest_version"]) + self.assertEqual(1000.0, state["checked_at"]) + + def test_read_returns_empty_dict_when_missing(self): + self.assertEqual({}, update_check.read_cached_state()) + + def test_read_returns_empty_dict_when_corrupt(self): + with open(self.cache_file, "w", encoding="utf-8") as cache: + cache.write("not json") + self.assertEqual({}, update_check.read_cached_state()) + + def test_read_returns_empty_dict_when_not_an_object(self): + with open(self.cache_file, "w", encoding="utf-8") as cache: + json.dump(["1.26.0"], cache) + self.assertEqual({}, update_check.read_cached_state()) + + +class TestCacheIsFresh(unittest.TestCase): + def test_fresh_within_interval(self): + state = {"checked_at": 1000.0, "latest_version": "1.26.0"} + self.assertTrue(update_check.cache_is_fresh(state, now=1000.0 + 3600)) + + def test_stale_after_interval(self): + state = {"checked_at": 1000.0, "latest_version": "1.26.0"} + stale_now = 1000.0 + update_check.DEFAULT_INTERVAL_SECONDS + 1 + self.assertFalse(update_check.cache_is_fresh(state, now=stale_now)) + + def test_stale_when_empty(self): + self.assertFalse(update_check.cache_is_fresh({}, now=1000.0)) + + def test_stale_when_checked_at_invalid(self): + state = {"checked_at": "bad", "latest_version": "1.26.0"} + self.assertFalse(update_check.cache_is_fresh(state, now=1000.0)) + + +class TestUpdateCheckEnabled(unittest.TestCase): + def setUp(self): + self.opts = mock.Mock() + self.opts.output = "pretty" + env_patcher = mock.patch.dict(os.environ, {}, clear=False) + self.addCleanup(env_patcher.stop) + env_patcher.start() + os.environ.pop("CLOUDSMITH_NO_UPDATE_CHECK", None) + os.environ.pop("CI", None) + tty_patcher = mock.patch.object( + update_check, "stderr_is_tty", return_value=True + ) + self.addCleanup(tty_patcher.stop) + tty_patcher.start() + + def test_enabled_by_default(self): + self.assertTrue(update_check.update_check_enabled(self.opts, "status")) + + def test_disabled_by_env_var(self): + os.environ["CLOUDSMITH_NO_UPDATE_CHECK"] = "1" + self.assertFalse(update_check.update_check_enabled(self.opts, "status")) + + def test_disabled_in_ci(self): + os.environ["CI"] = "true" + self.assertFalse(update_check.update_check_enabled(self.opts, "status")) + + def test_disabled_without_tty(self): + update_check.stderr_is_tty.return_value = False + self.assertFalse(update_check.update_check_enabled(self.opts, "status")) + + def test_disabled_for_json_output(self): + self.opts.output = "json" + self.assertFalse(update_check.update_check_enabled(self.opts, "status")) + + def test_disabled_for_mcp_and_upgrade(self): + self.assertFalse(update_check.update_check_enabled(self.opts, "mcp")) + self.assertFalse(update_check.update_check_enabled(self.opts, "upgrade")) + + +class TestGetAvailableUpdate(unittest.TestCase): + def _state(self, latest): + return {"checked_at": 1000.0, "latest_version": latest} + + def setUp(self): + patcher = mock.patch.object(version, "get_version", return_value="1.25.0") + self.addCleanup(patcher.stop) + patcher.start() + + def test_returns_newer_version(self): + with mock.patch.object( + update_check, "read_cached_state", return_value=self._state("1.26.0") + ): + self.assertEqual("1.26.0", update_check.get_available_update()) + + def test_returns_none_when_current(self): + with mock.patch.object( + update_check, "read_cached_state", return_value=self._state("1.25.0") + ): + self.assertIsNone(update_check.get_available_update()) + + def test_returns_none_when_older(self): + with mock.patch.object( + update_check, "read_cached_state", return_value=self._state("1.24.0") + ): + self.assertIsNone(update_check.get_available_update()) + + def test_returns_none_when_invalid(self): + with mock.patch.object( + update_check, "read_cached_state", return_value=self._state("nonsense") + ): + self.assertIsNone(update_check.get_available_update()) + + def test_returns_none_when_cache_empty(self): + with mock.patch.object(update_check, "read_cached_state", return_value={}): + self.assertIsNone(update_check.get_available_update()) + + +class TestStartBackgroundCheck(unittest.TestCase): + def test_skips_when_cache_fresh(self): + state = {"checked_at": 1000.0, "latest_version": "1.26.0"} + with ( + mock.patch.object(update_check, "read_cached_state", return_value=state), + mock.patch("time.time", return_value=1000.0 + 60), + ): + self.assertIsNone(update_check.start_background_check_if_stale()) + + def test_fetches_and_caches_when_stale(self): + with ( + mock.patch.object(update_check, "read_cached_state", return_value={}), + mock.patch.object( + update_check, + "fetch_latest_manifest", + return_value={"version": "1.26.0"}, + ), + mock.patch.object(update_check, "write_cached_state") as write_mock, + ): + thread = update_check.start_background_check_if_stale() + self.assertIsNotNone(thread) + thread.join(timeout=5) + write_mock.assert_called_with("1.26.0") + + def test_stamps_state_before_the_fetch(self): + state = {"checked_at": 0.0, "latest_version": "1.24.0"} + with ( + mock.patch.object(update_check, "read_cached_state", return_value=state), + mock.patch.object( + update_check, + "fetch_latest_manifest", + return_value={"version": "1.26.0"}, + ), + mock.patch.object(update_check, "write_cached_state") as write_mock, + ): + thread = update_check.start_background_check_if_stale() + thread.join(timeout=5) + self.assertEqual(mock.call("1.24.0"), write_mock.call_args_list[0]) + + def test_swallows_fetch_errors(self): + with ( + mock.patch.object(update_check, "read_cached_state", return_value={}), + mock.patch.object( + update_check, + "fetch_latest_manifest", + side_effect=OSError("network down"), + ), + mock.patch.object(update_check, "write_cached_state") as write_mock, + ): + thread = update_check.start_background_check_if_stale() + thread.join(timeout=5) + write_mock.assert_called_once_with(None) + + def test_skips_when_state_unwritable(self): + with ( + mock.patch.object(update_check, "read_cached_state", return_value={}), + mock.patch.object( + update_check, "write_cached_state", side_effect=OSError("denied") + ), + ): + self.assertIsNone(update_check.start_background_check_if_stale()) + + +class TestPrintUpdateNotice(unittest.TestCase): + @mock.patch("cloudsmith_cli.core.update_check.click.secho") + def test_prints_notice_when_update_available(self, secho_mock): + with ( + mock.patch.object( + update_check, "get_available_update", return_value="1.26.0" + ), + mock.patch.object(version, "get_version", return_value="1.25.0"), + ): + update_check.print_update_notice_if_available() + output = " ".join(str(call) for call in secho_mock.call_args_list) + self.assertIn("1.25.0", output) + self.assertIn("1.26.0", output) + self.assertIn("cloudsmith upgrade", output) + for call in secho_mock.call_args_list: + self.assertTrue(call.kwargs.get("err")) + + @mock.patch("cloudsmith_cli.core.update_check.click.secho") + def test_silent_when_no_update(self, secho_mock): + with mock.patch.object(update_check, "get_available_update", return_value=None): + update_check.print_update_notice_if_available() + secho_mock.assert_not_called() + + +class TestStderrIsTty(unittest.TestCase): + def test_false_when_stderr_is_missing(self): + with mock.patch.object(update_check.sys, "stderr", None): + self.assertFalse(update_check.stderr_is_tty()) + + +class TestArm(unittest.TestCase): + def setUp(self): + self.opts = mock.Mock() + self.opts.output = "pretty" + env_patcher = mock.patch.dict(os.environ, {}, clear=False) + self.addCleanup(env_patcher.stop) + env_patcher.start() + os.environ.pop("CLOUDSMITH_NO_UPDATE_CHECK", None) + os.environ.pop("CI", None) + tty_patcher = mock.patch.object( + update_check, "stderr_is_tty", return_value=True + ) + self.addCleanup(tty_patcher.stop) + tty_patcher.start() + start_patcher = mock.patch.object( + update_check, "start_background_check_if_stale", return_value=None + ) + self.addCleanup(start_patcher.stop) + self.start_mock = start_patcher.start() + + def _context(self, invoked="status", inverse=None): + ctx = mock.Mock() + ctx.invoked_subcommand = invoked + ctx.command.inverse = inverse if inverse is not None else {} + return ctx + + def test_arms_for_regular_command(self): + ctx = self._context() + update_check.arm(ctx, self.opts) + self.start_mock.assert_called_once_with() + ctx.call_on_close.assert_called_once() + + def test_suppresses_alias_of_suppressed_command(self): + ctx = self._context(invoked="update", inverse={"update": "upgrade"}) + update_check.arm(ctx, self.opts) + self.start_mock.assert_not_called() + + def test_close_callback_respects_late_json_output(self): + ctx = self._context() + update_check.arm(ctx, self.opts) + close_callback = ctx.call_on_close.call_args[0][0] + self.opts.output = "json" + with mock.patch.object( + update_check, "print_update_notice_if_available" + ) as notice_mock: + close_callback() + notice_mock.assert_not_called() + + def test_close_callback_joins_background_thread(self): + thread = mock.Mock() + self.start_mock.return_value = thread + ctx = self._context() + update_check.arm(ctx, self.opts) + close_callback = ctx.call_on_close.call_args[0][0] + with mock.patch.object(update_check, "print_update_notice_if_available"): + close_callback() + thread.join.assert_called_once_with( + update_check.BACKGROUND_JOIN_TIMEOUT_SECONDS + ) diff --git a/cloudsmith_cli/core/update_check.py b/cloudsmith_cli/core/update_check.py new file mode 100644 index 00000000..1abf4c17 --- /dev/null +++ b/cloudsmith_cli/core/update_check.py @@ -0,0 +1,187 @@ +"""Check for new CLI versions against Cloudsmith-hosted release manifests.""" + +import json +import logging +import os +import sys +import threading +import time + +import click + +from ..cli.config import get_default_config_path +from . import version + +logger = logging.getLogger(__name__) + +MANIFEST_URL_TEMPLATE = ( + "https://dl.cloudsmith.io/public/cloudsmith/cli/raw/names/" + "cloudsmith-cli-manifest-{target}/versions/latest/manifest.txt" +) +VERSION_PROBE_TARGET = "linux-x86_64-gnu" +DEFAULT_INTERVAL_SECONDS = 24 * 60 * 60 +BACKGROUND_JOIN_TIMEOUT_SECONDS = 1.0 +CACHE_FILE_NAME = "update_check.json" +MACHINE_OUTPUT_FORMATS = ("json", "pretty_json") +SUPPRESSED_SUBCOMMANDS = frozenset(("mcp", "upgrade")) + + +def parse_manifest(text): + """Parse the key=value lines of a release manifest into a dict.""" + manifest = {} + for line in text.splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, _, value = line.partition("=") + manifest[key.strip()] = value.strip() + return manifest + + +def fetch_latest_manifest(target=None, timeout=5.0): + """Fetch and parse the latest release manifest for a build target.""" + from .session import create_requests_session + + url = MANIFEST_URL_TEMPLATE.format(target=target or VERSION_PROBE_TARGET) + response = create_requests_session().get(url, timeout=timeout) + response.raise_for_status() + manifest = parse_manifest(response.text) + if "version" not in manifest: + raise ValueError(f"no version field in manifest from {url}") + return manifest + + +def get_cache_file_path(): + """Return the path of the update check cache file.""" + return os.path.join(get_default_config_path(), CACHE_FILE_NAME) + + +def read_cached_state(): + """Read the cached update check state, or an empty dict.""" + try: + with open(get_cache_file_path(), encoding="utf-8") as cache_file: + state = json.load(cache_file) + except (OSError, ValueError): + return {} + return state if isinstance(state, dict) else {} + + +def write_cached_state(latest_version): + """Write the cached update check state atomically.""" + from .cache_utils import atomic_write_json + + path = get_cache_file_path() + os.makedirs(os.path.dirname(path), mode=0o700, exist_ok=True) + state = {"checked_at": time.time(), "latest_version": latest_version} + atomic_write_json(path, state) + + +def store_latest_version(latest_version): + """Write the cached update check state; ignore storage errors.""" + try: + write_cached_state(latest_version) + except OSError: + logger.debug("Failed to store the update check state", exc_info=True) + + +def cache_is_fresh(state, now=None, interval=DEFAULT_INTERVAL_SECONDS): + """Tell whether the cached state is younger than the check interval.""" + try: + checked_at = float(state["checked_at"]) + except (KeyError, TypeError, ValueError): + return False + now = time.time() if now is None else now + return (now - checked_at) < interval + + +def stderr_is_tty(): + """Tell whether stderr is attached to a terminal.""" + return sys.stderr is not None and sys.stderr.isatty() + + +def update_check_enabled(opts, invoked_subcommand): + """Tell whether the update check runs for this invocation.""" + env_value = os.environ.get("CLOUDSMITH_NO_UPDATE_CHECK", "").strip().lower() + if env_value in ("1", "true", "yes"): + return False + if os.environ.get("CI"): + return False + if not stderr_is_tty(): + return False + if opts.output in MACHINE_OUTPUT_FORMATS: + return False + return invoked_subcommand not in SUPPRESSED_SUBCOMMANDS + + +def get_available_update(): + """Return the cached latest version if it is newer, else None.""" + latest = read_cached_state().get("latest_version") + if not latest: + return None + try: + newer = version.parse_version(latest) > version.get_version_info() + except (TypeError, ValueError): + return None + return latest if newer else None + + +def start_background_check_if_stale(): + """Refresh the cached latest version in a daemon thread if stale.""" + state = read_cached_state() + if cache_is_fresh(state): + return None + try: + write_cached_state(state.get("latest_version")) + except OSError: + logger.debug("Cannot write the update check state", exc_info=True) + return None + + def refresh(): + try: + from . import installation + + manifest = fetch_latest_manifest(target=installation.detect_target()) + write_cached_state(manifest["version"]) + except (OSError, ValueError): + logger.debug("Update check failed", exc_info=True) + + thread = threading.Thread( + target=refresh, name="cloudsmith-update-check", daemon=True + ) + thread.start() + return thread + + +def print_update_notice_if_available(): + """Print an update notice on stderr when a newer version is cached.""" + latest = get_available_update() + if latest is None: + return + current = version.get_version() + click.secho( + f"\nA new version of the Cloudsmith CLI is available: {current} → {latest}", + fg="yellow", + err=True, + ) + click.secho("Run `cloudsmith upgrade` to get it.", fg="yellow", err=True) + + +def arm(ctx, opts): + """Start the update check and register the exit notice, if enabled.""" + invoked = ctx.invoked_subcommand + invoked = getattr(ctx.command, "inverse", {}).get(invoked, invoked) + if not update_check_enabled(opts, invoked): + return + thread = start_background_check_if_stale() + ctx.call_on_close(lambda: _finish_and_print_notice(opts, thread)) + + +def _finish_and_print_notice(opts, thread): + if opts.output in MACHINE_OUTPUT_FORMATS: + return + try: + if thread is not None: + thread.join(BACKGROUND_JOIN_TIMEOUT_SECONDS) + print_update_notice_if_available() + except OSError: + logger.debug("Update notice failed", exc_info=True)