2Start the service in testsuite.
15from testsuite.logging
import logger
16from testsuite.utils
import url_util
18from ..utils
import colorize
19from ..utils
import net
22logger_testsuite = logging.getLogger(__name__)
27 self, *, writer: logger.LineLogger, verbose, colors_enabled,
29 super().__init__(writer)
31 verbose=verbose, colors_enabled=colors_enabled,
34 def log_service_line(self, line) -> None:
39 def log_entry(self, entry: dict) ->
None:
45def pytest_addoption(parser) -> None:
46 group = parser.getgroup(
'userver')
48 '--service-logs-file',
50 help=
'Write service output to specified file',
53 '--service-logs-pretty',
55 help=
'Enable pretty print and colorize service logs',
58 '--service-logs-pretty-verbose',
59 dest=
'service_logs_pretty',
62 help=
'Enable pretty print and colorize service logs in verbose mode',
65 '--service-logs-pretty-disable',
67 dest=
'service_logs_pretty',
68 help=
'Disable pretty print and colorize service logs',
72def pytest_override_testsuite_logger(
73 config, line_logger: logger.LineLogger, colors_enabled: bool,
74) -> typing.Optional[logger.Logger]:
75 pretty_logs = config.option.service_logs_pretty
80 verbose=pretty_logs ==
'verbose',
81 colors_enabled=colors_enabled,
85@pytest.fixture(scope='session')
88 Override this to pass extra environment variables to the service.
90 @snippet samples/redis_service/tests/conftest.py service_env
91 @ingroup userver_testsuite_fixtures
96@pytest.fixture(scope='session')
98 service_config, service_baseurl,
99) -> typing.Optional[str]:
101 Returns the service HTTP ping URL that is used by the testsuite to detect
102 that the service is ready to work. Returns None if there's no such URL.
104 By default attempts to find server::handlers::Ping component by
105 "handler-ping" name in static config. Override this fixture to change the
108 @ingroup userver_testsuite_fixtures
110 components = service_config[
'components_manager'][
'components']
111 ping_handler = components.get(
'handler-ping')
113 return url_util.join(service_baseurl, ping_handler[
'path'])
117@pytest.fixture(scope='session')
120) -> net.HealthChecks:
122 Returns a health checks info.
124 By default, returns pytest_userver.utils.net.get_health_checks_info().
126 Override this fixture to change the way testsuite detects the tested
129 @ingroup userver_testsuite_fixtures
132 return net.get_health_checks_info(service_config)
135@pytest.fixture(scope='session')
140 service_http_ping_url,
141 service_config_path_temp,
144 service_non_http_health_checks,
146 _userver_log_handler,
149 Configures the health checking to use service_http_ping_url fixture value
150 if it is not None; otherwise uses the service_non_http_health_checks info.
151 Starts the service daemon.
153 @ingroup userver_testsuite_fixtures
155 assert service_http_ping_url
or service_non_http_health_checks.tcp, (
156 '"service_http_ping_url" and "create_health_checker" fixtures '
157 'returned None. Testsuite is unable to detect if the service is ready '
158 'to accept requests.',
161 logger_testsuite.debug(
162 'userver fixture "service_daemon" would check for "%s"',
163 service_non_http_health_checks,
170 async def _checker(*, session, process) -> bool:
171 LocalCounters.attempts += 1
172 new_log_time = time.monotonic()
173 if new_log_time - LocalCounters.last_log_time > 1.0:
174 LocalCounters.last_log_time = new_log_time
175 logger_testsuite.debug(
176 'userver fixture "service_daemon" checking "%s", attempt %s',
177 service_non_http_health_checks,
178 LocalCounters.attempts,
181 return await net.check_availability(service_non_http_health_checks)
183 health_check = _checker
184 if service_http_ping_url:
187 async with create_daemon_scope(
191 str(service_config_path_temp),
193 ping_url=service_http_ping_url,
194 health_check=health_check,
196 stderr_handler=_userver_log_handler,
201@pytest.fixture(scope='session')
202def _userver_log_handler(pytestconfig, testsuite_logger, _uservice_logfile):
203 service_logs_pretty = pytestconfig.option.service_logs_pretty
204 if not service_logs_pretty
and not bool(_uservice_logfile):
207 if service_logs_pretty:
208 logger_plugin = pytestconfig.pluginmanager.getplugin(
211 logger_plugin.enable_logs_suspension()
213 def log_handler(line_binary):
214 if _uservice_logfile:
215 _uservice_logfile.write(line_binary)
217 line = line_binary.decode(
'utf-8').rstrip(
'\r\n')
218 testsuite_logger.log_service_line(line)
221 traceback.print_exc(file=sys.stderr)
226@pytest.fixture(scope='session')
227def _uservice_logfile(pytestconfig):
228 path = pytestconfig.option.service_logs_file
232 with path.open(
'wb')
as fp: