diff --git a/swh/deposit/cli/client.py b/swh/deposit/cli/client.py index debe3171..6965dcc6 100644 --- a/swh/deposit/cli/client.py +++ b/swh/deposit/cli/client.py @@ -1,520 +1,519 @@ # Copyright (C) 2017-2020 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information from __future__ import annotations import logging # WARNING: do not import unnecessary things here to keep cli startup time under # control import os import sys from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple import click from swh.deposit.cli import deposit logger = logging.getLogger(__name__) if TYPE_CHECKING: from swh.deposit.client import PublicApiDepositClient class InputError(ValueError): """Input script error """ pass def generate_slug() -> str: """Generate a slug (sample purposes). """ import uuid return str(uuid.uuid4()) def _url(url: str) -> str: """Force the /1 api version at the end of the url (avoiding confusing issues without it). Args: url (str): api url used by cli users Returns: Top level api url to actually request """ if not url.endswith("/1"): url = "%s/1" % url return url def generate_metadata_file( name: str, external_id: str, authors: List[str], temp_dir: str ) -> str: """Generate a temporary metadata file with the minimum required metadata This generates a xml file in a temporary location and returns the path to that file. This is up to the client of that function to clean up the temporary file. Args: name: Software's name external_id: External identifier (slug) or generated one authors: List of author names Returns: Filepath to the metadata generated file """ import xmltodict path = os.path.join(temp_dir, "metadata.xml") # generate a metadata file with the minimum required metadata codemetadata = { "entry": { "@xmlns": "http://www.w3.org/2005/Atom", "@xmlns:codemeta": "https://doi.org/10.5063/SCHEMA/CODEMETA-2.0", "codemeta:name": name, "codemeta:identifier": external_id, "codemeta:author": [ {"codemeta:name": author_name} for author_name in authors ], }, } logging.debug("Temporary file: %s", path) logging.debug("Metadata dict to generate as xml: %s", codemetadata) s = xmltodict.unparse(codemetadata, pretty=True) logging.debug("Metadata dict as xml generated: %s", s) with open(path, "w") as fp: fp.write(s) return path def _client(url: str, username: str, password: str) -> PublicApiDepositClient: """Instantiate a client to access the deposit api server Args: url (str): Deposit api server username (str): User password (str): User's password """ from swh.deposit.client import PublicApiDepositClient return PublicApiDepositClient( {"url": url, "auth": {"username": username, "password": password},} ) def _collection(client: PublicApiDepositClient) -> str: """Retrieve the client's collection """ # retrieve user's collection sd_content = client.service_document() if "error" in sd_content: raise InputError("Service document retrieval: %s" % (sd_content["error"],)) collection = sd_content["service"]["workspace"]["collection"]["sword:name"] return collection def client_command_parse_input( username: str, password: str, archive: Optional[str], metadata: Optional[str], archive_deposit: bool, metadata_deposit: bool, collection: Optional[str], slug: Optional[str], partial: bool, deposit_id: Optional[int], replace: bool, url: str, name: Optional[str], authors: List[str], temp_dir: str, ) -> Dict[str, Any]: """Parse the client subcommand options and make sure the combination is acceptable*. If not, an InputError exception is raised explaining the issue. By acceptable, we mean: - A multipart deposit (create or update) requires: - an existing software archive - an existing metadata file or author(s) and name provided in params - A binary deposit (create/update) requires an existing software archive - A metadata deposit (create/update) requires an existing metadata file or author(s) and name provided in params - A deposit update requires a deposit_id This will not prevent all failure cases though. The remaining errors are already dealt with by the underlying api client. Raises: InputError explaining the user input related issue MaintenanceError explaining the api status Returns: dict with the following keys: 'archive': the software archive to deposit 'username': username 'password': associated password 'metadata': the metadata file to deposit 'collection': the username's associated client 'slug': the slug or external id identifying the deposit to make 'partial': if the deposit is partial or not 'client': instantiated class 'url': deposit's server main entry point 'deposit_type': deposit's type (binary, multipart, metadata) 'deposit_id': optional deposit identifier """ if archive_deposit and metadata_deposit: # too many flags use, remove redundant ones (-> multipart deposit) archive_deposit = False metadata_deposit = False if not slug: # generate one as this is mandatory slug = generate_slug() if not metadata: + if metadata_deposit: + raise InputError( + "Metadata deposit must be provided for metadata " + "deposit, either a filepath with --metadata or --name and --author" + ) if name and authors: metadata = generate_metadata_file(name, slug, authors, temp_dir) elif not archive_deposit and not partial and not deposit_id: # If we meet all the following conditions: - # * there is not an archive-only deposit + # * this is not an archive-only deposit request # * it is not part of a multipart deposit (either create/update # or finish) # * it misses either name or authors raise InputError( - "Either a metadata file (--metadata) or both --author and " - "--name must be provided, unless this is an archive-only " - "deposit." + "For metadata deposit request, either a metadata file with " + "--metadata or both --author and --name must be provided. " + "If this is an archive deposit request, none is required." ) elif name or authors: # If we are generating metadata, then all mandatory metadata # must be present raise InputError( - "Either a metadata file (--metadata) or both --author and " - "--name must be provided." + "For metadata deposit request, either a metadata file with " + "--metadata or both --author and --name must be provided." ) else: # TODO: this is a multipart deposit, we might want to check that # metadata are deposited at some point pass elif name or authors: raise InputError( - "Using a metadata file (--metadata) is incompatible with " - "--author and --name, which are used to generate one." + "Using --metadata flag is incompatible with both " + "--author and --name (Those are used to generate one metadata file)." ) if metadata_deposit: archive = None if archive_deposit: metadata = None - if metadata_deposit and not metadata: - raise InputError( - "Metadata deposit must be provided for metadata " - "deposit (either a filepath or --name and --author)" - ) - if not archive and not metadata and partial: raise InputError( "Please provide an actionable command. See --help for more information" ) if replace and not deposit_id: raise InputError("To update an existing deposit, you must provide its id") client = _client(url, username, password) if not collection: collection = _collection(client) return { "archive": archive, "username": username, "password": password, "metadata": metadata, "collection": collection, "slug": slug, "in_progress": partial, "client": client, "url": url, "deposit_id": deposit_id, "replace": replace, } def _subdict(d: Dict[str, Any], keys: Tuple[str, ...]) -> Dict[str, Any]: "return a dict from d with only given keys" return {k: v for k, v in d.items() if k in keys} def deposit_create(config: Dict[str, Any]) -> Dict[str, Any]: """Delegate the actual deposit to the deposit client. """ logger.debug("Create deposit") client = config["client"] keys = ("collection", "archive", "metadata", "slug", "in_progress") return client.deposit_create(**_subdict(config, keys)) def deposit_update(config: Dict[str, Any]) -> Dict[str, Any]: """Delegate the actual deposit to the deposit client. """ logger.debug("Update deposit") client = config["client"] keys = ( "collection", "deposit_id", "archive", "metadata", "slug", "in_progress", "replace", ) return client.deposit_update(**_subdict(config, keys)) @deposit.command() @click.option("--username", required=True, help="(Mandatory) User's name") @click.option( "--password", required=True, help="(Mandatory) User's associated password" ) @click.option( "--archive", type=click.Path(exists=True), help="(Optional) Software archive to deposit", ) @click.option( "--metadata", type=click.Path(exists=True), help=( "(Optional) Path to xml metadata file. If not provided, " "this will use a file named .metadata.xml" ), ) # noqa @click.option( "--archive-deposit/--no-archive-deposit", default=False, help="(Optional) Software archive only deposit", ) @click.option( "--metadata-deposit/--no-metadata-deposit", default=False, help="(Optional) Metadata only deposit", ) @click.option( "--collection", help="(Optional) User's collection. If not provided, this will be fetched.", ) # noqa @click.option( "--slug", help=( "(Optional) External system information identifier. " "If not provided, it will be generated" ), ) # noqa @click.option( "--partial/--no-partial", default=False, help=( "(Optional) The deposit will be partial, other deposits " "will have to take place to finalize it." ), ) # noqa @click.option( "--deposit-id", default=None, help="(Optional) Update an existing partial deposit with its identifier", ) # noqa @click.option( "--replace/--no-replace", default=False, help="(Optional) Update by replacing existing metadata to a deposit", ) # noqa @click.option( "--url", default="https://deposit.softwareheritage.org", help=( "(Optional) Deposit server api endpoint. By default, " "https://deposit.softwareheritage.org/1" ), ) # noqa @click.option("--verbose/--no-verbose", default=False, help="Verbose mode") @click.option("--name", help="Software name") @click.option( "--author", multiple=True, help="Software author(s), this can be repeated as many times" " as there are authors", ) @click.option( "-f", "--format", "output_format", default="logging", type=click.Choice(["logging", "yaml", "json"]), help="Output format results.", ) @click.pass_context def upload( ctx, username: str, password: str, archive: Optional[str] = None, metadata: Optional[str] = None, archive_deposit: bool = False, metadata_deposit: bool = False, collection: Optional[str] = None, slug: Optional[str] = None, partial: bool = False, deposit_id: Optional[int] = None, replace: bool = False, url: str = "https://deposit.softwareheritage.org", verbose: bool = False, name: Optional[str] = None, author: List[str] = [], output_format: Optional[str] = None, ): """Software Heritage Public Deposit Client Create/Update deposit through the command line. More documentation can be found at https://docs.softwareheritage.org/devel/swh-deposit/getting-started.html. """ import tempfile from swh.deposit.client import MaintenanceError url = _url(url) config = {} with tempfile.TemporaryDirectory() as temp_dir: try: logger.debug("Parsing cli options") config = client_command_parse_input( username, password, archive, metadata, archive_deposit, metadata_deposit, collection, slug, partial, deposit_id, replace, url, name, author, temp_dir, ) except InputError as e: logger.error("Problem during parsing options: %s", e) sys.exit(1) except MaintenanceError as e: logger.error(e) sys.exit(1) if verbose: logger.info("Parsed configuration: %s", config) deposit_id = config["deposit_id"] if deposit_id: data = deposit_update(config) else: data = deposit_create(config) print_result(data, output_format) @deposit.command() @click.option( "--url", default="https://deposit.softwareheritage.org", help="(Optional) Deposit server api endpoint. By default, " "https://deposit.softwareheritage.org/1", ) @click.option("--username", required=True, help="(Mandatory) User's name") @click.option( "--password", required=True, help="(Mandatory) User's associated password" ) @click.option("--deposit-id", default=None, required=True, help="Deposit identifier.") @click.option( "-f", "--format", "output_format", default="logging", type=click.Choice(["logging", "yaml", "json"]), help="Output format results.", ) @click.pass_context def status(ctx, url, username, password, deposit_id, output_format): """Deposit's status """ from swh.deposit.client import MaintenanceError url = _url(url) logger.debug("Status deposit") try: client = _client(url, username, password) collection = _collection(client) except InputError as e: logger.error("Problem during parsing options: %s", e) sys.exit(1) except MaintenanceError as e: logger.error(e) sys.exit(1) print_result( client.deposit_status(collection=collection, deposit_id=deposit_id), output_format, ) def print_result(data: Dict[str, Any], output_format: Optional[str]) -> None: """Display the result data into a dedicated output format. """ import json import yaml if output_format == "json": click.echo(json.dumps(data)) elif output_format == "yaml": click.echo(yaml.dump(data)) else: logger.info(data) diff --git a/swh/deposit/tests/cli/test_client.py b/swh/deposit/tests/cli/test_client.py index f34d1805..d7bb5529 100644 --- a/swh/deposit/tests/cli/test_client.py +++ b/swh/deposit/tests/cli/test_client.py @@ -1,512 +1,639 @@ # Copyright (C) 2019-2020 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import ast import contextlib import json import logging import os from unittest.mock import MagicMock from click.testing import CliRunner import pytest import yaml from swh.deposit.cli import deposit as cli from swh.deposit.cli.client import InputError, _client, _collection, _url, generate_slug from swh.deposit.client import MaintenanceError, PublicApiDepositClient from swh.deposit.parsers import parse_xml from ..conftest import TEST_USER @pytest.fixture def deposit_config(): return { "url": "https://deposit.swh.test/1", "auth": {"username": "test", "password": "test",}, } @pytest.fixture def datadir(request): """Override default datadir to target main test datadir""" return os.path.join(os.path.dirname(str(request.fspath)), "../data") @pytest.fixture def slug(): return generate_slug() @pytest.fixture def client_mock_api_down(mocker, slug): """A mock client whose connection with api fails due to maintenance issue """ mocker.patch("swh.deposit.cli.client.generate_slug", return_value=slug) mock_client = MagicMock() mocker.patch("swh.deposit.cli.client._client", return_value=mock_client) mock_client.service_document.side_effect = MaintenanceError( "Database backend maintenance: Temporarily unavailable, try again later." ) return mock_client def test_cli_url(): assert _url("http://deposit") == "http://deposit/1" assert _url("https://other/1") == "https://other/1" def test_cli_client(): client = _client("http://deposit", "user", "pass") assert isinstance(client, PublicApiDepositClient) def test_cli_collection_error(): mock_client = MagicMock() mock_client.service_document.return_value = {"error": "something went wrong"} with pytest.raises(InputError) as e: _collection(mock_client) assert "Service document retrieval: something went wrong" == str(e.value) def test_cli_collection_ok(deposit_config, requests_mock_datadir): client = PublicApiDepositClient(deposit_config) collection_name = _collection(client) assert collection_name == "test" def test_cli_collection_ko_because_downtime(): mock_client = MagicMock() mock_client.service_document.side_effect = MaintenanceError("downtime") with pytest.raises(MaintenanceError, match="downtime"): _collection(mock_client) def test_cli_deposit_with_server_down_for_maintenance( sample_archive, mocker, caplog, client_mock_api_down, slug, tmp_path ): """ Deposit failure due to maintenance down time should be explicit """ runner = CliRunner() result = runner.invoke( cli, [ "upload", "--url", "https://deposit.swh.test/1", "--username", TEST_USER["username"], "--password", TEST_USER["password"], "--name", "test-project", "--archive", sample_archive["path"], "--author", "Jane Doe", ], ) assert result.exit_code == 1, result.output assert result.output == "" down_for_maintenance_log_record = ( "swh.deposit.cli.client", logging.ERROR, "Database backend maintenance: Temporarily unavailable, try again later.", ) assert down_for_maintenance_log_record in caplog.record_tuples client_mock_api_down.service_document.assert_called_once_with() def test_cli_single_minimal_deposit( sample_archive, mocker, slug, tmp_path, requests_mock_datadir ): """ This ensure a single deposit upload through the cli is fine, cf. https://docs.softwareheritage.org/devel/swh-deposit/getting-started.html#single-deposit """ # noqa metadata_path = os.path.join(tmp_path, "metadata.xml") mocker.patch( "tempfile.TemporaryDirectory", return_value=contextlib.nullcontext(str(tmp_path)), ) runner = CliRunner() result = runner.invoke( cli, [ "upload", "--url", "https://deposit.swh.test/1", "--username", TEST_USER["username"], "--password", TEST_USER["password"], "--name", "test-project", "--archive", sample_archive["path"], "--author", "Jane Doe", "--slug", slug, "--format", "json", ], ) assert result.exit_code == 0, result.output assert json.loads(result.output) == { "deposit_id": "615", "deposit_status": "partial", "deposit_status_detail": None, "deposit_date": "Oct. 8, 2020, 4:57 p.m.", } with open(metadata_path) as fd: assert ( fd.read() == f"""\ \ttest-project \t{slug} \t \t\tJane Doe \t """ ) -def test_cli_metadata_validation( - sample_archive, mocker, caplog, tmp_path, requests_mock_datadir -): +def test_cli_validation_metadata(sample_archive, mocker, caplog, tmp_path): """Multiple metadata flags scenario (missing, conflicts) properly fails the calls """ slug = generate_slug() metadata_path = os.path.join(tmp_path, "metadata.xml") mocker.patch( "tempfile.TemporaryDirectory", return_value=contextlib.nullcontext(str(tmp_path)), ) with open(metadata_path, "a"): pass # creates the file runner = CliRunner() - # Test missing author + for flag_title_or_name, author_or_name in [ + ("--author", "no one"), + ("--name", "test-project"), + ]: + # Test missing author then missing name + result = runner.invoke( + cli, + [ + "upload", + "--url", + "https://deposit.swh.test/1", + "--username", + TEST_USER["username"], + "--password", + TEST_USER["password"], + "--archive", + sample_archive["path"], + "--slug", + slug, + flag_title_or_name, + author_or_name, + ], + ) + + assert result.exit_code == 1, f"unexpected result: {result.output}" + assert result.output == "" + expected_error_log_record = ( + "swh.deposit.cli.client", + logging.ERROR, + ( + "Problem during parsing options: " + "For metadata deposit request, either a metadata file with " + "--metadata or both --author and --name must be provided. " + "If this is an archive deposit request, none is required." + ), + ) + assert expected_error_log_record in caplog.record_tuples + + # Clear mocking state + caplog.clear() + + # incompatible flags: Test both --metadata and --author, then --metadata and + # --name + result = runner.invoke( + cli, + [ + "upload", + "--url", + "https://deposit.swh.test/1", + "--username", + TEST_USER["username"], + "--password", + TEST_USER["password"], + "--name", + "test-project", + "--deposit-id", + 666, + "--archive", + sample_archive["path"], + "--slug", + slug, + ], + ) + assert result.exit_code == 1, f"unexpected result: {result.output}" + assert result.output == "" + expected_error_log_record = ( + "swh.deposit.cli.client", + logging.ERROR, + ( + "Problem during parsing options: " + "For metadata deposit request, either a metadata file with " + "--metadata or both --author and --name must be provided." + ), + ) + assert expected_error_log_record in caplog.record_tuples + + # Clear mocking state + caplog.clear() + + # incompatible flags check (Test both --metadata and --author, + # then --metadata and --name) + result = runner.invoke( + cli, + [ + "upload", + "--url", + "https://deposit.swh.test/1", + "--username", + TEST_USER["username"], + "--password", + TEST_USER["password"], + "--archive", + sample_archive["path"], + "--metadata", + metadata_path, + "--author", + "Jane Doe", + "--slug", + slug, + ], + ) + + assert result.exit_code == 1, result.output + assert result.output == "" + expected_error_log_record = ( + "swh.deposit.cli.client", + logging.ERROR, + ( + "Problem during parsing options: " + "Using --metadata flag is incompatible with both " + "--author and --name (Those are used to generate one metadata file)." + ), + ) + assert expected_error_log_record in caplog.record_tuples + caplog.clear() + + +def test_cli_validation_no_actionable_command(caplog): + """Multiple metadata flags scenario (missing, conflicts) properly fails the calls + + """ + runner = CliRunner() + # no actionable command result = runner.invoke( cli, [ "upload", "--url", "https://deposit.swh.test/1", "--username", TEST_USER["username"], "--password", TEST_USER["password"], - "--name", - "test-project", - "--archive", - sample_archive["path"], - "--slug", - slug, + "--partial", ], ) - assert result.exit_code == 1, f"unexpected result: {result.output}" + assert result.exit_code == 1, result.output assert result.output == "" expected_error_log_record = ( "swh.deposit.cli.client", logging.ERROR, ( - "Problem during parsing options: Either a metadata file" - " (--metadata) or both --author and --name must be provided, " - "unless this is an archive-only deposit." + "Problem during parsing options: " + "Please provide an actionable command. See --help for more information" ), ) assert expected_error_log_record in caplog.record_tuples - # Clear mocking state - caplog.clear() - # Test missing name - result = runner.invoke( - cli, - [ - "upload", - "--url", - "https://deposit.swh.test/1", - "--username", - TEST_USER["username"], - "--password", - TEST_USER["password"], - "--archive", - sample_archive["path"], - "--author", - "Jane Doe", - "--slug", - slug, - ], - ) +def test_cli_validation_missing_metadata_flag(caplog): + """--metadata-deposit requires --metadata (or --name and --author) otherwise fails - assert result.exit_code == 1, result.output - assert result.output == "" - assert expected_error_log_record in caplog.record_tuples - - # Clear mocking state - caplog.clear() - - # Test both --metadata and --author + """ + runner = CliRunner() + # --metadata-deposit without --metadata flag fails result = runner.invoke( cli, [ "upload", "--url", "https://deposit.swh.test/1", "--username", TEST_USER["username"], "--password", TEST_USER["password"], - "--archive", - sample_archive["path"], - "--metadata", - metadata_path, - "--author", - "Jane Doe", - "--slug", - slug, + "--metadata-deposit", # should fail because missing --metadata flag ], ) assert result.exit_code == 1, result.output assert result.output == "" - expected_error_log_record_2 = ( + expected_error_log_record = ( "swh.deposit.cli.client", logging.ERROR, ( - "Problem during parsing options: Using a metadata file " - "(--metadata) is incompatible with --author and --name, " - "which are used to generate one." + "Problem during parsing options: " + "Metadata deposit must be provided for metadata " + "deposit, either a filepath with --metadata or --name and --author" ), ) - assert expected_error_log_record_2 in caplog.record_tuples + assert expected_error_log_record in caplog.record_tuples + + +def test_cli_validation_replace_with_no_deposit_id_fails( + sample_archive, caplog, tmp_path, requests_mock_datadir, datadir +): + """--replace flags require --deposit-id otherwise fails + + """ + runner = CliRunner() + metadata_path = os.path.join(datadir, "atom", "entry-data-deposit-binary.xml") + + for flags in [ + ["--replace"], + ["--replace", "--metadata-deposit", "--archive-deposit"], + ]: + result = runner.invoke( + cli, + [ + "upload", + "--url", + "https://deposit.swh.test/1", + "--username", + TEST_USER["username"], + "--password", + TEST_USER["password"], + "--metadata", + metadata_path, + "--archive", + sample_archive["path"], + ] + + flags, + ) + + assert result.exit_code == 1, result.output + assert result.output == "" + expected_error_log_record = ( + "swh.deposit.cli.client", + logging.ERROR, + ( + "Problem during parsing options: " + "To update an existing deposit, you must provide its id" + ), + ) + assert expected_error_log_record in caplog.record_tuples def test_cli_single_deposit_slug_generation( sample_archive, mocker, tmp_path, requests_mock_datadir ): """Single deposit scenario without providing the slug, the slug is generated nonetheless https://docs.softwareheritage.org/devel/swh-deposit/getting-started.html#single-deposit """ # noqa metadata_path = os.path.join(tmp_path, "metadata.xml") mocker.patch( "tempfile.TemporaryDirectory", return_value=contextlib.nullcontext(str(tmp_path)), ) runner = CliRunner() result = runner.invoke( cli, [ "upload", "--url", "https://deposit.swh.test/1", "--username", TEST_USER["username"], "--password", TEST_USER["password"], "--name", "test-project", "--archive", sample_archive["path"], "--author", "Jane Doe", "--format", "json", ], ) assert result.exit_code == 0, result.output assert json.loads(result.output) == { "deposit_id": "615", "deposit_status": "partial", "deposit_status_detail": None, "deposit_date": "Oct. 8, 2020, 4:57 p.m.", } with open(metadata_path) as fd: metadata_xml = fd.read() actual_metadata = parse_xml(metadata_xml) assert actual_metadata["codemeta:identifier"] is not None def test_cli_multisteps_deposit(sample_archive, datadir, slug, requests_mock_datadir): """ First deposit a partial deposit (no metadata, only archive), then update the metadata part. https://docs.softwareheritage.org/devel/swh-deposit/getting-started.html#multisteps-deposit """ # noqa api_url = "https://deposit.test.metadata/1" deposit_id = 666 runner = CliRunner() # Create a partial deposit with only 1 archive result = runner.invoke( cli, [ "upload", "--url", api_url, "--username", TEST_USER["username"], "--password", TEST_USER["password"], "--archive", sample_archive["path"], "--partial", "--slug", slug, "--format", "json", ], ) assert result.exit_code == 0, f"unexpected output: {result.output}" actual_deposit = json.loads(result.output) assert actual_deposit == { "deposit_id": str(deposit_id), "deposit_status": "partial", "deposit_status_detail": None, "deposit_date": "Oct. 8, 2020, 4:57 p.m.", } # Update the partial deposit with only 1 archive runner = CliRunner() result = runner.invoke( cli, [ "upload", "--url", api_url, "--username", TEST_USER["username"], "--password", TEST_USER["password"], "--archive", sample_archive["path"], "--deposit-id", deposit_id, "--partial", # in-progress: True, because remains the metadata to upload "--slug", slug, "--format", "json", ], ) assert result.exit_code == 0, f"unexpected output: {result.output}" assert result.output is not None actual_deposit = json.loads(result.output) # deposit update scenario actually returns a deposit status dict assert actual_deposit["deposit_id"] == str(deposit_id) assert actual_deposit["deposit_status"] == "partial" # Update the partial deposit with only some metadata (and then finalize it) # https://docs.softwareheritage.org/devel/swh-deposit/getting-started.html#add-content-or-metadata-to-the-deposit metadata_path = os.path.join(datadir, "atom", "entry-data-deposit-binary.xml") # Update deposit with metadata result = runner.invoke( cli, [ "upload", "--url", api_url, "--username", TEST_USER["username"], "--password", TEST_USER["password"], "--metadata", metadata_path, "--deposit-id", deposit_id, "--slug", slug, "--format", "json", ], # this time, ^ we no longer flag it to partial, so the status changes to # in-progress false ) assert result.exit_code == 0, f"unexpected output: {result.output}" assert result.output is not None actual_deposit = json.loads(result.output) # deposit update scenario actually returns a deposit status dict assert actual_deposit["deposit_id"] == str(deposit_id) # FIXME: should be "deposited" but current limitation in the # requests_mock_datadir_visits use, cannot find a way to make it work right now assert actual_deposit["deposit_status"] == "partial" @pytest.mark.parametrize( "output_format,callable_fn", [ ("json", json.loads), ("yaml", yaml.safe_load), ( "logging", ast.literal_eval, ), # not enough though, the caplog fixture is needed ], ) def test_cli_deposit_status_json( output_format, callable_fn, datadir, slug, requests_mock_datadir, caplog ): """Check deposit status cli with all possible output formats """ api_url_basename = "deposit.test.status" deposit_id = 1033 deposit_status_xml_path = os.path.join( datadir, f"https_{api_url_basename}", f"1_test_{deposit_id}_status" ) with open(deposit_status_xml_path, "r") as f: deposit_status_xml = f.read() expected_deposit_dict = dict(parse_xml(deposit_status_xml)) runner = CliRunner() result = runner.invoke( cli, [ "status", "--url", f"https://{api_url_basename}/1", "--username", TEST_USER["username"], "--password", TEST_USER["password"], "--deposit-id", deposit_id, "--format", output_format, ], ) assert result.exit_code == 0, f"unexpected output: {result.output}" if output_format == "logging": assert len(caplog.record_tuples) == 1 # format: (, , ) _, _, result_output = caplog.record_tuples[0] else: result_output = result.output actual_deposit = callable_fn(result_output) assert actual_deposit == expected_deposit_dict