From 1a3ea64ccdb9450f155dfafb763c877b620caf2e Mon Sep 17 00:00:00 2001 From: Bartosz Blizniak Date: Mon, 11 May 2026 17:54:40 +0100 Subject: [PATCH] feat(eng-12002): Add metadata flag to push command --- cloudsmith_cli/cli/commands/push.py | 594 ++++++++++++- cloudsmith_cli/cli/exceptions.py | 7 + cloudsmith_cli/cli/tests/test_push.py | 1128 ++++++++++++++++++++++++- 3 files changed, 1714 insertions(+), 15 deletions(-) diff --git a/cloudsmith_cli/cli/commands/push.py b/cloudsmith_cli/cli/commands/push.py index 2cc3d663..ab7d0af1 100644 --- a/cloudsmith_cli/cli/commands/push.py +++ b/cloudsmith_cli/cli/commands/push.py @@ -1,7 +1,10 @@ """CLI/Commands - Push packages.""" +# pylint: disable=too-many-lines + import math import os +import shlex import time from datetime import datetime @@ -16,6 +19,10 @@ upload_file as api_upload_file, validate_request_file_upload, ) +from ...core.api.metadata import ( + create_metadata as api_create_metadata, + validate_metadata as api_validate_metadata, +) from ...core.api.packages import ( create_package as api_create_package, get_package_formats, @@ -24,10 +31,399 @@ ) from .. import command, decorators, utils, validators from ..exceptions import handle_api_exceptions +from ..metadata_common import ( + MetadataContentError, + ResolvedMetadata, + attach_metadata_options, + default_metadata_source_identity, + require_metadata_content_type, + resolve_metadata_content, + source_label_for, +) from ..types import ExpandPath from ..utils import maybe_spinner from .main import main +#: Env var that lets CI/CD wrappers (e.g. GHA) opt out of hard-failing the +#: push when push-time metadata attachment fails. Defaults to ``error`` so an +#: invalid metadata content aborts the upload (design requirement: metadata +#: pushes must surface failures by default). Set to ``0`` or ``warn`` to +#: downgrade failures to a warning and let the package upload regardless. +METADATA_FAILURE_MODE_ENV = "CLOUDSMITH_METADATA_FAILURE_MODE" +METADATA_FAILURE_MODE_WARN = {"0", "warn"} +#: Click option dest names for the push-time metadata flags. Used by the +#: push handler to split metadata flags off from the package-create payload +#: kwargs (the API client would otherwise reject the unknown keys). +METADATA_KWARG_NAMES = ( + "metadata_content_file", + "metadata_content", + "metadata_content_type", + "metadata_source_identity", +) + + +def _metadata_failure_is_warn(): + """Return True iff ``$CLOUDSMITH_METADATA_FAILURE_MODE`` downgrades failures. + + Single source of truth for the env-var parsing so the validation and + attach paths cannot drift (e.g. one accepting ``"false"`` and the + other not). + """ + mode = os.environ.get(METADATA_FAILURE_MODE_ENV, "error").strip().lower() + return mode in METADATA_FAILURE_MODE_WARN + + +def _metadata_content_failure_info(exc): + info = { + "status": "content_invalid", + "error": str(exc), + } + if getattr(exc, "source_label", None): + info["source"] = exc.source_label + return info + + +def _warn_metadata_failure(failure_info): + click.secho( + "Metadata content is invalid: %(error)s" % failure_info, + fg="yellow", + err=True, + ) + click.secho( + "Package upload will continue without metadata. " + f"Unset ${METADATA_FAILURE_MODE_ENV} (or set it to ``error``) to " + "fail the push instead.", + fg="yellow", + err=True, + ) + + +def resolve_push_metadata_options( + *, + metadata_content_file=None, + metadata_content=None, + metadata_content_type=None, + metadata_source_identity=None, +): + """Resolve push-time metadata flags once before package upload loops.""" + if metadata_content_file is not None and metadata_content is not None: + raise click.UsageError( + "--metadata-content-file and --metadata-content are mutually exclusive." + ) + + metadata_provided = ( + metadata_content_file is not None or metadata_content is not None + ) + if not metadata_provided: + if metadata_content_type or metadata_source_identity: + raise click.UsageError( + "Add --metadata-content-file or --metadata-content when using " + "--metadata-content-type or --metadata-source-identity." + ) + return ResolvedMetadata(provided=False, content=None), None + + require_metadata_content_type( + content_type=metadata_content_type, + content_provided=True, + option_name="--metadata-content-type", + ) + + try: + metadata = resolve_metadata_content( + content_file=metadata_content_file, + inline_content=metadata_content, + required=True, + file_option_name="--metadata-content-file", + content_option_name="--metadata-content", + ) + except MetadataContentError as exc: + if not _metadata_failure_is_warn(): + raise + + source_label = exc.source_label or source_label_for(metadata_content_file) + metadata = ResolvedMetadata( + provided=True, + content=None, + content_type=metadata_content_type, + source_identity=( + metadata_source_identity or default_metadata_source_identity() + ), + content_file=metadata_content_file, + source_label=source_label, + ) + return metadata, _metadata_content_failure_info(exc) + + return ( + attach_metadata_options( + metadata, + content_type=metadata_content_type, + source_identity=metadata_source_identity, + ), + None, + ) + + +def _handle_metadata_api_exception(ctx, opts, exc, context_msg, skip_errors=False): + """Route metadata API failures through the standard API exception handler.""" + with handle_api_exceptions( + ctx, + opts=opts, + context_msg=context_msg, + reraise_on_error=skip_errors, + ): + raise exc + + +def _print_metadata_retry_hint( + opts, + owner, + repo, + slug, + metadata_content_file, + cli_content_type, + cli_source_identity, + reason="attach_failed", +): + """Print a copy-paste ``cloudsmith metadata add`` line for failed attaches. + + Skipped in JSON output mode — the envelope already carries slugs and + failure context, so CI can reconstruct the command without text parsing. + Skipped for inline ``--metadata-content`` payloads, since they are not + safely reproducible as a single shell line (multi-line / quoting / size). + + ``reason`` distinguishes a transient/policy attach failure (``"attach_failed"``, + where retrying the same payload may succeed) from a pre-validation + failure (``"validation_failed"``, where the payload itself is broken and + must be fixed first). Wording changes accordingly. + """ + if utils.should_use_stderr(opts): + return + # Skip when no file path (inline ``--metadata-content``) or stdin ("-"), + # since neither is reproducible as a single shell line. + if not metadata_content_file or metadata_content_file == "-": + return + + parts = [ + f"cloudsmith metadata add {shlex.quote(f'{owner}/{repo}/{slug}')}", + f" --file {shlex.quote(metadata_content_file)}", + ] + if cli_source_identity: + parts.append(f" --source-identity {shlex.quote(cli_source_identity)}") + if cli_content_type: + parts.append(f" --content-type {shlex.quote(cli_content_type)}") + + if reason == "validation_failed": + heading = "Fix the metadata content, then run:" + else: + heading = "Run this command to attach metadata:" + + click.echo(err=True) + click.secho(heading, fg="yellow", err=True) + click.secho(" \\\n".join(parts), fg="yellow", err=True) + + +def validate_metadata_payload( + ctx, + opts, + content, + content_type, + source=None, + skip_errors=False, +): + """Validate metadata against ``POST /v2/metadata/validate/`` pre-upload. + + Runs before any file upload so a malformed payload does not produce an + orphan package. Returns ``None`` on success. Routes validation failure + through ``handle_api_exceptions`` by default so the push aborts before + any S3 traffic. + When ``$CLOUDSMITH_METADATA_FAILURE_MODE`` is ``warn``/``0`` it returns a + metadata-info dict instead so the caller can skip attachment but continue + the push. + + ``source`` is a human-readable label for the payload origin (file + basename, ``"stdin"``, ``"inline"``) — surfaced in the progress line so + users know which source is being validated. + """ + # pylint: disable=too-many-arguments + use_stderr = utils.should_use_stderr(opts) + + if source: + message = "Validating metadata content from {source} ... ".format( + source=click.style(source, bold=True), + ) + else: + message = "Validating metadata content ... " + + click.echo( + message, + nl=False, + err=use_stderr, + ) + + try: + with maybe_spinner(opts): + api_validate_metadata(content=content, content_type=content_type) + except ApiException as exc: + http_status = getattr(exc, "status", None) + detail = ( + getattr(exc, "detail", None) + or getattr(exc, "status_description", None) + or str(exc) + or "unknown error" + ) + + click.secho("FAILED", fg="red", err=use_stderr) + + message = ( + "Metadata content failed validation " + f"(HTTP {http_status if http_status is not None else '???'}): {detail}" + ) + failure_info = { + "status": "validation_failed", + "http_status": http_status, + "error": detail, + } + + if not _metadata_failure_is_warn(): + opts.push_metadata_info = failure_info + _handle_metadata_api_exception( + ctx, + opts, + exc, + context_msg=message, + skip_errors=skip_errors, + ) + + click.secho(message, fg="yellow", err=True) + click.secho( + "Package upload will continue without metadata. " + f"Unset ${METADATA_FAILURE_MODE_ENV} (or set it to ``error``) to " + "fail the push instead.", + fg="yellow", + err=True, + ) + return failure_info + + click.secho("OK", fg="green", err=use_stderr) + return None + + +def attach_metadata_to_package( + ctx, + opts, + owner, + repo, + slug, + slug_perm, + content, + content_type, + source_identity, + skip_errors=False, + metadata_content_file=None, + cli_content_type=None, + cli_source_identity=None, +): + """Attach a metadata entry to a freshly-created package. + + Failure is fatal by default: the API error is reported and the push + exits non-zero so CI/CD pipelines surface broken SBOM/BuildInfo uploads + instead of silently shipping a package without metadata. Wrappers that + explicitly want the legacy non-fatal behaviour can set + ``$CLOUDSMITH_METADATA_FAILURE_MODE`` to ``warn`` (or ``0``). + """ + # pylint: disable=too-many-arguments + use_stderr = utils.should_use_stderr(opts) + + click.echo( + "Attaching metadata to package %(slug)s ... " + % {"slug": click.style(slug_perm, bold=True)}, + nl=False, + err=use_stderr, + ) + + try: + with maybe_spinner(opts): + entry = api_create_metadata( + slug_perm, + content=content, + content_type=content_type, + source_identity=source_identity, + ) + except ApiException as exc: + click.secho("FAILED", fg="red", err=use_stderr) + + http_status = getattr(exc, "status", None) + detail = ( + getattr(exc, "detail", None) + or getattr(exc, "status_description", None) + or str(exc) + or "unknown error" + ) + message = ( + f"Could not attach metadata to package {slug_perm} " + f"(HTTP {http_status if http_status is not None else '???'}): {detail}" + ) + failure_info = { + "status": "attach_failed", + "http_status": http_status, + "error": detail, + } + + hint_kwargs = { + "opts": opts, + "owner": owner, + "repo": repo, + "slug": slug, + "metadata_content_file": metadata_content_file, + "cli_content_type": cli_content_type, + "cli_source_identity": cli_source_identity, + } + + if not _metadata_failure_is_warn(): + opts.push_metadata_info = failure_info + _print_metadata_retry_hint(**hint_kwargs) + _handle_metadata_api_exception( + ctx, + opts, + exc, + context_msg=message, + skip_errors=skip_errors, + ) + + click.secho(message, fg="yellow", err=True) + click.secho( + f"Package upload completed without metadata because " + f"${METADATA_FAILURE_MODE_ENV}=warn. Unset the env var " + "(or set it to ``error``) to fail the push instead.", + fg="yellow", + err=True, + ) + _print_metadata_retry_hint(**hint_kwargs) + return failure_info + + click.secho("OK", fg="green", err=use_stderr) + + metadata_slug_perm = (entry or {}).get("slug_perm") or "?" + package_path = "{owner}/{repo}/{slug}".format( + owner=click.style(owner, fg="magenta"), + repo=click.style(repo, fg="magenta"), + slug=click.style(slug, fg="green"), + ) + click.echo( + "Metadata attached: %(path)s/%(metadata)s" + % { + "path": package_path, + "metadata": click.style(metadata_slug_perm, bold=True), + }, + err=use_stderr, + ) + + return { + "status": "attached", + "slug_perm": (entry or {}).get("slug_perm"), + "entry": entry or None, + } + def validate_upload_file(ctx, opts, owner, repo, filepath, skip_errors): """Validate parameters for requesting a file upload.""" @@ -388,13 +784,39 @@ def upload_files_and_create_package( wait_interval, skip_errors, sync_attempts, + metadata_content_file=None, + metadata_content=None, + metadata_content_type=None, + metadata_source_identity=None, + metadata=None, + metadata_failure_info=None, **kwargs, ): """Upload package files and create a new package.""" - # pylint: disable=unused-argument + # pylint: disable=unused-argument,too-many-arguments,too-many-locals owner, repo = owner_repo - # 1. Validate package create parameters + # Reset push-time metadata state for this call. ``handle_api_exceptions`` + # consults this attribute to surface validation/attach context in the + # JSON error envelope; an unset value would leak prior state on retries. + opts.push_metadata_info = None + + # 0. Resolve push-time metadata before package work. The dynamic command + # handler resolves once for multi-file pushes so stdin is consumed once; + # direct callers can still pass the metadata flags for test coverage. + if metadata is None: + metadata, metadata_failure_info = resolve_push_metadata_options( + metadata_content_file=metadata_content_file, + metadata_content=metadata_content, + metadata_content_type=metadata_content_type, + metadata_source_identity=metadata_source_identity, + ) + + should_attach_metadata = metadata.provided and metadata_failure_info is None + + # 1. Validate package create parameters. This runs before the metadata + # pre-validation so a typo in --name/--version fails fast without + # burning a /v2/metadata/validate/ round-trip first. validate_create_package( ctx=ctx, opts=opts, @@ -405,6 +827,26 @@ def upload_files_and_create_package( **kwargs, ) + # 1b. Pre-validate metadata against the server-side schema endpoint so a + # malformed payload cannot produce an orphan package (the upload would + # succeed and only the attach would fail). + if metadata_failure_info is not None: + opts.push_metadata_info = metadata_failure_info + _warn_metadata_failure(metadata_failure_info) + elif should_attach_metadata: + validation_failure = validate_metadata_payload( + ctx=ctx, + opts=opts, + content=metadata.content, + content_type=metadata.content_type, + source=metadata.source_label, + skip_errors=skip_errors, + ) + if validation_failure is not None: + # Warn-mode validation failure: keep the push, drop the attach. + should_attach_metadata = False + opts.push_metadata_info = validation_failure + # 2. Validate file upload parameters md5_checksums = {} for k, v in kwargs.items(): @@ -484,10 +926,44 @@ def upload_files_and_create_package( **kwargs, ) + # 5. Attach push-time metadata, if provided AND it passed validation. + # Warn-mode metadata failures leave opts.push_metadata_info populated + # and should_attach_metadata=False; surface a retry hint now that we + # have the package slug. Skipped in JSON mode and for inline payloads. + if should_attach_metadata: + opts.push_metadata_info = attach_metadata_to_package( + ctx=ctx, + opts=opts, + owner=owner, + repo=repo, + slug=slug, + slug_perm=slug_perm, + content=metadata.content, + content_type=metadata.content_type, + source_identity=metadata.source_identity, + skip_errors=skip_errors, + metadata_content_file=metadata.content_file, + cli_content_type=metadata.content_type, + cli_source_identity=metadata_source_identity, + ) + elif metadata.provided: + # Metadata resolution/validation already warned the user; the payload + # is broken so a straight retry would fail. Use the "fix first" hint. + _print_metadata_retry_hint( + opts=opts, + owner=owner, + repo=repo, + slug=slug, + metadata_content_file=metadata.content_file, + cli_content_type=metadata.content_type, + cli_source_identity=metadata_source_identity, + reason="validation_failed", + ) + if no_wait_for_sync: return slug_perm, slug - # 5. (optionally) Wait for the package to synchronise + # 6. (optionally) Wait for the package to synchronise wait_for_package_sync( ctx=ctx, opts=opts, @@ -585,6 +1061,60 @@ def create_push_handlers(): is_flag=True, help="Execute in dry run mode (don't upload anything.)", ) + @click.option( + "--metadata-content-file", + "metadata_content_file", + type=click.Path( + exists=True, + dir_okay=False, + readable=True, + resolve_path=True, + allow_dash=True, + ), + default=None, + help=( + "Read metadata content from a JSON file " + "(for example, SBOM or BuildInfo). Use '-' for stdin. " + "Content must be a JSON object. " + "Mutually exclusive with --metadata-content. " + "Metadata failures abort the push by default; set " + "$CLOUDSMITH_METADATA_FAILURE_MODE=warn (or 0) to downgrade " + "to a warning and keep the package upload." + ), + ) + @click.option( + "--metadata-content", + "metadata_content", + default=None, + help=( + "Set metadata content from inline JSON. Content must be a " + "JSON object. " + "Mutually exclusive with --metadata-content-file. " + "Metadata failures abort the push by default; set " + "$CLOUDSMITH_METADATA_FAILURE_MODE=warn (or 0) to downgrade " + "to a warning and keep the package upload." + ), + ) + @click.option( + "--metadata-content-type", + "metadata_content_type", + default=None, + help=( + "Content type for metadata content " + "(for example, 'application/vnd.jfrog.buildinfo+json'). " + "Required when metadata content is supplied and determines " + "the schema used for validation." + ), + ) + @click.option( + "--metadata-source-identity", + "metadata_source_identity", + default=None, + help=( + "Identifier for the metadata source. " + "Defaults to 'cloudsmith-cli@'." + ), + ) @click.pass_context def push_handler(ctx, *args, **kwargs): """Handle upload for a specific package format.""" @@ -598,19 +1128,56 @@ def push_handler(ctx, *args, **kwargs): owner_repo = owner_repo[0:2] kwargs["owner_repo"] = owner_repo + # Metadata flags are not part of the package-create payload, so + # pop them and forward them as explicit kwargs so they don't leak + # into validate_create_package() / create_package(). + metadata_kwargs = { + key: kwargs.pop(key, None) for key in METADATA_KWARG_NAMES + } + package_files = kwargs.pop("package_file") if not isinstance(package_files, tuple): package_files = (package_files,) + # Reject multi-file push combined with metadata flags. A single + # metadata payload semantically belongs to one package; silently + # fanning it out across N packages (and validating + attaching it + # N times) is almost never what the user wants. Force them to + # push files individually with metadata, or drop the flags. + metadata_flags_set = any( + metadata_kwargs.get(k) for k in METADATA_KWARG_NAMES + ) + if len(package_files) > 1 and metadata_flags_set: + raise click.UsageError( + "Metadata flags (--metadata-content-file, --metadata-content, " + "--metadata-content-type, --metadata-source-identity) cannot " + "be combined with multiple package files. Push files " + "individually when attaching metadata." + ) + + metadata, metadata_failure_info = resolve_push_metadata_options( + **metadata_kwargs + ) + results = [] for package_file in package_files: kwargs["package_file"] = package_file try: click.echo(err=utils.should_use_stderr(opts)) - res = upload_files_and_create_package(ctx, *args, **kwargs) + res = upload_files_and_create_package( + ctx, + *args, + **kwargs, + **metadata_kwargs, + metadata=metadata, + metadata_failure_info=metadata_failure_info, + ) if res: - results.append(res) + # ``upload_files_and_create_package`` resets and then + # populates ``opts.push_metadata_info`` on every call, + # so reading it here always reflects this iteration. + results.append((res, opts.push_metadata_info)) except ApiException: click.secho( "Skipping error and moving on.", @@ -622,14 +1189,15 @@ def push_handler(ctx, *args, **kwargs): if utils.should_use_stderr(opts): data = [] - for slug_perm, slug in results: - data.append( - { - "slug_perm": slug_perm, - "slug": slug, - "status": "OK", # Assuming success if we got here - } - ) + for (slug_perm, slug), metadata_info in results: + entry = { + "slug_perm": slug_perm, + "slug": slug, + "status": "OK", # Assuming success if we got here + } + if metadata_info is not None: + entry["metadata_attachment"] = metadata_info + data.append(entry) if len(data) == 1: utils.maybe_print_as_json(opts, data[0]) diff --git a/cloudsmith_cli/cli/exceptions.py b/cloudsmith_cli/cli/exceptions.py index 162b8b8f..be12b733 100644 --- a/cloudsmith_cli/cli/exceptions.py +++ b/cloudsmith_cli/cli/exceptions.py @@ -45,6 +45,13 @@ def handle_api_exceptions( if fields: error_data["fields"] = fields + # Surface push-time metadata context (validation/attach result) + # in the same JSON envelope so a downstream package-create or + # sync failure does not lose the earlier metadata signal. + metadata_context = getattr(opts, "push_metadata_info", None) + if metadata_context is not None: + error_data["metadata_attachment"] = metadata_context + # Print to stdout import json diff --git a/cloudsmith_cli/cli/tests/test_push.py b/cloudsmith_cli/cli/tests/test_push.py index 247e62f9..ea4b6a71 100644 --- a/cloudsmith_cli/cli/tests/test_push.py +++ b/cloudsmith_cli/cli/tests/test_push.py @@ -1,10 +1,26 @@ +# pylint: disable=too-many-lines +import json +import os +import tempfile import unittest +from types import SimpleNamespace from unittest.mock import MagicMock, patch -from ..commands.push import upload_files_and_create_package +import click +import pytest +from ...core.api.exceptions import ApiException +from ..commands.push import ( + _print_metadata_retry_hint, + attach_metadata_to_package, + resolve_push_metadata_options, + upload_files_and_create_package, + validate_metadata_payload, +) +from ..metadata_common import ResolvedMetadata -# pylint: disable=too-many-instance-attributes + +# pylint: disable=too-many-instance-attributes,too-many-public-methods class TestPush(unittest.TestCase): def setUp(self): self.mock_ctx = MagicMock() @@ -121,6 +137,944 @@ def test_upload_files_and_create_package(self): **create_package_kwargs, ) + def test_upload_files_and_create_package_with_metadata(self): + """Successful push with metadata creates package + metadata entry.""" + input_kwargs = { + "package_file": "package/file/path", + "name": "test_package", + "version": "1.0.0", + } + metadata_kwargs = { + "metadata_content": '{"git_sha": "abc123"}', + "metadata_content_type": "application/vnd.jfrog.buildinfo+json", + "metadata_source_identity": "github-actions@example", + } + + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file" + ) as mock_validate_upload_file, + patch("cloudsmith_cli.cli.commands.push.upload_file") as mock_upload_file, + patch( + "cloudsmith_cli.cli.commands.push.create_package" + ) as mock_create_package, + patch("cloudsmith_cli.cli.commands.push.api_validate_metadata"), + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata" + ) as mock_create_metadata, + patch("cloudsmith_cli.cli.commands.push.wait_for_package_sync"), + ): + mock_validate_upload_file.return_value = "checksum" + mock_upload_file.return_value = "package_file_identifier" + mock_create_package.return_value = ("slug-perm-abc", "test_package_slug") + mock_create_metadata.return_value = {"slug_perm": "meta-slug-xyz"} + + upload_files_and_create_package( + self.mock_ctx, + self.mock_opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + **input_kwargs, + **metadata_kwargs, + ) + + mock_create_metadata.assert_called_once_with( + "slug-perm-abc", + content={"git_sha": "abc123"}, + content_type="application/vnd.jfrog.buildinfo+json", + source_identity="github-actions@example", + ) + + def test_upload_files_and_create_package_with_json_null_metadata(self): + """Explicit JSON null is rejected before upload by default.""" + with ( + patch( + "cloudsmith_cli.cli.commands.push.validate_create_package" + ) as mock_validate_create_package, + patch("cloudsmith_cli.cli.commands.push.api_validate_metadata"), + patch("cloudsmith_cli.cli.commands.push.api_create_metadata"), + ): + with pytest.raises(click.ClickException, match="JSON object"): + upload_files_and_create_package( + self.mock_ctx, + MagicMock(spec=[]), + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + package_file="path", + name="x", + version="1", + metadata_content="null", + metadata_content_type="application/json", + ) + + mock_validate_create_package.assert_not_called() + + def test_upload_attach_publishes_metadata_info_to_opts(self): + """Attach result is published on opts.push_metadata_info for JSON output.""" + api_entry = { + "slug_perm": "meta-slug-xyz", + "content_type": "application/json", + "classification": "GENERIC", + "source_kind": "CUSTOMER", + "source_identity": "github-actions@demo", + "content": {"git_sha": "abc123"}, + } + + opts = MagicMock(spec=[]) + + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file", + return_value="checksum", + ), + patch( + "cloudsmith_cli.cli.commands.push.upload_file", + return_value="package_file_identifier", + ), + patch( + "cloudsmith_cli.cli.commands.push.create_package", + return_value=("slug-perm-abc", "test_package_slug"), + ), + patch("cloudsmith_cli.cli.commands.push.api_validate_metadata"), + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata", + return_value=api_entry, + ), + patch("cloudsmith_cli.cli.commands.push.wait_for_package_sync"), + ): + result = upload_files_and_create_package( + self.mock_ctx, + opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + package_file="path", + name="x", + version="1", + metadata_content='{"git_sha": "abc123"}', + metadata_content_type="application/json", + ) + + assert result == ("slug-perm-abc", "test_package_slug") + assert opts.push_metadata_info == { + "status": "attached", + "slug_perm": "meta-slug-xyz", + "entry": api_entry, + } + + def test_upload_files_and_create_package_metadata_failure_warn_does_not_fail_push( + self, + ): + """With CLOUDSMITH_METADATA_FAILURE_MODE=warn the push survives a bad attach.""" + input_kwargs = { + "package_file": "package/file/path", + "name": "test_package", + "version": "1.0.0", + } + metadata_kwargs = { + "metadata_content": '{"git_sha": "abc123"}', + "metadata_content_type": "application/json", + } + + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file" + ) as mock_validate_upload_file, + patch("cloudsmith_cli.cli.commands.push.upload_file") as mock_upload_file, + patch( + "cloudsmith_cli.cli.commands.push.create_package" + ) as mock_create_package, + patch("cloudsmith_cli.cli.commands.push.api_validate_metadata"), + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata" + ) as mock_create_metadata, + patch( + "cloudsmith_cli.cli.commands.push.wait_for_package_sync" + ) as mock_wait_for_sync, + patch.dict( + "cloudsmith_cli.cli.commands.push.os.environ", + {"CLOUDSMITH_METADATA_FAILURE_MODE": "warn"}, + ), + ): + mock_validate_upload_file.return_value = "checksum" + mock_upload_file.return_value = "package_file_identifier" + mock_create_package.return_value = ("slug-perm-abc", "test_package_slug") + mock_create_metadata.side_effect = ApiException( + status=422, detail="Schema validation failed" + ) + + opts = MagicMock(spec=[]) + + # Push must complete without raising, returning the slug pair. + result = upload_files_and_create_package( + self.mock_ctx, + opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + **input_kwargs, + **metadata_kwargs, + ) + + assert result == ("slug-perm-abc", "test_package_slug") + assert opts.push_metadata_info == { + "status": "attach_failed", + "http_status": 422, + "error": "Schema validation failed", + } + mock_create_metadata.assert_called_once() + # Sync wait still happens — push behaviour unchanged otherwise. + mock_wait_for_sync.assert_called_once() + + def test_upload_files_and_create_package_metadata_failure_default_aborts_push(self): + """Default failure mode (no env override) aborts the push on a bad attach.""" + input_kwargs = { + "package_file": "package/file/path", + "name": "test_package", + "version": "1.0.0", + } + + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file", + return_value="checksum", + ), + patch( + "cloudsmith_cli.cli.commands.push.upload_file", + return_value="package_file_identifier", + ), + patch( + "cloudsmith_cli.cli.commands.push.create_package", + return_value=("slug-perm-abc", "test_package_slug"), + ), + patch("cloudsmith_cli.cli.commands.push.api_validate_metadata"), + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata", + side_effect=ApiException(status=422, detail="bad payload"), + ), + patch("cloudsmith_cli.cli.commands.push.wait_for_package_sync"), + patch.dict( + "cloudsmith_cli.cli.commands.push.os.environ", + {}, + clear=False, + ) as patched_env, + ): + patched_env.pop("CLOUDSMITH_METADATA_FAILURE_MODE", None) + opts = SimpleNamespace( + output=None, + push_metadata_info=None, + verbose=False, + debug=False, + api_key=None, + api_host=None, + ) + ctx = click.Context(click.Command("test")) + + with pytest.raises(click.exceptions.Exit) as exc_info: + upload_files_and_create_package( + ctx, + opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + metadata_content='{"x": 1}', + metadata_content_type="application/json", + **input_kwargs, + ) + + assert exc_info.value.exit_code == 422 + assert opts.push_metadata_info == { + "status": "attach_failed", + "http_status": 422, + "error": "bad payload", + } + + def test_prevalidate_metadata_404_aborts_before_upload_by_default(self): + """A validation endpoint API error aborts before upload by default.""" + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file", + return_value="checksum", + ) as mock_validate_upload_file, + patch( + "cloudsmith_cli.cli.commands.push.upload_file", + return_value="package_file_identifier", + ) as mock_upload_file, + patch( + "cloudsmith_cli.cli.commands.push.create_package", + return_value=("slug-perm-abc", "test_package_slug"), + ) as mock_create_package, + patch( + "cloudsmith_cli.cli.commands.push.api_validate_metadata", + side_effect=ApiException(status=404, detail="Not Found"), + ), + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata", + return_value={"slug_perm": "meta-slug-xyz"}, + ), + patch("cloudsmith_cli.cli.commands.push.wait_for_package_sync"), + ): + opts = SimpleNamespace( + output=None, + push_metadata_info=None, + verbose=False, + debug=False, + api_key=None, + api_host=None, + ) + ctx = click.Context(click.Command("test")) + with pytest.raises(click.exceptions.Exit) as exc_info: + upload_files_and_create_package( + ctx, + opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + package_file="path", + name="x", + version="1", + metadata_content='{"x": 1}', + metadata_content_type="application/json", + ) + + assert exc_info.value.exit_code == 404 + mock_validate_upload_file.assert_not_called() + mock_upload_file.assert_not_called() + mock_create_package.assert_not_called() + + def test_prevalidate_metadata_warn_skips_attach_but_uploads_package(self): + """Validation failure in warn mode drops attach but lets the package upload.""" + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file", + return_value="checksum", + ), + patch( + "cloudsmith_cli.cli.commands.push.upload_file", + return_value="package_file_identifier", + ), + patch( + "cloudsmith_cli.cli.commands.push.create_package", + return_value=("slug-perm-abc", "test_package_slug"), + ), + patch( + "cloudsmith_cli.cli.commands.push.api_validate_metadata", + side_effect=ApiException(status=422, detail="schema mismatch"), + ), + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata" + ) as mock_create_metadata, + patch("cloudsmith_cli.cli.commands.push.wait_for_package_sync"), + patch.dict( + "cloudsmith_cli.cli.commands.push.os.environ", + {"CLOUDSMITH_METADATA_FAILURE_MODE": "warn"}, + ), + ): + opts = MagicMock(spec=[]) + result = upload_files_and_create_package( + self.mock_ctx, + opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + package_file="path", + name="x", + version="1", + metadata_content='{"x": 1}', + metadata_content_type="application/json", + ) + + # Package created successfully, metadata attach call never reached. + assert result == ("slug-perm-abc", "test_package_slug") + assert opts.push_metadata_info == { + "status": "validation_failed", + "http_status": 422, + "error": "schema mismatch", + } + mock_create_metadata.assert_not_called() + + def test_local_metadata_content_warn_skips_attach_but_uploads_package(self): + """Local JSON object validation failure respects warn mode.""" + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file", + return_value="checksum", + ), + patch( + "cloudsmith_cli.cli.commands.push.upload_file", + return_value="package_file_identifier", + ), + patch( + "cloudsmith_cli.cli.commands.push.create_package", + return_value=("slug-perm-abc", "test_package_slug"), + ), + patch( + "cloudsmith_cli.cli.commands.push.api_validate_metadata" + ) as mock_validate_metadata, + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata" + ) as mock_create_metadata, + patch("cloudsmith_cli.cli.commands.push.wait_for_package_sync"), + patch.dict( + "cloudsmith_cli.cli.commands.push.os.environ", + {"CLOUDSMITH_METADATA_FAILURE_MODE": "warn"}, + ), + ): + opts = MagicMock(spec=[]) + result = upload_files_and_create_package( + self.mock_ctx, + opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + package_file="path", + name="x", + version="1", + metadata_content="[]", + metadata_content_type="application/json", + ) + + assert result == ("slug-perm-abc", "test_package_slug") + assert opts.push_metadata_info["status"] == "content_invalid" + mock_validate_metadata.assert_not_called() + mock_create_metadata.assert_not_called() + + def test_prevalidate_metadata_default_aborts_before_upload(self): + """Validation failure aborts before any file upload by default.""" + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file" + ) as mock_validate_upload_file, + patch("cloudsmith_cli.cli.commands.push.upload_file") as mock_upload_file, + patch( + "cloudsmith_cli.cli.commands.push.create_package" + ) as mock_create_package, + patch( + "cloudsmith_cli.cli.commands.push.api_validate_metadata", + side_effect=ApiException(status=422, detail="schema mismatch"), + ), + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata" + ) as mock_create_metadata, + patch.dict( + "cloudsmith_cli.cli.commands.push.os.environ", + {}, + clear=False, + ) as patched_env, + ): + patched_env.pop("CLOUDSMITH_METADATA_FAILURE_MODE", None) + opts = SimpleNamespace( + output=None, + push_metadata_info=None, + verbose=False, + debug=False, + api_key=None, + api_host=None, + ) + ctx = click.Context(click.Command("test")) + + with pytest.raises(click.exceptions.Exit) as exc_info: + upload_files_and_create_package( + ctx, + opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + package_file="path", + name="x", + version="1", + metadata_content='{"x": 1}', + metadata_content_type="application/json", + ) + + assert exc_info.value.exit_code == 422 + assert opts.push_metadata_info == { + "status": "validation_failed", + "http_status": 422, + "error": "schema mismatch", + } + + # No upload, no package, no attach. + mock_validate_upload_file.assert_not_called() + mock_upload_file.assert_not_called() + mock_create_package.assert_not_called() + mock_create_metadata.assert_not_called() + + def test_metadata_content_type_without_payload_errors(self): + """Setting only --metadata-content-type (no payload) is a usage error.""" + with pytest.raises(click.UsageError, match="Add --metadata-content"): + upload_files_and_create_package( + self.mock_ctx, + self.mock_opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + metadata_content_type="application/json", + package_file="path", + name="x", + version="1", + ) + + def test_metadata_source_identity_without_payload_errors(self): + """Setting only --metadata-source-identity is a usage error.""" + with pytest.raises(click.UsageError, match="Add --metadata-content"): + upload_files_and_create_package( + self.mock_ctx, + self.mock_opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + metadata_source_identity="ci@example", + package_file="path", + name="x", + version="1", + ) + + def test_metadata_content_without_content_type_errors(self): + """Metadata content requires --metadata-content-type.""" + with pytest.raises(click.UsageError, match="--metadata-content-type"): + upload_files_and_create_package( + self.mock_ctx, + self.mock_opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + metadata_content="{}", + package_file="path", + name="x", + version="1", + ) + + def test_resolve_push_metadata_options_inline_json(self): + metadata, failure = resolve_push_metadata_options( + metadata_content='{"k": "v"}', + metadata_content_type="application/json", + ) + + assert failure is None + assert metadata.content == {"k": "v"} + assert metadata.content_type == "application/json" + assert metadata.source_label == "inline" + + def test_resolve_push_metadata_options_json_null_rejected(self): + with pytest.raises(click.ClickException, match="JSON object"): + resolve_push_metadata_options( + metadata_content="null", + metadata_content_type="application/json", + ) + + def test_resolve_push_metadata_options_invalid_json_rejected(self): + with pytest.raises(click.ClickException, match="Invalid JSON"): + resolve_push_metadata_options( + metadata_content="not-json", + metadata_content_type="application/json", + ) + + def test_resolve_push_metadata_options_invalid_json_warn_returns_failure(self): + with patch.dict( + "cloudsmith_cli.cli.commands.push.os.environ", + {"CLOUDSMITH_METADATA_FAILURE_MODE": "warn"}, + ): + metadata, failure = resolve_push_metadata_options( + metadata_content="not-json", + metadata_content_type="application/json", + ) + + assert metadata.provided is True + assert metadata.content is None + assert failure["status"] == "content_invalid" + assert "Invalid JSON" in failure["error"] + + def test_resolve_push_metadata_options_neither_returns_not_provided(self): + metadata, failure = resolve_push_metadata_options() + + assert metadata.provided is False + assert metadata.content is None + assert failure is None + + def test_resolve_push_metadata_options_mutex(self): + with pytest.raises(click.UsageError, match="mutually exclusive"): + resolve_push_metadata_options( + metadata_content_file="/tmp/foo.json", + metadata_content='{"k": "v"}', + metadata_content_type="application/json", + ) + + def test_resolve_push_metadata_options_reads_stdin_via_dash(self): + """``--metadata-content-file -`` reads JSON from stdin once.""" + import io + + stdin = io.StringIO('{"git_sha": "abc123"}') + with patch( + "cloudsmith_cli.cli.metadata_common.click.get_text_stream", + return_value=stdin, + ): + metadata, failure = resolve_push_metadata_options( + metadata_content_file="-", + metadata_content_type="application/json", + ) + + assert failure is None + assert metadata.content == {"git_sha": "abc123"} + assert metadata.source_label == "stdin" + + def test_cached_stdin_metadata_payload_reused_across_uploads(self): + """The push command resolves ``-`` metadata before per-file uploads.""" + import io + + open_calls = [] + + def fake_get_text_stream(name): + assert name == "stdin" + open_calls.append(name) + return io.StringIO('{"git_sha": "abc123"}') + + metadata_kwargs = { + "metadata_content_file": "-", + "metadata_content": None, + "metadata_content_type": "application/json", + "metadata_source_identity": None, + } + + with patch( + "cloudsmith_cli.cli.metadata_common.click.get_text_stream", + side_effect=fake_get_text_stream, + ): + metadata, metadata_failure_info = resolve_push_metadata_options( + **metadata_kwargs + ) + + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file", + return_value="checksum", + ), + patch( + "cloudsmith_cli.cli.commands.push.upload_file", + return_value="package_file_identifier", + ), + patch( + "cloudsmith_cli.cli.commands.push.create_package", + side_effect=[ + ("slug-perm-abc", "test_package_slug"), + ("slug-perm-def", "test_package_slug_2"), + ], + ), + patch("cloudsmith_cli.cli.commands.push.api_validate_metadata"), + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata", + return_value={"slug_perm": "meta-slug-xyz"}, + ) as mock_create_metadata, + patch("cloudsmith_cli.cli.commands.push.wait_for_package_sync"), + ): + for package_file in ("path-1", "path-2"): + upload_files_and_create_package( + self.mock_ctx, + MagicMock(spec=[]), + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + package_file=package_file, + name="x", + version="1", + metadata=metadata, + metadata_failure_info=metadata_failure_info, + ) + + assert open_calls == ["stdin"] + assert mock_create_metadata.call_count == 2 + for call in mock_create_metadata.call_args_list: + assert call.kwargs["content"] == {"git_sha": "abc123"} + assert call.kwargs["content_type"] == "application/json" + + def test_metadata_content_file_round_trip(self): + """A JSON file given via --metadata-content-file reaches the API call.""" + payload = {"git_sha": "abc123", "build": 42} + fd, path = tempfile.mkstemp(suffix=".json") + try: + with os.fdopen(fd, "w", encoding="utf-8") as fh: + json.dump(payload, fh) + + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file", + return_value="checksum", + ), + patch( + "cloudsmith_cli.cli.commands.push.upload_file", + return_value="package_file_identifier", + ), + patch( + "cloudsmith_cli.cli.commands.push.create_package", + return_value=("slug-perm-abc", "test_package_slug"), + ), + patch("cloudsmith_cli.cli.commands.push.api_validate_metadata"), + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata", + return_value={"slug_perm": "meta-slug-xyz"}, + ) as mock_create_metadata, + patch("cloudsmith_cli.cli.commands.push.wait_for_package_sync"), + ): + upload_files_and_create_package( + self.mock_ctx, + MagicMock(spec=[]), + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + package_file="path", + name="x", + version="1", + metadata_content_file=path, + metadata_content_type="application/json", + ) + + kwargs = mock_create_metadata.call_args.kwargs + assert kwargs["content"] == payload + finally: + os.unlink(path) + + def test_metadata_default_source_identity(self): + """Without --metadata-source-identity the default is cloudsmith-cli@.""" + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file", + return_value="checksum", + ), + patch( + "cloudsmith_cli.cli.commands.push.upload_file", + return_value="package_file_identifier", + ), + patch( + "cloudsmith_cli.cli.commands.push.create_package", + return_value=("slug-perm-abc", "test_package_slug"), + ), + patch("cloudsmith_cli.cli.commands.push.api_validate_metadata"), + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata", + return_value={"slug_perm": "meta-slug-xyz"}, + ) as mock_create_metadata, + patch( + "cloudsmith_cli.cli.metadata_common.get_cli_version", + return_value="9.9.9", + ), + patch("cloudsmith_cli.cli.commands.push.wait_for_package_sync"), + ): + upload_files_and_create_package( + self.mock_ctx, + MagicMock(spec=[]), + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + package_file="path", + name="x", + version="1", + metadata_content='{"x": 1}', + metadata_content_type="application/json", + ) + + assert ( + mock_create_metadata.call_args.kwargs["source_identity"] + == "cloudsmith-cli@9.9.9" + ) + + def test_metadata_retry_hint_emitted_on_attach_warn_failure(self): + """Attach failure (warn mode, file payload) routes to the hint helper.""" + opts = SimpleNamespace(output=None, push_metadata_info=None) + metadata = ResolvedMetadata( + provided=True, + content={"x": 1}, + content_type="application/vnd.cyclonedx+json", + source_identity="ci@gha", + content_file="/tmp/sbom.json", + source_label="sbom.json", + ) + + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file", + return_value="checksum", + ), + patch( + "cloudsmith_cli.cli.commands.push.upload_file", + return_value="package_file_identifier", + ), + patch( + "cloudsmith_cli.cli.commands.push.create_package", + return_value=("slug-perm-abc", "test_package_slug"), + ), + patch("cloudsmith_cli.cli.commands.push.api_validate_metadata"), + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata", + side_effect=ApiException(status=422, detail="bad payload"), + ), + patch("cloudsmith_cli.cli.commands.push.wait_for_package_sync"), + patch("cloudsmith_cli.cli.commands.push._print_metadata_retry_hint") as spy, + patch.dict( + "cloudsmith_cli.cli.commands.push.os.environ", + {"CLOUDSMITH_METADATA_FAILURE_MODE": "warn"}, + ), + ): + upload_files_and_create_package( + self.mock_ctx, + opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + package_file="path", + name="x", + version="1", + metadata_content_type="application/vnd.cyclonedx+json", + metadata_source_identity="ci@gha", + metadata=metadata, + ) + + spy.assert_called_once() + kwargs = spy.call_args.kwargs + assert kwargs["owner"] == self.owner + assert kwargs["repo"] == self.repo + assert kwargs["slug"] == "test_package_slug" + assert kwargs["metadata_content_file"] == "/tmp/sbom.json" + assert kwargs["cli_content_type"] == "application/vnd.cyclonedx+json" + assert kwargs["cli_source_identity"] == "ci@gha" + + def test_metadata_retry_hint_emitted_on_validation_warn_failure(self): + """Pre-validation warn failure routes to the hint helper after create.""" + opts = SimpleNamespace(output=None, push_metadata_info=None) + metadata = ResolvedMetadata( + provided=True, + content={"x": 1}, + content_type="application/json", + source_identity="cloudsmith-cli@9.9.9", + content_file="/tmp/buildinfo.json", + source_label="buildinfo.json", + ) + + with ( + patch("cloudsmith_cli.cli.commands.push.validate_create_package"), + patch( + "cloudsmith_cli.cli.commands.push.validate_upload_file", + return_value="checksum", + ), + patch( + "cloudsmith_cli.cli.commands.push.upload_file", + return_value="package_file_identifier", + ), + patch( + "cloudsmith_cli.cli.commands.push.create_package", + return_value=("slug-perm-abc", "test_package_slug"), + ), + patch( + "cloudsmith_cli.cli.commands.push.api_validate_metadata", + side_effect=ApiException(status=422, detail="schema mismatch"), + ), + patch("cloudsmith_cli.cli.commands.push.api_create_metadata"), + patch("cloudsmith_cli.cli.commands.push.wait_for_package_sync"), + patch("cloudsmith_cli.cli.commands.push._print_metadata_retry_hint") as spy, + patch.dict( + "cloudsmith_cli.cli.commands.push.os.environ", + {"CLOUDSMITH_METADATA_FAILURE_MODE": "warn"}, + ), + ): + upload_files_and_create_package( + self.mock_ctx, + opts, + self.package_type, + [self.owner, self.repo], + self.dry_run, + self.no_wait_for_sync, + self.wait_interval, + self.skip_errors, + self.sync_attempts, + package_file="path", + name="x", + version="1", + metadata_content_type="application/json", + metadata=metadata, + ) + + spy.assert_called_once() + kwargs = spy.call_args.kwargs + assert kwargs["slug"] == "test_package_slug" + assert kwargs["metadata_content_file"] == "/tmp/buildinfo.json" + def test_upload_files_and_create_package_extra_files(self): # Values passed in from the command line input_kwargs = { @@ -243,3 +1197,173 @@ def test_upload_files_and_create_package_extra_files(self): skip_errors=self.skip_errors, **create_package_kwargs, ) + + +# Plain pytest functions for hint output — capsys is not auto-injected into +# unittest.TestCase methods, so these live outside the class. + + +def _hint_opts(output=None): + return SimpleNamespace(output=output, push_metadata_info=None) + + +def _api_opts(output="json"): + return SimpleNamespace( + output=output, + push_metadata_info=None, + verbose=False, + debug=False, + api_key=None, + api_host=None, + ) + + +def test_validate_metadata_payload_json_failure_uses_api_error_envelope(capsys): + ctx = click.Context(click.Command("push")) + opts = _api_opts(output="json") + + with ( + patch( + "cloudsmith_cli.cli.commands.push.api_validate_metadata", + side_effect=ApiException(status=422, detail="bad payload"), + ), + patch.dict( + "cloudsmith_cli.cli.commands.push.os.environ", + {}, + clear=False, + ) as patched_env, + ): + patched_env.pop("CLOUDSMITH_METADATA_FAILURE_MODE", None) + with pytest.raises(click.exceptions.Exit) as exc_info: + validate_metadata_payload( + ctx=ctx, + opts=opts, + content={"x": 1}, + content_type="application/json", + source="inline", + ) + + assert exc_info.value.exit_code == 422 + data = json.loads(capsys.readouterr().out) + assert data["detail"] == "bad payload" + assert data["help"]["context"].startswith("Metadata content failed validation") + assert data["metadata_attachment"] == { + "status": "validation_failed", + "http_status": 422, + "error": "bad payload", + } + + +def test_attach_metadata_json_failure_uses_api_error_envelope(capsys): + ctx = click.Context(click.Command("push")) + opts = _api_opts(output="pretty_json") + + with ( + patch( + "cloudsmith_cli.cli.commands.push.api_create_metadata", + side_effect=ApiException(status=422, detail="bad payload"), + ), + patch.dict( + "cloudsmith_cli.cli.commands.push.os.environ", + {}, + clear=False, + ) as patched_env, + ): + patched_env.pop("CLOUDSMITH_METADATA_FAILURE_MODE", None) + with pytest.raises(click.exceptions.Exit) as exc_info: + attach_metadata_to_package( + ctx=ctx, + opts=opts, + owner="acme", + repo="repo", + slug="hello-txt", + slug_perm="hello-txt-abc", + content={"x": 1}, + content_type="application/json", + source_identity="ci@example", + ) + + assert exc_info.value.exit_code == 422 + data = json.loads(capsys.readouterr().out) + assert data["detail"] == "bad payload" + assert data["help"]["context"].startswith("Could not attach metadata") + assert data["metadata_attachment"] == { + "status": "attach_failed", + "http_status": 422, + "error": "bad payload", + } + + +def test_print_metadata_retry_hint_emits_copy_paste_command(capsys): + _print_metadata_retry_hint( + opts=_hint_opts(), + owner="acme", + repo="repo", + slug="hello-txt-abc", + metadata_content_file="/tmp/sbom.json", + cli_content_type="application/vnd.cyclonedx+json", + cli_source_identity="ci@gha", + ) + err = capsys.readouterr().err + assert "Run this command to attach metadata:" in err + assert "cloudsmith metadata add acme/repo/hello-txt-abc" in err + assert "--file /tmp/sbom.json" in err + assert "--source-identity ci@gha" in err + assert "--content-type application/vnd.cyclonedx+json" in err + + +def test_print_metadata_retry_hint_omits_default_flags(capsys): + _print_metadata_retry_hint( + opts=_hint_opts(), + owner="acme", + repo="repo", + slug="hello-txt-abc", + metadata_content_file="/tmp/sbom.json", + cli_content_type=None, + cli_source_identity=None, + ) + err = capsys.readouterr().err + assert "cloudsmith metadata add acme/repo/hello-txt-abc" in err + assert "--file /tmp/sbom.json" in err + assert "--source-identity" not in err + assert "--content-type" not in err + + +def test_print_metadata_retry_hint_silent_for_inline_content(capsys): + _print_metadata_retry_hint( + opts=_hint_opts(), + owner="acme", + repo="repo", + slug="hello-txt-abc", + metadata_content_file=None, + cli_content_type=None, + cli_source_identity=None, + ) + assert capsys.readouterr().err == "" + + +def test_print_metadata_retry_hint_silent_for_stdin(capsys): + """Stdin payload (``-``) is not safely reproducible — suppress the hint.""" + _print_metadata_retry_hint( + opts=_hint_opts(), + owner="acme", + repo="repo", + slug="hello-txt-abc", + metadata_content_file="-", + cli_content_type=None, + cli_source_identity=None, + ) + assert capsys.readouterr().err == "" + + +def test_print_metadata_retry_hint_silent_in_json_mode(capsys): + _print_metadata_retry_hint( + opts=_hint_opts(output="json"), + owner="acme", + repo="repo", + slug="hello-txt-abc", + metadata_content_file="/tmp/sbom.json", + cli_content_type=None, + cli_source_identity=None, + ) + assert capsys.readouterr().err == ""