Changeset View
Changeset View
Standalone View
Standalone View
swh/icinga_plugins/tests/web_scenario.py
# Copyright (C) 2019 The Software Heritage developers | # Copyright (C) 2019-2022 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 | ||||
"""Wrapper around requests-mock to mock successive responses | """Wrapper around requests-mock to mock successive responses | ||||
from a web service. | from a web service. | ||||
Tests can build successive steps by calling :py:meth:`WebScenario.add_step` | Tests can build successive steps by calling :py:meth:`WebScenario.add_step` | ||||
with specifications of what endpoints should be called and in what order.""" | with specifications of what endpoints should be called and in what order.""" | ||||
from dataclasses import dataclass | import dataclasses | ||||
import json | import json | ||||
from typing import Callable, List, Optional, Set | from typing import Callable, Dict, List, Optional, Set, Union | ||||
import requests_mock | import requests_mock | ||||
@dataclass(frozen=True) | @dataclasses.dataclass(frozen=True) | ||||
class Step: | class Step: | ||||
expected_method: str | expected_method: str | ||||
expected_url: str | expected_url: str | ||||
response: object | response: Union[str, bytes, Dict, List] | ||||
status_code: int = 200 | status_code: int = 200 | ||||
headers: Dict[str, str] = dataclasses.field(default_factory=dict) | |||||
callback: Optional[Callable[[], int]] = None | callback: Optional[Callable[[], int]] = None | ||||
@dataclass(frozen=True) | @dataclasses.dataclass(frozen=True) | ||||
class Endpoint: | class Endpoint: | ||||
method: str | method: str | ||||
url: str | url: str | ||||
class WebScenario: | class WebScenario: | ||||
"""Stores the state of the successive calls to the web service | """Stores the state of the successive calls to the web service | ||||
expected by tests.""" | expected by tests.""" | ||||
Show All 26 Lines | def add_step(self, *args, **kwargs): | ||||
self.add_endpoint(step.expected_method, step.expected_url) | self.add_endpoint(step.expected_method, step.expected_url) | ||||
def install_mock(self, mocker: requests_mock.Mocker): | def install_mock(self, mocker: requests_mock.Mocker): | ||||
"""Mocks entrypoints registered with :py:meth:`add_endpoint` | """Mocks entrypoints registered with :py:meth:`add_endpoint` | ||||
(or :py:meth:`add_step`) using the provided mocker. | (or :py:meth:`add_step`) using the provided mocker. | ||||
""" | """ | ||||
for endpoint in self._endpoints: | for endpoint in self._endpoints: | ||||
mocker.register_uri( | mocker.register_uri( | ||||
endpoint.method.upper(), | endpoint.method.upper(), endpoint.url, content=self._request_callback, | ||||
endpoint.url, | |||||
text=self._request_callback, # type: ignore # stubs are too strict | |||||
) | ) | ||||
def _request_callback(self, request, context): | def _request_callback(self, request, context): | ||||
step = self._steps[self._current_step] | step = self._steps[self._current_step] | ||||
assert request.url == step.expected_url | assert request.url == step.expected_url | ||||
assert request.method.upper() == step.expected_method.upper() | assert request.method.upper() == step.expected_method.upper() | ||||
self._current_step += 1 | self._current_step += 1 | ||||
context.status_code = step.status_code | context.status_code = step.status_code | ||||
context.headers.update(step.headers) | |||||
if step.callback: | if step.callback: | ||||
step.callback() | step.callback() | ||||
if isinstance(step.response, str): | if isinstance(step.response, str): | ||||
return step.response.encode() | |||||
elif isinstance(step.response, bytes): | |||||
return step.response | return step.response | ||||
else: | else: | ||||
return json.dumps(step.response) | return json.dumps(step.response).encode() |