diff --git a/cloudsmith_cli/cli/command.py b/cloudsmith_cli/cli/command.py index 87ddc813..4e180be4 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 bcf76528..93f8038f 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 f52b8a8c..74452eee 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 ad8345af..60b21e61 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 db98b17d..c0172275 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 34475753..0ceb6388 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 603f5e5c..39e10963 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 c49008f6..c1f75dfa 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 d1654a4e..eac4dedd 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 00000000..ccc0f4eb --- /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 8f790458..78c5c1de 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 5fcfea28..21b8b574 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 00000000..39ff0433 --- /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 f89cf265..19d86fb7 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 00000000..fbe5cc5d --- /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 00000000..d4cfb715 --- /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 00000000..1758138b --- /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 5bb2f6fc..931eb061 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 5268dbed..75b4c618 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 0d7de3bb..edc6bbf4 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 2ba616b4..a9d155f1 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.