From 33f0c74f755bc28731e35e4f4231ac18c8c31bdc Mon Sep 17 00:00:00 2001 From: Bartosz Blizniak Date: Mon, 26 Oct 2026 11:50:02 +0000 Subject: [PATCH] feat(ENG-15367): capture usage facts and send one event at the end of every run A hook on the root command records the command path, the flags and aliases used, where each setting came from, the login method and the API status, all from the Click context, so new commands and flags need no telemetry code. AliasGroup.main hands the run to the sender when it ends. Credential helper commands and login use the tracked command classes so their runs are recorded (or excluded) consistently. Grouped short options (-vS, -vPname) are split up as Click does, so a typed -S or -P always reaches the send rules. The reported env var is the first non-empty one, as Click uses, and a typed --help only counts as help when the run succeeded. Co-Authored-By: Claude Opus 5.5 --- cloudsmith_cli/cli/command.py | 90 +++- .../commands/credential_helper/__init__.py | 3 +- .../cli/commands/credential_helper/cargo.py | 3 +- .../cli/commands/credential_helper/docker.py | 3 +- .../cli/commands/credential_helper/generic.py | 3 +- .../cli/commands/credential_helper/manage.py | 7 +- .../cli/commands/credential_helper/pnpm.py | 3 +- .../commands/credential_helper/terraform.py | 3 +- cloudsmith_cli/cli/commands/login.py | 3 +- cloudsmith_cli/cli/config_sources.py | 91 ++++ cloudsmith_cli/cli/decorators.py | 5 +- cloudsmith_cli/cli/exceptions.py | 3 + .../cli/tests/test_config_sources.py | 72 +++ .../cli/tests/test_startup_imports.py | 2 + .../cli/tests/test_usage_capture.py | 306 ++++++++++++ cloudsmith_cli/cli/tests/test_usage_main.py | 445 ++++++++++++++++++ cloudsmith_cli/cli/usage_capture.py | 358 ++++++++++++++ cloudsmith_cli/core/tests/test_usage_event.py | 11 +- cloudsmith_cli/core/tests/test_usage_run.py | 25 + cloudsmith_cli/core/usage/event.py | 15 +- cloudsmith_cli/core/usage/run.py | 19 +- 21 files changed, 1454 insertions(+), 16 deletions(-) create mode 100644 cloudsmith_cli/cli/config_sources.py create mode 100644 cloudsmith_cli/cli/tests/test_config_sources.py create mode 100644 cloudsmith_cli/cli/tests/test_usage_capture.py create mode 100644 cloudsmith_cli/cli/tests/test_usage_main.py create mode 100644 cloudsmith_cli/cli/usage_capture.py diff --git a/cloudsmith_cli/cli/command.py b/cloudsmith_cli/cli/command.py index 87ddc813e..4e180be49 100644 --- a/cloudsmith_cli/cli/command.py +++ b/cloudsmith_cli/cli/command.py @@ -8,6 +8,7 @@ from click_didyoumean import DYMGroup from ..core.certificate_errors import describe_certificate_error +from .usage_capture import TrackedCommand, TrackingMixin, record_alias def _is_json_output_requested(exception): @@ -74,7 +75,7 @@ def _echo_certificate_hint(exception): click.echo(f"Hint: {click.style(hint, fg='yellow')}", err=True) -class AliasGroup(DYMGroup): +class AliasGroup(TrackingMixin, DYMGroup): """A command group with DYM, alias and lazy-subcommand support. ``lazy_commands`` maps a command name to the module that registers it. @@ -83,6 +84,8 @@ class AliasGroup(DYMGroup): ``lazy_aliases`` supplies the alias map for those commands up front. """ + command_class = TrackedCommand + def __init__(self, *args, **kwargs): lazy_commands = kwargs.pop("lazy_commands", None) lazy_aliases = kwargs.pop("lazy_aliases", None) @@ -96,6 +99,8 @@ def __init__(self, *args, **kwargs): self.inverse[alias] = name def resolve_command(self, ctx, args): + if args and args[0] in self.inverse: + record_alias(ctx, args[0]) try: return super().resolve_command(ctx, args) except click.exceptions.UsageError: @@ -106,6 +111,7 @@ def resolve_command(self, ctx, args): cmds = self.list_commands(ctx) matched = [cmd for cmd in cmds if cmd.startswith(cmd_name)] if len(matched) == 1 and len(cmd_name) > 1: + record_alias(ctx, "") args[0] = matched[0] return super().resolve_command(ctx, args) @@ -177,21 +183,60 @@ def format_commands(self, ctx, formatter): ctx.showing_help = True return super().format_commands(ctx, formatter) + def _main_shell_completion(self, ctx_args, prog_name, complete_var=None): + """Mark shell completion runs so usage telemetry ignores them. + + Click handles ``*_source`` before any context exists, so the + ``resilient_parsing`` check in ``_finish_usage`` never sees it. + """ + import os + + from . import usage_capture + + name = complete_var + if name is None: + # Same derivation as Click's own _main_shell_completion. + name = prog_name.replace("-", "_").replace(".", "_") + name = f"_{name}_COMPLETE".upper() + if os.environ.get(name): + usage_capture.mark_completion_run() + return super()._main_shell_completion(ctx_args, prog_name, complete_var) + def main(self, *args, **kwargs): """Override main to intercept exceptions and format as JSON if requested.""" import sys + import time + + from . import usage_capture original_standalone_mode = kwargs.get("standalone_mode", True) kwargs["standalone_mode"] = False + usage = {"outcome": "unexpected_error", "exit_code": 1} + depth = usage_capture.enter_run() + started = time.monotonic() try: - return super().main(*args, **kwargs) + rv = super().main(*args, **kwargs) + usage.update( + outcome="success" if not rv else "error_exit", + exit_code=(rv if isinstance(rv, int) else 1) if rv else 0, + ) + return rv except click.exceptions.Abort: + usage.update(outcome="aborted", exit_code=1) if not original_standalone_mode: raise click.echo("Aborted!", err=True) sys.exit(1) except click.exceptions.ClickException as e: + usage.update( + outcome=( + "usage_error" + if isinstance(e, click.exceptions.UsageError) + else "click_error" + ), + exit_code=e.exit_code, + ) if _is_json_output_requested(e): import json @@ -205,15 +250,24 @@ def main(self, *args, **kwargs): sys.exit(e.exit_code) except BrokenPipeError: # Click does not catch a failed write in its Ctrl+C handler. + usage.update(outcome="aborted", exit_code=1) if not original_standalone_mode: raise sys.stdout = _PacifyFlushWrapper(sys.stdout) sys.stderr = _PacifyFlushWrapper(sys.stderr) sys.exit(1) + except SystemExit as e: + code = e.code + usage.update( + exit_code=code if isinstance(code, int) else (0 if code is None else 1), + outcome="success" if code in (0, None) else "error_exit", + ) + raise except Exception as e: # pylint: disable=broad-exception-caught # Anything reaching here is unexpected: click's own errors and # handled API errors (which exit via ctx.exit) are dealt with # above or never get this far. + usage.update(outcome="unexpected_error", exit_code=1) if not original_standalone_mode: raise main_args = kwargs.get("args", args[0] if args else None) @@ -228,3 +282,35 @@ def main(self, *args, **kwargs): click.echo(f"Error: {telemetry.format_exception_summary(e)}", err=True) _echo_certificate_hint(e) sys.exit(1) + finally: + _finish_usage(usage, started, nested=depth > 1) + usage_capture.exit_run() + + +def _finish_usage(usage, started, *, nested): + """Hand the run to usage telemetry; never raises, never prints.""" + try: + import time + + from ..core.usage import run + from . import usage_capture + + meta = usage_capture.current_root_meta() or {} + contexts = meta.get(usage_capture.META_KEY, {}).get("contexts", []) + # Shell completion parses with resilient_parsing (or, for the + # *_source script, not at all); never report it. + completing = usage_capture.is_completion_run() or any( + getattr(c, "resilient_parsing", False) for c in contexts + ) + facts = usage_capture.collect_facts( + meta, + outcome=usage["outcome"], + exit_code=usage["exit_code"], + duration_ms=int((time.monotonic() - started) * 1000), + nested=nested or completing, + ) + if facts.api_status is not None and facts.outcome == "error_exit": + facts.outcome = "api_error" + run.finish(facts) + except BaseException: # pylint: disable=broad-exception-caught + return diff --git a/cloudsmith_cli/cli/commands/credential_helper/__init__.py b/cloudsmith_cli/cli/commands/credential_helper/__init__.py index bcf76528e..93f8038fe 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/__init__.py +++ b/cloudsmith_cli/cli/commands/credential_helper/__init__.py @@ -8,6 +8,7 @@ import click +from ...usage_capture import TrackedGroup from ..main import main from .cargo import cargo as cargo_cmd from .docker import docker as docker_cmd @@ -17,7 +18,7 @@ from .terraform import terraform as terraform_cmd -@click.group() +@click.group(cls=TrackedGroup) def credential_helper(): """ Credential helpers for package managers. diff --git a/cloudsmith_cli/cli/commands/credential_helper/cargo.py b/cloudsmith_cli/cli/commands/credential_helper/cargo.py index f52b8a8c1..74452eeea 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/cargo.py +++ b/cloudsmith_cli/cli/commands/credential_helper/cargo.py @@ -13,9 +13,10 @@ from ....credential_helpers.cargo import execute from ...decorators import common_api_auth_options, resolve_credentials +from ...usage_capture import TrackedCommand -@click.command(context_settings={"ignore_unknown_options": True}) +@click.command(cls=TrackedCommand, context_settings={"ignore_unknown_options": True}) @click.option( "--cargo-plugin", is_flag=True, diff --git a/cloudsmith_cli/cli/commands/credential_helper/docker.py b/cloudsmith_cli/cli/commands/credential_helper/docker.py index ad8345af3..60b21e619 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/docker.py +++ b/cloudsmith_cli/cli/commands/credential_helper/docker.py @@ -13,9 +13,10 @@ from ....credential_helpers.docker import execute from ...decorators import common_api_auth_options, resolve_credentials +from ...usage_capture import TrackedCommand -@click.command() +@click.command(cls=TrackedCommand) @click.argument("operation", required=False, default="get") @common_api_auth_options @resolve_credentials diff --git a/cloudsmith_cli/cli/commands/credential_helper/generic.py b/cloudsmith_cli/cli/commands/credential_helper/generic.py index db98b17da..c01722750 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/generic.py +++ b/cloudsmith_cli/cli/commands/credential_helper/generic.py @@ -15,9 +15,10 @@ common_cli_config_options, resolve_credentials, ) +from ...usage_capture import TrackedCommand -@click.command() +@click.command(cls=TrackedCommand) @common_cli_config_options @common_api_auth_options @resolve_credentials diff --git a/cloudsmith_cli/cli/commands/credential_helper/manage.py b/cloudsmith_cli/cli/commands/credential_helper/manage.py index 344757537..0ceb6388b 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/manage.py +++ b/cloudsmith_cli/cli/commands/credential_helper/manage.py @@ -37,6 +37,7 @@ common_cli_output_options, resolve_credentials, ) +from ...usage_capture import TrackedCommand # --------------------------------------------------------------------------- # Helper registry — extend here when new helpers are added @@ -248,7 +249,7 @@ def _terraform_workspace_next_steps(helper_args: tuple[str, ...]) -> list[str]: # --------------------------------------------------------------------------- -@click.command("install") +@click.command("install", cls=TrackedCommand) @click.argument("helper") @click.option( "--bin-dir", @@ -420,7 +421,7 @@ def install_cmd( # --------------------------------------------------------------------------- -@click.command("uninstall") +@click.command("uninstall", cls=TrackedCommand) @click.argument("helper") @click.option( "--bin-dir", @@ -487,7 +488,7 @@ def uninstall_cmd(ctx, opts, helper: str, bin_dir: str | None, dry_run: bool) -> # --------------------------------------------------------------------------- -@click.command("list") +@click.command("list", cls=TrackedCommand) @common_cli_config_options @common_cli_output_options @click.pass_context diff --git a/cloudsmith_cli/cli/commands/credential_helper/pnpm.py b/cloudsmith_cli/cli/commands/credential_helper/pnpm.py index 603f5e5c3..39e10963a 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/pnpm.py +++ b/cloudsmith_cli/cli/commands/credential_helper/pnpm.py @@ -13,9 +13,10 @@ from ....credential_helpers.pnpm import execute from ...config import Options from ...decorators import common_api_auth_options, resolve_credentials +from ...usage_capture import TrackedCommand -@click.command() +@click.command(cls=TrackedCommand) @click.argument("repo", required=False, default="npm.cloudsmith.io") @common_api_auth_options @resolve_credentials diff --git a/cloudsmith_cli/cli/commands/credential_helper/terraform.py b/cloudsmith_cli/cli/commands/credential_helper/terraform.py index c49008f69..c1f75dfaf 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/terraform.py +++ b/cloudsmith_cli/cli/commands/credential_helper/terraform.py @@ -19,9 +19,10 @@ common_cli_config_options, resolve_credentials, ) +from ...usage_capture import TrackedCommand -@click.command(context_settings={"ignore_unknown_options": True}) +@click.command(cls=TrackedCommand, context_settings={"ignore_unknown_options": True}) @click.option( "-r", "--repo", diff --git a/cloudsmith_cli/cli/commands/login.py b/cloudsmith_cli/cli/commands/login.py index d1654a4e4..eac4dedd2 100644 --- a/cloudsmith_cli/cli/commands/login.py +++ b/cloudsmith_cli/cli/commands/login.py @@ -8,6 +8,7 @@ from ...core.config import create_config_files, new_config_messaging from .. import decorators, deprecation, utils from ..exceptions import handle_api_exceptions +from ..usage_capture import TrackedCommand from ..utils import maybe_spinner from .main import main @@ -61,7 +62,7 @@ def summarise_removed_login(exc, detail, fields): return None -class DeprecatedLoginCommand(click.Command): +class DeprecatedLoginCommand(TrackedCommand): """Command that prints the deprecation warning before any prompt.""" def parse_args(self, ctx, args): diff --git a/cloudsmith_cli/cli/config_sources.py b/cloudsmith_cli/cli/config_sources.py new file mode 100644 index 000000000..ccc0f4eb0 --- /dev/null +++ b/cloudsmith_cli/cli/config_sources.py @@ -0,0 +1,91 @@ +# Copyright 2026 Cloudsmith Ltd +"""Which settings were set in config.ini / credentials.ini, by section. + +``ConfigReader.read_config()`` fills schema defaults, so it cannot tell an +explicit key from a default; this reads the raw files and reports key +presence only. A value is only checked for being empty, and never kept. +""" + +import configparser +import os + +from ..core.usage.event import SETTING_ALIASES +from . import config + + +def _files(reader, explicit): + paths = [explicit] if explicit else [] + paths.extend( + name if os.path.isabs(name) else os.path.join(searchpath, name) + for searchpath in reader.config_searchpath + for name in reader.config_files + ) + return [p for p in paths if p and os.path.isfile(p)] + + +def _declared(schema): + """Key names the CLI reads for a section schema (never user-authored keys).""" + return { + param.name + for param in vars(schema).values() + if isinstance(param, config.ConfigParam) + } + + +def _keys(paths, section, declared): + keys = set() + for path in paths: + parser = configparser.RawConfigParser() + try: + parser.read(path, encoding="utf-8") + except (configparser.Error, UnicodeDecodeError, OSError): + continue + if parser.has_section(section): + # An empty key (`api_host=` in the default config.ini) sets nothing. + keys.update( + key + for key in parser.options(section) + if parser.get(section, key).strip() + ) + return keys & declared + + +def _present(config_file, creds_file, profile): + """Yield ``(key, source)`` for each schema key set in the ini files.""" + for reader, schema, explicit, base in ( + (config.ConfigReader, config.ConfigSchema.Default, config_file, "config"), + ( + config.CredentialsReader, + config.CredentialsSchema.Default, + creds_file, + "credentials", + ), + ): + paths = _files(reader, explicit) + declared = _declared(schema) + for key in _keys(paths, "default", declared): + yield key, f"{base}_file" + if profile and profile != "default": + for key in _keys(paths, f"profile:{profile}", declared): + yield key, f"{base}_profile" + + +def detect( + *, config_file: str | None, creds_file: str | None, profile: str | None +) -> dict[str, str]: + """Map each setting present in the ini files to where it was set.""" + found: dict[str, str] = {} + for key, source in _present(config_file, creds_file, profile): + found[SETTING_ALIASES.get(key, key)] = source + return found + + +def keys( + *, config_file: str | None, creds_file: str | None, profile: str | None +) -> list[str]: + """The key names set in the ini files, as written: `org`, not `workspace`. + + Shows which spelling of a renamed key is still in use. Only names the + schema declares, never values. + """ + return sorted({key for key, _ in _present(config_file, creds_file, profile)}) diff --git a/cloudsmith_cli/cli/decorators.py b/cloudsmith_cli/cli/decorators.py index 8f7904583..78c5c1de6 100644 --- a/cloudsmith_cli/cli/decorators.py +++ b/cloudsmith_cli/cli/decorators.py @@ -19,7 +19,7 @@ disabled_detectors_from_env, registered_detectors, ) -from . import config, deprecation, utils +from . import config, deprecation, usage_capture, utils def _expired_saml_renewal_message(error): @@ -242,6 +242,7 @@ def wrapper(ctx, *args, **kwargs): opts.load_config_file(path=config_file, profile=profile) opts.load_creds_file(path=creds_file, profile=profile) + usage_capture.record_config(ctx) opts.api_key_from_file = opts.api_key _warn_deprecated_config_keys(ctx, opts) kwargs["opts"] = opts @@ -627,6 +628,7 @@ def wrapper(ctx, *args, **kwargs): deprecation.warn_envvar(legacy_envvar, "CLOUDSMITH_SSL_VERIFY=0") ctx.params["ssl_verify"] = ssl_verify opts.ssl_verify = ssl_verify + usage_capture.record_session(ctx) opts.ssl_verify_strict = kwargs.pop("ssl_verify_strict") opts.api_user_agent = kwargs.pop("api_user_agent") opts.api_headers = kwargs.pop("api_headers") @@ -812,6 +814,7 @@ def wrapper(ctx, *args, **kwargs): click.secho(message, fg="yellow", err=True) opts.credential = credential + usage_capture.record_credential(ctx, credential) kwargs["opts"] = opts return ctx.invoke(f, *args, **kwargs) diff --git a/cloudsmith_cli/cli/exceptions.py b/cloudsmith_cli/cli/exceptions.py index 5fcfea284..21b8b574c 100644 --- a/cloudsmith_cli/cli/exceptions.py +++ b/cloudsmith_cli/cli/exceptions.py @@ -116,6 +116,9 @@ def handle_api_exceptions( raise if exit_on_error: + from .usage_capture import record_api_status + + record_api_status(ctx, exc.status) ctx.exit(exc.status or 1) diff --git a/cloudsmith_cli/cli/tests/test_config_sources.py b/cloudsmith_cli/cli/tests/test_config_sources.py new file mode 100644 index 000000000..39ff04335 --- /dev/null +++ b/cloudsmith_cli/cli/tests/test_config_sources.py @@ -0,0 +1,72 @@ +from cloudsmith_cli.cli import config, config_sources + + +def _point_readers(monkeypatch, path): + for reader in (config.ConfigReader, config.CredentialsReader): + monkeypatch.setattr(reader, "config_searchpath", [str(path)]) + monkeypatch.setattr(config.ConfigReader, "config_files", ["config.ini"]) + monkeypatch.setattr(config.CredentialsReader, "config_files", ["credentials.ini"]) + + +def test_detects_default_and_profile_keys_not_defaults(monkeypatch, tmp_path): + (tmp_path / "config.ini").write_text( + "[default]\nworkspace = acme\n\n[profile:ci]\ncheck_for_update = false\n" + ) + (tmp_path / "credentials.ini").write_text("[profile:ci]\napi_key = not-real\n") + _point_readers(monkeypatch, tmp_path) + found = config_sources.detect(config_file=None, creds_file=None, profile="ci") + assert found == { + "workspace": "config_file", + "check_for_update": "config_profile", + "api_key": "credentials_profile", + } + assert "telemetry" not in found and "ssl_verify" not in found + + +def test_reports_only_schema_declared_keys_for_each_file(monkeypatch, tmp_path): + (tmp_path / "config.ini").write_text( + "[default]\nworkspace = acme\nmy_secret_token = x\n" + ) + (tmp_path / "credentials.ini").write_text( + "[default]\napi_key = not-real\nworkspace = other\n" + ) + _point_readers(monkeypatch, tmp_path) + found = config_sources.detect(config_file=None, creds_file=None, profile=None) + assert found == {"workspace": "config_file", "api_key": "credentials_file"} + + +def test_unreadable_file_is_ignored(monkeypatch, tmp_path): + (tmp_path / "config.ini").write_text("[default\nbroken") + _point_readers(monkeypatch, tmp_path) + assert not config_sources.detect(config_file=None, creds_file=None, profile=None) + + +def test_empty_keys_of_the_default_config_set_nothing(monkeypatch, tmp_path): + # `cloudsmith auth` writes the packaged default config.ini, whose keys are + # mostly empty (`api_host=`); they must not count as set in the file. + from importlib import resources + + template = resources.files("cloudsmith_cli.data").joinpath("config.ini") + (tmp_path / "config.ini").write_text(template.read_text()) + _point_readers(monkeypatch, tmp_path) + found = config_sources.detect(config_file=None, creds_file=None, profile=None) + assert found == {"ssl_verify_strict": "config_file"} + + +def test_keys_are_the_names_as_written_in_default_and_profile(monkeypatch, tmp_path): + (tmp_path / "config.ini").write_text( + "[default]\norg = acme\napi_host =\nmy_secret_token = x\n\n" + "[profile:ci]\napi_proxy = http://proxy.example:3128\n" + "[profile:other]\nworkspace = other\n" + ) + (tmp_path / "credentials.ini").write_text("[profile:ci]\napi_key = not-real\n") + _point_readers(monkeypatch, tmp_path) + # The renamed key keeps its old name; empty and undeclared keys are left out. + assert config_sources.keys(config_file=None, creds_file=None, profile="ci") == [ + "api_key", + "api_proxy", + "org", + ] + assert config_sources.keys(config_file=None, creds_file=None, profile=None) == [ + "org" + ] diff --git a/cloudsmith_cli/cli/tests/test_startup_imports.py b/cloudsmith_cli/cli/tests/test_startup_imports.py index f89cf2658..19d86fb71 100644 --- a/cloudsmith_cli/cli/tests/test_startup_imports.py +++ b/cloudsmith_cli/cli/tests/test_startup_imports.py @@ -20,6 +20,8 @@ "ssl", # Error reporting imports it on the error path only. "sentry_sdk", + # Usage telemetry loads only at the end of a run. + "cloudsmith_cli.core.usage", ) diff --git a/cloudsmith_cli/cli/tests/test_usage_capture.py b/cloudsmith_cli/cli/tests/test_usage_capture.py new file mode 100644 index 000000000..fbe5cc5dc --- /dev/null +++ b/cloudsmith_cli/cli/tests/test_usage_capture.py @@ -0,0 +1,306 @@ +from types import SimpleNamespace + +import click +import pytest +from click.testing import CliRunner + +from cloudsmith_cli.cli import usage_capture +from cloudsmith_cli.cli.command import AliasGroup + +CAPTURED = {} + + +@click.group(cls=AliasGroup) +def app(): + pass + + +@app.command(aliases=["ls"]) +@click.option("-w", "--workspace", "--org", "workspace", envvar="CLOUDSMITH_WORKSPACE") +@click.option("-F", "--output-format", "output") +@click.argument("name", required=False) +@click.pass_context +def listing(ctx, workspace, output, name): + # Collect inside the run: env vars and contexts are live here. + CAPTURED["facts"] = usage_capture.collect_facts( + ctx.meta, outcome="success", exit_code=0, duration_ms=5, nested=False + ) + + +@pytest.fixture(autouse=True) +def isolated_config(telemetry_config_dir): # pylint: disable=unused-argument + CAPTURED.clear() + + +def _run(args, env=None): + CliRunner().invoke(app, args, env=env or {}, catch_exceptions=False) + return CAPTURED["facts"] + + +def test_typed_spellings_not_values(): + tokens = ["-w", "acme", "-Fjson", "--org=x", "--", "-w"] + assert usage_capture.match_typed_flags(listing.params, tokens) == [ + "--org", + "-F", + "-w", + ] + + +@click.command() +@click.option("-v", "--verbose", is_flag=True) +@click.option("-c", "--count", count=True) +@click.option("-S", "--without-api-ssl-verify", is_flag=True) +@click.option("-P", "--profile") +def clustered(verbose, count, without_api_ssl_verify, profile): + pass + + +@pytest.mark.parametrize( + "token, expected", + [ + # Flags in a cluster are all seen, as Click applies them all. + ("-vS", ["-S", "-v"]), + ("-cvc", ["-c", "-v"]), + # An option that takes a value consumes the rest of the token. + ("-vPprivate", ["-P", "-v"]), + ("-PvS", ["-P"]), + # An undeclared character stops the expansion (Click rejects it). + ("-vxS", ["-v"]), + ], +) +def test_short_option_clusters_are_expanded(token, expected): + assert usage_capture.match_typed_flags(clustered.params, [token]) == expected + + +def test_clustered_profile_and_ssl_off_reach_the_gates(): + typed = usage_capture.match_typed_flags(clustered.params, ["-vPprivate", "-vS"]) + assert usage_capture.CONFIG_SPELLINGS & set(typed) + assert usage_capture.resolve_ssl_verify(None, typed) is False + + +def test_facts_from_a_run(): + facts = _run(["ls", "-w", "secret-ws", "-F", "json", "pkg-name"]) + assert facts.command == "listing" + assert facts.aliases_used == ["ls"] + assert set(facts.flags) == {"workspace", "output"} + assert facts.flags_typed == ["-F", "-w"] + assert "name" not in facts.setting_sources # arguments are not settings + assert "secret-ws" not in repr(facts) and "pkg-name" not in repr(facts) + + +def test_env_sources_report_names_only(): + facts = _run(["listing"], env={"CLOUDSMITH_WORKSPACE": "secret-ws"}) + assert facts.envvars == ["CLOUDSMITH_WORKSPACE"] + assert facts.setting_sources.get("workspace") == "env" + assert "secret-ws" not in repr(facts) + + +@app.command() +@click.option("--proxy", envvar=["CLOUDSMITH_PROXY", "CLOUDSMITH_API_PROXY"]) +@click.pass_context +def proxied(ctx, proxy): + CAPTURED["facts"] = usage_capture.collect_facts( + ctx.meta, outcome="success", exit_code=0, duration_ms=5, nested=False + ) + + +def test_an_empty_env_var_is_not_reported_as_the_source(): + # Click skips the empty CLOUDSMITH_PROXY and uses CLOUDSMITH_API_PROXY. + facts = _run( + ["proxied"], env={"CLOUDSMITH_PROXY": "", "CLOUDSMITH_API_PROXY": "secret"} + ) + assert facts.envvars == ["CLOUDSMITH_API_PROXY"] + assert "secret" not in repr(facts) + + +def _facts_with_typed(typed, outcome, exit_code): + meta = {usage_capture.META_KEY: {"contexts": [], "typed": typed, "aliases": []}} + return usage_capture.collect_facts( + meta, outcome=outcome, exit_code=exit_code, duration_ms=1, nested=False + ) + + +def test_typed_help_after_a_successful_exit_is_help(): + assert _facts_with_typed(["--help"], "success", 0).outcome == "help" + + +@pytest.mark.parametrize("outcome", ["usage_error", "click_error", "api_error"]) +def test_typed_help_never_hides_a_failure(outcome): + # e.g. `whoami --no-such-flag --help` fails parsing before help is shown. + assert _facts_with_typed(["--help"], outcome, 2).outcome == outcome + + +def test_help_spelling_is_seen(): + tokens = ["--help"] + params = listing.get_params(click.Context(listing)) + assert usage_capture.match_typed_flags(params, tokens) == ["--help"] + + +def test_every_command_in_the_cli_is_tracked(): + import importlib + import pkgutil + + from cloudsmith_cli.cli import commands + from cloudsmith_cli.cli.commands.main import main + + for info in pkgutil.walk_packages( + commands.__path__, prefix=commands.__name__ + "." + ): + importlib.import_module(info.name) + + def walk(cmd, path): + assert isinstance(cmd, usage_capture.TrackingMixin), " ".join(path) + for name, sub in getattr(cmd, "commands", {}).items(): + walk(sub, path + [name]) + + walk(main, []) + + +def test_collect_facts_does_not_read_config_files(monkeypatch): + from cloudsmith_cli.cli import config_sources + + def explode(**_): + raise AssertionError("config files read before the telemetry gates") + + monkeypatch.setattr(config_sources, "detect", explode) + facts = _run(["listing"], env={"CLOUDSMITH_WORKSPACE": "x"}) + assert facts.setting_sources == {"workspace": "env"} + + +def test_deferred_sources_return_the_config_file_sources(monkeypatch): + from cloudsmith_cli.cli import config_sources + + monkeypatch.setattr( + config_sources, "detect", lambda **_: {"workspace": "config_file"} + ) + facts = _run(["listing"]) + assert facts.deferred_sources() == {"workspace": "config_file"} + + +def test_a_parse_outside_a_run_pins_nothing(): + assert usage_capture._RUN["depth"] == 0 # pylint: disable=protected-access + CliRunner().invoke(app, ["listing"], catch_exceptions=False) + assert usage_capture.current_root_meta() is None + + +def test_session_env_names_match_the_option_declarations(): + from cloudsmith_cli.cli.commands import list_ + + envvars = {p.name: p.envvar for p in list_.repos.params} + assert envvars["api_host"] == usage_capture.API_HOST_ENV + assert envvars["ssl_verify"] == usage_capture.SSL_VERIFY_ENV + assert ( + envvars["deprecated_without_api_ssl_verify"] + == usage_capture.WITHOUT_SSL_VERIFY_ENV + ) + spellings = {s for p in list_.repos.params for s in p.opts + p.secondary_opts} + assert usage_capture.API_HOST_SPELLINGS <= spellings + assert usage_capture.SSL_OFF_SPELLINGS <= spellings + + +def test_session_resolved_only_for_the_leaf_or_its_descendants(): + root = click.Context(click.Group("root")) + group = click.Context(click.Group("list"), parent=root) + leaf = click.Context(click.Command("repos"), parent=group) + forwarded = click.Context(click.Command("get"), parent=leaf) + unrelated = click.Context(click.Command("other"), parent=root) + contexts = [root, group, leaf] + + def resolved(session): + store = {"contexts": contexts, "session": session} + return usage_capture._session_resolved(store) # pylint: disable=protected-access + + assert resolved(leaf) is True + assert resolved(forwarded) is True # ctx.forward / ctx.invoke sub-context + assert resolved(group) is False # set up before the leaf parsed its flags + assert resolved(root) is False + assert resolved(unrelated) is False + assert resolved(None) is False + assert usage_capture._session_resolved({"session": leaf}) is False # pylint: disable=protected-access + + +def test_config_resolved_only_for_the_leaf_or_its_descendants(): + root = click.Context(click.Group("root")) + group = click.Context(click.Group("list"), parent=root) + leaf = click.Context(click.Command("repos"), parent=group) + forwarded = click.Context(click.Command("get"), parent=leaf) + contexts = [root, group, leaf] + + def resolved(config_ctx): + store = {"contexts": contexts, "config": config_ctx} + return usage_capture._config_resolved(store) # pylint: disable=protected-access + + assert resolved(leaf) is True + assert resolved(forwarded) is True + assert resolved(group) is False # loaded before the leaf parsed its flags + assert resolved(root) is False + assert resolved(None) is False + + +def test_config_spellings_match_the_option_declarations(): + from cloudsmith_cli.cli.commands import list_ + + spellings = { + s + for p in list_.repos.params + if p.name in ("config_file", "profile") + for s in p.opts + p.secondary_opts + } + assert usage_capture.CONFIG_SPELLINGS == spellings + + +#: Options of the shared decorators that are not settings, so they have no +#: ``source_``. Every other option must alias onto ALLOWED_SETTINGS. +NOT_SETTINGS = frozenset( + { + "debug", # per-run output switches + "verbose", + "no_color_env", # env plumbing for colour; the names land in envvars + "force_color_env", + "term_env", + "allowed_api_host_suffixes", # hidden, internal trust lists + "allowed_proxy_suffixes", + "deprecated_allowed_api_proxy_suffixes", + "oidc_discovery_disabled", # no config.ini key + "without_rate_limit", # API client tuning, no config.ini key + "rate_limit_warning", + "error_retry_max", + "error_retry_backoff", + "error_retry_codes", + } +) + + +@pytest.mark.parametrize( + "name", + [ + "common_cli_config_options", + "common_cli_output_options", + "common_api_auth_options", + "initialise_session", + "resolve_credentials", # holds -w/--workspace (``org``) + "initialise_api", + ], +) +def test_every_shared_option_is_a_setting_or_explicitly_not(name): + from cloudsmith_cli.cli import decorators + from cloudsmith_cli.core.usage import event + + decorator = getattr(decorators, name) + command = click.command()(decorator(lambda **kwargs: None)) + for param in command.params: + setting = event.SETTING_ALIASES.get(param.name, param.name) + assert setting in event.ALLOWED_SETTINGS or param.name in NOT_SETTINGS, ( + param.name + ) + + +def test_record_credential_keeps_the_last_resolved_source(): + store_ctx = click.Context(click.Command("x")) + usage_capture.record_credential(store_ctx, SimpleNamespace(source_name="env_var")) + usage_capture.record_credential(store_ctx, None) + assert store_ctx.meta[usage_capture.META_KEY]["auth_source"] == "env_var" + usage_capture.record_credential( + store_ctx, SimpleNamespace(source_name="credentials_file") + ) + assert store_ctx.meta[usage_capture.META_KEY]["auth_source"] == "credentials_file" diff --git a/cloudsmith_cli/cli/tests/test_usage_main.py b/cloudsmith_cli/cli/tests/test_usage_main.py new file mode 100644 index 000000000..d4cfb7155 --- /dev/null +++ b/cloudsmith_cli/cli/tests/test_usage_main.py @@ -0,0 +1,445 @@ +import json + +import pytest + +from cloudsmith_cli.cli.commands.main import main +from cloudsmith_cli.core.usage import notice, state + + +@pytest.fixture() +def notice_seen(): + state.write({"notice_version": notice.NOTICE_VERSION}) + + +def test_version_sends_success(usage_events, notice_seen): + assert main(args=["--version"], standalone_mode=False) in (0, None) + assert usage_events[-1]["properties"]["outcome"] == "success" + assert usage_events[-1]["properties"]["command"] == "" + + +def test_help_is_recorded_as_help(usage_events, notice_seen): + # Click returns 0 for --help when standalone_mode is False. + assert main(args=["repositories", "--help"], standalone_mode=False) == 0 + props = usage_events[-1]["properties"] + assert props["outcome"] == "help" and props["command"] == "repositories" + + +def test_usage_error_keeps_exit_code(usage_events, notice_seen): + with pytest.raises(SystemExit) as exit_info: + main(args=["repositories", "--no-such-flag"]) + assert exit_info.value.code == 2 + assert usage_events[-1]["properties"]["outcome"] == "usage_error" + + +def test_credential_helper_sends_nothing(usage_events, notice_seen): + with pytest.raises(SystemExit): + main(args=["credential-helper", "docker", "list"]) + assert usage_events == [] + + +def test_finish_failure_does_not_change_exit(usage_events, notice_seen, monkeypatch): + from cloudsmith_cli.core.usage import run + + monkeypatch.setattr(run, "_finish", lambda facts: 1 / 0) + assert main(args=["--version"], standalone_mode=False) in (0, None) + + +@pytest.mark.parametrize( + "env, args", + [ + ({"CLOUDSMITH_API_HOST": "https://artifacts.example.com"}, ["--version"]), + ({"CLOUDSMITH_SSL_VERIFY": "false"}, ["--version"]), + ({"CLOUDSMITH_SSL_VERIFY": "Off"}, ["--version"]), + ({"CLOUDSMITH_WITHOUT_API_SSL_VERIFY": "1"}, ["--version"]), + # Not `list repos`: test_deprecation runs that command object as a root + # command, and Click then caches its help option without `-h`. + ({}, ["repos", "get", "--api-host", "https://artifacts.example.com", "-h"]), + ( + {}, + ["list", "repos", "--api-host", "https://artifacts.example.com", "--help"], + ), + ({}, ["list", "repos", "--without-ssl-verify", "--help"]), + ({}, ["list", "repos", "-S", "--help"]), + ], +) +def test_host_and_ssl_gates_hold_before_session_setup( + usage_events, notice_seen, monkeypatch, env, args +): + # None of these runs reaches initialise_session, which is where the + # options normally land in opts. + for name, value in env.items(): + monkeypatch.setenv(name, value) + assert main(args=args, standalone_mode=False) in (0, None) + assert usage_events == [] + + +def test_cloudsmith_api_host_from_env_still_sends( + usage_events, notice_seen, monkeypatch +): + monkeypatch.setenv("CLOUDSMITH_API_HOST", "https://api.cloudsmith.io") + monkeypatch.setenv("CLOUDSMITH_SSL_VERIFY", "true") + assert main(args=["--version"], standalone_mode=False) in (0, None) + assert len(usage_events) == 1 + + +def test_empty_ssl_env_does_not_turn_ssl_off(usage_events, notice_seen, monkeypatch): + # Click treats an empty env var as unset. + monkeypatch.setenv("CLOUDSMITH_SSL_VERIFY", "") + monkeypatch.setenv("CLOUDSMITH_WITHOUT_API_SSL_VERIFY", "") + assert main(args=["--version"], standalone_mode=False) in (0, None) + assert len(usage_events) == 1 + + +@pytest.mark.parametrize( + "instruction, env", + [ + ("zsh_source", {}), + ("bash_complete", {"COMP_WORDS": "cloudsmith ", "COMP_CWORD": "1"}), + ], +) +def test_shell_completion_sends_nothing( + usage_events, notice_seen, monkeypatch, capsys, instruction, env +): + monkeypatch.setenv("_CLOUDSMITH_COMPLETE", instruction) + for name, value in env.items(): + monkeypatch.setenv(name, value) + with pytest.raises(SystemExit) as exit_info: + main(args=[], prog_name="cloudsmith") + assert exit_info.value.code == 0 + assert capsys.readouterr().out # the script or the completions + assert usage_events == [] + + +@pytest.fixture() +def reporting_enabled(sentry_events, monkeypatch): # pylint: disable=unused-argument + """Every opt-out cleared, as error reporting tests do (no usage_events).""" + for name in ( + "CLOUDSMITH_API_HOST", + "CLOUDSMITH_SSL_VERIFY", + "CLOUDSMITH_WITHOUT_API_SSL_VERIFY", + ): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("CLOUDSMITH_NO_UPDATE_CHECK", "1") + + +def test_exported_key_never_reaches_tests( + reporting_enabled, notice_seen, blocked_usage_posts +): + # Discriminates when CLOUDSMITH_TELEMETRY_POSTHOG_KEY is exported: the + # autouse guard removes it, so the run stops before any POST. + assert main(args=["--version"], standalone_mode=False) in (0, None) + assert blocked_usage_posts == [] + + +def test_a_test_that_sets_the_key_still_posts_nothing( + reporting_enabled, notice_seen, monkeypatch, blocked_usage_posts +): + from cloudsmith_cli.core.usage import settings + + monkeypatch.setenv(settings.KEY_ENV, "phc_test") + assert main(args=["--version"], standalone_mode=False) in (0, None) + # The run reached the POST, and the guard took it instead of the network. + assert len(blocked_usage_posts) == 1 + blocked_usage_posts.clear() # expected: don't fail at teardown + + +def test_guarded_post_raises(blocked_usage_posts): + from cloudsmith_cli.core.usage import sender + + with pytest.raises(AssertionError, match="must not send usage telemetry"): + sender._post(None, "https://eu.i.posthog.com/i/v0/e/", "{}", 1.0) + assert len(blocked_usage_posts) == 1 + blocked_usage_posts.clear() + + +@pytest.fixture() +def no_repos_api(monkeypatch): + """Answer ``list repos`` with an empty page instead of calling the API.""" + from cloudsmith_cli.core.api import repos + from cloudsmith_cli.core.pagination import PageInfo + + calls = [] + + def list_repos(**kwargs): + calls.append(kwargs) + return [], PageInfo() + + monkeypatch.setattr(repos, "list_repos", list_repos) + return calls + + +INSTALL_ID = "5b0d9c1e-6a2f-4d0e-9b7a-3f1c2e8d4a10" + + +def test_install_id_never_reaches_sentry( + usage_events, sentry_events, notice_seen, monkeypatch +): + from cloudsmith_cli.core.api import repos + + state.write({"install_id": INSTALL_ID, "created_at": 1792076242}) + + def crash(**kwargs): + raise RuntimeError("boom") + + monkeypatch.setattr(repos, "list_repos", crash) + with pytest.raises(SystemExit) as exit_info: + main(args=["list", "repos", "acme/x"]) + assert exit_info.value.code == 1 + + # One run produced both: the usage event carries the ID, the error does not. + assert usage_events[-1]["distinct_id"] == INSTALL_ID + assert usage_events[-1]["properties"]["outcome"] == "unexpected_error" + (error,) = sentry_events + serialised = json.dumps(error) + assert INSTALL_ID not in serialised + assert "1792076242" not in serialised + assert "install_id" not in serialised + + +@pytest.mark.parametrize( + "host, sent", + [("https://api.cloudsmith.io", 1), ("https://artifacts.example.com", 0)], +) +def test_forwarding_command_resolves_a_typed_api_host( + usage_events, notice_seen, no_repos_api, host, sent +): + # `list repos` hands off to `repositories get` (ctx.forward), whose + # session setup runs in a sub-context of the leaf. + args = ["list", "repos", "acme/x", "-F", "json", "--api-host", host] + assert main(args=args, standalone_mode=False) in (0, None) + assert len(no_repos_api) == 1 # the run reached the session and the API + assert len(usage_events) == sent + + +def _run_root(args): + """Run ``main()`` on a root path that ends before config is loaded.""" + from cloudsmith_cli.cli import config + + if args == ["--help"]: + assert main(args=args, standalone_mode=False) == 0 + else: + with pytest.raises(SystemExit) as exit_info: + main(args=args) + assert exit_info.value.code == 2 + # Root --help exits at parse and root usage errors in resolve_command, + # both before the root callback loads config. + assert getattr(config.OPTIONS, "value", None) is None + + +@pytest.mark.parametrize("args", [["--help"], ["no-such-command"]]) +@pytest.mark.parametrize( + "setting", + [ + "api_host = https://artifacts.example.com", + "ssl_verify = false", + "api_ssl_verify = false", + ], +) +def test_config_file_host_and_ssl_gates_hold_before_config_loads( + usage_events, notice_seen, telemetry_config_dir, args, setting +): + (telemetry_config_dir / "config.ini").write_text(f"[default]\n{setting}\n") + _run_root(args) + assert usage_events == [] + + +def test_config_file_profile_host_gate_holds_before_config_loads( + usage_events, notice_seen, telemetry_config_dir, monkeypatch +): + (telemetry_config_dir / "config.ini").write_text( + "[default]\n[profile:work]\napi_host = https://artifacts.example.com\n" + ) + monkeypatch.setenv("CLOUDSMITH_PROFILE", "work") + _run_root(["--help"]) + assert usage_events == [] + + +@pytest.mark.parametrize("args", [["--help"], ["no-such-command"]]) +def test_config_file_cloudsmith_host_still_sends_before_config_loads( + usage_events, notice_seen, telemetry_config_dir, args +): + (telemetry_config_dir / "config.ini").write_text( + "[default]\napi_host = https://api.cloudsmith.io\n" + ) + _run_root(args) + assert len(usage_events) == 1 + + +def test_env_api_host_beats_the_config_file_before_config_loads( + usage_events, notice_seen, telemetry_config_dir, monkeypatch +): + # As in initialise_session, CLOUDSMITH_API_HOST wins over config.ini. + (telemetry_config_dir / "config.ini").write_text( + "[default]\napi_host = https://artifacts.example.com\n" + ) + monkeypatch.setenv("CLOUDSMITH_API_HOST", "https://api.cloudsmith.io") + _run_root(["--help"]) + assert len(usage_events) == 1 + + +def test_unreadable_config_file_sends_nothing_before_config_loads( + usage_events, notice_seen, telemetry_config_dir +): + (telemetry_config_dir / "config.ini").write_text("[default\napi_host = x\n") + _run_root(["--help"]) + assert usage_events == [] + + +def _run_any(args): + """Run ``main()`` like a user would, whatever the exit code.""" + try: + main(args=args) + except SystemExit: + pass + + +@pytest.mark.parametrize( + "args", + [ + ["ls", "pkgs", "ws/r", "--config-file", "{file}", "--help"], + ["ls", "pkgs", "--config-file", "{file}"], # missing argument + ["whoami", "--config-file", "{file}", "--help"], + ["whoami", "-C", "{file}", "--help"], + ["--config-file", "{file}", "--help"], + ], +) +@pytest.mark.parametrize( + "setting", ["telemetry = false", "api_host = https://artifacts.example.com"] +) +def test_typed_config_file_holds_before_the_command_loads_it( + usage_events, notice_seen, tmp_path, args, setting +): + path = tmp_path / "optout.ini" + path.write_text(f"[default]\n{setting}\n") + _run_any([arg.format(file=path) for arg in args]) + assert usage_events == [] + + +@pytest.mark.parametrize("args", [["--profile", "p", "--help"], ["-P", "p", "-h"]]) +def test_typed_profile_opt_out_holds_before_the_command_loads_it( + usage_events, notice_seen, telemetry_config_dir, args +): + (telemetry_config_dir / "config.ini").write_text( + "[default]\n[profile:p]\ntelemetry = false\n" + ) + _run_any(args) + assert usage_events == [] + + +@pytest.mark.parametrize( + "setting", ["telemetry = false", "api_host = https://artifacts.example.com"] +) +def test_config_file_env_holds_before_config_loads( + usage_events, notice_seen, tmp_path, monkeypatch, setting +): + path = tmp_path / "optout.ini" + path.write_text(f"[default]\n{setting}\n") + monkeypatch.setenv("CLOUDSMITH_CONFIG_FILE", str(path)) + _run_root(["--help"]) + assert usage_events == [] + + +def test_config_file_env_without_an_opt_out_still_sends( + usage_events, notice_seen, tmp_path, monkeypatch +): + path = tmp_path / "plain.ini" + path.write_text("[default]\ncheck_for_update = true\n") + monkeypatch.setenv("CLOUDSMITH_CONFIG_FILE", str(path)) + _run_root(["--help"]) + assert len(usage_events) == 1 + + +def test_typed_config_file_without_an_opt_out_sends_once_loaded( + usage_events, notice_seen, no_repos_api, tmp_path +): + # The leaf loads the file itself (and hands off through ctx.forward), so + # the opt-out check sees what the file sets. + path = tmp_path / "plain.ini" + path.write_text("[default]\ncheck_for_update = true\n") + args = ["list", "repos", "acme/x", "-F", "json", "--config-file", str(path)] + assert main(args=args, standalone_mode=False) in (0, None) + assert len(no_repos_api) == 1 + assert len(usage_events) == 1 + + +def test_typed_config_file_with_an_opt_out_sends_nothing_once_loaded( + usage_events, notice_seen, no_repos_api, tmp_path +): + path = tmp_path / "optout.ini" + path.write_text("[default]\ntelemetry = false\n") + args = ["list", "repos", "acme/x", "-F", "json", "--config-file", str(path)] + assert main(args=args, standalone_mode=False) in (0, None) + assert usage_events == [] + + +@pytest.mark.parametrize( + "env, extra, setting, source, flag", + [ + ({"CLOUDSMITH_WORKSPACE": "acme"}, [], "workspace", "env", None), + ({}, ["-w", "acme"], "workspace", "flag", "org"), + ({}, ["--config-file", "{file}"], "config_file_path", "flag", "config_file"), + ], +) +def test_setting_sources_use_the_setting_names( + usage_events, + notice_seen, + no_repos_api, + tmp_path, + monkeypatch, + env, + extra, + setting, + source, + flag, +): + path = tmp_path / "plain.ini" + path.write_text("[default]\ncheck_for_update = true\n") + for name, value in env.items(): + monkeypatch.setenv(name, value) + args = ["list", "repos", "acme/x", "-F", "json"] + args += [arg.format(file=path) for arg in extra] + assert main(args=args, standalone_mode=False) in (0, None) + props = usage_events[-1]["properties"] + assert props[f"source_{setting}"] == source + if flag: # flags keep the Click parameter names + assert flag in props["flags"] + + +def test_forwarding_command_keeps_the_login_method( + usage_events, notice_seen, no_repos_api, monkeypatch +): + # `list repos` resolves credentials again inside `repositories get` + # (ctx.forward), where the env key is not passed on; that later None + # must not hide the login method the run used. + monkeypatch.setenv("CLOUDSMITH_API_KEY", "not-a-real-key") + args = ["list", "repos", "acme/x", "-F", "json"] + assert main(args=args, standalone_mode=False) in (0, None) + assert usage_events[-1]["properties"]["auth_source"] == "env_var" + + +def test_command_that_never_logs_in_reports_no_login_method( + usage_events, notice_seen, monkeypatch +): + monkeypatch.setenv("CLOUDSMITH_API_KEY", "not-a-real-key") + main(args=["credential-helper", "list"], standalone_mode=False) + assert "auth_source" not in usage_events[-1]["properties"] + + +@pytest.mark.parametrize( + "config, keys, source", + [ + ("", [], None), + ("workspace = acme", ["workspace"], "config_file"), + ("org = acme", ["org"], "config_file"), # the renamed key keeps its name + ("api_host =", [], None), # an empty key sets nothing + ], +) +def test_config_keys_name_what_config_ini_sets( + usage_events, notice_seen, no_repos_api, telemetry_config_dir, config, keys, source +): + (telemetry_config_dir / "config.ini").write_text(f"[default]\n{config}\n") + args = ["list", "repos", "acme", "-F", "json"] + assert main(args=args, standalone_mode=False) in (0, None) + (event,) = usage_events + assert event["properties"]["config_keys"] == keys + assert event["properties"].get("source_workspace") == source diff --git a/cloudsmith_cli/cli/usage_capture.py b/cloudsmith_cli/cli/usage_capture.py new file mode 100644 index 000000000..1758138b0 --- /dev/null +++ b/cloudsmith_cli/cli/usage_capture.py @@ -0,0 +1,358 @@ +# Copyright 2026 Cloudsmith Ltd +"""Record usage facts from the live Click context. + +Only names defined in code leave this module: command names, parameter +names, the option spellings declared on each option, env var names declared +on each option, and alias names. Argument and option values are never read. +""" + +import os +import sys +from typing import Any + +import click +from click.core import ParameterSource + +META_KEY = "cloudsmith_usage" +_HELP_SPELLINGS = frozenset({"-h", "--help"}) +_SOURCE_BY_CLICK = { + ParameterSource.COMMANDLINE: "flag", + ParameterSource.ENVIRONMENT: "env", +} +#: Click env vars that carry no information (set on almost every run). +_IGNORED_ENVVARS = frozenset({"TERM"}) + +# The env vars and spellings of the host and SSL options declared in +# ``decorators.initialise_session``. Their values reach ``opts`` only when +# that decorator runs, which ``--version``, ``--help`` and usage errors never +# reach, so the host and SSL gates also read them here. +API_HOST_ENV = "CLOUDSMITH_API_HOST" +SSL_VERIFY_ENV = "CLOUDSMITH_SSL_VERIFY" +WITHOUT_SSL_VERIFY_ENV = "CLOUDSMITH_WITHOUT_API_SSL_VERIFY" +API_HOST_SPELLINGS = frozenset({"--api-host"}) +SSL_OFF_SPELLINGS = frozenset( + {"--without-ssl-verify", "--without-api-ssl-verify", "-S"} +) +#: The config file and profile options of ``decorators.common_cli_config_options``. +#: A typed one names a file or profile only that decorator loads. +CONFIG_SPELLINGS = frozenset({"-C", "--config-file", "-P", "--profile"}) +#: Not a host: a typed ``--api-host`` that never reached opts fails the gate. +UNRESOLVED_HOST = "unresolved" +#: Click's boolean spellings (``BoolParamType``); an empty env var is unset. +_FALSE = frozenset({"0", "false", "no", "off", "n", "f"}) +_TRUE = frozenset({"1", "true", "yes", "on", "y", "t"}) + + +def _store(ctx): + return ctx.meta.setdefault(META_KEY, {"contexts": [], "typed": [], "aliases": []}) + + +def _takes_value(param): + return not ( + param.is_flag or param.count or getattr(param, "_flag_needs_value", False) + ) + + +def _short_cluster(token, spellings): + """Expand ``-vSPname`` the way Click does: ``-v``, ``-S``, ``-P``. + + Stops at the first option that takes a value (the rest of the token is + that value) or at an undeclared character. + """ + found = [] + for char in token[1:]: + spelling = f"-{char}" + if spelling not in spellings: + break + found.append(spelling) + if spellings[spelling]: + break + return found + + +def match_typed_flags(params, args): + """Return the declared option spellings present in ``args`` (no values).""" + # spelling -> whether the option takes a value + spellings = {} + for param in params: + if isinstance(param, click.Option): + for spelling in param.opts: + spellings[spelling] = _takes_value(param) + for spelling in param.secondary_opts: + spellings[spelling] = False + found = set() + for token in args: + if token == "--": + break + if not token.startswith("-") or token == "-": + continue + name = token.split("=", 1)[0] + if name in spellings: + found.add(name) + elif not name.startswith("--"): + found.update(_short_cluster(token, spellings)) + return sorted(found) + + +_RUN: dict[str, Any] = {"depth": 0, "root_meta": None, "completion": False} + + +def enter_run(): + """Mark the start of a top-level ``main()``; return the nesting depth.""" + _RUN["depth"] += 1 + if _RUN["depth"] == 1: + _RUN["root_meta"] = None + _RUN["completion"] = False + return _RUN["depth"] + + +def exit_run(): + """Mark the end of a ``main()`` run.""" + _RUN["depth"] = max(0, _RUN["depth"] - 1) + if _RUN["depth"] == 0: + _RUN["root_meta"] = None + _RUN["completion"] = False + + +def mark_completion_run(): + """Record that this run only serves shell completion.""" + _RUN["completion"] = True + + +def is_completion_run(): + """Tell whether this run only serves shell completion.""" + return _RUN["completion"] + + +def current_root_meta(): + """Return the ``ctx.meta`` of the root context of the current run.""" + return _RUN["root_meta"] + + +class TrackingMixin(click.Command): + """Record each context and its typed option spellings before parsing. + + Recording happens before ``super().parse_args`` so usage errors and + ``--help`` (which exit during parsing) still have a command path. + """ + + def parse_args(self, ctx, args): + if _RUN["depth"] >= 1 and ctx.parent is None and _RUN["root_meta"] is None: + _RUN["root_meta"] = ctx.meta + store = _store(ctx) + store["contexts"].append(ctx) + store["typed"].extend(match_typed_flags(self.get_params(ctx), args)) + return super().parse_args(ctx, args) + + +class TrackedCommand(TrackingMixin, click.Command): + """A command whose invocation is recorded for usage facts.""" + + +class TrackedGroup(TrackingMixin, click.Group): + """A group whose invocation is recorded; subcommands default to tracked.""" + + command_class = TrackedCommand + + +def record_alias(ctx, name): + """Record that a command was reached through an alias (or ````).""" + _store(ctx)["aliases"].append(name) + + +def record_api_status(ctx, status): + """Record the HTTP status of the API error that ended the run. + + Best effort: this runs on the error path, which usage capture must + never change, so a context that cannot hold the status is ignored. + """ + try: + _store(ctx)["api_status"] = status + except (AttributeError, TypeError): + pass + + +def _command_path(contexts): + leaf = contexts[-1] if contexts else None + names = [] + while leaf is not None and leaf.parent is not None: + if leaf.command.name: + names.append(leaf.command.name) + leaf = leaf.parent + return " ".join(reversed(names)) or "" + + +def _env_names(param): + if isinstance(param.envvar, str): + return [param.envvar] + return list(param.envvar or []) + + +def _param_facts(contexts): + from ..core.usage.event import SETTING_ALIASES + + flags, envvars, sources = set(), set(), {} + for ctx in contexts: + for param in ctx.command.params: + if not param.name or not isinstance(param, click.Option): + continue # arguments are neither flags nor settings + source = ctx.get_parameter_source(param.name) + if source == ParameterSource.COMMANDLINE: + flags.add(param.name) + if source == ParameterSource.ENVIRONMENT: + # Click uses the first non-empty variable, not the first set one. + used = next((n for n in _env_names(param) if os.environ.get(n)), None) + if used and used not in _IGNORED_ENVVARS: + envvars.add(used) + if source in _SOURCE_BY_CLICK: + setting = SETTING_ALIASES.get(param.name, param.name) + sources[setting] = _SOURCE_BY_CLICK[source] + return sorted(flags), sorted(envvars), sources + + +def _config_sources(meta): + from . import config_sources + + return config_sources.detect( + config_file=meta.get("config_file"), + creds_file=meta.get("creds_file"), + profile=meta.get("profile"), + ) + + +def _config_keys(meta): + from . import config_sources + + return config_sources.keys( + config_file=meta.get("config_file"), + creds_file=meta.get("creds_file"), + profile=meta.get("profile"), + ) + + +def _options(): + from . import config + + return getattr(config.OPTIONS, "value", None) + + +def record_session(ctx): + """Record that ``initialise_session`` put the host and SSL options in opts.""" + _store(ctx)["session"] = ctx + + +def record_credential(ctx, credential): + """Record how the run logged in, by provider name only. + + A command that hands off (``list repos`` to ``repositories get``) + resolves credentials again where the env key is not passed on, so a + later ``None`` must not replace the login the run used. + """ + source = getattr(credential, "source_name", None) + if source: + _store(ctx)["auth_source"] = source + + +def record_config(ctx): + """Record that ``common_cli_config_options`` loaded the config files.""" + _store(ctx)["config"] = ctx + + +def _set_up_by_leaf(store, key): + # A group (e.g. `list`) loads config and sets up the session before its + # subcommand parses, so only work done by the leaf counts: in the leaf's + # own context, or in a sub-context below it when the leaf hands off to + # another command (`list repos` runs `repositories get` via ctx.forward). + contexts = store.get("contexts", []) + ctx = store.get(key) + while contexts and ctx is not None: + if ctx is contexts[-1]: + return True + ctx = ctx.parent + return False + + +def _session_resolved(store): + return _set_up_by_leaf(store, "session") + + +def _config_resolved(store): + return _set_up_by_leaf(store, "config") + + +def resolve_api_host(opts, *, resolved, typed=()): + """Return the API host the run targets, for the host gate (never sent).""" + if resolved: + return getattr(opts, "api_host", None) + if API_HOST_SPELLINGS & set(typed): + return UNRESOLVED_HOST + # The env var beats config.ini, as it does in initialise_session. + return os.environ.get(API_HOST_ENV) or getattr(opts, "api_host", None) + + +def resolve_ssl_verify(opts, typed=()): + """Tell whether SSL verification is on; any sign of it being off wins.""" + if getattr(opts, "ssl_verify", True) is False or SSL_OFF_SPELLINGS & set(typed): + return False + verify = os.environ.get(SSL_VERIFY_ENV, "").strip().lower() + without = os.environ.get(WITHOUT_SSL_VERIFY_ENV, "").strip().lower() + return verify not in _FALSE and without not in _TRUE + + +def _ssl_verify_strict(opts): + """The resolved ``--without-ssl-verify-strict`` setting, or None if unset. + + Set by ``initialise_session`` even for commands that never load the SDK + (``update``), whose Configuration the sender would otherwise fall back to. + """ + value = getattr(opts, "ssl_verify_strict", None) + return value if isinstance(value, bool) else None + + +def typed_flags(meta): + """The option spellings typed in this run, from its root ``ctx.meta``.""" + return sorted(set((meta or {}).get(META_KEY, {}).get("typed", []))) + + +def collect_facts(meta, *, outcome, exit_code, duration_ms, nested): + """Build the ``RunFacts`` for a run from its recorded context metadata.""" + from ..core.usage.event import RunFacts + + store = meta.get(META_KEY, {}) + contexts = store.get("contexts", []) + flags, envvars, sources = _param_facts(contexts) + opts = _options() + typed = typed_flags(meta) + credential = getattr(opts, "credential", None) + return RunFacts( + command=_command_path(contexts), + flags=flags, + flags_typed=typed, + envvars=envvars, + aliases_used=list(store.get("aliases", [])), + output_format=getattr(opts, "output", None), + # A typed --help only means help was shown if the run then succeeded: + # `whoami --no-such-flag --help` is still a usage error. + outcome="help" + if _HELP_SPELLINGS & set(typed) and outcome == "success" + else outcome, + exit_code=exit_code, + api_status=store.get("api_status"), + duration_ms=duration_ms, + setting_sources=sources, + auth_source=store.get("auth_source") + or getattr(credential, "source_name", None), + api_host=resolve_api_host(opts, resolved=_session_resolved(store), typed=typed), + ssl_verify=resolve_ssl_verify(opts, typed), + ssl_verify_strict=_ssl_verify_strict(opts), + stdout_tty=bool(sys.stdout and sys.stdout.isatty()), + stdin_tty=bool(sys.stdin and sys.stdin.isatty()), + nested=nested, + # A typed config file or profile that the leaf never loaded may hold + # an opt-out or a host this run cannot see: never send. + config_unresolved=bool(CONFIG_SPELLINGS & set(typed)) + and not _config_resolved(store), + # Reads the ini files, so it runs only once a run is known to send. + deferred_sources=lambda: _config_sources(meta), + deferred_config_keys=lambda: _config_keys(meta), + ) diff --git a/cloudsmith_cli/core/tests/test_usage_event.py b/cloudsmith_cli/core/tests/test_usage_event.py index 5bb2f6fc3..931eb0614 100644 --- a/cloudsmith_cli/core/tests/test_usage_event.py +++ b/cloudsmith_cli/core/tests/test_usage_event.py @@ -95,7 +95,9 @@ def test_wrong_type_is_dropped(): assert "duration_ms" not in props -@pytest.mark.parametrize("name", ["flags", "flags_typed", "envvars", "aliases_used"]) +@pytest.mark.parametrize( + "name", ["flags", "flags_typed", "envvars", "aliases_used", "config_keys"] +) @pytest.mark.parametrize("bad", ["--page-size", None, ["ok", 3], {"a": 1}, 7]) def test_malformed_name_lists_are_dropped_not_split(name, bad): # A bare string must not be sent one character at a time, and None or @@ -113,3 +115,10 @@ def test_name_lists_are_deduplicated_and_sorted_but_aliases_keep_order(): )["properties"] assert props["flags"] == ["output_format", "page_size"] assert props["aliases_used"] == ["ls", "repos"] + + +def test_config_keys_are_deduplicated_and_sorted(): + props = _build(_facts(config_keys=["workspace", "api_proxy", "workspace"]))[ + "properties" + ] + assert props["config_keys"] == ["api_proxy", "workspace"] diff --git a/cloudsmith_cli/core/tests/test_usage_run.py b/cloudsmith_cli/core/tests/test_usage_run.py index 5268dbede..75b4c618c 100644 --- a/cloudsmith_cli/core/tests/test_usage_run.py +++ b/cloudsmith_cli/core/tests/test_usage_run.py @@ -268,6 +268,31 @@ def test_failing_loader_still_sends_the_event(usage_events, interactive): assert usage_events[0]["properties"]["source_workspace"] == "env" +def test_deferred_config_keys_load_only_for_a_run_that_sends( + usage_events, interactive, monkeypatch +): + keys = _Loader(["org", "proxy"]) + monkeypatch.setenv("DO_NOT_TRACK", "1") + run.finish(facts(deferred_config_keys=keys)) + assert keys.calls == 0 and usage_events == [] + monkeypatch.delenv("DO_NOT_TRACK") + run.finish(facts(deferred_config_keys=keys)) + assert keys.calls == 1 + assert usage_events[0]["properties"]["config_keys"] == ["org", "proxy"] + + +def test_failing_config_keys_loader_keeps_the_sources(usage_events, interactive): + run.finish( + facts( + deferred_sources=_Loader({"workspace": "config_file"}), + deferred_config_keys=_Loader(error=RuntimeError("boom")), + ) + ) + props = usage_events[0]["properties"] + assert props["source_workspace"] == "config_file" + assert props["config_keys"] == [] + + @pytest.fixture() def config_reads(monkeypatch): """Count config.ini reads; no command loaded config (no options yet).""" diff --git a/cloudsmith_cli/core/usage/event.py b/cloudsmith_cli/core/usage/event.py index 0d7de3bb8..edc6bbf42 100644 --- a/cloudsmith_cli/core/usage/event.py +++ b/cloudsmith_cli/core/usage/event.py @@ -49,7 +49,15 @@ "oidc_org": "workspace", "config_file": "config_file_path", "credentials_file": "credentials_file_path", - "without_api_ssl_verify": "ssl_verify", + # The hidden options of deprecated spellings (``deprecation.alias_options``). + "deprecated_api_proxy": "proxy", + "deprecated_without_api_ssl_verify": "ssl_verify", + "deprecated_S": "ssl_verify", + "deprecated_org": "workspace", + "deprecated_organization": "workspace", + "deprecated_oidc_org": "workspace", + "deprecated_owner": "workspace", + "deprecated_o": "workspace", } SOURCE_VALUES = frozenset( { @@ -91,6 +99,9 @@ class RunFacts: #: Loads config-file setting sources after the telemetry gates, so a run #: that sends nothing does no file I/O. Not an event property. deferred_sources: Callable[[], dict[str, str]] | None = None + #: config.ini / credentials.ini key names set, loaded with the sources. + config_keys: list[str] = field(default_factory=list) + deferred_config_keys: Callable[[], list[str]] | None = None @dataclass(frozen=True) @@ -106,6 +117,7 @@ class Property: Property("flags", list, "command and flag use; configuration habits"), Property("flags_typed", list, "command and flag use"), Property("envvars", list, "configuration habits; login methods"), + Property("config_keys", list, "configuration habits; renamed config keys in use"), Property("aliases_used", list, "command and flag use"), Property("output_format", str, "command and flag use; persona"), Property("outcome", str, "reliability and speed"), @@ -163,6 +175,7 @@ def _values( "flags": _names(facts.flags), "flags_typed": _names(facts.flags_typed), "envvars": _names(facts.envvars), + "config_keys": _names(facts.config_keys), "aliases_used": _names(facts.aliases_used, ordered=True), "output_format": facts.output_format, "outcome": facts.outcome, diff --git a/cloudsmith_cli/core/usage/run.py b/cloudsmith_cli/core/usage/run.py index 2ba616b41..a9d155f12 100644 --- a/cloudsmith_cli/core/usage/run.py +++ b/cloudsmith_cli/core/usage/run.py @@ -32,7 +32,15 @@ def _excluded(command): def _load_deferred_sources(facts): - """Merge config-file setting sources in; Click flag/env sources win.""" + """Merge config-file setting sources in; Click flag/env sources win. + + Also loads the config key names set. Each part fails on its own. + """ + _merge_config_sources(facts) + _load_config_keys(facts) + + +def _merge_config_sources(facts): if facts.deferred_sources is None: return try: @@ -42,6 +50,15 @@ def _load_deferred_sources(facts): return +def _load_config_keys(facts): + if facts.deferred_config_keys is None: + return + try: + facts.config_keys = facts.deferred_config_keys() + except Exception: # pylint: disable=broad-exception-caught + return + + def _config_file_blocks(facts, read_config): """Apply the host and SSL gates to config.ini for a run that never loaded it.