2Configure the service in testsuite.
7from typing
import Callable
8from typing
import Optional
12import testsuite.plugins.network
14USERVER_CONFIG_HOOKS = [
'userver_base_prepare_service_config']
17def pytest_addoption(parser) -> None:
18 group = parser.getgroup(
'userver')
22 help=
'Path to service build directory.',
25 group = parser.getgroup(
'Test service')
29 help=
'Path to service binary.',
33 help=(
'Main HTTP port of the service (default: use the port from the static config)'),
39 help=(
'Monitor HTTP port of the service (default: use the port from the static config)'),
44 '--service-source-dir',
46 help=
'Path to service source directory.',
47 default=pathlib.Path(
'.'),
51def pytest_configure(config):
52 config.option.asyncio_mode =
'auto'
55@pytest.fixture(scope='session')
58 Returns the path to the service source directory that is set by command
59 line `--service-source-dir` option.
61 Override this fixture to change the way the path to the service
62 source directory is detected by testsuite.
64 @ingroup userver_testsuite_fixtures
66 return pytestconfig.option.service_source_dir
69@pytest.fixture(scope='session')
72 Returns the build directory set by command line `--build-dir` option.
74 Override this fixture to change the way the build directory is
75 detected by the testsuite.
77 @ingroup userver_testsuite_fixtures
79 return pytestconfig.option.build_dir
82@pytest.fixture(scope='session')
85 Returns the path to service binary set by command line `--service-binary`
88 Override this fixture to change the way the path to service binary is
89 detected by the testsuite.
91 @ingroup userver_testsuite_fixtures
93 return pytestconfig.option.service_binary
96@pytest.fixture(scope='session')
97def service_port(pytestconfig, _original_service_config, choose_free_port) -> int:
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.
104 Override this fixture to change the way the main listener port number is
105 detected by the testsuite.
107 @ingroup userver_testsuite_fixtures
109 return pytestconfig.option.service_port
or _get_port(
110 _original_service_config,
118@pytest.fixture(scope='session')
119def monitor_port(pytestconfig, _original_service_config, choose_free_port) -> int:
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.
126 Override this fixture to change the way the monitor listener port number
127 is detected by testsuite.
129 @ingroup userver_testsuite_fixtures
131 return pytestconfig.option.monitor_port
or _get_port(
132 _original_service_config,
141 original_service_config,
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, {})
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(
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'
169_allocated_ports = set()
172@pytest.fixture(scope='session')
174 pytestconfig, get_free_port, _testsuite_socket_cleanup, _testsuite_default_af
175) -> Callable[[Optional[int]], int]:
177 A function that chooses a free port based on the optional hint given in the parameter.
179 @ingroup userver_testsuite_fixtures
182 family, address = _testsuite_default_af
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)
190 port = _get_free_port_not_allocated(get_free_port)
191 _allocated_ports.add(port)
197def _get_free_port_not_allocated(get_free_port) -> int:
199 port = get_free_port()
200 if port
not in _allocated_ports:
202 raise testsuite.plugins.network.NoEnabledPorts()
205def _is_port_free(port_num: int, family: int, address: str) -> bool:
207 with socket.socket(family, socket.SOCK_STREAM)
as sock:
208 sock.bind((address, port_num))
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'] = {}
223 components[
'congestion-control'][
'fake-mode'] =
True