userver: /data/code/userver/testsuite/pytest_plugins/pytest_userver/plugins/dumps.py Source File
Loading...
Searching...
No Matches
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