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}