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() | ||||