diff --git a/swh/icinga_plugins/tests/test_deposit.py b/swh/icinga_plugins/tests/test_deposit.py index de4327c..60658c4 100644 --- a/swh/icinga_plugins/tests/test_deposit.py +++ b/swh/icinga_plugins/tests/test_deposit.py @@ -1,467 +1,467 @@ # Copyright (C) 2019 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 io import os import tarfile import time from click.testing import CliRunner import pytest from swh.icinga_plugins.cli import icinga_cli_group from .web_scenario import WebScenario BASE_URL = "http://swh-deposit.example.org/1" COMMON_OPTIONS = [ "--server", BASE_URL, "--username", "test", "--password", "test", "--collection", "testcol", ] SAMPLE_METADATA = """ Test Software swh test-software No One """ ENTRY_TEMPLATE = """ 42 2019-12-19 18:11:00 foo.tar.gz {status} http://purl.org/net/sword/package/SimpleZip """ STATUS_TEMPLATE = """ 42 {status} {status_detail} """ @pytest.fixture(scope="session") def tmp_path(tmp_path_factory): return tmp_path_factory.mktemp(__name__) @pytest.fixture(scope="session") def sample_metadata(tmp_path): """Returns a sample metadata file's path """ path = os.path.join(tmp_path, "metadata.xml") with open(path, "w") as fd: fd.write(SAMPLE_METADATA) return path @pytest.fixture(scope="session") def sample_archive(tmp_path): """Returns a sample archive's path """ path = os.path.join(tmp_path, "archive.tar.gz") with tarfile.open(path, "w:gz") as tf: tf.addfile(tarfile.TarInfo("hello.py"), io.BytesIO(b'print("Hello world")')) return path def invoke(args, catch_exceptions=False): runner = CliRunner() result = runner.invoke(icinga_cli_group, args) if not catch_exceptions and result.exception: print(result.output) raise result.exception return result def test_deposit_immediate_success( requests_mock, mocker, sample_archive, sample_metadata, mocked_time ): scenario = WebScenario() scenario.add_step( - "post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="done") + "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="done") ) scenario.install_mock(requests_mock) result = invoke( [ "check-deposit", *COMMON_OPTIONS, "single", "--archive", sample_archive, "--metadata", sample_metadata, ] ) assert result.output == ( "DEPOSIT OK - Deposit took 0.00s and succeeded.\n" "| 'load_time' = 0.00s\n" "| 'total_time' = 0.00s\n" "| 'upload_time' = 0.00s\n" "| 'validation_time' = 0.00s\n" ) assert result.exit_code == 0, result.output def test_deposit_delays( requests_mock, mocker, sample_archive, sample_metadata, mocked_time ): scenario = WebScenario() scenario.add_step( - "post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="deposited") + "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited") ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="verified", status_detail=""), ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="loading", status_detail=""), ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="done", status_detail=""), ) scenario.install_mock(requests_mock) result = invoke( [ "check-deposit", *COMMON_OPTIONS, "single", "--archive", sample_archive, "--metadata", sample_metadata, ] ) assert result.output == ( "DEPOSIT OK - Deposit took 30.00s and succeeded.\n" "| 'load_time' = 20.00s\n" "| 'total_time' = 30.00s\n" "| 'upload_time' = 0.00s\n" "| 'validation_time' = 10.00s\n" ) assert result.exit_code == 0, result.output def test_deposit_delay_warning( requests_mock, mocker, sample_archive, sample_metadata, mocked_time ): scenario = WebScenario() scenario.add_step( - "post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="deposited") + "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited") ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="verified", status_detail=""), ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="done", status_detail=""), ) scenario.install_mock(requests_mock) result = invoke( [ "--warning", "15", "check-deposit", *COMMON_OPTIONS, "single", "--archive", sample_archive, "--metadata", sample_metadata, ], catch_exceptions=True, ) assert result.output == ( "DEPOSIT WARNING - Deposit took 20.00s and succeeded.\n" "| 'load_time' = 10.00s\n" "| 'total_time' = 20.00s\n" "| 'upload_time' = 0.00s\n" "| 'validation_time' = 10.00s\n" ) assert result.exit_code == 1, result.output def test_deposit_delay_critical( requests_mock, mocker, sample_archive, sample_metadata, mocked_time ): scenario = WebScenario() scenario.add_step( - "post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="deposited") + "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited") ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="verified", status_detail=""), ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="done", status_detail=""), callback=lambda: time.sleep(60), ) scenario.install_mock(requests_mock) result = invoke( [ "--critical", "50", "check-deposit", *COMMON_OPTIONS, "single", "--archive", sample_archive, "--metadata", sample_metadata, ], catch_exceptions=True, ) assert result.output == ( "DEPOSIT CRITICAL - Deposit took 80.00s and succeeded.\n" "| 'load_time' = 70.00s\n" "| 'total_time' = 80.00s\n" "| 'upload_time' = 0.00s\n" "| 'validation_time' = 10.00s\n" ) assert result.exit_code == 2, result.output def test_deposit_timeout( requests_mock, mocker, sample_archive, sample_metadata, mocked_time ): scenario = WebScenario() scenario.add_step( "post", - BASE_URL + "/testcol/", + f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited"), callback=lambda: time.sleep(1500), ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="verified", status_detail=""), callback=lambda: time.sleep(1500), ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="loading", status_detail=""), callback=lambda: time.sleep(1500), ) scenario.install_mock(requests_mock) result = invoke( [ "check-deposit", *COMMON_OPTIONS, "single", "--archive", sample_archive, "--metadata", sample_metadata, ], catch_exceptions=True, ) assert result.output == ( "DEPOSIT CRITICAL - Timed out while in status loading " "(4520.0s seconds since deposit started)\n" "| 'total_time' = 4520.00s\n" "| 'upload_time' = 1500.00s\n" "| 'validation_time' = 1510.00s\n" ) assert result.exit_code == 2, result.output def test_deposit_rejected( requests_mock, mocker, sample_archive, sample_metadata, mocked_time ): scenario = WebScenario() scenario.add_step( - "post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="deposited") + "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited") ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="rejected", status_detail="booo"), ) scenario.install_mock(requests_mock) result = invoke( [ "check-deposit", *COMMON_OPTIONS, "single", "--archive", sample_archive, "--metadata", sample_metadata, ], catch_exceptions=True, ) assert result.output == ( "DEPOSIT CRITICAL - Deposit was rejected: booo\n" "| 'total_time' = 10.00s\n" "| 'upload_time' = 0.00s\n" "| 'validation_time' = 10.00s\n" ) assert result.exit_code == 2, result.output def test_deposit_failed( requests_mock, mocker, sample_archive, sample_metadata, mocked_time ): scenario = WebScenario() scenario.add_step( - "post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="deposited") + "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited") ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="verified", status_detail=""), ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="loading", status_detail=""), ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="failed", status_detail="booo"), ) scenario.install_mock(requests_mock) result = invoke( [ "check-deposit", *COMMON_OPTIONS, "single", "--archive", sample_archive, "--metadata", sample_metadata, ], catch_exceptions=True, ) assert result.output == ( "DEPOSIT CRITICAL - Deposit loading failed: booo\n" "| 'load_time' = 20.00s\n" "| 'total_time' = 30.00s\n" "| 'upload_time' = 0.00s\n" "| 'validation_time' = 10.00s\n" ) assert result.exit_code == 2, result.output def test_deposit_unexpected_status( requests_mock, mocker, sample_archive, sample_metadata, mocked_time ): scenario = WebScenario() scenario.add_step( - "post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="deposited") + "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited") ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="verified", status_detail=""), ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="loading", status_detail=""), ) scenario.add_step( "get", - BASE_URL + "/testcol/42/status/", + f"{BASE_URL}/testcol/42/status/", STATUS_TEMPLATE.format(status="what", status_detail="booo"), ) scenario.install_mock(requests_mock) result = invoke( [ "check-deposit", *COMMON_OPTIONS, "single", "--archive", sample_archive, "--metadata", sample_metadata, ], catch_exceptions=True, ) assert result.output == ( "DEPOSIT CRITICAL - Deposit got unexpected status: what (booo)\n" "| 'load_time' = 20.00s\n" "| 'total_time' = 30.00s\n" "| 'upload_time' = 0.00s\n" "| 'validation_time' = 10.00s\n" ) assert result.exit_code == 2, result.output