From dfbf8ca01f398669aaaef07f31d3c5bac263f6f3 Mon Sep 17 00:00:00 2001 From: Bartosz Blizniak Date: Tue, 25 Aug 2026 15:50:21 +0100 Subject: [PATCH] feat(no-ticket): add `cloudsmith repos privileges` command group Adds list/set/revoke/replace for the explicit team, user and service account privileges on a repository. Only revoke and replace can take access away, so only those two confirm before writing. The API returns every privilege in one response and ignores page parameters, so list offers no page options. It also has no way to delete a single privilege, so revoke reads the current set and writes back what is being kept. confirm_operation() now accepts an explicit empty prefix so these commands can ask their question directly, and handle_api_exceptions() takes an optional summarise_error callable so the API's field-indexed 422 reads as one sentence naming the repository. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 4 + cloudsmith_cli/cli/commands/repos.py | 586 ++++++++++++++ cloudsmith_cli/cli/exceptions.py | 36 +- .../tests/commands/test_repos_privileges.py | 762 ++++++++++++++++++ cloudsmith_cli/cli/tests/test_exceptions.py | 65 +- cloudsmith_cli/cli/tests/test_utils.py | 37 +- cloudsmith_cli/cli/utils.py | 14 +- cloudsmith_cli/core/api/repos.py | 56 ++ .../core/tests/test_repos_privileges.py | 208 +++++ 9 files changed, 1753 insertions(+), 15 deletions(-) create mode 100644 cloudsmith_cli/cli/tests/commands/test_repos_privileges.py create mode 100644 cloudsmith_cli/core/tests/test_repos_privileges.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c08c2ba0..68abb93d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ## [Unreleased] +### Added + +- Added `cloudsmith repos privileges` for managing explicit repository access from the terminal. `list` shows the teams, users and service accounts that were granted access explicitly; `set` grants or raises access for any number of them and leaves everyone else untouched; `revoke` takes access away from the ones named, skipping any that had none; and `replace` makes a JSON file (or stdin) the complete truth for the repository. `revoke` and `replace` ask for confirmation first unless `-y` is passed. + ## [1.25.0] - 2026-08-24 ### Added diff --git a/cloudsmith_cli/cli/commands/repos.py b/cloudsmith_cli/cli/commands/repos.py index 4308abda..83dea345 100644 --- a/cloudsmith_cli/cli/commands/repos.py +++ b/cloudsmith_cli/cli/commands/repos.py @@ -60,6 +60,65 @@ def print_repositories(opts, data, page_info=None, show_list_info=True, page_all ) +PRIVILEGE_LEVELS = ("read", "write", "admin") + +# A privilege names a team, a user or a service account, never more than one, +# which is why the table has a Type/Name pair rather than a column per kind. +TARGET_KINDS = ("team", "user", "service") + + +def get_privilege_target(entry): + """Get the (kind, name) pair a privilege entry applies to, if any.""" + for kind in TARGET_KINDS: + name = entry.get(kind) + if name: + return kind, name + return None + + +def as_privilege_entry(target, privilege): + """Build the compact entry shape the API accepts on write. + + A listed privilege carries every target key, with null for the two that + don't apply, and the write endpoints reject those nulls outright. + """ + kind, name = target + return {"privilege": privilege, kind: name} + + +def print_privileges(opts, data, show_list_info=True): + """Print repository privileges as a table or output in another format.""" + headers = ["Type", "Name", "Privilege"] + + targeted = [(get_privilege_target(entry), entry) for entry in data] + rows = [] + for target, entry in sorted(targeted, key=lambda item: item[0] or ("", "")): + kind, name = target or ("", "") + rows.append( + [ + click.style(kind.capitalize(), fg="yellow"), + click.style(name, fg="magenta"), + click.style(entry.get("privilege") or "", fg="cyan"), + ] + ) + + if rows: + click.echo() + utils.pretty_print_table(headers, rows) + + click.echo() + + if not show_list_info: + return + + num_results = len(rows) + utils.pretty_print_list_info( + num_results=num_results, + suffix="privilege%s" % ("s" if num_results != 1 else ""), + page_all=True, + ) + + @main.group(cls=command.AliasGroup, name="repositories", aliases=["repos"]) @decorators.common_cli_config_options @decorators.common_cli_output_options @@ -330,3 +389,530 @@ def delete(ctx, opts, owner_repo, yes): api.delete_repo(owner=owner, repo=repo) click.secho("OK", fg="green") + + +@repositories.group(cls=command.AliasGroup, name="privileges", aliases=["privilege"]) +def privileges(): + """ + Manage explicit team/user/service privileges on a repository. + + See the help for subcommands for more information on each. + """ + + +def collect_privilege_targets(teams, users, services): + """Turn the repeated --team/--user/--service options into targets.""" + targets = [] + seen = set() + + for kind, names in zip(TARGET_KINDS, (teams, users, services)): + for name in names: + if not name.strip(): + raise click.UsageError(f"Specify a slug for --{kind}.") + + if (kind, name) in seen: + raise click.UsageError(f"Specified more than once: {kind} {name}.") + seen.add((kind, name)) + targets.append((kind, name)) + + if not targets: + raise click.UsageError("Specify at least one of --team, --user or --service.") + + return targets + + +def describe_privilege_targets(targets): + """Describe targets for a message, e.g. 'team eng, service ci'.""" + return ", ".join(f"{kind} {click.style(name, bold=True)}" for kind, name in targets) + + +def summarise_privileges_error(action, repo): + """Build a summariser that renders privilege rejections as one sentence. + + The API reports a rejected privilege as a field-indexed 422, which reads + as three lines of machine detail. The person running the command only + needs to know which repository was not changed and why. Every other + status keeps the standard rendering, because the status code is the part + that matters when the request failed for some other reason. + """ + + def summarise(exc, detail, fields): + # Without fields there is nothing better to say than the standard + # rendering: `detail` falls back to the status description, which + # would read as "unprocessable Entity" and lose the status code. + if exc.status != 422 or not fields: + return None + + messages = [] + for value in fields.values(): + if isinstance(value, (list, tuple)): + value = " ".join(str(item) for item in value) + messages.append(str(value)) + + message = " ".join(messages).strip().rstrip(".") + if not message: + return None + + # The API capitalises these as standalone sentences; lower the first + # letter to join it onto ours, unless it starts an acronym. + if message[:2].istitle() or len(message) == 1: + message = message[0].lower() + message[1:] + + return f"Could not {action} privileges for {repo}: {message}" + + return summarise + + +@privileges.command(name="list", aliases=["ls", "get"]) +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.argument( + "owner_repo", metavar="OWNER/REPO", callback=validators.validate_owner_repo +) +@click.pass_context +def privileges_list(ctx, opts, owner_repo): + """ + List the explicit team/user/service privileges on a repository. + + - OWNER/REPO: Specify the OWNER namespace (i.e. user or org), and the + REPO name to list privileges for. All separated by a slash. + + Example: 'your-org/your-repo' + + Only explicitly granted privileges are listed here; this does not include + access implied by organisation role, team membership or ownership, none of + which these commands can change. The endpoint returns every privilege at + once, so there is nothing to page through. + + Full CLI example: + + $ cloudsmith repos privileges list your-org/your-repo + """ + owner, repo = owner_repo + + # Use stderr for messages if the output is something else (e.g. JSON) + use_stderr = utils.should_use_stderr(opts) + + click.echo("Getting list of repository privileges ... ", nl=False, err=use_stderr) + + context_msg = "Failed to get list of repository privileges!" + with ( + handle_api_exceptions(ctx, opts=opts, context_msg=context_msg), + maybe_spinner(opts), + ): + privileges_ = api.list_repo_privileges(owner=owner, repo=repo) + + click.secho("OK", fg="green", err=use_stderr) + + if utils.maybe_print_as_json(opts, privileges_): + return + + print_privileges(opts=opts, data=privileges_) + + +@privileges.command(name="set") +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.argument( + "owner_repo", metavar="OWNER/REPO", callback=validators.validate_owner_repo +) +@click.option( + "--team", + "teams", + multiple=True, + help="A team (slug) to grant the privilege to. Repeatable.", +) +@click.option( + "--user", + "users", + multiple=True, + help="A user (slug) to grant the privilege to. Repeatable.", +) +@click.option( + "--service", + "services", + multiple=True, + help="A service account (slug) to grant the privilege to. Repeatable.", +) +@click.option( + "--privilege", + required=True, + type=click.Choice(PRIVILEGE_LEVELS, case_sensitive=False), + help="The privilege level to grant.", +) +@click.pass_context +def privileges_set(ctx, opts, owner_repo, teams, users, services, privilege): + """ + Grant a privilege to teams, users and/or service accounts. + + - OWNER/REPO: Specify the OWNER namespace (i.e. user or org), and the + REPO name to set privileges on. All separated by a slash. + + Example: 'your-org/your-repo' + + At least one of --team, --user or --service must be given, and each may be + repeated to give several targets the same privilege in one call. + + This only ever adds or raises access: a target that already has an + explicit privilege is updated in place, and anything not named is left + exactly as it was. That makes it safe to run in a pipeline without reading + the current privileges first, and safe to run twice. + + Full CLI example: + + $ cloudsmith repos privileges set your-org/your-repo --team your-team --privilege write + """ + owner, repo = owner_repo + targets = collect_privilege_targets(teams, users, services) + + # The API accepts any casing but always stores and echoes its own, so + # normalise here to keep the CLI's own output consistent with a later list. + privilege = privilege.capitalize() + entries = [as_privilege_entry(target, privilege) for target in targets] + + # Use stderr for messages if the output is something else (e.g. JSON) + use_stderr = utils.should_use_stderr(opts) + + click.echo( + f"Granting {click.style(privilege, bold=True)} on " + f"{click.style(repo, bold=True)} in the {click.style(owner, bold=True)} " + f"namespace to {describe_privilege_targets(targets)} ... ", + nl=False, + err=use_stderr, + ) + + with ( + handle_api_exceptions( + ctx, + opts=opts, + context_msg="Failed to set the repository privileges!", + summarise_error=summarise_privileges_error("set", repo), + ), + maybe_spinner(opts), + ): + api.update_repo_privileges(owner, repo, entries) + + click.secho("OK", fg="green", err=use_stderr) + + if utils.maybe_print_as_json(opts, entries): + return + + print_privileges(opts=opts, data=entries) + + +@privileges.command(name="revoke") +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.argument( + "owner_repo", metavar="OWNER/REPO", callback=validators.validate_owner_repo +) +@click.option( + "--team", + "teams", + multiple=True, + help="A team (slug) to revoke the privilege of. Repeatable.", +) +@click.option( + "--user", + "users", + multiple=True, + help="A user (slug) to revoke the privilege of. Repeatable.", +) +@click.option( + "--service", + "services", + multiple=True, + help="A service account (slug) to revoke the privilege of. Repeatable.", +) +@click.option( + "-y", + "--yes", + default=False, + is_flag=True, + help="Assume yes as default answer to questions (this is dangerous!)", +) +@click.pass_context +def privileges_revoke(ctx, opts, owner_repo, teams, users, services, yes): + """ + Revoke the explicit privileges of teams, users and/or service accounts. + + - OWNER/REPO: Specify the OWNER namespace (i.e. user or org), and the + REPO name to revoke privileges on. All separated by a slash. + + Example: 'your-org/your-repo' + + At least one of --team, --user or --service must be given, and each may be + repeated. Targets with no explicit privilege are named and skipped, so + running this twice is a no-op rather than a failure. + + The API cannot delete a single privilege, so this reads the current + privileges and writes back the ones being kept. A change made by someone + else in between can therefore be lost. + + Full CLI example: + + $ cloudsmith repos privileges revoke your-org/your-repo --team your-team + """ + owner, repo = owner_repo + targets = collect_privilege_targets(teams, users, services) + + # Use stderr for messages if the output is something else (e.g. JSON) + use_stderr = utils.should_use_stderr(opts) + + click.echo("Getting list of repository privileges ... ", nl=False, err=use_stderr) + + context_msg = "Failed to get list of repository privileges!" + with ( + handle_api_exceptions(ctx, opts=opts, context_msg=context_msg), + maybe_spinner(opts), + ): + current = api.list_repo_privileges(owner=owner, repo=repo) + + click.secho("OK", fg="green", err=use_stderr) + + # Revoking means writing the whole list back, so every entry that stays + # has to be one this CLI can express. Rather than silently dropping an + # entry it can't read, or writing back the nulls the endpoint rejects, + # say so and point at the command that can express anything. + for entry in current: + if get_privilege_target(entry) is None or not entry.get("privilege"): + raise click.ClickException( + "This repository has a privilege this version of the CLI " + "doesn't understand, and revoking would drop it. Use " + "'cloudsmith repos privileges replace' to state the whole " + "list explicitly, or upgrade the CLI." + ) + + classified = [(get_privilege_target(entry), entry) for entry in current] + existing = {target for target, _ in classified} + found = [target for target in targets if target in existing] + + for kind, name in targets: + if (kind, name) not in existing: + click.secho( + f"No explicit privilege for {kind} {name}, skipping.", + fg="yellow", + err=use_stderr, + ) + + if not found: + click.secho("Nothing to revoke.", fg="green", err=use_stderr) + # The command ran to completion, so a `-F json` consumer still gets a + # document to parse: the privileges, unchanged. + utils.maybe_print_as_json( + opts, + [ + as_privilege_entry(target, entry.get("privilege")) + for target, entry in classified + ], + ) + return + + prompt = ( + f"Revoke the privileges of {describe_privilege_targets(found)} on " + f"{click.style(repo, bold=True)} in the {click.style(owner, bold=True)} " + "namespace" + ) + # Declining writes nothing, so there is nothing to report: both revoke and + # replace leave stdout empty rather than implying a result. + if not utils.confirm_operation(prompt, prefix="", assume_yes=yes, err=use_stderr): + return + + kept = [ + as_privilege_entry(target, entry.get("privilege")) + for target, entry in classified + if target not in found + ] + + click.echo( + f"Revoking the privileges of {describe_privilege_targets(found)} ... ", + nl=False, + err=use_stderr, + ) + + with ( + handle_api_exceptions( + ctx, + opts=opts, + context_msg="Failed to revoke the repository privileges!", + summarise_error=summarise_privileges_error("revoke", repo), + ), + maybe_spinner(opts), + ): + api.replace_repo_privileges(owner, repo, kept) + + click.secho("OK", fg="green", err=use_stderr) + + if utils.maybe_print_as_json(opts, kept): + return + + print_privileges(opts=opts, data=kept) + + +def read_privileges_file(privileges_file): + """Read and validate the privileges declared in a JSON file.""" + param_hint = "PRIVILEGES_FILE" + + try: + document = json.load(privileges_file) + except ValueError as exc: + raise click.BadParameter(f"Invalid JSON: {exc}", param_hint=param_hint) + + if isinstance(document, dict): + document = document.get("privileges") + + if not isinstance(document, list): + raise click.BadParameter( + "Expected a list of privileges, or an object with a 'privileges' list.", + param_hint=param_hint, + ) + + entries = [] + seen = set() + for entry in document: + if not isinstance(entry, dict): + raise click.BadParameter( + "Each privilege must be an object.", param_hint=param_hint + ) + + named = [kind for kind in TARGET_KINDS if entry.get(kind)] + if len(named) != 1: + raise click.BadParameter( + "Each privilege needs exactly one of 'team', 'user' or 'service'.", + param_hint=param_hint, + ) + + kind = named[0] + name = entry[kind] + if not isinstance(name, str): + raise click.BadParameter( + f"The '{kind}' of a privilege must be a slug, not {name!r}.", + param_hint=param_hint, + ) + + if (kind, name) in seen: + raise click.BadParameter( + f"Specified more than once: {kind} {name}.", param_hint=param_hint + ) + seen.add((kind, name)) + + privilege = str(entry.get("privilege") or "") + if privilege.lower() not in PRIVILEGE_LEVELS: + raise click.BadParameter( + f"'{privilege}' is not one of " + + ", ".join(f"'{level}'" for level in PRIVILEGE_LEVELS) + + ".", + param_hint=param_hint, + ) + + entries.append(as_privilege_entry((kind, name), privilege.capitalize())) + + return entries + + +@privileges.command(name="replace") +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.argument( + "owner_repo", metavar="OWNER/REPO", callback=validators.validate_owner_repo +) +@click.argument("privileges_file", metavar="PRIVILEGES_FILE", type=click.File("r")) +@click.option( + "-y", + "--yes", + default=False, + is_flag=True, + help="Assume yes as default answer to questions (this is dangerous!)", +) +@click.pass_context +def privileges_replace(ctx, opts, owner_repo, privileges_file, yes): + """ + Replace every explicit privilege on a repository with those in a file. + + - OWNER/REPO: Specify the OWNER namespace (i.e. user or org), and the + REPO name to replace privileges on. All separated by a slash. + + Example: 'your-org/your-repo' + + - PRIVILEGES_FILE: A JSON file holding either a list of privileges or an + object with a 'privileges' list. Use '-' to read it from stdin, which + needs -y because the confirmation has nowhere left to read an answer. + + The file becomes the complete truth for the repository, so anything absent + from it loses its explicit access, including you. Each entry names exactly + one of 'team', 'user' or 'service', plus a 'privilege' of 'read', 'write' + or 'admin'. A file listing nothing revokes every explicit privilege, which + the confirmation says in those words. + + Full CLI example: + + $ cloudsmith repos privileges replace your-org/your-repo privileges.json + """ + owner, repo = owner_repo + + # The file and the answer to the prompt would come from the same stream, + # so reading one leaves nothing to read the other from. + if not yes and getattr(privileges_file, "name", None) in ("-", ""): + raise click.UsageError( + "Reading the privileges from stdin leaves nothing to answer the " + "confirmation with. Pass -y to confirm up front." + ) + + entries = read_privileges_file(privileges_file) + + # Use stderr for messages if the output is something else (e.g. JSON) + use_stderr = utils.should_use_stderr(opts) + + if entries: + prompt = ( + "Replace all {count} privilege{plural} on {repo} in the {owner} " + "namespace, removing any not listed".format( + count=len(entries), + plural="" if len(entries) == 1 else "s", + repo=click.style(repo, bold=True), + owner=click.style(owner, bold=True), + ) + ) + else: + # "Replace all 0 privileges" reads as a no-op and is the opposite: + # an empty file revokes every explicit privilege on the repository. + prompt = ( + "The file lists no privileges. Revoke all explicit access to " + f"{click.style(repo, bold=True)} in the " + f"{click.style(owner, bold=True)} namespace" + ) + if not utils.confirm_operation(prompt, prefix="", assume_yes=yes, err=use_stderr): + return + + click.echo( + f"Replacing the privileges on {click.style(repo, bold=True)} in the " + f"{click.style(owner, bold=True)} namespace ... ", + nl=False, + err=use_stderr, + ) + + with ( + handle_api_exceptions( + ctx, + opts=opts, + context_msg="Failed to replace the repository privileges!", + summarise_error=summarise_privileges_error("replace", repo), + ), + maybe_spinner(opts), + ): + api.replace_repo_privileges(owner, repo, entries) + + click.secho("OK", fg="green", err=use_stderr) + + if utils.maybe_print_as_json(opts, entries): + return + + print_privileges(opts=opts, data=entries) diff --git a/cloudsmith_cli/cli/exceptions.py b/cloudsmith_cli/cli/exceptions.py index 7fe1b51e..703e69c6 100644 --- a/cloudsmith_cli/cli/exceptions.py +++ b/cloudsmith_cli/cli/exceptions.py @@ -11,9 +11,23 @@ @contextlib.contextmanager def handle_api_exceptions( - ctx, opts, context_msg=None, nl=False, exit_on_error=True, reraise_on_error=False + ctx, + opts, + context_msg=None, + nl=False, + exit_on_error=True, + reraise_on_error=False, + summarise_error=None, ): - """Context manager that handles API exceptions.""" + """Context manager that handles API exceptions. + + ``summarise_error`` is an optional callable taking ``(exc, detail, + fields)`` and returning a single sentence to show instead of the default + context/detail/field block, or ``None`` to keep the default. Commands use + it where the API's field-indexed errors read poorly next to the rest of + their output; returning ``None`` for statuses they don't recognise keeps + the status code visible where it still matters. + """ # flake8: ignore=C901 # Use stderr for messages if the output is something else (e.g. # JSON) @@ -26,11 +40,12 @@ def handle_api_exceptions( context_msg = context_msg or "Failed to perform operation!" detail, fields = get_details(exc) hint = get_error_hint(ctx, opts, exc) + summary = summarise_error(exc, detail, fields) if summarise_error else None if is_json_output: # Construct JSON error object error_data = { - "detail": detail or exc.status_description, + "detail": summary or detail or exc.status_description, "help": { "context": context_msg, "hint": hint, @@ -68,13 +83,16 @@ def handle_api_exceptions( else: click.secho("ERROR", fg="red", err=use_stderr) - click.secho( - f"{context_msg} (status: {exc.status} - {exc.status_description})", - fg="red", - err=use_stderr, - ) + if summary: + click.secho(summary, fg="red", err=use_stderr) + else: + click.secho( + f"{context_msg} (status: {exc.status} - {exc.status_description})", + fg="red", + err=use_stderr, + ) - if detail or fields: + if not summary and (detail or fields): click.echo(err=use_stderr) if detail: diff --git a/cloudsmith_cli/cli/tests/commands/test_repos_privileges.py b/cloudsmith_cli/cli/tests/commands/test_repos_privileges.py new file mode 100644 index 00000000..f864faae --- /dev/null +++ b/cloudsmith_cli/cli/tests/commands/test_repos_privileges.py @@ -0,0 +1,762 @@ +"""Tests for the `cloudsmith repos privileges` commands.""" + +import json + +import httpretty +import httpretty.core +import pytest + +from ....cli.commands.main import main + +API_HOST = "https://api.cloudsmith.io" +OWNER = "test-org" +REPO = "test-repo" +OWNER_REPO = f"{OWNER}/{REPO}" +PRIVILEGES_URL = f"{API_HOST}/repos/{OWNER}/{REPO}/privileges" +HERMETIC_ARGS = ["--api-key", "fake-api-key", "--api-host", API_HOST] +PRIVILEGES_COMMAND = ["repos", "privileges"] + + +@pytest.fixture(autouse=True) +def hermetic_environment(monkeypatch): + """Keep stray environment/config from influencing these commands. + + An inherited CLOUDSMITH_ORG/CLOUDSMITH_API_HOST would change which host + or org the command resolves to without the test noticing. + """ + monkeypatch.delenv("CLOUDSMITH_ORG", raising=False) + monkeypatch.delenv("CLOUDSMITH_API_HOST", raising=False) + monkeypatch.delenv("CLOUDSMITH_API_KEY", raising=False) + monkeypatch.setattr( + httpretty.core.fakesock.socket, + "shutdown", + lambda self, how: None, + raising=False, + ) + + +def register_list(privileges, status=200): + """Register a GET response holding the given privileges.""" + httpretty.register_uri( + httpretty.GET, + PRIVILEGES_URL, + body=json.dumps({"privileges": privileges}), + status=status, + content_type="application/json", + ) + + +def register_write(method, status=200, body=None): + """Register a PATCH/PUT response for the privileges endpoint.""" + httpretty.register_uri( + method, + PRIVILEGES_URL, + body=json.dumps(body) if body is not None else "", + status=status, + content_type="application/json", + ) + + +def last_request_body(): + """Get the JSON body of the last request httpretty captured.""" + return json.loads(httpretty.last_request().body.decode("utf-8")) + + +class TestPrivilegesList: + @httpretty.activate(allow_net_connect=False) + def test_lists_privileges_by_type_and_name(self, runner): + register_list( + [ + {"privilege": "Read", "team": None, "user": "alice", "service": None}, + {"privilege": "Admin", "team": None, "user": None, "service": "ci"}, + {"privilege": "Write", "team": "eng", "user": None, "service": None}, + ] + ) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["list"] + HERMETIC_ARGS + [OWNER_REPO], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert "Getting list of repository privileges" in result.output + assert "Type" in result.output and "Name" in result.output + # Sorted by type then name, so service/team/user in that order. + rows = [ + line + for line in result.output.splitlines() + if line.startswith(("Service", "Team", "User")) + ] + assert [row.split("|")[0].strip() for row in rows] == [ + "Service", + "Team", + "User", + ] + assert "Results: 3 privileges" in result.output + + @httpretty.activate(allow_net_connect=False) + def test_singular_result_count(self, runner): + register_list([{"privilege": "Read", "team": "eng"}]) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["list"] + HERMETIC_ARGS + [OWNER_REPO], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert "Results: 1 privilege\n" in result.output + + @httpretty.activate(allow_net_connect=False) + def test_empty_list(self, runner): + register_list([]) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["list"] + HERMETIC_ARGS + [OWNER_REPO], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert "Results: 0 privileges" in result.output + + @httpretty.activate(allow_net_connect=False) + def test_json_output_is_clean_on_stdout(self, runner): + register_list([{"privilege": "Read", "team": "eng"}]) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["list"] + HERMETIC_ARGS + ["-F", "json", OWNER_REPO], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + # The progress message goes to stderr in JSON mode, so stdout on its + # own has to be a single valid document a script can parse. + payload = json.loads(result.stdout) + assert payload["data"] == [ + {"privilege": "Read", "team": "eng", "user": None, "service": None} + ] + + @httpretty.activate(allow_net_connect=False) + def test_group_and_command_aliases(self, runner): + register_list([{"privilege": "Read", "team": "eng"}]) + + result = runner.invoke( + main, + ["repos", "privilege", "ls"] + HERMETIC_ARGS + [OWNER_REPO], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert "Results: 1 privilege" in result.output + + def test_no_page_options(self, runner): + """The endpoint returns everything at once, so paging isn't offered.""" + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["list", "--help"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert "--page" not in result.output + + +class TestPrivilegesSet: + @httpretty.activate(allow_net_connect=False) + def test_grants_to_several_targets_in_one_call(self, runner): + register_write(httpretty.PATCH) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["set"] + + HERMETIC_ARGS + + [ + OWNER_REPO, + "--team", + "eng", + "--user", + "alice", + "--service", + "ci", + "--privilege", + "write", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert last_request_body() == { + "privileges": [ + {"privilege": "Write", "team": "eng"}, + {"privilege": "Write", "user": "alice"}, + {"privilege": "Write", "service": "ci"}, + ] + } + assert ( + f"Granting Write on {REPO} in the {OWNER} namespace to " + "team eng, user alice, service ci" in result.output + ) + assert "Results: 3 privileges" in result.output + + @httpretty.activate(allow_net_connect=False) + def test_privilege_is_case_insensitive_and_echoed_in_api_casing(self, runner): + register_write(httpretty.PATCH) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["set"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng", "--privilege", "ADMIN"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert last_request_body()["privileges"] == [ + {"privilege": "Admin", "team": "eng"} + ] + assert "Granting Admin" in result.output + + def test_rejects_an_unknown_privilege(self, runner): + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["set"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng", "--privilege", "Owner"], + ) + + assert result.exit_code != 0 + assert "'Owner' is not one of 'read', 'write', 'admin'" in result.output + + def test_requires_at_least_one_target(self, runner): + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["set"] + + HERMETIC_ARGS + + [OWNER_REPO, "--privilege", "read"], + ) + + assert result.exit_code != 0 + assert "Specify at least one of --team, --user or --service." in result.output + + def test_rejects_a_repeated_target(self, runner): + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["set"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng", "--team", "eng", "--privilege", "read"], + ) + + assert result.exit_code != 0 + assert "Specified more than once: team eng." in result.output + + @httpretty.activate(allow_net_connect=False) + def test_api_rejection_is_one_sentence(self, runner): + register_write( + httpretty.PATCH, + status=422, + body={ + "detail": "Invalid input.", + "fields": { + "privileges": "Invalid team(s) specified ['no-such-team']", + }, + }, + ) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["set"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "no-such-team", "--privilege", "read"], + ) + + # AliasGroup.main runs click with standalone_mode=False, so the exit + # code comes back as the return value rather than via SystemExit. + assert result.return_value == 422 + assert ( + f"Could not set privileges for {REPO}: " + "invalid team(s) specified ['no-such-team']" in result.output + ) + assert "Privileges Field:" not in result.output + assert "status: 422" not in result.output + + @httpretty.activate(allow_net_connect=False) + def test_other_statuses_keep_the_status_code(self, runner): + register_write( + httpretty.PATCH, + status=403, + body={"detail": "You do not have permission to perform this action."}, + ) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["set"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng", "--privilege", "read"], + ) + + assert result.return_value == 403 + assert "status: 403" in result.output + assert "Could not set privileges" not in result.output + + @httpretty.activate(allow_net_connect=False) + def test_a_leading_acronym_keeps_its_casing(self, runner): + register_write( + httpretty.PATCH, + status=422, + body={"detail": "Invalid input.", "fields": {"privileges": "URL rejected"}}, + ) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["set"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng", "--privilege", "read"], + ) + + assert f"Could not set privileges for {REPO}: URL rejected" in result.output + + @httpretty.activate(allow_net_connect=False) + def test_a_422_without_fields_keeps_the_status_code(self, runner): + """`detail` falls back to the status description, which reads badly.""" + register_write(httpretty.PATCH, status=422, body={}) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["set"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng", "--privilege", "read"], + ) + + assert result.return_value == 422 + assert "status: 422" in result.output + assert "unprocessable" not in result.output.lower().split("status: 422")[0] + + @httpretty.activate(allow_net_connect=False) + def test_summarised_error_in_json_mode(self, runner): + register_write( + httpretty.PATCH, + status=422, + body={ + "detail": "Invalid input.", + "fields": {"privileges": "Invalid team(s) specified ['nope']"}, + }, + ) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["set"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "nope", "--privilege", "read", "-F", "json"], + ) + + payload = json.loads(result.stdout) + assert payload["detail"] == ( + f"Could not set privileges for {REPO}: invalid team(s) specified ['nope']" + ) + + def test_rejects_an_empty_target_name(self, runner): + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["set"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "", "--privilege", "read"], + ) + + assert result.exit_code != 0 + assert "Specify a slug for --team." in result.output + + +class TestPrivilegesRevoke: + @httpretty.activate(allow_net_connect=False) + def test_revokes_named_targets_and_keeps_the_rest(self, runner): + register_list( + [ + {"privilege": "Write", "team": "eng", "user": None, "service": None}, + {"privilege": "Read", "team": None, "user": "alice", "service": None}, + ] + ) + register_write(httpretty.PUT) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["revoke"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng"], + input="y\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert ( + f"Revoke the privileges of team eng on {REPO} in the {OWNER} namespace?" + in result.output + ) + assert "Are you absolutely certain" not in result.output + # The listed entries carry a null for every kind that doesn't apply, + # and the write endpoint rejects those nulls, so they're dropped. + assert last_request_body() == { + "privileges": [{"privilege": "Read", "user": "alice"}] + } + assert "Results: 1 privilege" in result.output + + @httpretty.activate(allow_net_connect=False) + def test_skips_targets_without_an_explicit_privilege(self, runner): + register_list( + [{"privilege": "Write", "team": "eng", "user": None, "service": None}] + ) + register_write(httpretty.PUT) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["revoke"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng", "--user", "someone-else", "-y"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert "No explicit privilege for user someone-else, skipping." in result.output + assert last_request_body() == {"privileges": []} + + @httpretty.activate(allow_net_connect=False) + def test_second_run_is_a_no_op(self, runner): + register_list([]) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["revoke"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng", "-y"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert "No explicit privilege for team eng, skipping." in result.output + assert "Nothing to revoke." in result.output + assert httpretty.last_request().method == "GET" + + @httpretty.activate(allow_net_connect=False) + def test_declining_the_prompt_writes_nothing(self, runner): + register_list( + [{"privilege": "Write", "team": "eng", "user": None, "service": None}] + ) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["revoke"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng"], + input="n\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert httpretty.last_request().method == "GET" + + @httpretty.activate(allow_net_connect=False) + def test_api_rejection_names_the_revoke(self, runner): + register_list([{"privilege": "Read", "team": "eng"}]) + register_write( + httpretty.PUT, + status=422, + body={"detail": "Invalid input.", "fields": {"privileges": "Nope"}}, + ) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["revoke"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng", "-y"], + ) + + assert f"Could not revoke privileges for {REPO}: nope" in result.output + + def test_requires_at_least_one_target(self, runner): + result = runner.invoke( + main, PRIVILEGES_COMMAND + ["revoke"] + HERMETIC_ARGS + [OWNER_REPO] + ) + + assert result.exit_code != 0 + assert "Specify at least one of --team, --user or --service." in result.output + + @httpretty.activate(allow_net_connect=False) + def test_no_op_still_emits_a_json_document(self, runner): + """A `-F json` consumer must get something to parse on every path.""" + register_list([{"privilege": "Read", "team": "eng"}]) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["revoke"] + + HERMETIC_ARGS + + [OWNER_REPO, "--user", "nobody", "-y", "-F", "json"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + # The same compact shape the success path emits, so a consumer sees + # one entry shape whichever path ran. + assert json.loads(result.stdout)["data"] == [ + {"privilege": "Read", "team": "eng"} + ] + + @httpretty.activate(allow_net_connect=False) + def test_declining_writes_and_says_nothing(self, runner): + register_list([{"privilege": "Read", "team": "eng"}]) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["revoke"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng", "-F", "json"], + input="n\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert result.stdout == "" + assert httpretty.last_request().method == "GET" + + @httpretty.activate(allow_net_connect=False) + def test_refuses_when_it_cannot_read_a_current_privilege(self, runner): + """Writing the list back would drop what the CLI can't express.""" + register_list( + [ + {"privilege": "Write", "team": "eng", "user": None, "service": None}, + {"privilege": "Read", "team": None, "user": None, "service": None}, + ] + ) + register_write(httpretty.PUT) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["revoke"] + + HERMETIC_ARGS + + [OWNER_REPO, "--team", "eng", "-y"], + ) + + assert result.exit_code != 0 + assert "doesn't understand" in result.output + assert "privileges replace" in result.output + assert httpretty.last_request().method == "GET" + + +class TestPrivilegesReplace: + @httpretty.activate(allow_net_connect=False) + def test_replaces_from_a_file(self, runner, tmp_path): + register_write(httpretty.PUT) + path = tmp_path / "privileges.json" + path.write_text( + json.dumps( + { + "privileges": [ + {"team": "eng", "privilege": "Write"}, + {"user": "alice", "privilege": "read"}, + ] + } + ) + ) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["replace"] + HERMETIC_ARGS + [OWNER_REPO, str(path)], + input="y\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert ( + f"Replace all 2 privileges on {REPO} in the {OWNER} namespace, " + "removing any not listed?" in result.output + ) + assert "Are you absolutely certain" not in result.output + assert last_request_body() == { + "privileges": [ + {"privilege": "Write", "team": "eng"}, + {"privilege": "Read", "user": "alice"}, + ] + } + + @httpretty.activate(allow_net_connect=False) + def test_accepts_a_bare_list_on_stdin(self, runner): + register_write(httpretty.PUT) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["replace"] + HERMETIC_ARGS + [OWNER_REPO, "-", "-y"], + input=json.dumps([{"service": "ci", "privilege": "admin"}]), + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert last_request_body() == { + "privileges": [{"privilege": "Admin", "service": "ci"}] + } + + def test_rejects_an_entry_naming_two_kinds(self, runner): + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["replace"] + HERMETIC_ARGS + [OWNER_REPO, "-", "-y"], + input=json.dumps([{"team": "eng", "user": "alice", "privilege": "read"}]), + ) + + assert result.exit_code != 0 + assert "Invalid value for PRIVILEGES_FILE" in result.output + assert ( + "Each privilege needs exactly one of 'team', 'user' or 'service'." + in result.output + ) + + def test_rejects_an_unknown_privilege(self, runner): + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["replace"] + HERMETIC_ARGS + [OWNER_REPO, "-", "-y"], + input=json.dumps([{"team": "eng", "privilege": "Owner"}]), + ) + + assert result.exit_code != 0 + assert "'Owner' is not one of 'read', 'write', 'admin'." in result.output + + def test_stdin_without_yes_is_refused(self, runner): + """The document and the y/N answer would come from the same stream.""" + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["replace"] + HERMETIC_ARGS + [OWNER_REPO, "-"], + input=json.dumps([{"team": "eng", "privilege": "read"}]), + ) + + assert result.exit_code != 0 + assert "Pass -y to confirm up front." in result.output + + @httpretty.activate(allow_net_connect=False) + def test_an_empty_file_says_it_revokes_everything(self, runner, tmp_path): + register_write(httpretty.PUT) + path = tmp_path / "privileges.json" + path.write_text("[]") + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["replace"] + HERMETIC_ARGS + [OWNER_REPO, str(path)], + input="y\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert ( + "The file lists no privileges. Revoke all explicit access to " + f"{REPO} in the {OWNER} namespace?" in result.output + ) + assert "Replace all 0 privileges" not in result.output + assert last_request_body() == {"privileges": []} + + def test_rejects_a_non_object_entry(self, runner): + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["replace"] + HERMETIC_ARGS + [OWNER_REPO, "-", "-y"], + input=json.dumps(["eng"]), + ) + + assert result.exit_code != 0 + assert "Each privilege must be an object." in result.output + + def test_rejects_a_document_that_is_not_a_list(self, runner): + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["replace"] + HERMETIC_ARGS + [OWNER_REPO, "-", "-y"], + input=json.dumps({"teams": []}), + ) + + assert result.exit_code != 0 + assert "Expected a list of privileges" in result.output + + def test_rejects_a_non_string_name(self, runner): + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["replace"] + HERMETIC_ARGS + [OWNER_REPO, "-", "-y"], + input=json.dumps([{"team": 123, "privilege": "read"}]), + ) + + assert result.exit_code != 0 + assert "The 'team' of a privilege must be a slug, not 123." in result.output + + def test_rejects_a_repeated_target(self, runner): + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["replace"] + HERMETIC_ARGS + [OWNER_REPO, "-", "-y"], + input=json.dumps( + [ + {"team": "eng", "privilege": "read"}, + {"team": "eng", "privilege": "admin"}, + ] + ), + ) + + assert result.exit_code != 0 + assert "Specified more than once: team eng." in result.output + + @httpretty.activate(allow_net_connect=False) + def test_api_rejection_names_the_replace(self, runner): + register_write( + httpretty.PUT, + status=422, + body={"detail": "Invalid input.", "fields": {"privileges": "Nope"}}, + ) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["replace"] + HERMETIC_ARGS + [OWNER_REPO, "-", "-y"], + input=json.dumps([{"team": "eng", "privilege": "read"}]), + ) + + assert f"Could not replace privileges for {REPO}: nope" in result.output + + @httpretty.activate(allow_net_connect=False) + def test_declining_writes_and_says_nothing(self, runner, tmp_path): + register_write(httpretty.PUT) + path = tmp_path / "privileges.json" + path.write_text(json.dumps([{"team": "eng", "privilege": "read"}])) + + result = runner.invoke( + main, + PRIVILEGES_COMMAND + + ["replace"] + + HERMETIC_ARGS + + [OWNER_REPO, str(path), "-F", "json"], + input="n\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert result.stdout == "" + assert httpretty.has_request() is False + + def test_rejects_invalid_json(self, runner): + result = runner.invoke( + main, + PRIVILEGES_COMMAND + ["replace"] + HERMETIC_ARGS + [OWNER_REPO, "-", "-y"], + input="not json", + ) + + assert result.exit_code != 0 + assert "Invalid JSON" in result.output diff --git a/cloudsmith_cli/cli/tests/test_exceptions.py b/cloudsmith_cli/cli/tests/test_exceptions.py index 56455ffc..67f143d9 100644 --- a/cloudsmith_cli/cli/tests/test_exceptions.py +++ b/cloudsmith_cli/cli/tests/test_exceptions.py @@ -4,7 +4,7 @@ from unittest.mock import Mock, patch from cloudsmith_cli.cli.commands.main import main -from cloudsmith_cli.cli.exceptions import get_401_error_hint +from cloudsmith_cli.cli.exceptions import get_401_error_hint, handle_api_exceptions from cloudsmith_cli.core.api.exceptions import ApiException from cloudsmith_cli.core.credentials.models import CredentialResult @@ -92,3 +92,66 @@ def test_json_output_renders_the_hint(self, runner, tmp_path): error = json.loads(result.stdout) assert error["meta"] == {"code": 401, "description": "Unauthorized"} assert error["help"]["hint"] == API_KEY_HINT + + +class Opts: + """The handful of attributes the renderer reads off opts.""" + + def __init__(self, output="pretty"): + self.output = output + self.verbose = False + self.debug = False + self.push_metadata_info = None + + +class TestSummariseError: + """The opt-in one-sentence rendering, in isolation from its callers.""" + + @staticmethod + def run(summarise, opts, status=422): + ctx = Mock() + ctx.exit.side_effect = SystemExit + exc = ApiException(status=status, detail="Invalid input.") + exc.fields = {"privileges": "Nope."} + + try: + with handle_api_exceptions( + ctx, opts=opts, context_msg="Boom!", summarise_error=summarise + ): + raise exc + except SystemExit: + pass + + def test_a_summary_replaces_the_context_and_field_block(self, capsys): + self.run(lambda exc, detail, fields: "Could not do the thing: nope", Opts()) + + out = capsys.readouterr().out + assert "Could not do the thing: nope" in out + assert "status: 422" not in out + assert "Privileges Field:" not in out + + def test_declining_keeps_the_default_rendering(self, capsys): + self.run(lambda exc, detail, fields: None, Opts()) + + out = capsys.readouterr().out + assert "Boom! (status: 422" in out + assert "Privileges Field: Nope." in out + + def test_the_summary_becomes_the_json_detail(self, capsys): + self.run( + lambda exc, detail, fields: "Could not do the thing: nope", + Opts("json"), + ) + + error = json.loads(capsys.readouterr().out) + assert error["detail"] == "Could not do the thing: nope" + assert error["meta"]["code"] == 422 + + def test_the_callable_sees_the_exception(self): + seen = [] + self.run( + lambda exc, detail, fields: seen.append((exc.status, detail, fields)), + Opts(), + ) + + assert seen == [(422, "Invalid input.", {"privileges": "Nope."})] diff --git a/cloudsmith_cli/cli/tests/test_utils.py b/cloudsmith_cli/cli/tests/test_utils.py index 04dd2fcc..80314df8 100644 --- a/cloudsmith_cli/cli/tests/test_utils.py +++ b/cloudsmith_cli/cli/tests/test_utils.py @@ -1,6 +1,7 @@ +import click import pytest -from ..utils import maybe_truncate_list, maybe_truncate_string +from ..utils import confirm_operation, maybe_truncate_list, maybe_truncate_string @pytest.mark.parametrize( @@ -33,3 +34,37 @@ def test_maybe_truncate_string(data, max_length, expected_len): if expected_len > max_length: assert truncated[-4:-1] == "..." + + +class TestConfirmOperation: + """The prompt's preamble is opt-out, so a command can ask directly.""" + + @staticmethod + def prompt_for(monkeypatch, **kwargs): + asked = [] + monkeypatch.setattr( + "click.confirm", lambda text, **_: asked.append(text) or True + ) + monkeypatch.setattr( + "click.get_text_stream", + lambda name: type("S", (), {"isatty": lambda self: True})(), + ) + confirm_operation("do the thing", **kwargs) + return click.unstyle(asked[0]) + + def test_the_default_preamble_is_unchanged(self, monkeypatch): + assert self.prompt_for(monkeypatch) == ( + "Are you absolutely certain you want to do the thing?" + ) + + def test_an_empty_prefix_asks_the_question_on_its_own(self, monkeypatch): + assert self.prompt_for(monkeypatch, prefix="") == "do the thing?" + + def test_a_given_prefix_still_wins(self, monkeypatch): + assert self.prompt_for(monkeypatch, prefix="Really") == "Really do the thing?" + + def test_assume_yes_never_asks(self, monkeypatch): + monkeypatch.setattr( + "click.confirm", lambda *a, **k: pytest.fail("should not have asked") + ) + assert confirm_operation("do the thing", prefix="", assume_yes=True) is True diff --git a/cloudsmith_cli/cli/utils.py b/cloudsmith_cli/cli/utils.py index 7fc749f6..435f719a 100644 --- a/cloudsmith_cli/cli/utils.py +++ b/cloudsmith_cli/cli/utils.py @@ -199,11 +199,17 @@ def confirm_operation(prompt, prefix=None, assume_yes=False, err=False): if assume_yes: return True - prefix = prefix or click.style( - "Are you {} certain you want to".format(click.style("absolutely", bold=True)) - ) + if prefix is None: + prefix = click.style( + "Are you {} certain you want to".format( + click.style("absolutely", bold=True) + ) + ) - prompt = maybe_unstyle_prompt(f"{prefix} {prompt}?", err=err) + # An explicit empty prefix asks the question on its own, without the + # "Are you absolutely certain..." preamble. + question = f"{prefix} {prompt}?" if prefix else f"{prompt}?" + prompt = maybe_unstyle_prompt(question, err=err) answered = click.confirm(prompt, err=err) diff --git a/cloudsmith_cli/core/api/repos.py b/cloudsmith_cli/core/api/repos.py index 8591155d..35b04444 100644 --- a/cloudsmith_cli/core/api/repos.py +++ b/cloudsmith_cli/core/api/repos.py @@ -79,3 +79,59 @@ def delete_repo(owner, repo): _, _, headers = client.repos_delete_with_http_info(owner, repo) ratelimits.maybe_rate_limit(client, headers) + + +def list_repo_privileges(owner, repo): + """Get the explicit team/user/service privileges on a repository. + + The endpoint returns every privilege in a single response and ignores + page parameters, so there is nothing to paginate over here. + """ + client = get_repos_api() + + with catch_raise_api_exception(): + data, _, headers = client.repos_privileges_list_with_http_info(owner, repo) + + ratelimits.maybe_rate_limit(client, headers) + return [privilege.to_dict() for privilege in data.privileges] + + +def update_repo_privileges(owner, repo, privileges): + """Add or raise one or more explicit privileges on a repository. + + ``privileges`` is a list of dicts, each shaped like + ``{"privilege": "Read"|"Write"|"Admin", "team": }`` (or ``"user"`` + or ``"service"`` in place of ``"team"``). This calls the + ``PATCH .../privileges`` endpoint, which the API documents (and manual + verification against a live org confirmed) as an upsert: each entry is + matched against the repository's existing privileges by its + team/user/service key and updated in place, or added if no match exists. + Entries not mentioned are left untouched, so callers don't need to read + the existing list first. + """ + client = get_repos_api() + + with catch_raise_api_exception(): + _, _, headers = client.repos_privileges_partial_update_with_http_info( + owner, repo, data={"privileges": privileges} + ) + + ratelimits.maybe_rate_limit(client, headers) + + +def replace_repo_privileges(owner, repo, privileges): + """Replace every explicit privilege on a repository with ``privileges``. + + This calls the ``PUT .../privileges`` endpoint, which is a whole-list + write: anything absent from ``privileges`` loses its explicit access. + The API has no way to delete a single privilege, so revoking is also + expressed as a replace of the entries being kept. + """ + client = get_repos_api() + + with catch_raise_api_exception(): + _, _, headers = client.repos_privileges_update_with_http_info( + owner, repo, data={"privileges": privileges} + ) + + ratelimits.maybe_rate_limit(client, headers) diff --git a/cloudsmith_cli/core/tests/test_repos_privileges.py b/cloudsmith_cli/core/tests/test_repos_privileges.py new file mode 100644 index 00000000..fd26da56 --- /dev/null +++ b/cloudsmith_cli/core/tests/test_repos_privileges.py @@ -0,0 +1,208 @@ +"""Tests for the repository privileges API client.""" + +import json + +import httpretty +import httpretty.core +import pytest + +from .. import keyring +from ..api import repos +from ..api.exceptions import ApiException +from ..api.init import initialise_api +from ..credentials.models import CredentialResult + +API_HOST = "https://api.cloudsmith.io" +OWNER = "test-org" +REPO = "test-repo" +PRIVILEGES_URL = f"{API_HOST}/repos/{OWNER}/{REPO}/privileges" + + +@pytest.fixture(autouse=True) +def _setup_api(monkeypatch): + """Initialise the SDK Configuration and stub keyring lookups. + + See ``core/tests/test_metadata.py`` for why each of these is required: + initialise_api() registers retry attributes the REST client expects, and + keyring is stubbed so tests never touch the real OS keyring. + """ + monkeypatch.setattr(keyring, "get_access_token", lambda host: None) + monkeypatch.setattr(keyring, "get_refresh_token", lambda host: None) + monkeypatch.setattr(keyring, "should_refresh_access_token", lambda host: False) + monkeypatch.setattr( + httpretty.core.fakesock.socket, + "shutdown", + lambda self, how: None, + raising=False, + ) + initialise_api( + host=API_HOST, + credential=CredentialResult( + api_key="test-api-key", + source_name="test", + auth_type="api_key", + ), + ) + + +def _last_request(): + return httpretty.last_request() + + +class TestListRepoPrivileges: + @httpretty.activate(allow_net_connect=False) + def test_success_returns_privileges(self): + body = { + "privileges": [ + {"privilege": "Admin", "team": None, "user": None, "service": "ci"}, + {"privilege": "Read", "team": "eng", "user": None, "service": None}, + ] + } + httpretty.register_uri( + httpretty.GET, + PRIVILEGES_URL, + body=json.dumps(body), + status=200, + content_type="application/json", + ) + + privileges = repos.list_repo_privileges(OWNER, REPO) + + assert privileges == body["privileges"] + + sent = _last_request() + assert sent.headers.get("X-Api-Key") == "test-api-key" + assert sent.path == f"/repos/{OWNER}/{REPO}/privileges" + + @httpretty.activate(allow_net_connect=False) + def test_empty_privileges_list(self): + httpretty.register_uri( + httpretty.GET, + PRIVILEGES_URL, + body=json.dumps({"privileges": []}), + status=200, + content_type="application/json", + ) + + privileges = repos.list_repo_privileges(OWNER, REPO) + + assert privileges == [] + + @httpretty.activate(allow_net_connect=False) + def test_404_raises_api_exception(self): + httpretty.register_uri( + httpretty.GET, + PRIVILEGES_URL, + body=json.dumps({"detail": "Not found."}), + status=404, + content_type="application/json", + ) + + with pytest.raises(ApiException) as exc_info: + repos.list_repo_privileges(OWNER, REPO) + + assert exc_info.value.status == 404 + assert exc_info.value.detail == "Not found." + + +class TestUpdateRepoPrivileges: + @httpretty.activate(allow_net_connect=False) + def test_success_sends_patch_with_privileges_body(self): + httpretty.register_uri( + httpretty.PATCH, + PRIVILEGES_URL, + body="", + status=204, + ) + + result = repos.update_repo_privileges( + OWNER, REPO, [{"privilege": "Write", "team": "eng"}] + ) + + assert result is None + + sent = _last_request() + assert sent.method == "PATCH" + assert json.loads(sent.body) == { + "privileges": [{"privilege": "Write", "team": "eng"}] + } + + @httpretty.activate(allow_net_connect=False) + def test_422_raises_api_exception_with_fields(self): + message = "bogus is not valid for privilege - must be one of ['Admin', 'Write', 'Read']" + body = { + "detail": "Invalid data.", + "fields": {"privilege": [message]}, + } + httpretty.register_uri( + httpretty.PATCH, + PRIVILEGES_URL, + body=json.dumps(body), + status=422, + content_type="application/json", + ) + + with pytest.raises(ApiException) as exc_info: + repos.update_repo_privileges( + OWNER, REPO, [{"privilege": "bogus", "team": "eng"}] + ) + + assert exc_info.value.status == 422 + assert exc_info.value.fields == {"privilege": [message]} + + +class TestReplaceRepoPrivileges: + @httpretty.activate(allow_net_connect=False) + def test_success_sends_put_with_privileges_body(self): + httpretty.register_uri( + httpretty.PUT, + PRIVILEGES_URL, + body="", + status=204, + ) + + result = repos.replace_repo_privileges( + OWNER, REPO, [{"privilege": "Write", "team": "eng"}] + ) + + assert result is None + + sent = _last_request() + assert sent.method == "PUT" + assert json.loads(sent.body) == { + "privileges": [{"privilege": "Write", "team": "eng"}] + } + + @httpretty.activate(allow_net_connect=False) + def test_empty_list_revokes_everything(self): + httpretty.register_uri( + httpretty.PUT, + PRIVILEGES_URL, + body="", + status=204, + ) + + repos.replace_repo_privileges(OWNER, REPO, []) + + assert json.loads(_last_request().body) == {"privileges": []} + + @httpretty.activate(allow_net_connect=False) + def test_422_raises_api_exception_with_fields(self): + message = "Invalid team(s) specified ['no-such-team']" + httpretty.register_uri( + httpretty.PUT, + PRIVILEGES_URL, + body=json.dumps( + {"detail": "Invalid input.", "fields": {"privileges": message}} + ), + status=422, + content_type="application/json", + ) + + with pytest.raises(ApiException) as exc_info: + repos.replace_repo_privileges( + OWNER, REPO, [{"privilege": "Read", "team": "no-such-team"}] + ) + + assert exc_info.value.status == 422 + assert exc_info.value.fields == {"privileges": message}