Changeset View
Changeset View
Standalone View
Standalone View
swh/icinga_plugins/tests/test_deposit.py
Show First 20 Lines • Show All 108 Lines • ▼ Show 20 Lines | |||||
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() | ||||
scenario.add_step( | 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) | scenario.install_mock(requests_mock) | ||||
result = invoke( | result = invoke( | ||||
[ | [ | ||||
"check-deposit", | "check-deposit", | ||||
*COMMON_OPTIONS, | *COMMON_OPTIONS, | ||||
Show All 16 Lines | |||||
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( | ||||
"post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="deposited") | "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited") | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="verified", status_detail=""), | STATUS_TEMPLATE.format(status="verified", status_detail=""), | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="loading", status_detail=""), | STATUS_TEMPLATE.format(status="loading", status_detail=""), | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="done", status_detail=""), | STATUS_TEMPLATE.format(status="done", status_detail=""), | ||||
) | ) | ||||
scenario.install_mock(requests_mock) | scenario.install_mock(requests_mock) | ||||
result = invoke( | result = invoke( | ||||
[ | [ | ||||
"check-deposit", | "check-deposit", | ||||
Show All 17 Lines | |||||
def test_deposit_delay_warning( | def test_deposit_delay_warning( | ||||
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( | ||||
"post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="deposited") | "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited") | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="verified", status_detail=""), | STATUS_TEMPLATE.format(status="verified", status_detail=""), | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="done", status_detail=""), | STATUS_TEMPLATE.format(status="done", status_detail=""), | ||||
) | ) | ||||
scenario.install_mock(requests_mock) | scenario.install_mock(requests_mock) | ||||
result = invoke( | result = invoke( | ||||
[ | [ | ||||
"--warning", | "--warning", | ||||
Show All 20 Lines | |||||
def test_deposit_delay_critical( | def test_deposit_delay_critical( | ||||
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( | ||||
"post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="deposited") | "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited") | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="verified", status_detail=""), | STATUS_TEMPLATE.format(status="verified", status_detail=""), | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="done", status_detail=""), | STATUS_TEMPLATE.format(status="done", status_detail=""), | ||||
callback=lambda: time.sleep(60), | callback=lambda: time.sleep(60), | ||||
) | ) | ||||
scenario.install_mock(requests_mock) | scenario.install_mock(requests_mock) | ||||
result = invoke( | result = invoke( | ||||
[ | [ | ||||
Show All 22 Lines | |||||
def test_deposit_timeout( | def test_deposit_timeout( | ||||
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( | ||||
"post", | "post", | ||||
BASE_URL + "/testcol/", | f"{BASE_URL}/testcol/", | ||||
ENTRY_TEMPLATE.format(status="deposited"), | ENTRY_TEMPLATE.format(status="deposited"), | ||||
callback=lambda: time.sleep(1500), | callback=lambda: time.sleep(1500), | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="verified", status_detail=""), | STATUS_TEMPLATE.format(status="verified", status_detail=""), | ||||
callback=lambda: time.sleep(1500), | callback=lambda: time.sleep(1500), | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="loading", status_detail=""), | STATUS_TEMPLATE.format(status="loading", status_detail=""), | ||||
callback=lambda: time.sleep(1500), | callback=lambda: time.sleep(1500), | ||||
) | ) | ||||
scenario.install_mock(requests_mock) | scenario.install_mock(requests_mock) | ||||
result = invoke( | result = invoke( | ||||
[ | [ | ||||
Show All 19 Lines | |||||
def test_deposit_rejected( | def test_deposit_rejected( | ||||
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( | ||||
"post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="deposited") | "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited") | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="rejected", status_detail="booo"), | STATUS_TEMPLATE.format(status="rejected", status_detail="booo"), | ||||
) | ) | ||||
scenario.install_mock(requests_mock) | scenario.install_mock(requests_mock) | ||||
result = invoke( | result = invoke( | ||||
[ | [ | ||||
"check-deposit", | "check-deposit", | ||||
Show All 17 Lines | |||||
def test_deposit_failed( | def test_deposit_failed( | ||||
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( | ||||
"post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="deposited") | "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited") | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="verified", status_detail=""), | STATUS_TEMPLATE.format(status="verified", status_detail=""), | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="loading", status_detail=""), | STATUS_TEMPLATE.format(status="loading", status_detail=""), | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="failed", status_detail="booo"), | STATUS_TEMPLATE.format(status="failed", status_detail="booo"), | ||||
) | ) | ||||
scenario.install_mock(requests_mock) | scenario.install_mock(requests_mock) | ||||
result = invoke( | result = invoke( | ||||
[ | [ | ||||
"check-deposit", | "check-deposit", | ||||
Show All 18 Lines | |||||
def test_deposit_unexpected_status( | def test_deposit_unexpected_status( | ||||
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( | ||||
"post", BASE_URL + "/testcol/", ENTRY_TEMPLATE.format(status="deposited") | "post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited") | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="verified", status_detail=""), | STATUS_TEMPLATE.format(status="verified", status_detail=""), | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="loading", status_detail=""), | STATUS_TEMPLATE.format(status="loading", status_detail=""), | ||||
) | ) | ||||
scenario.add_step( | scenario.add_step( | ||||
"get", | "get", | ||||
BASE_URL + "/testcol/42/status/", | f"{BASE_URL}/testcol/42/status/", | ||||
STATUS_TEMPLATE.format(status="what", status_detail="booo"), | STATUS_TEMPLATE.format(status="what", status_detail="booo"), | ||||
) | ) | ||||
scenario.install_mock(requests_mock) | scenario.install_mock(requests_mock) | ||||
result = invoke( | result = invoke( | ||||
[ | [ | ||||
"check-deposit", | "check-deposit", | ||||
Show All 18 Lines |