Changeset View
Changeset View
Standalone View
Standalone View
swh/icinga_plugins/tests/web_scenario.py
- This file was added.
| # 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 | |||||
| from dataclasses import dataclass | |||||
| import json | |||||
| from typing import List, Set, Optional, Callable | |||||
| @dataclass(frozen=True) | |||||
| class ScenarioStep: | |||||
| 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: | |||||
| _steps: List[ScenarioStep] | |||||
| _endpoints: Set[Endpoint] | |||||
| _current_step: int | |||||
| def __init__(self): | |||||
| self._steps = [] | |||||
| self._endpoints = set() | |||||
| self._current_step = 0 | |||||
| def add_endpoint(self, *args, **kwargs): | |||||
| self._endpoints.add(Endpoint(*args, **kwargs)) | |||||
| def add_step(self, *args, **kwargs): | |||||
| step = ScenarioStep(*args, **kwargs) | |||||
| self._steps.append(step) | |||||
| self.add_endpoint(step.expected_method, step.expected_url) | |||||
| def install_mock(self, 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) | |||||