userver: /data/code/userver/testsuite/pytest_plugins/pytest_userver/plugins/service.py Source File
Loading...
Searching...
No Matches
service.py
1"""
2Start the service in testsuite.
3"""
4
5# pylint: disable=redefined-outer-name
6import logging
7import pathlib
8import time
9import typing
10
11import pytest
12
13from testsuite.utils import url_util
14
15from ..utils import net
16
17
18logger_testsuite = logging.getLogger(__name__)
19
20
21def pytest_addoption(parser) -> None:
22 group = parser.getgroup('userver')
23 group.addoption(
24 '--service-logs-file',
25 type=pathlib.Path,
26 help='Write service output to specified file',
27 )
28 group.addoption(
29 '--service-logs-pretty',
30 action='store_true',
31 help='Enable pretty print and colorize service logs',
32 )
33 group.addoption(
34 '--service-logs-pretty-verbose',
35 dest='service_logs_pretty',
36 action='store_const',
37 const='verbose',
38 help='Enable pretty print and colorize service logs in verbose mode',
39 )
40 group.addoption(
41 '--service-logs-pretty-disable',
42 action='store_false',
43 dest='service_logs_pretty',
44 help='Disable pretty print and colorize service logs',
45 )
46
47
48@pytest.fixture(scope='session')
50 """
51 Override this to pass extra environment variables to the service.
52
53 @snippet samples/redis_service/tests/conftest.py service_env
54 @ingroup userver_testsuite_fixtures
55 """
56 return None
57
58
59@pytest.fixture(scope='session')
61 service_config, service_baseurl,
62) -> typing.Optional[str]:
63 """
64 Returns the service HTTP ping URL that is used by the testsuite to detect
65 that the service is ready to work. Returns None if there's no such URL.
66
67 By default attempts to find server::handlers::Ping component by
68 "handler-ping" name in static config. Override this fixture to change the
69 behavior.
70
71 @ingroup userver_testsuite_fixtures
72 """
73 components = service_config['components_manager']['components']
74 ping_handler = components.get('handler-ping')
75 if ping_handler:
76 return url_util.join(service_baseurl, ping_handler['path'])
77 return None
78
79
80@pytest.fixture(scope='session')
81def service_non_http_health_checks( # pylint: disable=invalid-name
82 service_config,
83) -> net.HealthChecks:
84 """
85 Returns a health checks info.
86
87 By default, returns pytest_userver.utils.net.get_health_checks_info().
88
89 Override this fixture to change the way testsuite detects the tested
90 service being alive.
91
92 @ingroup userver_testsuite_fixtures
93 """
94
95 return net.get_health_checks_info(service_config)
96
97
98@pytest.fixture(scope='session')
100 pytestconfig,
101 create_daemon_scope,
102 service_env,
103 service_http_ping_url,
104 service_config_path_temp,
105 service_config,
106 service_binary,
107 service_non_http_health_checks,
108 testsuite_logger,
109):
110 """
111 Configures the health checking to use service_http_ping_url fixture value
112 if it is not None; otherwise uses the service_non_http_health_checks info.
113 Starts the service daemon.
114
115 @ingroup userver_testsuite_fixtures
116 """
117 assert service_http_ping_url or service_non_http_health_checks.tcp, (
118 '"service_http_ping_url" and "create_health_checker" fixtures '
119 'returned None. Testsuite is unable to detect if the service is ready '
120 'to accept requests.',
121 )
122
123 logger_testsuite.debug(
124 'userver fixture "service_daemon" would check for "%s"',
125 service_non_http_health_checks,
126 )
127
128 class LocalCounters:
129 last_log_time = 0.0
130 attempts = 0
131
132 async def _checker(*, session, process) -> bool:
133 LocalCounters.attempts += 1
134 new_log_time = time.monotonic()
135 if new_log_time - LocalCounters.last_log_time > 1.0:
136 LocalCounters.last_log_time = new_log_time
137 logger_testsuite.debug(
138 'userver fixture "service_daemon" checking "%s", attempt %s',
139 service_non_http_health_checks,
140 LocalCounters.attempts,
141 )
142
143 return await net.check_availability(service_non_http_health_checks)
144
145 health_check = _checker
146 if service_http_ping_url:
147 health_check = None
148
149 async with create_daemon_scope(
150 args=[
151 str(service_binary),
152 '--config',
153 str(service_config_path_temp),
154 ],
155 ping_url=service_http_ping_url,
156 health_check=health_check,
157 env=service_env,
158 ) as scope:
159 yield scope