diff --git a/cloudsmith_cli/cli/commands/__init__.py b/cloudsmith_cli/cli/commands/__init__.py index af10c90e..ed80eac5 100644 --- a/cloudsmith_cli/cli/commands/__init__.py +++ b/cloudsmith_cli/cli/commands/__init__.py @@ -14,6 +14,7 @@ login, logout, mcp, + metadata, metrics, move, policy, diff --git a/cloudsmith_cli/cli/commands/metadata.py b/cloudsmith_cli/cli/commands/metadata.py new file mode 100644 index 00000000..7574ebb9 --- /dev/null +++ b/cloudsmith_cli/cli/commands/metadata.py @@ -0,0 +1,535 @@ +"""CLI/Commands - Manage metadata attached to packages.""" + +import json + +import click + +from ...core.api.metadata import ( + create_metadata as api_create_metadata, + delete_metadata as api_delete_metadata, + get_metadata as api_get_metadata, + list_metadata as api_list_metadata, + normalise_classification, + normalise_source_kind, + update_metadata as api_update_metadata, +) +from ...core.api.packages import get_package_slug_perm as api_get_package_slug_perm +from ...core.pagination import paginate_results +from ...core.version import get_version as get_cli_version +from .. import command, decorators, utils, validators +from ..exceptions import handle_api_exceptions +from ..utils import maybe_spinner +from .main import main + +_METADATA_HEADERS = [ + "Slug", + "Content Type", + "Classification", + "Source Kind", + "Source Identity", +] + + +def _default_source_identity(): + """Return the default value for --source-identity.""" + return f"cloudsmith-cli@{get_cli_version()}" + + +def _format_metadata_row(entry): + return [ + click.style(entry.get("slug_perm") or "", fg="cyan"), + click.style(entry.get("content_type") or "", fg="yellow"), + click.style(str(entry.get("classification", "")), fg="magenta"), + click.style(str(entry.get("source_kind", "")), fg="blue"), + click.style(entry.get("source_identity") or "", fg="green"), + ] + + +def _echo_action(message, use_stderr): + """Print an in-progress status message.""" + click.echo(message, nl=False, err=use_stderr) + + +def _print_metadata_table(opts, entries, page_info=None, page_all=False): + """Print a list of metadata entries as a table or JSON.""" + if utils.maybe_print_as_json( + opts, list(entries), page_info=None if page_all else page_info + ): + return + + rows = [ + _format_metadata_row(e) + for e in sorted(entries, key=lambda e: e.get("slug_perm") or "") + ] + + if rows: + click.echo() + utils.pretty_print_table(_METADATA_HEADERS, rows) + + click.echo() + + num_results = len(rows) + list_suffix = f"metadata entr{'ies' if num_results != 1 else 'y'}" + utils.pretty_print_list_info( + num_results=num_results, + page_info=None if page_all else page_info, + suffix=f"{list_suffix} retrieved" if page_all else f"{list_suffix} visible", + page_all=page_all, + ) + + +def _print_metadata_entry(opts, entry): + """Print a single metadata entry as a table + indented JSON content.""" + if utils.maybe_print_as_json(opts, entry): + return + + click.echo() + utils.pretty_print_table(_METADATA_HEADERS, [_format_metadata_row(entry)]) + click.echo() + + content = entry.get("content") + if content is not None: + click.secho("Content:", bold=True) + click.echo(json.dumps(content, indent=2, sort_keys=True)) + + +def _load_content(content_file, inline_content, *, required): + """Resolve --file / --content into a parsed object. + + Enforces the XOR between the two sources. When `required` is True (used + by `add`), at least one source must be provided. When False (used by + `update`), a missing source means "do not change content". + """ + if content_file is not None and inline_content is not None: + raise click.UsageError("--file and --content are mutually exclusive.") + + if content_file is not None: + if content_file == "-": + raw, source = click.get_text_stream("stdin").read(), "stdin" + else: + with open(content_file, encoding="utf-8") as fh: + raw, source = fh.read(), "--file" + elif inline_content is not None: + raw, source = inline_content, "--content" + elif required: + raise click.UsageError("One of --file or --content is required.") + else: + return None + + try: + return json.loads(raw) + except ValueError as exc: + raise click.UsageError(f"Invalid JSON in {source}: {exc}") from exc + + +@main.group(name="metadata", cls=command.AliasGroup) +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.pass_context +def metadata_(ctx, opts): # pylint: disable=unused-argument + """ + Manage metadata attached to packages in a repository. + + See the help for subcommands for more information on each. + """ + + +@metadata_.command(name="list", aliases=["ls"]) +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_cli_list_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.argument( + "owner_repo_package", + metavar="OWNER/REPO/PACKAGE", + callback=validators.validate_owner_repo_package, +) +@click.argument("metadata_slug_perm", required=False, default=None) +@click.option( + "--source-kind", + "source_kind", + default=None, + help=( + "Filter by metadata source kind. Accepts an integer or a name " + "(e.g. 'customer', 'third_party'). Ignored when METADATA_SLUG_PERM is given." + ), +) +@click.option( + "--classification", + "classification", + default=None, + help=( + "Filter by metadata classification. Accepts an integer or a name " + "(e.g. 'provenance', 'sbom'). Ignored when METADATA_SLUG_PERM is given." + ), +) +@click.pass_context +def list_metadata( + ctx, + opts, + owner_repo_package, + metadata_slug_perm, + page, + page_size, + page_all, + source_kind, + classification, +): + """ + List metadata entries attached to a package. + + OWNER/REPO/PACKAGE: identifies the package whose metadata you want to list. + + METADATA_SLUG_PERM (optional): if given, fetch and display only that single + metadata entry. Pagination and filter flags are ignored in this case. + + \b + Examples: + $ cloudsmith metadata list your-org/awesome-repo/better-pkg + $ cloudsmith metadata list your-org/awesome-repo/better-pkg --classification provenance + $ cloudsmith metadata list your-org/awesome-repo/better-pkg meta-slug-perm + """ + owner, repo, package = owner_repo_package + use_stderr = utils.should_use_stderr(opts) + + if metadata_slug_perm: + _echo_action( + "Fetching metadata entry %(metadata)s for the '%(package)s' package ... " + % { + "metadata": click.style(metadata_slug_perm, bold=True), + "package": click.style(package, bold=True), + }, + use_stderr, + ) + + context_msg = "Failed to fetch metadata for the package!" + with handle_api_exceptions(ctx, opts=opts, context_msg=context_msg): + with maybe_spinner(opts): + slug_perm = api_get_package_slug_perm( + owner=owner, repo=repo, identifier=package + ) + entry = api_get_metadata(slug_perm, metadata_slug_perm) + + click.secho("OK", fg="green", err=use_stderr) + _print_metadata_entry(opts, entry) + return + + # Validate filter values up-front for a friendlier error than what the + # API would return (the normalisers raise ValueError on invalid values). + try: + normalise_source_kind(source_kind) + normalise_classification(classification) + except ValueError as exc: + raise click.UsageError(str(exc)) from exc + + _echo_action( + "Listing metadata for the '%(package)s' package ... " + % {"package": click.style(package, bold=True)}, + use_stderr, + ) + + context_msg = "Failed to list metadata for the package!" + with handle_api_exceptions(ctx, opts=opts, context_msg=context_msg): + with maybe_spinner(opts): + slug_perm = api_get_package_slug_perm( + owner=owner, repo=repo, identifier=package + ) + entries, page_info = paginate_results( + api_list_metadata, + page_all=page_all, + page=page, + page_size=page_size, + package_slug_perm=slug_perm, + source_kind=source_kind, + classification=classification, + ) + + click.secho("OK", fg="green", err=use_stderr) + _print_metadata_table(opts, entries, page_info=page_info, page_all=page_all) + + +@metadata_.command(name="add") +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.argument( + "owner_repo_package", + metavar="OWNER/REPO/PACKAGE", + callback=validators.validate_owner_repo_package, +) +@click.option( + "--content-type", + "content_type", + required=True, + help=( + "The content type of the metadata payload (e.g. 'application/json'). " + "Content type is immutable after creation." + ), +) +@click.option( + "--file", + "content_file", + type=click.Path( + exists=True, + dir_okay=False, + readable=True, + resolve_path=True, + allow_dash=True, + ), + default=None, + help="Path to a JSON file containing the metadata content. Use '-' for stdin.", +) +@click.option( + "--content", + "inline_content", + default=None, + help=("Inline JSON content for the metadata. Mutually exclusive with --file."), +) +@click.option( + "--source-identity", + "source_identity", + default=None, + help=( + "Free-text identifier indicating where this metadata originated. " + "Defaults to 'cloudsmith-cli@'." + ), +) +@click.pass_context +def add_metadata( + ctx, + opts, + owner_repo_package, + content_type, + content_file, + inline_content, + source_identity, +): + """ + Attach a new metadata entry to a package. + + OWNER/REPO/PACKAGE: the package the metadata should be attached to. + + Exactly one of --file or --content must be supplied. + Content type is set on creation and cannot be changed later. + + \b + Examples: + $ cloudsmith metadata add your-org/awesome-repo/better-pkg \\ + --content-type application/json \\ + --content '{"foo": "bar"}' + $ cat payload.json | cloudsmith metadata add your-org/awesome-repo/better-pkg \\ + --content-type application/json \\ + --file - + $ cloudsmith metadata add your-org/awesome-repo/better-pkg \\ + --content-type application/vnd.jfrog.buildinfo+json \\ + --file buildinfo.json + """ + owner, repo, package = owner_repo_package + use_stderr = utils.should_use_stderr(opts) + + content = _load_content(content_file, inline_content, required=True) + source_identity = source_identity or _default_source_identity() + + _echo_action( + "Attaching metadata to the '%(package)s' package ... " + % {"package": click.style(package, bold=True)}, + use_stderr, + ) + + context_msg = "Failed to attach metadata to the package!" + with handle_api_exceptions(ctx, opts=opts, context_msg=context_msg): + with maybe_spinner(opts): + slug_perm = api_get_package_slug_perm( + owner=owner, repo=repo, identifier=package + ) + entry = api_create_metadata( + slug_perm, + content=content, + content_type=content_type, + source_identity=source_identity, + ) + + click.secho("OK", fg="green", err=use_stderr) + _print_metadata_entry(opts, entry) + + +@metadata_.command(name="update") +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.argument( + "owner_repo_package", + metavar="OWNER/REPO/PACKAGE", + callback=validators.validate_owner_repo_package, +) +@click.argument("metadata_slug_perm") +@click.option( + "--file", + "content_file", + type=click.Path( + exists=True, + dir_okay=False, + readable=True, + resolve_path=True, + allow_dash=True, + ), + default=None, + help=( + "Path to a JSON file containing replacement metadata content. " + "Use '-' for stdin." + ), +) +@click.option( + "--content", + "inline_content", + default=None, + help=( + "Inline JSON replacement content for the metadata. Mutually exclusive " + "with --file." + ), +) +@click.option( + "--source-identity", + "source_identity", + default=None, + help="Update the free-text source identity for this metadata entry.", +) +@click.pass_context +def update_metadata( + ctx, + opts, + owner_repo_package, + metadata_slug_perm, + content_file, + inline_content, + source_identity, +): + """ + Patch an existing metadata entry on a package. + + OWNER/REPO/PACKAGE: the package the metadata is attached to. + METADATA_SLUG_PERM: the permanent slug of the metadata entry to update. + + Content type cannot be changed after creation. + + \b + Examples: + $ cloudsmith metadata update your-org/awesome-repo/better-pkg meta-slug \\ + --content '{"foo": "baz"}' + $ cat payload.json | cloudsmith metadata update your-org/awesome-repo/better-pkg meta-slug \\ + --file - + """ + owner, repo, package = owner_repo_package + use_stderr = utils.should_use_stderr(opts) + + content = _load_content(content_file, inline_content, required=False) + + patch_kwargs = { + key: value + for key, value in ( + ("content", content), + ("source_identity", source_identity), + ) + if value is not None + } + if not patch_kwargs: + raise click.UsageError( + "Nothing to update. Provide --file, --content, or --source-identity." + ) + + _echo_action( + "Updating metadata entry %(metadata)s on the '%(package)s' package ... " + % { + "metadata": click.style(metadata_slug_perm, bold=True), + "package": click.style(package, bold=True), + }, + use_stderr, + ) + + context_msg = "Failed to update metadata on the package!" + with handle_api_exceptions(ctx, opts=opts, context_msg=context_msg): + with maybe_spinner(opts): + slug_perm = api_get_package_slug_perm( + owner=owner, repo=repo, identifier=package + ) + entry = api_update_metadata(slug_perm, metadata_slug_perm, **patch_kwargs) + + click.secho("OK", fg="green", err=use_stderr) + _print_metadata_entry(opts, entry) + + +@metadata_.command(name="remove", aliases=["rm"]) +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.argument( + "owner_repo_package", + metavar="OWNER/REPO/PACKAGE", + callback=validators.validate_owner_repo_package, +) +@click.argument("metadata_slug_perm") +@click.option( + "-y", + "--yes", + default=False, + is_flag=True, + help="Assume yes as default answer to questions (this is dangerous!)", +) +@click.pass_context +def remove_metadata(ctx, opts, owner_repo_package, metadata_slug_perm, yes): + """ + Remove a metadata entry from a package. + + OWNER/REPO/PACKAGE: the package the metadata is attached to. + METADATA_SLUG_PERM: the permanent slug of the metadata entry to delete. + + \b + Example: + $ cloudsmith metadata remove your-org/awesome-repo/better-pkg meta-slug + """ + owner, repo, package = owner_repo_package + use_stderr = utils.should_use_stderr(opts) + + remove_args = { + "metadata": click.style(metadata_slug_perm, bold=True), + "package": click.style(package, bold=True), + } + + prompt = ( + "remove the %(metadata)s metadata entry from the %(package)s package" + % remove_args + ) + if not utils.confirm_operation(prompt, assume_yes=yes, err=use_stderr): + return + + _echo_action( + "Removing metadata entry %(metadata)s from the '%(package)s' package ... " + % remove_args, + use_stderr, + ) + + context_msg = "Failed to remove metadata from the package!" + with handle_api_exceptions(ctx, opts=opts, context_msg=context_msg): + with maybe_spinner(opts): + slug_perm = api_get_package_slug_perm( + owner=owner, repo=repo, identifier=package + ) + api_delete_metadata(slug_perm, metadata_slug_perm) + + click.secho("OK", fg="green", err=use_stderr) + + result_payload = {"deleted": True, "slug_perm": metadata_slug_perm} + if utils.maybe_print_as_json(opts, result_payload): + return + + click.echo() + click.secho( + "Removed metadata entry %(slug)s." + % {"slug": click.style(metadata_slug_perm, bold=True)} + ) diff --git a/cloudsmith_cli/cli/tests/commands/test_metadata.py b/cloudsmith_cli/cli/tests/commands/test_metadata.py new file mode 100644 index 00000000..afb5d316 --- /dev/null +++ b/cloudsmith_cli/cli/tests/commands/test_metadata.py @@ -0,0 +1,651 @@ +"""CLI tests for the `cloudsmith metadata` command group.""" + +import json +import unittest +from unittest.mock import patch + +from click.testing import CliRunner + +from cloudsmith_cli.cli.commands.metadata import metadata_ +from cloudsmith_cli.core.pagination import MAX_PAGE_SIZE, PageInfo + + +def _empty_page_info(): + """Return an invalid PageInfo, matching current v2 API responses.""" + return PageInfo() + + +def _page_info(*, page, page_total, count, page_size=MAX_PAGE_SIZE): + info = PageInfo() + info.count = count + info.page = page + info.page_size = page_size + info.page_total = page_total + return info + + +class TestMetadataGroupSmoke(unittest.TestCase): + def setUp(self): + self.runner = CliRunner() + + def test_help_lists_subcommands(self): + result = self.runner.invoke(metadata_, ["--help"]) + self.assertEqual(result.exit_code, 0, msg=result.output) + self.assertIn("list", result.output) + self.assertIn("add", result.output) + self.assertIn("update", result.output) + self.assertIn("remove", result.output) + + def test_help_preserves_example_lines(self): + result = self.runner.invoke(metadata_, ["list", "--help"]) + + self.assertEqual(result.exit_code, 0, msg=result.output) + self.assertIn( + "$ cloudsmith metadata list your-org/awesome-repo/better-pkg\n", + result.output, + ) + self.assertIn( + "$ cloudsmith metadata list your-org/awesome-repo/better-pkg " + "--classification provenance\n", + result.output, + ) + self.assertIn( + "$ cloudsmith metadata list your-org/awesome-repo/better-pkg meta-slug-perm\n", + result.output, + ) + + def test_add_help_preserves_multiline_example(self): + result = self.runner.invoke(metadata_, ["add", "--help"]) + + self.assertEqual(result.exit_code, 0, msg=result.output) + self.assertIn( + "$ cloudsmith metadata add your-org/awesome-repo/better-pkg \\\n", + result.output, + ) + self.assertIn("--content-type application/json \\\n", result.output) + self.assertIn('--content \'{"foo": "bar"}\'', result.output) + self.assertIn("cat payload.json | cloudsmith metadata add", result.output) + self.assertIn("--file -", result.output) + self.assertIn("application/vnd.jfrog.buildinfo+json", result.output) + self.assertIn("--file buildinfo.json", result.output) + + def test_update_help_preserves_stdin_example(self): + result = self.runner.invoke(metadata_, ["update", "--help"]) + + self.assertEqual(result.exit_code, 0, msg=result.output) + self.assertIn("cat payload.json | cloudsmith metadata update", result.output) + self.assertIn("--file -", result.output) + + +class TestMetadataList(unittest.TestCase): + def setUp(self): + self.runner = CliRunner() + + @patch("cloudsmith_cli.cli.commands.metadata.api_list_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_list_resolves_slug_perm_and_calls_list(self, mock_resolve, mock_list): + mock_resolve.return_value = "pkg-slug-perm" + mock_list.return_value = ([], _empty_page_info()) + + result = self.runner.invoke(metadata_, ["list", "myorg/myrepo/mypkg"]) + + self.assertEqual(result.exit_code, 0, msg=result.output) + mock_resolve.assert_called_once_with( + owner="myorg", repo="myrepo", identifier="mypkg" + ) + mock_list.assert_called_once() + kwargs = mock_list.call_args.kwargs + self.assertEqual(kwargs["package_slug_perm"], "pkg-slug-perm") + self.assertIsNone(kwargs.get("source_kind")) + self.assertIsNone(kwargs.get("classification")) + + @patch("cloudsmith_cli.cli.commands.metadata.api_list_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_list_passes_filters(self, mock_resolve, mock_list): + mock_resolve.return_value = "pkg-slug-perm" + mock_list.return_value = ([], _empty_page_info()) + + result = self.runner.invoke( + metadata_, + [ + "list", + "myorg/myrepo/mypkg", + "--source-kind", + "customer", + "--classification", + "4", + ], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + kwargs = mock_list.call_args.kwargs + self.assertEqual(kwargs["source_kind"], "customer") + self.assertEqual(kwargs["classification"], "4") + + @patch("cloudsmith_cli.cli.commands.metadata.api_list_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_list_json_output(self, mock_resolve, mock_list): + mock_resolve.return_value = "pkg-slug-perm" + mock_list.return_value = ( + [ + { + "slug_perm": "abc", + "content_type": "application/json", + "classification": "GENERIC", + "source_kind": "CUSTOMER", + "source_identity": "cloudsmith-cli@1.16.0", + } + ], + _empty_page_info(), + ) + + result = self.runner.invoke( + metadata_, ["list", "-F", "json", "myorg/myrepo/mypkg"] + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + payload = json.loads(result.stdout) + self.assertEqual(len(payload["data"]), 1) + self.assertEqual(payload["data"][0]["slug_perm"], "abc") + + @patch("cloudsmith_cli.cli.commands.metadata.api_list_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_list_invalid_filter_value_is_usage_error(self, mock_resolve, mock_list): + mock_resolve.return_value = "pkg-slug-perm" + + result = self.runner.invoke( + metadata_, + ["list", "myorg/myrepo/mypkg", "--source-kind", "not-a-kind"], + ) + + self.assertNotEqual(result.exit_code, 0) + self.assertIn("source_kind", result.output.lower()) + mock_list.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.metadata.api_list_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_list_page_all_aggregates_all_pages(self, mock_resolve, mock_list): + mock_resolve.return_value = "pkg-slug-perm" + mock_list.side_effect = [ + ( + [ + { + "slug_perm": "first", + "content_type": "application/json", + } + ], + _page_info(page=1, page_total=2, count=2), + ), + ( + [ + { + "slug_perm": "second", + "content_type": "application/json", + } + ], + _page_info(page=2, page_total=2, count=2), + ), + ] + + result = self.runner.invoke( + metadata_, ["list", "-F", "json", "myorg/myrepo/mypkg", "--page-all"] + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + payload = json.loads(result.stdout) + self.assertEqual( + [item["slug_perm"] for item in payload["data"]], ["first", "second"] + ) + self.assertNotIn("meta", payload) + self.assertEqual(mock_list.call_count, 2) + self.assertEqual( + [call.kwargs["page"] for call in mock_list.call_args_list], [1, 2] + ) + self.assertTrue( + all( + call.kwargs["page_size"] == MAX_PAGE_SIZE + for call in mock_list.call_args_list + ) + ) + + @patch("cloudsmith_cli.cli.commands.metadata.api_list_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_list_ls_alias(self, mock_resolve, mock_list): + mock_resolve.return_value = "pkg-slug-perm" + mock_list.return_value = ([], _empty_page_info()) + + result = self.runner.invoke(metadata_, ["ls", "myorg/myrepo/mypkg"]) + + self.assertEqual(result.exit_code, 0, msg=result.output) + mock_list.assert_called_once() + + +class TestMetadataListSingle(unittest.TestCase): + def setUp(self): + self.runner = CliRunner() + + @patch("cloudsmith_cli.cli.commands.metadata.api_get_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_list_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_single_fetch_calls_get_and_skips_list( + self, mock_resolve, mock_list, mock_get + ): + mock_resolve.return_value = "pkg-slug-perm" + mock_get.return_value = { + "slug_perm": "meta-slug", + "content_type": "application/json", + "content": {"hello": "world"}, + } + + result = self.runner.invoke( + metadata_, ["list", "myorg/myrepo/mypkg", "meta-slug"] + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + mock_get.assert_called_once_with("pkg-slug-perm", "meta-slug") + mock_list.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.metadata.api_get_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_single_fetch_json_output(self, mock_resolve, mock_get): + mock_resolve.return_value = "pkg-slug-perm" + mock_get.return_value = { + "slug_perm": "meta-slug", + "content_type": "application/json", + "content": {"hello": "world"}, + } + + result = self.runner.invoke( + metadata_, + ["list", "-F", "json", "myorg/myrepo/mypkg", "meta-slug"], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + payload = json.loads(result.stdout) + self.assertEqual(payload["data"]["slug_perm"], "meta-slug") + self.assertEqual(payload["data"]["content"], {"hello": "world"}) + + +class TestMetadataAdd(unittest.TestCase): + def setUp(self): + self.runner = CliRunner() + + @patch("cloudsmith_cli.cli.commands.metadata.api_create_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_add_with_inline_content(self, mock_resolve, mock_create): + mock_resolve.return_value = "pkg-slug-perm" + mock_create.return_value = {"slug_perm": "new-slug"} + + result = self.runner.invoke( + metadata_, + [ + "add", + "myorg/myrepo/mypkg", + "--content-type", + "application/json", + "--content", + '{"foo": "bar"}', + ], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + mock_create.assert_called_once() + kwargs = mock_create.call_args.kwargs + self.assertEqual(kwargs["content"], {"foo": "bar"}) + self.assertEqual(kwargs["content_type"], "application/json") + self.assertTrue(kwargs["source_identity"].startswith("cloudsmith-cli@")) + # First positional arg is the resolved slug_perm. + self.assertEqual(mock_create.call_args.args[0], "pkg-slug-perm") + + @patch("cloudsmith_cli.cli.commands.metadata.api_create_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_add_with_file(self, mock_resolve, mock_create): + mock_resolve.return_value = "pkg-slug-perm" + mock_create.return_value = {"slug_perm": "new-slug"} + + with self.runner.isolated_filesystem(): + with open("payload.json", "w", encoding="utf-8") as fh: + fh.write('{"hello": "world"}') + + result = self.runner.invoke( + metadata_, + [ + "add", + "myorg/myrepo/mypkg", + "--content-type", + "application/json", + "--file", + "payload.json", + ], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + kwargs = mock_create.call_args.kwargs + self.assertEqual(kwargs["content"], {"hello": "world"}) + + @patch("cloudsmith_cli.cli.commands.metadata.api_create_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_add_with_stdin_file(self, mock_resolve, mock_create): + mock_resolve.return_value = "pkg-slug-perm" + mock_create.return_value = {"slug_perm": "new-slug"} + + result = self.runner.invoke( + metadata_, + [ + "add", + "myorg/myrepo/mypkg", + "--content-type", + "application/json", + "--file", + "-", + ], + input='{"from": "stdin"}', + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + kwargs = mock_create.call_args.kwargs + self.assertEqual(kwargs["content"], {"from": "stdin"}) + + @patch("cloudsmith_cli.cli.commands.metadata.api_create_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_add_rejects_both_sources(self, mock_resolve, mock_create): + mock_resolve.return_value = "pkg-slug-perm" + + with self.runner.isolated_filesystem(): + with open("payload.json", "w", encoding="utf-8") as fh: + fh.write("{}") + + result = self.runner.invoke( + metadata_, + [ + "add", + "myorg/myrepo/mypkg", + "--content-type", + "application/json", + "--file", + "payload.json", + "--content", + "{}", + ], + ) + + self.assertNotEqual(result.exit_code, 0) + self.assertIn("mutually exclusive", result.output.lower()) + mock_create.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.metadata.api_create_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_add_requires_one_source(self, mock_resolve, mock_create): + mock_resolve.return_value = "pkg-slug-perm" + + result = self.runner.invoke( + metadata_, + [ + "add", + "myorg/myrepo/mypkg", + "--content-type", + "application/json", + ], + ) + + self.assertNotEqual(result.exit_code, 0) + self.assertIn("--file", result.output) + mock_create.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.metadata.api_create_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_add_invalid_json_is_usage_error(self, mock_resolve, mock_create): + mock_resolve.return_value = "pkg-slug-perm" + + result = self.runner.invoke( + metadata_, + [ + "add", + "myorg/myrepo/mypkg", + "--content-type", + "application/json", + "--content", + "{not json", + ], + ) + + self.assertNotEqual(result.exit_code, 0) + self.assertIn("invalid", result.output.lower()) + mock_create.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.metadata.api_create_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_add_invalid_stdin_json_is_usage_error(self, mock_resolve, mock_create): + mock_resolve.return_value = "pkg-slug-perm" + + result = self.runner.invoke( + metadata_, + [ + "add", + "myorg/myrepo/mypkg", + "--content-type", + "application/json", + "--file", + "-", + ], + input="{not json", + ) + + self.assertNotEqual(result.exit_code, 0) + self.assertIn("invalid json in stdin", result.output.lower()) + mock_create.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.metadata.api_create_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_add_uses_explicit_source_identity(self, mock_resolve, mock_create): + mock_resolve.return_value = "pkg-slug-perm" + mock_create.return_value = {"slug_perm": "new-slug"} + + result = self.runner.invoke( + metadata_, + [ + "add", + "myorg/myrepo/mypkg", + "--content-type", + "application/json", + "--content", + "{}", + "--source-identity", + "ci-pipeline:42", + ], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + kwargs = mock_create.call_args.kwargs + self.assertEqual(kwargs["source_identity"], "ci-pipeline:42") + + +class TestMetadataUpdate(unittest.TestCase): + def setUp(self): + self.runner = CliRunner() + + @patch("cloudsmith_cli.cli.commands.metadata.api_update_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_update_patches_content(self, mock_resolve, mock_update): + mock_resolve.return_value = "pkg-slug-perm" + mock_update.return_value = {"slug_perm": "meta-slug"} + + result = self.runner.invoke( + metadata_, + [ + "update", + "myorg/myrepo/mypkg", + "meta-slug", + "--content", + '{"foo": "baz"}', + ], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + mock_update.assert_called_once() + args = mock_update.call_args.args + kwargs = mock_update.call_args.kwargs + self.assertEqual(args, ("pkg-slug-perm", "meta-slug")) + self.assertEqual(kwargs["content"], {"foo": "baz"}) + self.assertNotIn("source_identity", kwargs) + + @patch("cloudsmith_cli.cli.commands.metadata.api_update_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_update_patches_stdin_file(self, mock_resolve, mock_update): + mock_resolve.return_value = "pkg-slug-perm" + mock_update.return_value = {"slug_perm": "meta-slug"} + + result = self.runner.invoke( + metadata_, + [ + "update", + "myorg/myrepo/mypkg", + "meta-slug", + "--file", + "-", + ], + input='{"foo": "from-stdin"}', + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + kwargs = mock_update.call_args.kwargs + self.assertEqual(kwargs["content"], {"foo": "from-stdin"}) + self.assertNotIn("source_identity", kwargs) + + @patch("cloudsmith_cli.cli.commands.metadata.api_update_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_update_patches_source_identity_only(self, mock_resolve, mock_update): + mock_resolve.return_value = "pkg-slug-perm" + mock_update.return_value = {"slug_perm": "meta-slug"} + + result = self.runner.invoke( + metadata_, + [ + "update", + "myorg/myrepo/mypkg", + "meta-slug", + "--source-identity", + "ci-pipeline:99", + ], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + kwargs = mock_update.call_args.kwargs + self.assertEqual(kwargs["source_identity"], "ci-pipeline:99") + self.assertNotIn("content", kwargs) + + @patch("cloudsmith_cli.cli.commands.metadata.api_update_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_update_rejects_both_content_sources(self, mock_resolve, mock_update): + mock_resolve.return_value = "pkg-slug-perm" + + with self.runner.isolated_filesystem(): + with open("payload.json", "w", encoding="utf-8") as fh: + fh.write("{}") + + result = self.runner.invoke( + metadata_, + [ + "update", + "myorg/myrepo/mypkg", + "meta-slug", + "--file", + "payload.json", + "--content", + "{}", + ], + ) + + self.assertNotEqual(result.exit_code, 0) + self.assertIn("mutually exclusive", result.output.lower()) + mock_update.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.metadata.api_update_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_update_requires_some_field(self, mock_resolve, mock_update): + mock_resolve.return_value = "pkg-slug-perm" + + result = self.runner.invoke( + metadata_, + ["update", "myorg/myrepo/mypkg", "meta-slug"], + ) + + self.assertNotEqual(result.exit_code, 0) + self.assertIn("nothing to update", result.output.lower()) + mock_update.assert_not_called() + + def test_update_rejects_content_type_flag(self): + result = self.runner.invoke( + metadata_, + [ + "update", + "myorg/myrepo/mypkg", + "meta-slug", + "--content-type", + "application/json", + ], + ) + + self.assertNotEqual(result.exit_code, 0) + # Click's default is "no such option" + self.assertIn("--content-type", result.output) + + +class TestMetadataRemove(unittest.TestCase): + def setUp(self): + self.runner = CliRunner() + + @patch("cloudsmith_cli.cli.commands.metadata.api_delete_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_remove_calls_delete(self, mock_resolve, mock_delete): + mock_resolve.return_value = "pkg-slug-perm" + + result = self.runner.invoke( + metadata_, ["remove", "-y", "myorg/myrepo/mypkg", "meta-slug"] + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + mock_delete.assert_called_once_with("pkg-slug-perm", "meta-slug") + + @patch("cloudsmith_cli.cli.commands.metadata.api_delete_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_remove_prompts_and_aborts(self, mock_resolve, mock_delete): + result = self.runner.invoke( + metadata_, ["remove", "myorg/myrepo/mypkg", "meta-slug"], input="N\n" + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + self.assertIn("Are you absolutely certain", result.output) + mock_resolve.assert_not_called() + mock_delete.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.metadata.api_delete_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_remove_alias_rm(self, mock_resolve, mock_delete): + mock_resolve.return_value = "pkg-slug-perm" + + result = self.runner.invoke( + metadata_, ["rm", "-y", "myorg/myrepo/mypkg", "meta-slug"] + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + mock_delete.assert_called_once_with("pkg-slug-perm", "meta-slug") + + @patch("cloudsmith_cli.cli.commands.metadata.api_delete_metadata") + @patch("cloudsmith_cli.cli.commands.metadata.api_get_package_slug_perm") + def test_remove_json_output(self, mock_resolve, mock_delete): + mock_resolve.return_value = "pkg-slug-perm" + + result = self.runner.invoke( + metadata_, + ["remove", "-F", "json", "-y", "myorg/myrepo/mypkg", "meta-slug"], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + payload = json.loads(result.stdout) + self.assertTrue(payload["data"]["deleted"]) + self.assertEqual(payload["data"]["slug_perm"], "meta-slug") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloudsmith_cli/core/api/packages.py b/cloudsmith_cli/core/api/packages.py index cdd237ab..94879046 100644 --- a/cloudsmith_cli/core/api/packages.py +++ b/cloudsmith_cli/core/api/packages.py @@ -215,6 +215,24 @@ def get_package_tags(owner, repo, identifier): return (data.tags, data.tags_immutable) +def get_package_slug_perm(owner, repo, identifier): + """Resolve a package's permanent slug from owner/repo/identifier. + + Used by metadata commands that address packages by slug_perm. + """ + client = get_packages_api() + + with catch_raise_api_exception(): + data, _, headers = client.packages_read_with_http_info( + owner=owner, repo=repo, identifier=identifier + ) + + ratelimits.maybe_rate_limit(client, headers) + + # pylint: disable=no-member + return data.slug_perm + + def list_packages(owner, repo, **kwargs): """List packages for a repository.""" client = get_packages_api()