Changeset View
Changeset View
Standalone View
Standalone View
swh/core/pytest_plugin.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 | ||||
| from collections import deque | |||||
| from functools import partial | from functools import partial | ||||
| import logging | import logging | ||||
| from os import path | from os import path | ||||
| import re | import re | ||||
| from typing import Dict, List, Optional | from typing import Dict, List, Optional | ||||
| from urllib.parse import unquote, urlparse | from urllib.parse import unquote, urlparse | ||||
| from _pytest.fixtures import FixtureRequest | from _pytest.fixtures import FixtureRequest | ||||
| ▲ Show 20 Lines • Show All 308 Lines • ▼ Show 20 Lines | def _push_request_context(request: FixtureRequest): | ||||
| app = request.getfixturevalue("app") | app = request.getfixturevalue("app") | ||||
| ctx = app.test_request_context() | ctx = app.test_request_context() | ||||
| ctx.push() | ctx.push() | ||||
| def teardown(): | def teardown(): | ||||
| ctx.pop() | ctx.pop() | ||||
| request.addfinalizer(teardown) | request.addfinalizer(teardown) | ||||
| class FakeSocket(object): | |||||
| """ A fake socket for testing. """ | |||||
| def __init__(self): | |||||
| self.payloads = deque() | |||||
| def send(self, payload): | |||||
| assert type(payload) == bytes | |||||
| self.payloads.append(payload) | |||||
| def recv(self): | |||||
| try: | |||||
| return self.payloads.popleft().decode("utf-8") | |||||
| except IndexError: | |||||
| return None | |||||
| def close(self): | |||||
| pass | |||||
| def __repr__(self): | |||||
| return str(self.payloads) | |||||
| # simple fixture giving a Statsd instance | |||||
| @pytest.fixture | |||||
| def statsd(): | |||||
| from swh.core.statsd import Statsd | |||||
| statsd = Statsd() | |||||
| statsd._socket = FakeSocket() | |||||
| yield statsd | |||||