Changeset View
Changeset View
Standalone View
Standalone View
swh/icinga_plugins/tests/test_vault.py
| Show All 29 Lines | |||||
| response_failed = { | response_failed = { | ||||
| "obj_id": dir_id, | "obj_id": dir_id, | ||||
| "obj_type": "directory", | "obj_type": "directory", | ||||
| "progress_message": "foobar", | "progress_message": "foobar", | ||||
| "status": "failed" | "status": "failed" | ||||
| } | } | ||||
| response_unknown_status = { | |||||
| "obj_id": dir_id, | |||||
| "obj_type": "directory", | |||||
| "progress_message": "what", | |||||
| "status": "boo" | |||||
| } | |||||
| class FakeStorage: | class FakeStorage: | ||||
| def __init__(self, foo, **kwargs): | def __init__(self, foo, **kwargs): | ||||
| pass | pass | ||||
| def directory_get_random(self): | def directory_get_random(self): | ||||
| return bytes.fromhex(dir_id) | return bytes.fromhex(dir_id) | ||||
| ▲ Show 20 Lines • Show All 87 Lines • ▼ Show 20 Lines | def test_vault_failure(requests_mock, mocker, mocked_time): | ||||
| assert result.output == ( | assert result.output == ( | ||||
| f"VAULT CRITICAL - cooking directory {dir_id} took " | f"VAULT CRITICAL - cooking directory {dir_id} took " | ||||
| f"10.00s and failed with: foobar\n" | f"10.00s and failed with: foobar\n" | ||||
| f"| 'total_time' = 10.00s\n") | f"| 'total_time' = 10.00s\n") | ||||
| assert result.exit_code == 2, result.output | assert result.exit_code == 2, result.output | ||||
| def test_vault_unknown_status(requests_mock, mocker, mocked_time): | |||||
| scenario = WebScenario() | |||||
| url = f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/' | |||||
| scenario.add_step('get', url, {}, status_code=404) | |||||
| scenario.add_step('post', url, response_pending) | |||||
| scenario.add_step('get', url, response_unknown_status) | |||||
| scenario.install_mock(requests_mock) | |||||
| get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') | |||||
| get_storage_mock.side_effect = FakeStorage | |||||
| result = invoke([ | |||||
| 'check-vault', | |||||
| '--swh-web-url', 'mock://swh-web.example.org', | |||||
| '--swh-storage-url', 'foo://example.org', | |||||
| 'directory', | |||||
| ], catch_exceptions=True) | |||||
| assert result.output == ( | |||||
| f"VAULT CRITICAL - cooking directory {dir_id} took " | |||||
| f"10.00s and resulted in unknown status: boo\n" | |||||
| f"| 'total_time' = 10.00s\n") | |||||
| assert result.exit_code == 2, result.output | |||||
| def test_vault_timeout(requests_mock, mocker, mocked_time): | def test_vault_timeout(requests_mock, mocker, mocked_time): | ||||
| scenario = WebScenario() | scenario = WebScenario() | ||||
| url = f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/' | url = f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/' | ||||
| scenario.add_step('get', url, {}, status_code=404) | scenario.add_step('get', url, {}, status_code=404) | ||||
| scenario.add_step('post', url, response_pending) | scenario.add_step('post', url, response_pending) | ||||
| scenario.add_step('get', url, response_pending) | scenario.add_step('get', url, response_pending) | ||||
| Show All 12 Lines | result = invoke([ | ||||
| 'directory', | 'directory', | ||||
| ], catch_exceptions=True) | ], catch_exceptions=True) | ||||
| assert result.output == ( | assert result.output == ( | ||||
| f"VAULT CRITICAL - cooking directory {dir_id} took more than " | f"VAULT CRITICAL - cooking directory {dir_id} took more than " | ||||
| f"4020.00s and has status: foo\n" | f"4020.00s and has status: foo\n" | ||||
| f"| 'total_time' = 4020.00s\n") | f"| 'total_time' = 4020.00s\n") | ||||
| assert result.exit_code == 2, result.output | assert result.exit_code == 2, result.output | ||||
| def test_vault_cached_directory(requests_mock, mocker, mocked_time): | |||||
| """First serves a directory that's already in the cache, to | |||||
| test that vault_check requests another one.""" | |||||
| scenario = WebScenario() | |||||
| url = f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/' | |||||
| scenario.add_step('get', url, {}, status_code=200) | |||||
| scenario.add_step('get', url, {}, status_code=404) | |||||
| scenario.add_step('post', url, response_pending) | |||||
| scenario.add_step('get', url, response_done) | |||||
| scenario.install_mock(requests_mock) | |||||
| get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') | |||||
| get_storage_mock.side_effect = FakeStorage | |||||
| result = invoke([ | |||||
| 'check-vault', | |||||
| '--swh-web-url', 'mock://swh-web.example.org', | |||||
| '--swh-storage-url', 'foo://example.org', | |||||
| 'directory', | |||||
| ]) | |||||
| assert result.output == ( | |||||
| f"VAULT OK - cooking directory {dir_id} took " | |||||
| f"10.00s and succeeded.\n" | |||||
| f"| 'total_time' = 10.00s\n") | |||||
| assert result.exit_code == 0, result.output | |||||
| def test_vault_no_directory(requests_mock, mocker, mocked_time): | |||||
| """Tests with an empty storage""" | |||||
| scenario = WebScenario() | |||||
| scenario.install_mock(requests_mock) | |||||
| get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') | |||||
| get_storage_mock.side_effect = FakeStorage | |||||
| mocker.patch( | |||||
| f'{__name__}.FakeStorage.directory_get_random', return_value=None) | |||||
| result = invoke([ | |||||
| 'check-vault', | |||||
| '--swh-web-url', 'mock://swh-web.example.org', | |||||
| '--swh-storage-url', 'foo://example.org', | |||||
| 'directory', | |||||
| ], catch_exceptions=True) | |||||
| assert result.output == ( | |||||
| "VAULT CRITICAL - No directory exists in the archive.\n") | |||||
| assert result.exit_code == 2, result.output | |||||