95 lines
2.8 KiB
Python
95 lines
2.8 KiB
Python
from typing import Generic, TypeVar
|
|
|
|
from rainbowadn.chain.states import *
|
|
from rainbowadn.collection.pair import *
|
|
from rainbowadn.core import *
|
|
from rainbowadn.flow.core import *
|
|
from rainbowadn.flow.verification.core import *
|
|
from rainbowadn.flow.verification.stateverification import *
|
|
from rainbowadn.nullability import *
|
|
|
|
__all__ = ('StateBridgeM', 'StateBridgeV', 'state_bridge',)
|
|
|
|
HeaderT = TypeVar('HeaderT')
|
|
StateT = TypeVar('StateT')
|
|
|
|
|
|
class StateBridgeM(
|
|
Mapper[
|
|
HashPoint[Pair[HeaderT, StateT]],
|
|
tuple[HashPoint[HeaderT], HashPoint[StateT]],
|
|
],
|
|
Generic[HeaderT, StateT]
|
|
):
|
|
async def map(
|
|
self,
|
|
element: HashPoint[Pair[HeaderT, StateT]]
|
|
) -> tuple[HashPoint[HeaderT], HashPoint[StateT]]:
|
|
assert isinstance(element, HashPoint)
|
|
resolved = await element.resolve()
|
|
header: HashPoint[HeaderT] = resolved.element0
|
|
assert isinstance(header, HashPoint)
|
|
state: HashPoint[StateT] = resolved.element1
|
|
assert isinstance(state, HashPoint)
|
|
return header, state
|
|
|
|
|
|
class StateBridgeV(
|
|
Verification[
|
|
tuple[Nullable[HashPoint[StateT]], HashPoint[HeaderT], HashPoint[StateT]]
|
|
],
|
|
Generic[HeaderT, StateT]
|
|
):
|
|
def __init__(self, protocol: StateProtocol[HeaderT, StateT]):
|
|
assert isinstance(protocol, StateProtocol)
|
|
self.protocol = protocol
|
|
|
|
async def verify(
|
|
self,
|
|
element: tuple[Nullable[HashPoint[StateT]], HashPoint[HeaderT], HashPoint[StateT]]
|
|
) -> bool:
|
|
assert isinstance(element, tuple)
|
|
previous: Nullable[HashPoint[StateT]]
|
|
header: HashPoint[HeaderT]
|
|
state: HashPoint[StateT]
|
|
previous, header, state = element
|
|
assert isinstance(previous, Nullable)
|
|
assert isinstance(header, HashPoint)
|
|
assert isinstance(state, HashPoint)
|
|
assert_true(
|
|
await self.protocol.verify(
|
|
NullableReference(previous, state.factory),
|
|
header,
|
|
state
|
|
)
|
|
)
|
|
return True
|
|
|
|
|
|
def state_bridge(
|
|
protocol: StateProtocol[HeaderT, StateT]
|
|
) -> Verification[
|
|
tuple[
|
|
Nullable[HashPoint[Pair[HeaderT, StateT]]],
|
|
HashPoint[Pair[HeaderT, StateT]]
|
|
]
|
|
]:
|
|
assert isinstance(protocol, StateProtocol)
|
|
m: Mapper[
|
|
HashPoint[Pair[HeaderT, StateT]],
|
|
tuple[HashPoint[HeaderT], HashPoint[StateT]],
|
|
] = StateBridgeM()
|
|
assert isinstance(m, Mapper)
|
|
sv: Verification[
|
|
tuple[Nullable[HashPoint[StateT]], HashPoint[HeaderT], HashPoint[StateT]]
|
|
] = StateBridgeV(protocol)
|
|
assert isinstance(sv, Verification)
|
|
v: Verification[
|
|
tuple[
|
|
Nullable[HashPoint[Pair[HeaderT, StateT]]],
|
|
HashPoint[Pair[HeaderT, StateT]]
|
|
]
|
|
] = StateVerification(m, sv).loose()
|
|
assert isinstance(v, Verification)
|
|
return v
|