67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
from typing import Any, Generic, TypeVar
|
|
|
|
from rainbowadn.core import *
|
|
from rainbowadn.flow.core import *
|
|
from rainbowadn.flow.verification.core import *
|
|
|
|
__all__ = ('FlowTree',)
|
|
|
|
Key = TypeVar('Key')
|
|
Tree = TypeVar('Tree')
|
|
|
|
|
|
class FlowTree(Generic[Key, Tree]):
|
|
async def contains(self, key: Key, *, exact: bool) -> bool:
|
|
raise NotImplementedError
|
|
|
|
async def verify_contains_all(self, keys: Reducer[Key, bool]) -> bool:
|
|
assert isinstance(keys, Reducer)
|
|
key_verification: Verification[Key] = ContainsVerification(self)
|
|
assert isinstance(key_verification, Verification)
|
|
assert_true(await ReduceVerification(key_verification).loose().verify(keys))
|
|
return True
|
|
|
|
async def verify_does_not_intersect(self, keys: Reducer[Key, bool]) -> bool:
|
|
assert isinstance(keys, Reducer)
|
|
key_verification: Verification[Key] = DoesNotContainVerification(self)
|
|
assert isinstance(key_verification, Verification)
|
|
assert_true(await ReduceVerification(key_verification).loose().verify(keys))
|
|
return True
|
|
|
|
async def verify_subset(self: Tree, trees: Reducer['FlowTree[Key, Tree]', Any]) -> bool:
|
|
raise NotImplementedError
|
|
|
|
async def reducer(self) -> Reducer[Key, Any]:
|
|
raise NotImplementedError
|
|
|
|
async def verify(self, verification: Verification[Key]) -> bool:
|
|
assert isinstance(verification, Verification)
|
|
assert_true(
|
|
await ReduceVerification(
|
|
verification
|
|
).loose().verify(
|
|
await self.reducer()
|
|
)
|
|
)
|
|
return True
|
|
|
|
|
|
class ContainsVerification(Verification[Key], Generic[Key, Tree]):
|
|
def __init__(self, tree: FlowTree[Key, Tree]):
|
|
assert isinstance(tree, FlowTree)
|
|
self.tree = tree
|
|
|
|
async def verify(self, element: Key) -> bool:
|
|
assert_true(await self.tree.contains(element, exact=True))
|
|
return True
|
|
|
|
|
|
class DoesNotContainVerification(Verification[Key], Generic[Key, Tree]):
|
|
def __init__(self, tree: FlowTree[Key, Tree]):
|
|
assert isinstance(tree, FlowTree)
|
|
self.tree = tree
|
|
|
|
async def verify(self, element: Key) -> bool:
|
|
assert_false(await self.tree.contains(element, exact=False))
|
|
return True
|