Changeset View
Changeset View
Standalone View
Standalone View
swh/icinga_plugins/tests/test_deposit.py
Show First 20 Lines • Show All 58 Lines • ▼ Show 20 Lines | |||||
STATUS_TEMPLATE = """ | STATUS_TEMPLATE = """ | ||||
<entry xmlns="http://www.w3.org/2005/Atom" | <entry xmlns="http://www.w3.org/2005/Atom" | ||||
xmlns:sword="http://purl.org/net/sword/" | xmlns:sword="http://purl.org/net/sword/" | ||||
xmlns:dcterms="http://purl.org/dc/terms/"> | xmlns:dcterms="http://purl.org/dc/terms/"> | ||||
<deposit_id>42</deposit_id> | <deposit_id>42</deposit_id> | ||||
<deposit_status>{status}</deposit_status> | <deposit_status>{status}</deposit_status> | ||||
<deposit_status_detail>{status_detail}</deposit_status_detail> | <deposit_status_detail>{status_detail}</deposit_status_detail> | ||||
<deposit_swh_id>{swhid}</deposit_swh_id> | |||||
</entry> | </entry> | ||||
""" | """ | ||||
@pytest.fixture(scope="session") | @pytest.fixture(scope="session") | ||||
def tmp_path(tmp_path_factory): | def tmp_path(tmp_path_factory): | ||||
return tmp_path_factory.mktemp(__name__) | return tmp_path_factory.mktemp(__name__) | ||||
@pytest.fixture(scope="session") | @pytest.fixture(scope="session") | ||||
def sample_metadata(tmp_path): | def sample_metadata(tmp_path): | ||||
"""Returns a sample metadata file's path | """Returns a sample metadata file's path | ||||
""" | """ | ||||
path = os.path.join(tmp_path, "metadata.xml") | path = os.path.join(tmp_path, "metadata.xml") | ||||
with open(path, "w") as fd: | with open(path, "w") as fd: | ||||
fd.write(SAMPLE_METADATA) | fd.write(SAMPLE_METADATA) | ||||
return path | return path | ||||
@pytest.fixture(scope="session") | @pytest.fixture(scope="session") | ||||
def sample_archive(tmp_path): | def sample_archive(tmp_path): | ||||
"""Returns a sample archive's path | """Returns a sample archive's path | ||||
""" | """ | ||||
vlorentz: ugh, we really have to do something about these tag names that don't belong in the Atom… | |||||
Done Inline Actionsyes, and yes. does that mean, we need to start defining our own namespace and use it? ardumont: yes, and yes.
does that mean, we need to start defining our own namespace and use it?
(if so… | |||||
path = os.path.join(tmp_path, "archive.tar.gz") | path = os.path.join(tmp_path, "archive.tar.gz") | ||||
with tarfile.open(path, "w:gz") as tf: | with tarfile.open(path, "w:gz") as tf: | ||||
tf.addfile(tarfile.TarInfo("hello.py"), io.BytesIO(b'print("Hello world")')) | tf.addfile(tarfile.TarInfo("hello.py"), io.BytesIO(b'print("Hello world")')) | ||||
return path | return path | ||||
def invoke(args, catch_exceptions=False): | def invoke(args, catch_exceptions=False): | ||||
runner = CliRunner() | runner = CliRunner() | ||||
result = runner.invoke(icinga_cli_group, args) | result = runner.invoke(icinga_cli_group, args) | ||||
if not catch_exceptions and result.exception: | if not catch_exceptions and result.exception: | ||||
print(result.output) | print(result.output) | ||||
raise result.exception | raise result.exception | ||||
return result | return result | ||||
def test_deposit_immediate_success( | def test_deposit_immediate_success( | ||||
requests_mock, mocker, sample_archive, sample_metadata, mocked_time | requests_mock, mocker, sample_archive, sample_metadata, mocked_time | ||||
): | ): | ||||
scenario = WebScenario() | scenario = WebScenario() | ||||
status_xml = STATUS_TEMPLATE.format( | |||||
status="done", | |||||
status_detail="", | |||||
swhid="swh:1:dir:02ed6084fb0e8384ac58980e07548a547431cf74", | |||||
) | |||||
# Initial deposit | |||||
scenario.add_step( | |||||
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="done") | |||||
) | |||||
# Then metadata update | |||||
scenario.add_step( | |||||
"get", f"{BASE_URL}/testcol/42/status/", status_xml, | |||||
) | |||||
# internal deposit client does call status, then update metadata then status api | |||||
scenario.add_step( | |||||
"get", f"{BASE_URL}/testcol/42/status/", status_xml, | |||||
) | |||||
scenario.add_step( | scenario.add_step( | ||||
"post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="done") | "put", f"{BASE_URL}/testcol/42/metadata/", status_xml, | ||||
) | |||||
scenario.add_step( | |||||
"get", f"{BASE_URL}/testcol/42/status/", status_xml, | |||||
) | ) | ||||
scenario.install_mock(requests_mock) | scenario.install_mock(requests_mock) | ||||
result = invoke( | result = invoke( | ||||
[ | [ | ||||
"check-deposit", | "check-deposit", | ||||
*COMMON_OPTIONS, | *COMMON_OPTIONS, | ||||
"single", | "single", | ||||
"--archive", | "--archive", | ||||
sample_archive, | sample_archive, | ||||
"--metadata", | "--metadata", | ||||
sample_metadata, | sample_metadata, | ||||
] | ] | ||||
) | ) | ||||
assert result.output == ( | assert result.output == ( | ||||
"DEPOSIT OK - Deposit took 0.00s and succeeded.\n" | "DEPOSIT OK - Deposit took 0.00s and succeeded.\n" | ||||
"| 'load_time' = 0.00s\n" | "| 'load_time' = 0.00s\n" | ||||
"| 'total_time' = 0.00s\n" | "| 'total_time' = 0.00s\n" | ||||
"| 'upload_time' = 0.00s\n" | "| 'upload_time' = 0.00s\n" | ||||
"| 'validation_time' = 0.00s\n" | "| 'validation_time' = 0.00s\n" | ||||
"DEPOSIT OK - Deposit Metadata update took 0.00s and succeeded.\n" | |||||
"| 'load_time' = 0.00s\n" | |||||
"| 'total_time' = 0.00s\n" | |||||
"| 'update_time' = 0.00s\n" | |||||
"| 'upload_time' = 0.00s\n" | |||||
"| 'validation_time' = 0.00s\n" | |||||
) | ) | ||||
assert result.exit_code == 0, result.output | assert result.exit_code == 0, f"Unexpected output: {result.output}" | ||||
def test_deposit_delays( | def test_deposit_delays( | ||||
requests_mock, mocker, sample_archive, sample_metadata, mocked_time | requests_mock, mocker, sample_archive, sample_metadata, mocked_time | ||||
): | ): | ||||
scenario = WebScenario() | scenario = WebScenario() | ||||
scenario.add_step( | scenario.add_step( | ||||
▲ Show 20 Lines • Show All 318 Lines • Show Last 20 Lines |
ugh, we really have to do something about these tag names that don't belong in the Atom namespace.
But out of scope for this diff