72 Returns the service HTTP ping URL that is used by the testsuite to detect
73 that the service is ready to work. Returns None if there's no such URL.
75 By default, attempts to find server::handlers::Ping component by
76 "handler-ping" name in static config. Override this fixture to change the
79 @ingroup userver_testsuite_fixtures
81 components = service_config[
'components_manager'][
'components']
82 ping_handler = components.get(
'handler-ping')
84 return url_util.join(service_baseurl, ping_handler[
'path'])
88@pytest.fixture(scope='session')
111 service_http_ping_url,
112 service_config_path_temp,
114 service_non_http_health_checks,
117 Prepares the start of the service daemon.
118 Configures the health checking to use service_http_ping_url fixture value
119 if it is not None; otherwise uses the service_non_http_health_checks info.
121 @see @ref pytest_userver.plugins.service.service_daemon_instance "service_daemon_instance"
122 @ingroup userver_testsuite_fixtures
124 assert service_http_ping_url
or service_non_http_health_checks.tcp, (
125 '"service_http_ping_url" and "create_health_checker" fixtures '
126 'returned None. Testsuite is unable to detect if the service is ready '
127 'to accept requests.',
131 'userver fixture "service_daemon_scope" would check for "%s"',
132 service_non_http_health_checks,
139 async def _checker(*, session, process) -> bool:
140 LocalCounters.attempts += 1
141 new_log_time = time.monotonic()
142 if new_log_time - LocalCounters.last_log_time > 1.0:
143 LocalCounters.last_log_time = new_log_time
145 'userver fixture "service_daemon_scope" checking "%s", attempt %s',
146 service_non_http_health_checks,
147 LocalCounters.attempts,
150 return await net.check_availability(service_non_http_health_checks)
152 health_check = _checker
153 if service_http_ping_url:
156 async with create_daemon_scope(
157 args=[str(service_binary),
'--config', str(service_config_path_temp)],
158 ping_url=service_http_ping_url,
159 health_check=health_check,
290 Depend on this fixture directly or transitively to make your fixture a per-daemon fixture.
295 @pytest.fixture(scope='session')
296 def users_cache_state(daemon_scoped_mark, ...):
297 return UsersCacheState(users_list=[])
300 For tests marked with `@pytest.mark.uservice_oneshot(...)`, the service will be restarted,
301 and all the per-daemon fixtures will be recreated.
303 This fixture returns kwargs passed to the `uservice_oneshot` mark (which may be an empty dict).
304 For normal tests, this fixture returns `None`.
306 @ingroup userver_testsuite_fixtures
314 return getattr(request,
'param',
None)
320def pytest_configure(config):
321 config.addinivalue_line(
323 'uservice_oneshot: use a per-test service daemon instance',
327def _contains_oneshot_marker(parametrize: Iterable[pytest.Mark]) -> bool:
329 Check if at least one of 'parametrize' marks is of the form:
331 @pytest.mark.parametrize(
335 pytest.param("b", 20, marks=pytest.mark.uservice_oneshot), # <====
341 for parametrize_mark
in parametrize
342 if len(parametrize_mark.args) >= 2
343 for parameter_set
in parametrize_mark.args[1]
344 if hasattr(parameter_set,
'marks')
345 for mark
in parameter_set.marks
346 if mark.name ==
'uservice_oneshot'
350def pytest_generate_tests(metafunc: pytest.Metafunc) ->
None:
351 oneshot_marker = metafunc.definition.get_closest_marker(
'uservice_oneshot')
352 parametrize_markers = metafunc.definition.iter_markers(
'parametrize')
353 if oneshot_marker
is not None or _contains_oneshot_marker(parametrize_markers):
355 metafunc.parametrize(
356 (daemon_scoped_mark.__name__,),
361 ids=[
'uservice_oneshot'],
370def pytest_collection_modifyitems(items: List[pytest.Item]):
372 oneshot_marker = item.get_closest_marker(
'uservice_oneshot')
373 if oneshot_marker
and isinstance(item, pytest.Function):
374 func_item = typing.cast(pytest.Function, item)
375 func_item.callspec.params[daemon_scoped_mark.__name__] = dict(oneshot_marker.kwargs, function=func_item)