userver: /data/code/service_template/third_party/userver/testsuite/pytest_plugins/pytest_userver/plugins/caches.py Source File
Loading...
Searching...
No Matches
caches.py
1"""
2Fixtures for controlling userver caches.
3"""
4import copy
5import typing
6
7import pytest
8
9
11 def __init__(self):
12 self.incremental_caches = set()
13
14 def register_incremental_cache(self, *names):
15 self.incremental_caches.update(names)
16
17
19 def pytest_userver_caches_setup(
20 self, userver_cache_config: CacheConfig,
21 ) -> None:
22 pass
23
24
26 def __init__(self, cache_config: CacheConfig):
27 # None means that we should update all caches.
28 # We invalidate all caches at the start of each test.
29 self._invalidated_caches: typing.Optional[typing.Set[str]] = None
30 self._cache_config = cache_config
31
32 def invalidate_all(self) -> None:
33 self._invalidated_caches = None
34
35 def invalidate(self, caches: typing.Iterable[str]) -> None:
36 if self._invalidated_caches is not None:
37 self._invalidated_caches.update(caches)
38
39 @property
40 def should_update_all_caches(self) -> bool:
41 return self._invalidated_caches is None
42
43 @property
44 def caches_to_update(self) -> typing.FrozenSet[str]:
45 assert self._invalidated_caches is not None
46 return frozenset(self._invalidated_caches)
47
48 @property
49 def has_caches_to_update(self) -> bool:
50 caches = self._invalidated_caches
51 return caches is None or bool(caches)
52
53 @property
54 def incremental_caches(self):
55 return tuple(self._cache_config.incremental_caches)
56
57 def on_caches_updated(self, caches: typing.Iterable[str]) -> None:
58 if self._invalidated_caches is not None:
59 self._invalidated_caches.difference_update(caches)
60
61 def on_all_caches_updated(self) -> None:
62 self._invalidated_caches = set()
63
64 def assign_copy(self, other: 'InvalidationState') -> None:
65 # pylint: disable=protected-access
66 self._invalidated_caches = copy.deepcopy(other._invalidated_caches)
67
68
69@pytest.fixture
71 userver_cache_config: CacheConfig,
72) -> InvalidationState:
73 """
74 A fixture for notifying the service of changes in cache data sources.
75
76 Intended to be used by other fixtures that represent those data sources,
77 not by tests directly.
78
79 @ingroup userver_testsuite_fixtures
80 """
81 return InvalidationState(userver_cache_config)
82
83
84@pytest.fixture(name='userver_cache_config', scope='session')
85def _userver_cache_config(pytestconfig) -> CacheConfig:
86 cache_config = CacheConfig()
87 pytestconfig.pluginmanager.hook.pytest_userver_caches_setup(
88 userver_cache_config=cache_config,
89 )
90 return cache_config
91
92
93def pytest_addhooks(pluginmanager):
94 pluginmanager.add_hookspecs(UserverCacheHookspec)