test(TransactionHorizon): cover basic cases

Author: emptybutton

pyproject.toml

@@ -62,7 +62,7 @@ select = [
     "Q", "RET", "SLF", "SLOT", "SIM", "TID252", "TCH", "ARG", "PTH", "ERA", "TRY",
     "PERF", "INP", "I", "S", "FAST", "TID", "TCH", "INT"
 ]
-ignore = ["N818", "RUF009", "UP018", "PLR6301", "PLR0913", "PLW0108", "TC006"]
+ignore = ["N818", "RUF009", "UP018", "PLR6301", "PLR0913", "PLW0108", "TC006", "S101"]
 
 [tool.ruff.lint.isort]
 lines-after-imports = 2
@@ -71,9 +71,8 @@ lines-after-imports = 2
 "src/tgdb/entities/*" = ["PLR2004"]
 "src/tgdb/application/*" = ["PLR0917"]
 "src/tgdb/infrastructure/adapters/*" = ["RUF029"]
-"src/tgdb/infrastructure/*" = ["S101"]
 "src/tgdb/entities/transaction.py" = ["SLF001"]
-"tests/*" = ["S101", "PLR0124", "PLR0917", "S106", "C901"]
+"tests/*" = ["PLR0124", "PLR0917", "S106", "C901", "PLR2004"]
 "__init__.py" = ["PLC0414"]
 
 [tool.pytest.ini_options]

src/tgdb/application/input_operator.py

@@ -0,0 +1,37 @@
+from dataclasses import dataclass
+
+from tgdb.application.ports.async_queque import AsyncQueque
+from tgdb.application.ports.logic_clock import LogicClock
+from tgdb.application.ports.operator_serialization import OperatorSerialization
+from tgdb.entities.operator import AppliedOperator, applied_operator
+
+
+class InputOperatorsError(Exception): ...
+
+
+@dataclass(frozen=True)
+class InputOperators[SerializedOperatorsT]:
+    clock: LogicClock
+    input_operators: AsyncQueque[AppliedOperator]
+    operator_serialization: OperatorSerialization[SerializedOperatorsT]
+
+    async def __call__(
+        self, serialized_operators: SerializedOperatorsT
+    ) -> None:
+        """
+        :raises tgdb.application.input_operator.InputOperatorsError:
+        """
+
+        input_operators = await self.operator_serialization.deserialized(
+            serialized_operators
+        )
+
+        if input_operators is None:
+            raise InputOperatorsError
+
+        times = await self.clock.times(len(input_operators))
+
+        await self.input_operators.push(*(
+            applied_operator(input_operator, time)
+            for time, input_operator in zip(times, input_operators, strict=True)
+        ))

src/tgdb/application/input_operators.py

@@ -4,7 +4,7 @@
 
 class AsyncQueque[ValueT](ABC):
     @abstractmethod
-    async def push(self, value: ValueT) -> None: ...
+    async def push(self, *values: ValueT) -> None: ...
 
     @abstractmethod
     async def iter(self) -> AsyncIterator[ValueT]: ...

src/tgdb/application/ports/async_queque.py

@@ -9,7 +9,7 @@
 
 class Log(ABC):
     @abstractmethod
-    async def push(self, operator: AppliedOperator, /) -> None: ...
+    async def push(self, *operators: AppliedOperator) -> None: ...
 
     @abstractmethod
     async def truncate(self, offset: LogOffset, /) -> None: ...

src/tgdb/application/ports/log.py

@@ -1,7 +1,12 @@
-from abc import ABC
-from collections.abc import Awaitable
+from abc import ABC, abstractmethod
+from collections.abc import Sequence
 
 from tgdb.entities.logic_time import LogicTime
 
 
-class LogicClock(ABC, Awaitable[LogicTime]): ...
+class LogicClock(ABC):
+    @abstractmethod
+    async def times(self, count: int) -> Sequence[LogicTime]: ...
+
+    @abstractmethod
+    async def time(self) -> LogicTime: ...

src/tgdb/application/ports/logic_clock.py

@@ -1,10 +1,11 @@
 from abc import ABC, abstractmethod
+from collections.abc import Sequence
 
 from tgdb.entities.operator import Operator
 
 
-class OperatorSerialization[SerializedOperatorT](ABC):
+class OperatorSerialization[SerializedOperatorsT](ABC):
     @abstractmethod
     async def deserialized(
-        self, operator: SerializedOperatorT, /
-    ) -> Operator | None: ...
+        self, operators: SerializedOperatorsT, /
+    ) -> Sequence[Operator] | None: ...

src/tgdb/application/ports/operator_serialization.py

@@ -4,7 +4,7 @@
 
 class SyncQueque[ValueT](ABC):
     @abstractmethod
-    async def push(self, value: ValueT) -> None: ...
+    async def push(self, *values: ValueT) -> None: ...
 
     @abstractmethod
     async def sync(self) -> None: ...

src/tgdb/application/ports/sync_queque.py

@@ -11,53 +11,9 @@
 type RowAttribute = bool | int | str | datetime | UUID | StrEnum
 
 
-@dataclass(frozen=True)
-class RowSchema(Sequence[type[RowAttribute]]):
-    name: str
-    id_type: type[RowAttribute]
-    body_types: tuple[type[RowAttribute], ...] = tuple()
-
-    def __iter__(self) -> Iterator[type[RowAttribute]]:
-        yield self.id_type
-        yield from self.body_types
-
-    @overload
-    def __getitem__(self, index: int, /) -> type[RowAttribute]: ...
-
-    @overload
-    def __getitem__(
-        self, slice_: "slice[Any, Any, Any]", /
-    ) -> Sequence[type[RowAttribute]]: ...
-
-    def __getitem__(
-        self, value: "int | slice[Any, Any, Any]", /
-    ) -> Sequence[type[RowAttribute]] | type[RowAttribute]:
-        return tuple(self)[value]
-
-    def __len__(self) -> int:
-        return len(self.body_types) + 1
-
-
-class RowSchemaError(Exception):
-    def __init__(self, schema: RowSchema) -> None:
-        self.schema = schema
-        super().__init__()
-
-
 @dataclass(frozen=True)
 class Row(IdentifiedValue[RowAttribute], Sequence[RowAttribute]):
     body: tuple[RowAttribute, ...]
-    schema: RowSchema
-
-    def __post_init__(self) -> None:
-        for attribute_and_type in zip(self, self.schema, strict=False):
-            if len(attribute_and_type) != 2:
-                raise RowSchemaError(self.schema)
-
-            attribute, type = attribute_and_type
-
-            if not isinstance(attribute, type):
-                raise RowSchemaError(self.schema)
 
     def __iter__(self) -> Iterator[RowAttribute]:
         yield self.id
@@ -80,4 +36,8 @@ def __len__(self) -> int:
         return len(self.body) + 1
 
 
+def row(*attrs: RowAttribute) -> Row:
+    return Row(attrs[0], attrs[1:])
+
+
 type RowEffect = New[Row] | Mutated[Row] | Dead[Row]

src/tgdb/entities/row.py

@@ -5,6 +5,7 @@
 from uuid import UUID
 
 from effect import Effect, IdentifiedValueSet
+from effect.state_transition import InvalidStateTransitionError
 
 from tgdb.entities.logic_time import LogicTime
 from tgdb.entities.row import Row, RowEffect
@@ -32,7 +33,7 @@ class TransactionOkCommit:
 @dataclass(frozen=True)
 class TransactionFailedCommit:
     transaction_id: UUID
-    conflict: TransactionConflict
+    conflict: TransactionConflict | None
 
 
 type TransactionCommit = TransactionOkCommit | TransactionFailedCommit
@@ -55,6 +56,10 @@ def is_readonly(self) -> bool:
         return not any(self._effect) and not self._uniqueness_marks
 
     def consider(self, effect: RowEffect) -> None:
+        """
+        :raises effect.state_transition.InvalidStateTransitionError:
+        """
+
         self._effect &= effect
 
     def add_uniqueness_mark(
@@ -68,9 +73,6 @@ def add_viewed_row_mark(
         self._viewed_row_marks.add(viewed_row_mark)
 
     def beginning(self) -> LogicTime | None:
-        if self._beginning is None:
-            return None
-
         return self._beginning
 
     def rollback(self) -> None: