diff --git a/mypy.ini b/mypy.ini --- a/mypy.ini +++ b/mypy.ini @@ -11,5 +11,8 @@ [mypy-pytest.*] ignore_missing_imports = True +[mypy-requests_mock.*] +ignore_missing_imports = True + # [mypy-add_your_lib_here.*] # ignore_missing_imports = True diff --git a/swh/icinga_plugins/tests/test_vault.py b/swh/icinga_plugins/tests/test_vault.py --- a/swh/icinga_plugins/tests/test_vault.py +++ b/swh/icinga_plugins/tests/test_vault.py @@ -3,14 +3,13 @@ # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information -import enum -import json import re import time from click.testing import CliRunner from swh.icinga_plugins.cli import cli +from .web_scenario import WebScenario dir_id = 'ab'*20 @@ -56,41 +55,15 @@ def test_vault_immediate_success(requests_mock, mocker): + scenario = WebScenario() - class Step(enum.Enum): - NOTHING_DONE = 0 - CHECKED_UNCOOKED = 1 - REQUESTED_COOKING = 2 - - step = Step.NOTHING_DONE - - def post_callback(request, context): - nonlocal step - if step == Step.CHECKED_UNCOOKED: - step = Step.REQUESTED_COOKING - return json.dumps(response_pending) - else: - assert False, step - - def get_callback(request, context): - context.json = True - nonlocal step - if step == Step.NOTHING_DONE: - context.status_code = 404 - step = Step.CHECKED_UNCOOKED - elif step == Step.CHECKED_UNCOOKED: - assert False - elif step == Step.REQUESTED_COOKING: - return json.dumps(response_done) - else: - assert False, step - - requests_mock.get( - f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/', - text=get_callback) - requests_mock.post( - f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/', - text=post_callback) + 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_done) + + scenario.install_mock(requests_mock) get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') get_storage_mock.side_effect = FakeStorage @@ -114,45 +87,16 @@ def test_vault_delayed_success(requests_mock, mocker): + scenario = WebScenario() + + url = f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/' - class Step(enum.Enum): - NOTHING_DONE = 0 - CHECKED_UNCOOKED = 1 - REQUESTED_COOKING = 2 - PENDING = 3 - - step = Step.NOTHING_DONE - - def post_callback(request, context): - nonlocal step - if step == Step.CHECKED_UNCOOKED: - step = Step.REQUESTED_COOKING - return json.dumps(response_pending) - else: - assert False, step - - def get_callback(request, context): - context.json = True - nonlocal step - if step == Step.NOTHING_DONE: - context.status_code = 404 - step = Step.CHECKED_UNCOOKED - elif step == Step.CHECKED_UNCOOKED: - assert False - elif step == Step.REQUESTED_COOKING: - step = Step.PENDING - return json.dumps(response_pending) - elif step == Step.PENDING: - return json.dumps(response_done) - else: - assert False, step - - requests_mock.get( - f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/', - text=get_callback) - requests_mock.post( - f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/', - text=post_callback) + scenario.add_step('get', url, {}, status_code=404) + scenario.add_step('post', url, response_pending) + scenario.add_step('get', 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 @@ -176,41 +120,15 @@ def test_vault_failure(requests_mock, mocker): + 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_failed) - class Step(enum.Enum): - NOTHING_DONE = 0 - CHECKED_UNCOOKED = 1 - REQUESTED_COOKING = 2 - - step = Step.NOTHING_DONE - - def post_callback(request, context): - nonlocal step - if step == Step.CHECKED_UNCOOKED: - step = Step.REQUESTED_COOKING - return json.dumps(response_pending) - else: - assert False, step - - def get_callback(request, context): - context.json = True - nonlocal step - if step == Step.NOTHING_DONE: - context.status_code = 404 - step = Step.CHECKED_UNCOOKED - elif step == Step.CHECKED_UNCOOKED: - assert False - elif step == Step.REQUESTED_COOKING: - return json.dumps(response_failed) - else: - assert False, step - - requests_mock.get( - f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/', - text=get_callback) - requests_mock.post( - f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/', - text=post_callback) + scenario.install_mock(requests_mock) get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') get_storage_mock.side_effect = FakeStorage @@ -234,46 +152,23 @@ def test_vault_timeout(requests_mock, mocker): + time_offset = 0 + + def increment_time(): + nonlocal time_offset + time_offset += 4000 + + scenario = WebScenario() - class Step(enum.Enum): - NOTHING_DONE = 0 - CHECKED_UNCOOKED = 1 - REQUESTED_COOKING = 2 - PENDING = 3 - - step = Step.NOTHING_DONE - - def post_callback(request, context): - nonlocal step - if step == Step.CHECKED_UNCOOKED: - step = Step.REQUESTED_COOKING - return json.dumps(response_pending) - else: - assert False, step - - def get_callback(request, context): - context.json = True - nonlocal step, time_offset - if step == Step.NOTHING_DONE: - context.status_code = 404 - step = Step.CHECKED_UNCOOKED - elif step == Step.CHECKED_UNCOOKED: - assert False - elif step == Step.REQUESTED_COOKING: - step = Step.PENDING - return json.dumps(response_pending) - elif step == Step.PENDING: - time_offset += 4000 # jump forward in time more than 1h - return json.dumps(response_pending) - else: - assert False, step - - requests_mock.get( - f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/', - text=get_callback) - requests_mock.post( - f'mock://swh-web.example.org/api/1/vault/directory/{dir_id}/', - text=post_callback) + 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_pending) + scenario.add_step('get', url, response_pending, + callback=increment_time) + + scenario.install_mock(requests_mock) get_storage_mock = mocker.patch('swh.icinga_plugins.vault.get_storage') get_storage_mock.side_effect = FakeStorage @@ -281,7 +176,6 @@ sleep_mock = mocker.patch('time.sleep') real_time = time.time - time_offset = 0 mocker.patch( 'time.time', side_effect=lambda: real_time() + time_offset) diff --git a/swh/icinga_plugins/tests/web_scenario.py b/swh/icinga_plugins/tests/web_scenario.py new file mode 100644 --- /dev/null +++ b/swh/icinga_plugins/tests/web_scenario.py @@ -0,0 +1,87 @@ +# 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 + +"""Wrapper around requests-mock to mock successive responses +from a web service. + +Tests can build successive steps by calling :py:meth:`WebScenario.add_step` +with specifications of what endpoints should be called and in what order.""" + +from dataclasses import dataclass +import json +from typing import List, Set, Optional, Callable + +import requests_mock + + +@dataclass(frozen=True) +class Step: + expected_method: str + expected_url: str + response: object + status_code: int = 200 + callback: Optional[Callable[[], int]] = None + + +@dataclass(frozen=True) +class Endpoint: + method: str + url: str + + +class WebScenario: + """Stores the state of the successive calls to the web service + expected by tests.""" + + _steps: List[Step] + _endpoints: Set[Endpoint] + _current_step: int + + def __init__(self): + self._steps = [] + self._endpoints = set() + self._current_step = 0 + + def add_endpoint(self, *args, **kwargs): + """Adds an endpoint to be mocked. + + Arguments are the same as :py:class:Endpoint. + """ + self._endpoints.add(Endpoint(*args, **kwargs)) + + def add_step(self, *args, **kwargs): + """Adds an expected call to the list of expected calls. + Also automatically calls :py:meth:`add_endpoint` so the + associated endpoint is mocked. + + Arguments are the same as :py:class:`Step`. + """ + step = Step(*args, **kwargs) + self._steps.append(step) + self.add_endpoint(step.expected_method, step.expected_url) + + def install_mock(self, mocker: requests_mock.Mocker): + """Mocks entrypoints registered with :py:meth:`add_endpoint` + (or :py:meth:`add_step`) using the provided mocker. + """ + for endpoint in self._endpoints: + mocker.register_uri( + endpoint.method.upper(), endpoint.url, + text=self._request_callback) + + def _request_callback(self, request, context): + step = self._steps[self._current_step] + + assert request.url == step.expected_url + assert request.method.upper() == step.expected_method.upper() + + self._current_step += 1 + + context.status_code = step.status_code + + if step.callback: + step.callback() + + return json.dumps(step.response)