userver: /home/antonyzhilin/arcadia/taxi/uservices/userver/testsuite/pytest_plugins/pytest_userver/plugins/service_runner.py Source File
All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Modules Pages Concepts
service_runner.py
1"""
2Helpers to make the `make start-*` commands work.
3"""
4
5# pylint: disable=no-member,missing-kwoa
6import pathlib
7
8import pytest
9
10
11class ServiceRunnerModule(pytest.Module):
13 def __init__(self, path):
14 self.__file__ = path
15
16 def __init__(self, **kwargs) -> None:
17 super().__init__(**kwargs)
18 self._module = self.FakeModule(path=str(self.path))
19
20 @property
21 def obj(self):
22 return self._module
23
24
26 @pytest.hookimpl(tryfirst=True)
27 def pytest_collection_modifyitems(self, session, config, items):
28 paths = set()
29
30 # Is there servicetest chosen
31 for item in items:
32 skip = False
33 for marker in item.own_markers:
34 if marker.name == 'servicetest':
35 return
36 elif marker.name == 'includetest':
37 skip = True
38
39 if not skip:
40 paths.add(pathlib.Path(item.module.__file__).parent)
41
42 if not paths:
43 return
44
45 tests_root = min(paths, key=lambda p: len(p.parts))
46
47 module = ServiceRunnerModule.from_parent(
48 parent=session,
49 path=pathlib.Path(tests_root).resolve(),
50 )
51 function = pytest.Function.from_parent(
52 parent=module,
53 name=test_service_default.__name__,
54 callobj=test_service_default,
55 )
56
57 items.append(function)
58
59
60@pytest.mark.servicetest
62 service_client,
63 service_baseurl,
64 monitor_baseurl,
65 request,
66) -> None:
67 """
68 This is default service runner testcase. Feel free to override it
69 in your own tests, e.g.:
70
71 @code
72 @pytest.mark.servicetest
73 def test_service(service_client):
74 ...
75 @endcode
76
77 @ingroup userver_testsuite
78 """
79 # TODO: use service_client.base_url() and monitor_client.base_url()
80 delimiter = '=' * 100
81 message = f'\n{delimiter}\nStarted service at {service_baseurl}'
82 if monitor_baseurl:
83 message += f'\nMonitor URL is {monitor_baseurl}'
84 if 'grpc_service_endpoint' in request.fixturenames:
85 grpc_endpoint = request.getfixturevalue('grpc_service_endpoint')
86 message += f'\ngRPC endpoint is {grpc_endpoint}'
87 message += f'\n{delimiter}\n'
88 print(message)
89
90
91def pytest_configure(config):
92 if config.option.service_runner_mode:
93 runner = UserviceRunner()
94 config.pluginmanager.register(runner, 'uservice_runner')