userver: /data/code/service_template/third_party/userver/testsuite/pytest_plugins/pytest_userver/plugins/dumps.py Source File
⚠️ This is the documentation for an old userver version. Click here to switch to the latest version.
All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Modules Pages Concepts
dumps.py
1import pathlib
2import shutil
3import typing
4
5import pytest
6
7USERVER_CONFIG_HOOKS = ['_userver_config_dumps']
8
9
10@pytest.fixture(scope='session')
11def userver_dumps_root(tmp_path_factory) -> pathlib.Path:
12 """
13 The directory which the service will use for cache dumps.
14 Dumps of individual components will be stored in
15 `{userver_dumps_root}/{component-name}/{datetime}-v{format-version}`
16
17 @see userver::dump::Dumper
18 @ingroup userver_testsuite_fixtures
19 """
20 path = tmp_path_factory.mktemp('userver-cache-dumps', numbered=True)
21 return path.resolve()
22
23
24@pytest.fixture
25def read_latest_dump(userver_dumps_root):
26 """
27 Read the latest dump produced by a specified dumper.
28
29 @see userver::dump::Dumper
30 @ingroup userver_testsuite_fixtures
31 """
32
33 def read(dumper_name: str) -> typing.Optional[bytes]:
34 specific_dir = userver_dumps_root.joinpath(dumper_name)
35 if not specific_dir.is_dir():
36 return None
37 latest_dump_filename = max(
38 (
39 f
40 for f in specific_dir.iterdir()
41 if specific_dir.joinpath(f).is_file()
42 ),
43 default=None,
44 )
45 if not latest_dump_filename:
46 return None
47 with open(specific_dir.joinpath(latest_dump_filename), 'rb') as dump:
48 return dump.read()
49
50 return read
51
52
53@pytest.fixture
54def cleanup_userver_dumps(userver_dumps_root, request):
55 """
56 To avoid leaking dumps between tests, cache_dump_dir must be cleaned after
57 each test. To observe the dumps, add a final `time.sleep(1000000)` to your
58 test locally. The returned function may also be used to clean dumps
59 manually as appropriate.
60
61 @see userver::dump::Dumper
62 @ingroup userver_testsuite_fixtures
63 """
64
65 def cleanup() -> None:
66 shutil.rmtree(userver_dumps_root, ignore_errors=True)
67 userver_dumps_root.mkdir()
68
69 request.addfinalizer(cleanup)
70 return cleanup
71
72
73@pytest.fixture(scope='session')
74def _userver_config_dumps(pytestconfig, userver_dumps_root):
75 def patch_config(_config, config_vars) -> None:
76 config_vars['userver-dumps-root'] = str(userver_dumps_root)
77 if not pytestconfig.getoption('--service-runner-mode', False):
78 config_vars['userver-dumps-periodic'] = False
79
80 return patch_config