10logger = logging.getLogger(__name__)
16@dataclasses.dataclass(frozen=True)
19 Class that holds a host and port.
21 @ingroup userver_testsuite
28@dataclasses.dataclass(frozen=False)
31 Class that holds all the info for health checks.
33 @ingroup userver_testsuite
36 tcp: typing.List[HostPort] = dataclasses.field(default_factory=list)
39async def _check_tcp_port_availability(tcp: HostPort) -> bool:
41 _, writer = await asyncio.open_connection(tcp.host, tcp.port)
43 await writer.wait_closed()
44 except (OSError, asyncio.TimeoutError):
51 Checks availability for each provided entry.
53 @ingroup userver_testsuite
56 done, pending = await asyncio.wait(
58 asyncio.Task(_check_tcp_port_availability(val))
62 return_when=asyncio.ALL_COMPLETED,
71 return not pending
and all(task.result()
for task
in done)
76 Returns a health checks info that for server.listener, grpc-server.port
77 and server.listener-monitor.
79 @see pytest_userver.plugins.base.service_non_http_health_checks()
81 @ingroup userver_testsuite
85 components = service_config[
'components_manager'][
'components']
86 server = components.get(
'server')
88 for listener_name
in (
'listener-monitor',
'listener'):
89 listener_config = server.get(listener_name, {})
90 port = int(listener_config.get(
'port', 0))
92 host = listener_config.get(
'address',
'localhost')
95 checks.tcp.append(
HostPort(host, port))
97 port = int(components.get(
'grpc-server', {}).get(
'port', 0))
99 checks.tcp.append(
HostPort(
'localhost', port))