19 def userver_cache_control_hooks(self) -> typing.Dict[str, str]:
22 def pytest_plugin_registered(self, plugin, manager):
23 if not isinstance(plugin, types.ModuleType):
25 uhooks = getattr(plugin,
'USERVER_CACHE_CONTROL_HOOKS',
None)
28 if not isinstance(uhooks, dict):
30 f
'USERVER_CACHE_CONTROL_HOOKS must be dictionary: '
31 f
'{{cache_name: fixture_name}}, got {uhooks} instead',
33 for cache_name, fixture_name
in uhooks.items():
34 if cache_name
in self.
_hooks:
36 f
'USERVER_CACHE_CONTROL_HOOKS: hook already registered '
37 f
'for cache {cache_name}',
39 self.
_hooks[cache_name] = fixture_name
48 def invalidate_all(self) -> None:
51 def invalidate(self, caches: typing.Iterable[str]) ->
None:
56 def should_update_all_caches(self) -> bool:
60 def caches_to_update(self) -> typing.FrozenSet[str]:
65 def has_caches_to_update(self) -> bool:
67 return caches
is None or bool(caches)
69 def on_caches_updated(self, caches: typing.Iterable[str]) ->
None:
73 def on_all_caches_updated(self) -> None:
76 def assign_copy(self, other:
'InvalidationState') ->
None:
185 _userver_cache_control_context, _userver_cache_fixtures, request,
186) -> typing.Callable[[DaemonInstance], CacheControl]:
187 """Userver cache control handler.
189 To install per cache handler use USERVER_CACHE_CONTROL_HOOKS variable
190 in your pytest plugin:
193 USERVER_CACHE_CONTROL_HOOKS = {
194 'my-cache-name': 'my_cache_cc',
198 def my_cache_cc(my_cache_context):
199 def cache_control(request, state):
200 new_state = my_cache_context.get_state()
201 if state == new_state:
202 # Cache is already up to date, no need to update
205 # Request incremental update, if you cache supports it
206 request.incremental()
211 @ingroup userver_testsuite_fixtures
214 caches_disabled = set()
216 def userver_cache_control_disabled(
217 caches: typing.Sequence[str] =
None, *, reason: str,
219 if caches
is not None:
220 caches_disabled.update(caches)
224 for mark
in request.node.iter_markers(
'userver_cache_control_disabled'):
225 enabled = userver_cache_control_disabled(*mark.args, **mark.kwargs)
227 def get_cache_control(daemon: DaemonInstance):
228 context = _userver_cache_control_context.setdefault(daemon.id, {})
231 fixtures=_userver_cache_fixtures,
233 caches_disabled=caches_disabled,
236 return get_cache_control