userver: /home/antonyzhilin/arcadia/taxi/uservices/userver/testsuite/pytest_plugins/pytest_userver/plugins/base.py Source File
Loading...
Searching...
No Matches
base.py
1"""
2Configure the service in testsuite.
3"""
4
5import pathlib
6import socket
7from typing import Callable
8from typing import Optional
9
10import pytest
11
12import testsuite.plugins.network
13
14USERVER_CONFIG_HOOKS = ['userver_base_prepare_service_config']
15
16
17def pytest_addoption(parser) -> None:
18 group = parser.getgroup('userver')
19 group.addoption(
20 '--build-dir',
21 type=pathlib.Path,
22 help='Path to service build directory.',
23 )
24
25 group = parser.getgroup('Test service')
26 group.addoption(
27 '--service-binary',
28 type=pathlib.Path,
29 help='Path to service binary.',
30 )
31 group.addoption(
32 '--service-port',
33 help=('Main HTTP port of the service (default: use the port from the static config)'),
34 default=None,
35 type=int,
36 )
37 group.addoption(
38 '--monitor-port',
39 help=('Monitor HTTP port of the service (default: use the port from the static config)'),
40 default=None,
41 type=int,
42 )
43 group.addoption(
44 '--service-source-dir',
45 type=pathlib.Path,
46 help='Path to service source directory.',
47 default=pathlib.Path('.'),
48 )
49
50
51def pytest_configure(config):
52 config.option.asyncio_mode = 'auto'
53
54
55@pytest.fixture(scope='session')
56def service_source_dir(pytestconfig) -> pathlib.Path:
57 """
58 Returns the path to the service source directory that is set by command
59 line `--service-source-dir` option.
60
61 Override this fixture to change the way the path to the service
62 source directory is detected by testsuite.
63
64 @ingroup userver_testsuite_fixtures
65 """
66 return pytestconfig.option.service_source_dir
67
68
69@pytest.fixture(scope='session')
70def build_dir(pytestconfig) -> pathlib.Path:
71 """
72 Returns the build directory set by command line `--build-dir` option.
73
74 Override this fixture to change the way the build directory is
75 detected by the testsuite.
76
77 @ingroup userver_testsuite_fixtures
78 """
79 return pytestconfig.option.build_dir
80
81
82@pytest.fixture(scope='session')
83def service_binary(pytestconfig) -> pathlib.Path:
84 """
85 Returns the path to service binary set by command line `--service-binary`
86 option.
87
88 Override this fixture to change the way the path to service binary is
89 detected by the testsuite.
90
91 @ingroup userver_testsuite_fixtures
92 """
93 return pytestconfig.option.service_binary
94
95
96@pytest.fixture(scope='session')
97def service_port(pytestconfig, _original_service_config, choose_free_port) -> int:
98 """
99 Returns the main listener port number of the service set by command line
100 `--service-port` option.
101 If no port is specified in the command line option, keeps the original port
102 specified in the static config.
103
104 Override this fixture to change the way the main listener port number is
105 detected by the testsuite.
106
107 @ingroup userver_testsuite_fixtures
108 """
109 return pytestconfig.option.service_port or _get_port(
110 _original_service_config,
111 choose_free_port,
112 'listener',
113 service_port,
114 '--service-port',
115 )
116
117
118@pytest.fixture(scope='session')
119def monitor_port(pytestconfig, _original_service_config, choose_free_port) -> int:
120 """
121 Returns the monitor listener port number of the service set by command line
122 `--monitor-port` option.
123 If no port is specified in the command line option, keeps the original port
124 specified in the static config.
125
126 Override this fixture to change the way the monitor listener port number
127 is detected by testsuite.
128
129 @ingroup userver_testsuite_fixtures
130 """
131 return pytestconfig.option.monitor_port or _get_port(
132 _original_service_config,
133 choose_free_port,
134 'listener-monitor',
135 monitor_port,
136 '--service-port',
137 )
138
139
140def _get_port(
141 original_service_config,
142 choose_free_port,
143 listener_name,
144 port_fixture,
145 option_name,
146) -> int:
147 config_yaml = original_service_config.config_yaml
148 config_vars = original_service_config.config_vars
149 components = config_yaml['components_manager']['components']
150 listener = components.get('server', {}).get(listener_name, {})
151 if not listener:
152 return -1
153 port = listener.get('port', None)
154 if isinstance(port, str) and port.startswith('$'):
155 port = config_vars.get(port[1:], None) or listener.get(
156 'port#fallback',
157 None,
158 )
159 assert port, (
160 f'Please specify '
161 f'components_manager.components.server.{listener_name}.port '
162 f'in the static config, or pass {option_name} pytest option, '
163 f'or override the {port_fixture.__name__} fixture'
164 )
165 return choose_free_port(port)
166
167
168# Beware: global variable
169_allocated_ports = set()
170
171
172@pytest.fixture(scope='session')
174 pytestconfig, get_free_port, _testsuite_socket_cleanup, _testsuite_default_af
175) -> Callable[[Optional[int]], int]:
176 """
177 A function that chooses a free port based on the optional hint given in the parameter.
178
179 @ingroup userver_testsuite_fixtures
180 """
181
182 family, address = _testsuite_default_af
183
184 def choose(port_hint: Optional[int] = None, /) -> int:
185 should_not_randomize_ports = pytestconfig.option.service_runner_mode
186 if should_not_randomize_ports and port_hint is not None and port_hint != 0:
187 if _is_port_free(port_hint, family, address):
188 _allocated_ports.add(port_hint)
189 return port_hint
190 port = _get_free_port_not_allocated(get_free_port)
191 _allocated_ports.add(port)
192 return port
193
194 return choose
195
196
197def _get_free_port_not_allocated(get_free_port) -> int:
198 for _ in range(100):
199 port = get_free_port()
200 if port not in _allocated_ports:
201 return port
202 raise testsuite.plugins.network.NoEnabledPorts()
203
204
205def _is_port_free(port_num: int, family: int, address: str) -> bool:
206 try:
207 with socket.socket(family, socket.SOCK_STREAM) as sock:
208 sock.bind((address, port_num))
209 except OSError:
210 return False
211 else:
212 return True
213
214
215@pytest.fixture(scope='session')
216def userver_base_prepare_service_config():
217 def patch_config(config, config_vars):
218 components = config['components_manager']['components']
219 if 'congestion-control' in components:
220 if components['congestion-control'] is None:
221 components['congestion-control'] = {}
222
223 components['congestion-control']['fake-mode'] = True
224
225 return patch_config