Changeset View
Changeset View
Standalone View
Standalone View
swh/icinga_plugins/tests/test_vault.py
| # Copyright (C) 2019 The Software Heritage developers | # Copyright (C) 2019 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import re | |||||
| import time | import time | ||||
| from click.testing import CliRunner | from click.testing import CliRunner | ||||
| from swh.icinga_plugins.cli import cli | from swh.icinga_plugins.cli import cli | ||||
| from .web_scenario import WebScenario | from .web_scenario import WebScenario | ||||
| Show All 34 Lines | def invoke(args, catch_exceptions=False): | ||||
| runner = CliRunner() | runner = CliRunner() | ||||
| result = runner.invoke(cli, args) | result = runner.invoke(cli, 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_vault_immediate_success(requests_mock, mocker): | def test_vault_immediate_success(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_done) | scenario.add_step('get', url, response_done) | ||||
| scenario.install_mock(requests_mock) | scenario.install_mock(requests_mock) | ||||
| get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') | get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') | ||||
| get_storage_mock.side_effect = FakeStorage | get_storage_mock.side_effect = FakeStorage | ||||
| sleep_mock = mocker.patch('time.sleep') | |||||
| result = invoke([ | result = invoke([ | ||||
| 'check-vault', | 'check-vault', | ||||
| '--swh-web-url', 'mock://swh-web.example.org', | '--swh-web-url', 'mock://swh-web.example.org', | ||||
| '--swh-storage-url', 'foo://example.org', | '--swh-storage-url', 'foo://example.org', | ||||
| 'directory', | 'directory', | ||||
| ]) | ]) | ||||
| assert re.match( | assert result.output == ( | ||||
| rf'^VAULT OK - cooking directory {dir_id} took ' | f"VAULT OK - cooking directory {dir_id} took " | ||||
| r'[0-9]\.[0-9]{2}s and succeeded.\n' | f"10.00s and succeeded.\n" | ||||
| r"\| 'total_time' = [0-9]\.[0-9]{2}s$", | f"| 'total_time' = 10.00s\n") | ||||
| result.output) | |||||
| assert result.exit_code == 0, result.output | assert result.exit_code == 0, result.output | ||||
| sleep_mock.assert_called_once_with(10) | |||||
| def test_vault_delayed_success(requests_mock, mocker): | def test_vault_delayed_success(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) | ||||
| scenario.add_step('get', url, response_done) | scenario.add_step('get', url, response_done) | ||||
| scenario.install_mock(requests_mock) | scenario.install_mock(requests_mock) | ||||
| get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') | get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') | ||||
| get_storage_mock.side_effect = FakeStorage | get_storage_mock.side_effect = FakeStorage | ||||
| sleep_mock = mocker.patch('time.sleep') | |||||
| result = invoke([ | result = invoke([ | ||||
| 'check-vault', | 'check-vault', | ||||
| '--swh-web-url', 'mock://swh-web.example.org', | '--swh-web-url', 'mock://swh-web.example.org', | ||||
| '--swh-storage-url', 'foo://example.org', | '--swh-storage-url', 'foo://example.org', | ||||
| 'directory', | 'directory', | ||||
| ]) | ]) | ||||
| assert re.match( | assert result.output == ( | ||||
| rf'^VAULT OK - cooking directory {dir_id} took ' | f"VAULT OK - cooking directory {dir_id} took " | ||||
| r'[0-9]\.[0-9]{2}s and succeeded.\n' | f"20.00s and succeeded.\n" | ||||
| r"\| 'total_time' = [0-9]\.[0-9]{2}s$", | f"| 'total_time' = 20.00s\n") | ||||
| result.output) | |||||
| assert result.exit_code == 0, result.output | assert result.exit_code == 0, result.output | ||||
| assert sleep_mock.call_count == 2 | |||||
| def test_vault_failure(requests_mock, mocker, mocked_time): | |||||
| def test_vault_failure(requests_mock, mocker): | |||||
| 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_failed) | scenario.add_step('get', url, response_failed) | ||||
| scenario.install_mock(requests_mock) | scenario.install_mock(requests_mock) | ||||
| get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') | get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') | ||||
| get_storage_mock.side_effect = FakeStorage | get_storage_mock.side_effect = FakeStorage | ||||
| sleep_mock = mocker.patch('time.sleep') | |||||
| result = invoke([ | result = invoke([ | ||||
| 'check-vault', | 'check-vault', | ||||
| '--swh-web-url', 'mock://swh-web.example.org', | '--swh-web-url', 'mock://swh-web.example.org', | ||||
| '--swh-storage-url', 'foo://example.org', | '--swh-storage-url', 'foo://example.org', | ||||
| 'directory', | 'directory', | ||||
| ], catch_exceptions=True) | ], catch_exceptions=True) | ||||
| assert re.match( | assert result.output == ( | ||||
| rf'^VAULT CRITICAL - cooking directory {dir_id} took ' | f"VAULT CRITICAL - cooking directory {dir_id} took " | ||||
| r'[0-9]\.[0-9]{2}s and failed with: foobar\n' | f"10.00s and failed with: foobar\n" | ||||
| r"\| 'total_time' = [0-9]\.[0-9]{2}s\n$", | f"| 'total_time' = 10.00s\n") | ||||
| result.output) | |||||
| assert result.exit_code == 2, result.output | assert result.exit_code == 2, result.output | ||||
| sleep_mock.assert_called_once_with(10) | |||||
| def test_vault_timeout(requests_mock, mocker): | |||||
| time_offset = 0 | |||||
| def increment_time(): | |||||
| nonlocal time_offset | |||||
| time_offset += 4000 | |||||
| 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) | ||||
| scenario.add_step('get', url, response_pending, | scenario.add_step('get', url, response_pending, | ||||
| callback=increment_time) | callback=lambda: time.sleep(4000)) | ||||
| scenario.install_mock(requests_mock) | scenario.install_mock(requests_mock) | ||||
| get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') | get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') | ||||
| get_storage_mock.side_effect = FakeStorage | get_storage_mock.side_effect = FakeStorage | ||||
| sleep_mock = mocker.patch('time.sleep') | |||||
| real_time = time.time | |||||
| mocker.patch( | |||||
| 'time.time', side_effect=lambda: real_time() + time_offset) | |||||
| result = invoke([ | result = invoke([ | ||||
| 'check-vault', | 'check-vault', | ||||
| '--swh-web-url', 'mock://swh-web.example.org', | '--swh-web-url', 'mock://swh-web.example.org', | ||||
| '--swh-storage-url', 'foo://example.org', | '--swh-storage-url', 'foo://example.org', | ||||
| 'directory', | 'directory', | ||||
| ], catch_exceptions=True) | ], catch_exceptions=True) | ||||
| assert re.match( | assert result.output == ( | ||||
| rf'^VAULT CRITICAL - cooking directory {dir_id} took more than ' | f"VAULT CRITICAL - cooking directory {dir_id} took more than " | ||||
| r'[0-9]+\.[0-9]{2}s and has status: foo\n' | f"4020.00s and has status: foo\n" | ||||
| r"\| 'total_time' = [0-9]{4}\.[0-9]{2}s\n$", | f"| 'total_time' = 4020.00s\n") | ||||
| result.output) | |||||
| assert result.exit_code == 2, result.output | assert result.exit_code == 2, result.output | ||||
| assert sleep_mock.call_count == 2 | |||||