OpenSlides/tests/unit/utils/cache_provider.py

83 lines
2.3 KiB
Python
Raw Normal View History

2018-08-22 22:00:08 +02:00
import asyncio
from typing import Any, Callable, Dict, List
from openslides.utils.cache_providers import Cachable, MemmoryCacheProvider
def restrict_elements(elements: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Adds the prefix 'restricted_' to all values except id.
"""
out = []
for element in elements:
restricted_element = {}
for key, value in element.items():
2019-01-06 16:22:33 +01:00
if key == "id":
restricted_element[key] = value
else:
2019-01-12 23:01:42 +01:00
restricted_element[key] = f"restricted_{value}"
out.append(restricted_element)
return out
class Collection1:
def get_collection_string(self) -> str:
2019-01-06 16:22:33 +01:00
return "app/collection1"
def get_elements(self) -> List[Dict[str, Any]]:
2019-01-06 16:22:33 +01:00
return [{"id": 1, "value": "value1"}, {"id": 2, "value": "value2"}]
2019-01-06 16:22:33 +01:00
async def restrict_elements(
self, user_id: int, elements: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
return restrict_elements(elements)
class Collection2:
def get_collection_string(self) -> str:
2019-01-06 16:22:33 +01:00
return "app/collection2"
def get_elements(self) -> List[Dict[str, Any]]:
2019-01-06 16:22:33 +01:00
return [{"id": 1, "key": "value1"}, {"id": 2, "key": "value2"}]
2019-01-06 16:22:33 +01:00
async def restrict_elements(
self, user_id: int, elements: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
return restrict_elements(elements)
2019-01-06 16:22:33 +01:00
def get_cachable_provider(
cachables: List[Cachable] = [Collection1(), Collection2()]
) -> Callable[[], List[Cachable]]:
"""
Returns a cachable_provider.
"""
return lambda: cachables
def example_data():
return {
2019-01-06 16:22:33 +01:00
"app/collection1": [{"id": 1, "value": "value1"}, {"id": 2, "value": "value2"}],
"app/collection2": [{"id": 1, "key": "value1"}, {"id": 2, "key": "value2"}],
}
class TTestCacheProvider(MemmoryCacheProvider):
"""
CacheProvider simular to the MemmoryCacheProvider with special methods for
testing.
"""
2019-01-06 16:22:33 +01:00
async def del_lock_after_wait(
self, lock_name: str, future: asyncio.Future = None
) -> None:
if future is None:
2018-09-01 08:00:00 +02:00
asyncio.ensure_future(self.del_lock(lock_name))
else:
2019-01-06 16:22:33 +01:00
async def set_future() -> None:
2018-09-01 08:00:00 +02:00
await self.del_lock(lock_name)
future.set_result(1) # type: ignore
2019-01-06 16:22:33 +01:00
asyncio.ensure_future(set_future())