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