dephell/tests/conftest.py at master · Python-Repository-Hub/dephell · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
# built-in
import shutil
import socket
from pathlib import Path

# external
import pytest
from aioresponses import aioresponses


true_socket = socket.socket
true_connect = socket.socket.connect


class SocketBlockedError(RuntimeError):
    pass


class SocketConnectBlockedError(RuntimeError):
    pass


def fake_socket(*args, **kwargs):
    raise SocketBlockedError('use @pytest.mark.allow_hosts to unblock some hosts')


def pytest_addoption(parser):
    group = parser.getgroup('socket')
    group.addoption(
        '--no-network',
        action='store_true',
        help='skip tests with network interactions',
    )


@pytest.fixture(autouse=True)
def no_network(request):
    _setup(request)
    yield
    socket.socket.connect = true_connect
    socket.socket = true_socket


def _setup(request):
    """
    + socket disabled by default
    + @pytest.mark.allow_hosts() -- allow all connections
    + @pytest.mark.allow_hosts(['pypi,org']) -- allow pypi.org connections
    """
    marker = request.node.get_closest_marker('allow_hosts')

    # block network if no marker
    if not marker:
        socket.socket = fake_socket
        return

    # skip marked tests if no network connection
    if request.config.getoption('--no-network'):
        pytest.skip('test requires network connection')

    # if test marked with allow_hosts and hosts specified
    # allow only these hosts. Otherwise allow all hosts
    if marker.args:
        hosts = marker.args[0]
        socket_allow_hosts(hosts)
        return


def socket_allow_hosts(allowed=None):
    def guarded_connect(inst, *args):
        host = args[0][0]
        if host and host in allowed:
            return true_connect(inst, *args)
        raise SocketConnectBlockedError(allowed, host)

    socket.socket.connect = guarded_connect


@pytest.fixture()
def temp_path(tmp_path: Path):
    for path in tmp_path.iterdir():
        if path.is_file():
            path.unlink()
        else:
            shutil.rmtree(str(path))
    yield tmp_path


@pytest.fixture()
def temp_cache(temp_path):
    # project
    from dephell.config import config

    config.attach({'cache': {'path': str(temp_path)}})


@pytest.fixture
def tests_path() -> Path:
    """ Return the absolute Path to 'tests' directory """
    return Path(__file__).parent


@pytest.fixture
def requirements_path(tests_path) -> Path:
    """ Return the absolute Path to 'tests/requirements' """
    return tests_path / Path('requirements')


@pytest.fixture
def repository_path(tests_path) -> Path:
    """ Return the absolute Path to 'tests/repository' """
    return tests_path / Path('repository')


@pytest.fixture
def fixtures_path(tests_path) -> Path:
    return tests_path / Path('fixtures')


class PatchedAIOResponses(aioresponses):
    async def _request_mock(self, orig_self, method, url, *args, **kwargs):
        # add `orig_self` (original `ClientSession`) into `args`
        return await super()._request_mock(orig_self, method, url, orig_self, *args, **kwargs)


@pytest.fixture
def asyncio_mock() -> PatchedAIOResponses:
    with PatchedAIOResponses() as mock:
        yield mock