Differential D4227 Diff 14985 swh/deposit/tests/cli/test_client.py

Changeset View

Standalone View

swh/deposit/tests/cli/test_client.py

Show All 35 Lines	def datadir(request):
return os.path.join(os.path.dirname(str(request.fspath)), "../data")		return os.path.join(os.path.dirname(str(request.fspath)), "../data")


@pytest.fixture		@pytest.fixture
def slug():		def slug():
return generate_slug()		return generate_slug()


@pytest.fixture		@pytest.fixture
		vlorentzUnsubmitted Done Inline Actions it's really unclear from the name of the fixture that it patches `tempfile.TemporaryDirectory` vlorentz: it's really unclear from the name of the fixture that it patches `tempfile.TemporaryDirectory`
		ardumontAuthorUnsubmitted Done Inline Actions does `patched_tmp_path` sound better ? ardumont: does `patched_tmp_path` sound better ?
		ardumontAuthorUnsubmitted Done Inline Actions i gave it a shot, nothing else comes to mind... ardumont: i gave it a shot, nothing else comes to mind...
		def patched_tmp_path(tmp_path, mocker):
		mocker.patch(
		"tempfile.TemporaryDirectory",
		return_value=contextlib.nullcontext(str(tmp_path)),
		)
		ardumontAuthorUnsubmitted Done Inline Actions tmp_path here ;) ardumont: tmp_path here ;)
		return tmp_path


		@pytest.fixture
		def cli_runner():
		return CliRunner()


		@pytest.fixture
def client_mock_api_down(mocker, slug):		def client_mock_api_down(mocker, slug):
"""A mock client whose connection with api fails due to maintenance issue		"""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()		mock_client = MagicMock()
mocker.patch("swh.deposit.cli.client._client", return_value=mock_client)		mocker.patch("swh.deposit.cli.client._client", return_value=mock_client)
mock_client.service_document.side_effect = MaintenanceError(		mock_client.service_document.side_effect = MaintenanceError(
"Database backend maintenance: Temporarily unavailable, try again later."		"Database backend maintenance: Temporarily unavailable, try again later."
)		)
return mock_client		return mock_client


Show All 26 Lines
def test_cli_collection_ko_because_downtime():		def test_cli_collection_ko_because_downtime():
mock_client = MagicMock()		mock_client = MagicMock()
mock_client.service_document.side_effect = MaintenanceError("downtime")		mock_client.service_document.side_effect = MaintenanceError("downtime")
with pytest.raises(MaintenanceError, match="downtime"):		with pytest.raises(MaintenanceError, match="downtime"):
_collection(mock_client)		_collection(mock_client)


def test_cli_deposit_with_server_down_for_maintenance(		def test_cli_deposit_with_server_down_for_maintenance(
sample_archive, mocker, caplog, client_mock_api_down, slug, tmp_path		sample_archive, caplog, client_mock_api_down, slug, patched_tmp_path, cli_runner
):		):
""" Deposit failure due to maintenance down time should be explicit		""" Deposit failure due to maintenance down time should be explicit

"""		"""
runner = CliRunner()		result = cli_runner.invoke(
result = runner.invoke(
cli,		cli,
[		[
"upload",		"upload",
"--url",		"--url",
"https://deposit.swh.test/1",		"https://deposit.swh.test/1",
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 15 Lines	down_for_maintenance_log_record = (
"Database backend maintenance: Temporarily unavailable, try again later.",		"Database backend maintenance: Temporarily unavailable, try again later.",
)		)
assert down_for_maintenance_log_record in caplog.record_tuples		assert down_for_maintenance_log_record in caplog.record_tuples

client_mock_api_down.service_document.assert_called_once_with()		client_mock_api_down.service_document.assert_called_once_with()


def test_cli_single_minimal_deposit(		def test_cli_single_minimal_deposit(
sample_archive, mocker, slug, tmp_path, requests_mock_datadir		sample_archive, slug, patched_tmp_path, requests_mock_datadir, cli_runner
):		):
""" This ensure a single deposit upload through the cli is fine, cf.		""" This ensure a single deposit upload through the cli is fine, cf.
https://docs.softwareheritage.org/devel/swh-deposit/getting-started.html#single-deposit		https://docs.softwareheritage.org/devel/swh-deposit/getting-started.html#single-deposit
""" # noqa		""" # noqa

metadata_path = os.path.join(tmp_path, "metadata.xml")		metadata_path = os.path.join(patched_tmp_path, "metadata.xml")
mocker.patch(		result = cli_runner.invoke(
"tempfile.TemporaryDirectory",
return_value=contextlib.nullcontext(str(tmp_path)),
)

runner = CliRunner()
result = runner.invoke(
cli,		cli,
[		[
"upload",		"upload",
"--url",		"--url",
"https://deposit.swh.test/1",		"https://deposit.swh.test/1",
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 30 Lines
\t<codemeta:identifier>{slug}</codemeta:identifier>		\t<codemeta:identifier>{slug}</codemeta:identifier>
\t<codemeta:author>		\t<codemeta:author>
\t\t<codemeta:name>Jane Doe</codemeta:name>		\t\t<codemeta:name>Jane Doe</codemeta:name>
\t</codemeta:author>		\t</codemeta:author>
</entry>"""		</entry>"""
)		)


def test_cli_validation_metadata(sample_archive, mocker, caplog, tmp_path):		def test_cli_validation_metadata(
		sample_archive, caplog, patched_tmp_path, cli_runner, slug
		):
"""Multiple metadata flags scenario (missing, conflicts) properly fails the calls		"""Multiple metadata flags scenario (missing, conflicts) properly fails the calls

"""		"""
slug = generate_slug()

metadata_path = os.path.join(tmp_path, "metadata.xml")		metadata_path = os.path.join(patched_tmp_path, "metadata.xml")
mocker.patch(
"tempfile.TemporaryDirectory",
return_value=contextlib.nullcontext(str(tmp_path)),
)
with open(metadata_path, "a"):		with open(metadata_path, "a"):
pass # creates the file		pass # creates the file

runner = CliRunner()

for flag_title_or_name, author_or_name in [		for flag_title_or_name, author_or_name in [
("--author", "no one"),		("--author", "no one"),
("--name", "test-project"),		("--name", "test-project"),
]:		]:
# Test missing author then missing name		# Test missing author then missing name
result = runner.invoke(		result = cli_runner.invoke(
cli,		cli,
[		[
"upload",		"upload",
"--url",		"--url",
"https://deposit.swh.test/1",		"https://deposit.swh.test/1",
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 21 Lines	]:
)		)
assert expected_error_log_record in caplog.record_tuples		assert expected_error_log_record in caplog.record_tuples

# Clear mocking state		# Clear mocking state
caplog.clear()		caplog.clear()

# incompatible flags: Test both --metadata and --author, then --metadata and		# incompatible flags: Test both --metadata and --author, then --metadata and
# --name		# --name
result = runner.invoke(		result = cli_runner.invoke(
cli,		cli,
[		[
"upload",		"upload",
"--url",		"--url",
"https://deposit.swh.test/1",		"https://deposit.swh.test/1",
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 21 Lines	]:
)		)
assert expected_error_log_record in caplog.record_tuples		assert expected_error_log_record in caplog.record_tuples

# Clear mocking state		# Clear mocking state
caplog.clear()		caplog.clear()

# incompatible flags check (Test both --metadata and --author,		# incompatible flags check (Test both --metadata and --author,
# then --metadata and --name)		# then --metadata and --name)
result = runner.invoke(		result = cli_runner.invoke(
cli,		cli,
[		[
"upload",		"upload",
"--url",		"--url",
"https://deposit.swh.test/1",		"https://deposit.swh.test/1",
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 19 Lines	]:
"Using --metadata flag is incompatible with both "		"Using --metadata flag is incompatible with both "
"--author and --name (Those are used to generate one metadata file)."		"--author and --name (Those are used to generate one metadata file)."
),		),
)		)
assert expected_error_log_record in caplog.record_tuples		assert expected_error_log_record in caplog.record_tuples
caplog.clear()		caplog.clear()


def test_cli_validation_no_actionable_command(caplog):		def test_cli_validation_no_actionable_command(caplog, cli_runner):
"""Multiple metadata flags scenario (missing, conflicts) properly fails the calls		"""Multiple metadata flags scenario (missing, conflicts) properly fails the calls

"""		"""
runner = CliRunner()
# no actionable command		# no actionable command
result = runner.invoke(		result = cli_runner.invoke(
cli,		cli,
[		[
"upload",		"upload",
"--url",		"--url",
"https://deposit.swh.test/1",		"https://deposit.swh.test/1",
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 10 Lines	expected_error_log_record = (
(		(
"Problem during parsing options: "		"Problem during parsing options: "
"Please provide an actionable command. See --help for more information"		"Please provide an actionable command. See --help for more information"
),		),
)		)
assert expected_error_log_record in caplog.record_tuples		assert expected_error_log_record in caplog.record_tuples


def test_cli_validation_missing_metadata_flag(caplog):		def test_cli_validation_missing_metadata_flag(caplog, cli_runner):
"""--metadata-deposit requires --metadata (or --name and --author) otherwise fails		"""--metadata-deposit requires --metadata (or --name and --author) otherwise fails

"""		"""
runner = CliRunner()
# --metadata-deposit without --metadata flag fails		# --metadata-deposit without --metadata flag fails
result = runner.invoke(		result = cli_runner.invoke(
cli,		cli,
[		[
"upload",		"upload",
"--url",		"--url",
"https://deposit.swh.test/1",		"https://deposit.swh.test/1",
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 12 Lines	expected_error_log_record = (
"Metadata deposit must be provided for metadata "		"Metadata deposit must be provided for metadata "
"deposit, either a filepath with --metadata or --name and --author"		"deposit, either a filepath with --metadata or --name and --author"
),		),
)		)
assert expected_error_log_record in caplog.record_tuples		assert expected_error_log_record in caplog.record_tuples


def test_cli_validation_replace_with_no_deposit_id_fails(		def test_cli_validation_replace_with_no_deposit_id_fails(
sample_archive, caplog, tmp_path, requests_mock_datadir, datadir		sample_archive, caplog, patched_tmp_path, requests_mock_datadir, datadir, cli_runner
):		):
"""--replace flags require --deposit-id otherwise fails		"""--replace flags require --deposit-id otherwise fails

"""		"""
runner = CliRunner()
metadata_path = os.path.join(datadir, "atom", "entry-data-deposit-binary.xml")		metadata_path = os.path.join(datadir, "atom", "entry-data-deposit-binary.xml")

for flags in [		for flags in [
["--replace"],		["--replace"],
["--replace", "--metadata-deposit", "--archive-deposit"],		["--replace", "--metadata-deposit", "--archive-deposit"],
]:		]:
result = runner.invoke(		result = cli_runner.invoke(
cli,		cli,
[		[
"upload",		"upload",
"--url",		"--url",
"https://deposit.swh.test/1",		"https://deposit.swh.test/1",
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 15 Lines	]:
"Problem during parsing options: "		"Problem during parsing options: "
"To update an existing deposit, you must provide its id"		"To update an existing deposit, you must provide its id"
),		),
)		)
assert expected_error_log_record in caplog.record_tuples		assert expected_error_log_record in caplog.record_tuples


def test_cli_single_deposit_slug_generation(		def test_cli_single_deposit_slug_generation(
sample_archive, mocker, tmp_path, requests_mock_datadir		sample_archive, patched_tmp_path, requests_mock_datadir, cli_runner
):		):
"""Single deposit scenario without providing the slug, the slug is generated nonetheless		"""Single deposit scenario without providing the slug, the slug is generated nonetheless
https://docs.softwareheritage.org/devel/swh-deposit/getting-started.html#single-deposit		https://docs.softwareheritage.org/devel/swh-deposit/getting-started.html#single-deposit
""" # noqa		""" # noqa
metadata_path = os.path.join(tmp_path, "metadata.xml")		metadata_path = os.path.join(patched_tmp_path, "metadata.xml")
mocker.patch(		result = cli_runner.invoke(
"tempfile.TemporaryDirectory",
return_value=contextlib.nullcontext(str(tmp_path)),
)

runner = CliRunner()
result = runner.invoke(
cli,		cli,
[		[
"upload",		"upload",
"--url",		"--url",
"https://deposit.swh.test/1",		"https://deposit.swh.test/1",
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 17 Lines	):
}		}

with open(metadata_path) as fd:		with open(metadata_path) as fd:
metadata_xml = fd.read()		metadata_xml = fd.read()
actual_metadata = parse_xml(metadata_xml)		actual_metadata = parse_xml(metadata_xml)
assert actual_metadata["codemeta:identifier"] is not None		assert actual_metadata["codemeta:identifier"] is not None


def test_cli_multisteps_deposit(sample_archive, datadir, slug, requests_mock_datadir):		def test_cli_multisteps_deposit(
		sample_archive, datadir, slug, requests_mock_datadir, cli_runner
		):
""" First deposit a partial deposit (no metadata, only archive), then update the metadata part.		""" 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		https://docs.softwareheritage.org/devel/swh-deposit/getting-started.html#multisteps-deposit
""" # noqa		""" # noqa
api_url = "https://deposit.test.metadata/1"		api_url = "https://deposit.test.metadata/1"
deposit_id = 666		deposit_id = 666

runner = CliRunner()
# Create a partial deposit with only 1 archive		# Create a partial deposit with only 1 archive
result = runner.invoke(		result = cli_runner.invoke(
cli,		cli,
[		[
"upload",		"upload",
"--url",		"--url",
api_url,		api_url,
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 13 Lines	):
assert actual_deposit == {		assert actual_deposit == {
"deposit_id": str(deposit_id),		"deposit_id": str(deposit_id),
"deposit_status": "partial",		"deposit_status": "partial",
"deposit_status_detail": None,		"deposit_status_detail": None,
"deposit_date": "Oct. 8, 2020, 4:57 p.m.",		"deposit_date": "Oct. 8, 2020, 4:57 p.m.",
}		}

# Update the partial deposit with only 1 archive		# Update the partial deposit with only 1 archive
runner = CliRunner()		result = cli_runner.invoke(
result = runner.invoke(
cli,		cli,
[		[
"upload",		"upload",
"--url",		"--url",
api_url,		api_url,
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 16 Lines	):
assert actual_deposit["deposit_id"] == str(deposit_id)		assert actual_deposit["deposit_id"] == str(deposit_id)
assert actual_deposit["deposit_status"] == "partial"		assert actual_deposit["deposit_status"] == "partial"

# Update the partial deposit with only some metadata (and then finalize it)		# 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		# 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")		metadata_path = os.path.join(datadir, "atom", "entry-data-deposit-binary.xml")

# Update deposit with metadata		# Update deposit with metadata
result = runner.invoke(		result = cli_runner.invoke(
cli,		cli,
[		[
"upload",		"upload",
"--url",		"--url",
api_url,		api_url,
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 27 Lines	[
("yaml", yaml.safe_load),		("yaml", yaml.safe_load),
(		(
"logging",		"logging",
ast.literal_eval,		ast.literal_eval,
), # not enough though, the caplog fixture is needed		), # not enough though, the caplog fixture is needed
],		],
)		)
def test_cli_deposit_status_with_output_format(		def test_cli_deposit_status_with_output_format(
output_format, callable_fn, datadir, slug, requests_mock_datadir, caplog		output_format, callable_fn, datadir, slug, requests_mock_datadir, caplog, cli_runner
):		):
"""Check deposit status cli with all possible output formats (json, yaml, logging).		"""Check deposit status cli with all possible output formats (json, yaml, logging).

"""		"""
api_url_basename = "deposit.test.status"		api_url_basename = "deposit.test.status"
deposit_id = 1033		deposit_id = 1033
deposit_status_xml_path = os.path.join(		deposit_status_xml_path = os.path.join(
datadir, f"https_{api_url_basename}", f"1_test_{deposit_id}_status"		datadir, f"https_{api_url_basename}", f"1_test_{deposit_id}_status"
)		)
with open(deposit_status_xml_path, "r") as f:		with open(deposit_status_xml_path, "r") as f:
deposit_status_xml = f.read()		deposit_status_xml = f.read()
expected_deposit_dict = dict(parse_xml(deposit_status_xml))		expected_deposit_dict = dict(parse_xml(deposit_status_xml))

runner = CliRunner()		result = cli_runner.invoke(
result = runner.invoke(
cli,		cli,
[		[
"status",		"status",
"--url",		"--url",
f"https://{api_url_basename}/1",		f"https://{api_url_basename}/1",
"--username",		"--username",
TEST_USER["username"],		TEST_USER["username"],
"--password",		"--password",
Show All 18 Lines