67 Returns the service HTTP ping URL that is used by the testsuite to detect
68 that the service is ready to work. Returns None if there's no such URL.
70 By default, attempts to find server::handlers::Ping component by
71 "handler-ping" name in static config. Override this fixture to change the
74 @ingroup userver_testsuite_fixtures
76 components = service_config[
'components_manager'][
'components']
77 ping_handler = components.get(
'handler-ping')
79 return url_util.join(service_baseurl, ping_handler[
'path'])
83@pytest.fixture(scope='session')
106 service_http_ping_url,
107 service_config_path_temp,
109 service_non_http_health_checks,
112 Prepares the start of the service daemon.
113 Configures the health checking to use service_http_ping_url fixture value
114 if it is not None; otherwise uses the service_non_http_health_checks info.
116 @see @ref pytest_userver.plugins.service.service_daemon_instance "service_daemon_instance"
117 @ingroup userver_testsuite_fixtures
119 assert service_http_ping_url
or service_non_http_health_checks.tcp, (
120 '"service_http_ping_url" and "create_health_checker" fixtures '
121 'returned None. Testsuite is unable to detect if the service is ready '
122 'to accept requests.',
126 'userver fixture "service_daemon_scope" would check for "%s"',
127 service_non_http_health_checks,
134 async def _checker(*, session, process) -> bool:
135 LocalCounters.attempts += 1
136 new_log_time = time.monotonic()
137 if new_log_time - LocalCounters.last_log_time > 1.0:
138 LocalCounters.last_log_time = new_log_time
140 'userver fixture "service_daemon_scope" checking "%s", attempt %s',
141 service_non_http_health_checks,
142 LocalCounters.attempts,
145 return await net.check_availability(service_non_http_health_checks)
147 health_check = _checker
148 if service_http_ping_url:
151 async with create_daemon_scope(
152 args=[str(service_binary),
'--config', str(service_config_path_temp)],
153 ping_url=service_http_ping_url,
154 health_check=health_check,
285 Depend on this fixture directly or transitively to make your fixture a per-daemon fixture.
290 @pytest.fixture(scope='session')
291 def users_cache_state(daemon_scoped_mark, ...):
292 return UsersCacheState(users_list=[])
295 For tests marked with `@pytest.mark.uservice_oneshot(...)`, the service will be restarted,
296 and all the per-daemon fixtures will be recreated.
298 This fixture returns kwargs passed to the `uservice_oneshot` mark (which may be an empty dict).
299 For normal tests, this fixture returns `None`.
301 @ingroup userver_testsuite_fixtures
309 return getattr(request,
'param',
None)
315def pytest_configure(config):
316 config.addinivalue_line(
318 'uservice_oneshot: use a per-test service daemon instance',
322def _contains_oneshot_marker(parametrize: Iterable[pytest.Mark]) -> bool:
324 Check if at least one of 'parametrize' marks is of the form:
326 @pytest.mark.parametrize(
330 pytest.param("b", 20, marks=pytest.mark.uservice_oneshot), # <====
336 for parametrize_mark
in parametrize
337 if len(parametrize_mark.args) >= 2
338 for parameter_set
in parametrize_mark.args[1]
339 if hasattr(parameter_set,
'marks')
340 for mark
in parameter_set.marks
341 if mark.name ==
'uservice_oneshot'
345def pytest_generate_tests(metafunc: pytest.Metafunc) ->
None:
346 oneshot_marker = metafunc.definition.get_closest_marker(
'uservice_oneshot')
347 parametrize_markers = metafunc.definition.iter_markers(
'parametrize')
348 if oneshot_marker
is not None or _contains_oneshot_marker(parametrize_markers):
350 metafunc.parametrize(
351 (daemon_scoped_mark.__name__,),
356 ids=[
'uservice_oneshot'],
365def pytest_collection_modifyitems(items: List[pytest.Item]):
367 oneshot_marker = item.get_closest_marker(
'uservice_oneshot')
368 if oneshot_marker
and isinstance(item, pytest.Function):
369 func_item = typing.cast(pytest.Function, item)
370 func_item.callspec.params[daemon_scoped_mark.__name__] = dict(oneshot_marker.kwargs, function=func_item)