ISSUE #99

* Add `minos.saga.testing` module.
* Implement `LmdbSagaExecutionDatabaseOperationFactory`.

Author: Sergio García Prado

packages/core/minos-microservice-saga/minos/saga/definitions/steps/abc.py

@@ -55,6 +55,8 @@ def from_raw(cls, raw: Union[dict[str, Any], SagaStep], **kwargs) -> SagaStep:
         if isinstance(raw, cls):
             return raw
 
+        raw = raw.copy()
+
         if "cls" in raw:
             # noinspection PyTypeChecker
             step_cls: type = import_module(raw.pop("cls"))

packages/core/minos-microservice-saga/minos/saga/executions/repositories/database/factories.py

@@ -1,11 +1,31 @@
 from abc import (
     ABC,
+    abstractmethod,
+)
+from uuid import (
+    UUID,
 )
 
 from minos.common import (
+    DatabaseOperation,
     DatabaseOperationFactory,
 )
 
 
 class SagaExecutionDatabaseOperationFactory(DatabaseOperationFactory, ABC):
     """TODO"""
+
+    @abstractmethod
+    def build_store(self, uuid: UUID, **kwargs) -> DatabaseOperation:
+        """TODO"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def build_load(self, uuid: UUID) -> DatabaseOperation:
+        """TODO"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def build_delete(self, uuid: UUID) -> DatabaseOperation:
+        """TODO"""
+        raise NotImplementedError

packages/core/minos-microservice-saga/minos/saga/executions/repositories/database/impl.py

@@ -2,22 +2,15 @@
     annotations,
 )
 
-from contextlib import (
-    suppress,
-)
 from typing import (
-    Type,
+    Optional,
 )
 from uuid import (
     UUID,
 )
 
 from minos.common import (
-    Config,
-    MinosConfigException,
-    MinosJsonBinaryProtocol,
-    MinosStorage,
-    MinosStorageLmdb,
+    DatabaseMixin,
 )
 
 from ....exceptions import (
@@ -29,39 +22,31 @@
 from ..abc import (
     SagaExecutionRepository,
 )
+from .factories import (
+    SagaExecutionDatabaseOperationFactory,
+)
 
 
-class DatabaseSagaExecutionRepository(SagaExecutionRepository):
+class DatabaseSagaExecutionRepository(SagaExecutionRepository, DatabaseMixin[SagaExecutionDatabaseOperationFactory]):
     """Saga Execution Storage class."""
 
-    def __init__(
-        self,
-        storage_cls: Type[MinosStorage] = MinosStorageLmdb,
-        protocol=MinosJsonBinaryProtocol,
-        db_name: str = "LocalState",
-        *args,
-        **kwargs,
-    ):
-        super().__init__(*args, **kwargs)
-        self.db_name = db_name
-        self._storage = storage_cls.build(protocol=protocol, **kwargs)
-
-    @classmethod
-    def _from_config(cls, config: Config, **kwargs) -> SagaExecutionRepository:
-        with suppress(MinosConfigException):
-            kwargs |= config.get_database_by_name("saga")
-        return cls(**kwargs)
+    def __init__(self, *args, database_key: Optional[tuple[str]] = None, **kwargs):
+        if database_key is None:
+            database_key = ("saga",)
+        super().__init__(*args, database_key=database_key, **kwargs)
 
     async def _store(self, execution: SagaExecution) -> None:
-        key = str(execution.uuid)
-        value = execution.raw
-        self._storage.update(table=self.db_name, key=key, value=value)
+        operation = self.database_operation_factory.build_store(**execution.raw)
+        await self.execute_on_database(operation)
 
     async def _delete(self, uuid: UUID) -> None:
-        self._storage.delete(table=self.db_name, key=str(uuid))
+        operation = self.database_operation_factory.build_delete(uuid)
+        await self.execute_on_database(operation)
 
     async def _load(self, uuid: UUID) -> SagaExecution:
-        value = self._storage.get(table=self.db_name, key=str(uuid))
+        operation = self.database_operation_factory.build_load(uuid)
+
+        value = await self.execute_on_database_and_fetch_one(operation)
         if value is None:
             raise SagaExecutionNotFoundException(f"The execution identified by {uuid} was not found.")
         execution = SagaExecution.from_raw(value)

packages/core/minos-microservice-saga/minos/saga/executions/saga.py

@@ -93,6 +93,8 @@ def from_raw(cls, raw: Union[dict[str, Any], SagaExecution], **kwargs) -> SagaEx
         if isinstance(raw, cls):
             return raw
 
+        raw = raw.copy()
+
         current = raw | kwargs
         current["definition"] = Saga.from_raw(current["definition"])
         current["status"] = SagaStatus.from_raw(current["status"])

packages/core/minos-microservice-saga/minos/saga/manager.py

@@ -85,7 +85,7 @@ def __init__(
         self.broker_pool = broker_pool
 
     @classmethod
-    def _from_config(cls, *args, config: Config, **kwargs) -> SagaManager:
+    def _from_config(cls, config: Config, **kwargs) -> SagaManager:
         """Build an instance from config.
 
         :param args: Additional positional arguments.

packages/core/minos-microservice-saga/minos/saga/testing.py

@@ -0,0 +1,136 @@
+from abc import (
+    ABC,
+    abstractmethod,
+)
+from uuid import (
+    UUID,
+)
+
+from minos.common import (
+    DatabaseOperation,
+    DeclarativeModel,
+)
+from minos.common.testing import (
+    MinosTestCase,
+    MockedDatabaseClient,
+    MockedDatabaseOperation,
+)
+
+from .context import (
+    SagaContext,
+)
+from .definitions import (
+    Saga,
+)
+from .exceptions import (
+    SagaExecutionNotFoundException,
+)
+from .executions import (
+    SagaExecution,
+    SagaExecutionDatabaseOperationFactory,
+    SagaExecutionRepository,
+)
+
+
+class MockedSagaExecutionDatabaseOperationFactory(SagaExecutionDatabaseOperationFactory):
+    """For testing purposes"""
+
+    def build_store(self, uuid: UUID, **kwargs) -> DatabaseOperation:
+        """For testing purposes"""
+        return MockedDatabaseOperation("create_table")
+
+    def build_load(self, uuid: UUID) -> DatabaseOperation:
+        """For testing purposes"""
+        return MockedDatabaseOperation("create_table")
+
+    def build_delete(self, uuid: UUID) -> DatabaseOperation:
+        """For testing purposes"""
+        return MockedDatabaseOperation("create_table")
+
+
+MockedDatabaseClient.set_factory(
+    SagaExecutionDatabaseOperationFactory,
+    MockedSagaExecutionDatabaseOperationFactory,
+)
+
+
+class Foo(DeclarativeModel):
+    """For testing purposes."""
+
+    foo: str
+
+
+def _fn1(context: SagaContext) -> SagaContext:  # pragma: no cover
+    context["payment"] = "payment"
+    return context
+
+
+def _fn2(context: SagaContext) -> SagaContext:  # pragma: no cover
+    context["payment"] = None
+    return context
+
+
+_SAGA = Saga().local_step(_fn1).on_failure(_fn2).commit()
+
+
+class SagaExecutionRepositoryTestCase(MinosTestCase, ABC):
+    __test__ = False
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.saga_execution_repository = self.build_saga_execution_repository()
+
+    async def asyncSetUp(self) -> None:
+        await super().asyncSetUp()
+
+        await self.saga_execution_repository.setup()
+
+        execution = SagaExecution.from_definition(_SAGA)
+        await execution.execute(autocommit=False)
+
+        self.execution = execution
+
+        self.another = SagaExecution.from_definition(_SAGA)
+
+    async def asyncTearDown(self):
+        await self.saga_execution_repository.destroy()
+        await super().asyncTearDown()
+
+    @abstractmethod
+    def build_saga_execution_repository(self) -> SagaExecutionRepository:
+        """For testing purposes."""
+
+    async def test_store(self):
+        await self.saga_execution_repository.store(self.execution)
+
+        self.assertEqual(self.execution, await self.saga_execution_repository.load(self.execution.uuid))
+
+    async def test_store_overwrite(self):
+        await self.saga_execution_repository.store(self.execution)
+        self.assertEqual(self.execution, await self.saga_execution_repository.load(self.execution.uuid))
+
+        self.another.uuid = self.execution.uuid
+        await self.saga_execution_repository.store(self.another)
+
+        self.assertNotEqual(self.execution, await self.saga_execution_repository.load(self.execution.uuid))
+        self.assertEqual(self.another, await self.saga_execution_repository.load(self.execution.uuid))
+
+    async def test_load_from_str(self):
+        await self.saga_execution_repository.store(self.execution)
+        self.assertEqual(self.execution, await self.saga_execution_repository.load(str(self.execution.uuid)))
+
+    async def test_load_raises(self):
+        with self.assertRaises(SagaExecutionNotFoundException):
+            await self.saga_execution_repository.load(self.execution.uuid)
+
+    async def test_delete(self):
+        await self.saga_execution_repository.store(self.execution)
+        await self.saga_execution_repository.delete(self.execution)
+        with self.assertRaises(SagaExecutionNotFoundException):
+            await self.saga_execution_repository.load(self.execution.uuid)
+
+    async def test_delete_from_str(self):
+        await self.saga_execution_repository.store(self.execution)
+        await self.saga_execution_repository.delete(str(self.execution.uuid))
+        with self.assertRaises(SagaExecutionNotFoundException):
+            await self.saga_execution_repository.load(self.execution.uuid)

packages/core/minos-microservice-saga/tests/config.yml

@@ -1,6 +1,10 @@
-service:
-    name: order
-    aggregate: tests.utils.Order
-saga:
-    storage:
-        path: "./order.lmdb"
+version: 2
+name: order
+
+databases:
+  saga:
+    client: minos.common.testing.MockedDatabaseClient
+    path: "./order.lmdb"
+
+pools:
+  database: minos.common.DatabaseClientPool