Merge pull request #17 from RnDAO/dev_amin_lib_update

Updated DB collection!

setup.py

@@ -6,7 +6,7 @@
 
 setup(
     name="tc-messageBroker",
-    version="1.1.0",
+    version="1.1.1",
     author="Mohammad Amin Dadgar, RnDAO",
     maintainer="Mohammad Amin Dadgar",
     maintainer_email="[email protected]",

tc_messageBroker/rabbit_mq/db_operations/mongodb.py

@@ -10,9 +10,7 @@
 
 
 class MongoDB:
-    def __init__(
-        self, connection_str: str, db_name: str = "RnDAO", collection_name: str = "Saga"
-    ) -> None:
+    def __init__(self, connection_str: str, db_name: str, collection_name: str) -> None:
         """
         initialize the mongodb class
 

tc_messageBroker/rabbit_mq/event/events_microservice.py

@@ -5,6 +5,7 @@ class BotBaseEvent:
 class AnalyzerBaseEvent:
     RUN = "RUN"
     SAVE = "SAVE"
+    RUN_ONCE = "RUN_ONCE"
 
 
 class ServerEvent:
@@ -17,5 +18,6 @@ class DiscordBotEvent:
 
 
 class DiscordAnalyzerEvent:
-    RUN = AnalyzerBaseEvent.RUN
+    RUN = AnalyzerBaseEvent.RUN  ## RECOMPUTE
+    RUN_ONCE = AnalyzerBaseEvent.RUN_ONCE
     # SAVE = AnalyzerBaseEvent.SAVE

tc_messageBroker/rabbit_mq/saga/choreography.py

@@ -1,12 +1,21 @@
 from .choreography_base import IChoreography
-from .transactions import DISCORD_UPDATE_CHANNELS_TRANSACTIONS
+from .transactions import (
+    DISCORD_UPDATE_CHANNELS_TRANSACTIONS,
+    DISCORD_SCHEDULED_JOB_TRANSACTIONS,
+)
 
 
 DISCORD_UPDATE_CHANNELS = IChoreography(
     name="DISCORD_UPDATE_CHANNELS",
     transactions=DISCORD_UPDATE_CHANNELS_TRANSACTIONS,
 )
 
+DISCORD_SCHEDULED_JOB = IChoreography(
+    name="DISCORD_UPDATE_CHANNELS",
+    transactions=DISCORD_SCHEDULED_JOB_TRANSACTIONS,
+)
+
 
 class ChoreographyDict:
     DISCORD_UPDATE_CHANNELS = DISCORD_UPDATE_CHANNELS
+    DISCORD_SCHEDULED_JOB = DISCORD_SCHEDULED_JOB

tc_messageBroker/rabbit_mq/saga/saga.py

@@ -11,5 +11,14 @@ def __init__(self, data: any) -> None:
             Status.NOT_STARTED,
             data=data,
             created_at=datetime.now(),
-            next=None,
+        )
+
+
+class DiscordScheculedJob(Saga):
+    def __init__(self, data: any) -> None:
+        super().__init__(
+            ChoreographyDict.DISCORD_SCHEDULED_JOB,
+            Status.NOT_STARTED,
+            data=data,
+            created_at=datetime.now(),
         )

tc_messageBroker/rabbit_mq/saga/saga_base.py

@@ -22,6 +22,31 @@ def __init__(
         self.data = data
         self.created_at = created_at
 
+    def start(
+        self,
+        publish_method: callable,
+        mongo_connection: str,
+        test_mode=False,
+    ):
+        """
+        just publish the first transaction
+        """
+        tx_sorted, _ = self._sort_transactions(self.choreography.transactions)
+        current_tx = tx_sorted[0]
+        self.status = Status.IN_PROGRESS
+
+        self._update_save(
+            transactions=tx_sorted,
+            mongo_connection=mongo_connection,
+            test=test_mode,
+        )
+
+        publish_method(
+            queue_name=current_tx.queue,
+            event=current_tx.event,
+            content={"uuid": self.uuid},
+        )
+
     def next(
         self,
         publish_method: callable,
@@ -206,7 +231,7 @@ def _create_data(self) -> dict[str, any]:
         return data
 
 
-def get_saga(guildId, connection_url):
+def get_saga(guildId: str, connection_url: str, db_name: str, collection: str):
     """
     get saga object for a special guild
 
@@ -216,13 +241,19 @@ def get_saga(guildId, connection_url):
         the guildId which the saga belongs to
     connection_url : str
         the connection to db which the saga architecture is saved
+    db_name : str
+        the database name to use
+    collection : str
+        the collection which the saga is saved
 
     Returns:
     ----------
     saga_obj : Saga
         the saga object to use
     """
-    mongodb = MongoDB(connection_str=connection_url)
+    mongodb = MongoDB(
+        connection_str=connection_url, db_name=db_name, collection_name=collection
+    )
     mongodb.connect()
 
     data = mongodb.read(query={"data.guildId": guildId}, count=1)

tc_messageBroker/rabbit_mq/saga/transactions.py

@@ -23,10 +23,25 @@
     #     order=3,
     #     status=Status.NOT_STARTED,
     # ),
+    # ITransaction(
+    #     Queue.SERVER_API,
+    #     Event.SERVER_API.UPDATE_GUILD,
+    #     order=3,
+    #     status=Status.NOT_STARTED,
+    # ),
+]
+
+DISCORD_SCHEDULED_JOB_TRANSACTIONS = [
+    # ITransaction(
+    #     Queue.DISCORD_BOT,
+    #     Event.DISCORD_BOT.FETCH,
+    #     order=1,
+    #     status=Status.NOT_STARTED,
+    # ),
     ITransaction(
-        Queue.SERVER_API,
-        Event.SERVER_API.UPDATE_GUILD,
-        order=3,
+        Queue.DISCORD_ANALYZER,
+        Event.DISCORD_ANALYZER.RUN_ONCE,
+        order=2,
         status=Status.NOT_STARTED,
     ),
 ]

tests/integration/test_saga.py

@@ -20,7 +20,12 @@ def test_inputs():
     connection_url = "mongodb://127.0.0.1:27017/"
 
     ## we should have this data before running this test in db
-    saga = get_saga(guildId="993163081939165234", connection_url=connection_url)
+    saga = get_saga(
+        guildId="993163081939165234",
+        connection_url=connection_url,
+        db_name="Saga",
+        collection="saga",
+    )
 
     assert saga.choreography is not None
     assert saga.status in [

tests/unit/test_choreagraphy_base.py

@@ -1,13 +1,39 @@
 from tc_messageBroker.rabbit_mq.saga.choreography_base import IChoreography
 from tc_messageBroker.rabbit_mq.saga.transactions import (
     DISCORD_UPDATE_CHANNELS_TRANSACTIONS,
+    DISCORD_SCHEDULED_JOB_TRANSACTIONS,
 )
 
 
-def test_choreography():
+def test_choreography_discord_update_channels_no_inputs():
+    choreography = IChoreography(name=None, transactions=None)
+
+    assert choreography.name is None
+    assert choreography.transactions is None
+
+
+def test_choreography_discord_update_channels():
     choreography = IChoreography(
         name="sample", transactions=DISCORD_UPDATE_CHANNELS_TRANSACTIONS
     )
 
     assert choreography.name == "sample"
     assert choreography.transactions == DISCORD_UPDATE_CHANNELS_TRANSACTIONS
+
+
+def test_choreography_discord_update_channels_wrong_input():
+    choreography = IChoreography(
+        name="sample", transactions=DISCORD_UPDATE_CHANNELS_TRANSACTIONS
+    )
+
+    assert choreography.name != "some_choreography"
+    assert choreography.transactions != DISCORD_SCHEDULED_JOB_TRANSACTIONS
+
+
+def test_choreography_discord_job_transactions():
+    choreography = IChoreography(
+        name="sample", transactions=DISCORD_SCHEDULED_JOB_TRANSACTIONS
+    )
+
+    assert choreography.name == "sample"
+    assert choreography.transactions == DISCORD_SCHEDULED_JOB_TRANSACTIONS

tests/unit/test_event.py

@@ -7,4 +7,5 @@ def test_enum_event_default():
     assert Event.DISCORD_BOT.FETCH == "FETCH"
     assert Event.DISCORD_BOT.SEND_MESSAGE == "SEND_MESSAGE"
     assert Event.DISCORD_ANALYZER.RUN == "RUN"
+    assert Event.DISCORD_ANALYZER.RUN_ONCE == "RUN_ONCE"
     # assert Event.DISCORD_ANALYZER.SAVE == "SAVE"

tests/unit/test_predefined_transactions.py

@@ -0,0 +1,37 @@
+from tc_messageBroker.rabbit_mq.saga.transactions import (
+    DISCORD_SCHEDULED_JOB_TRANSACTIONS,
+    DISCORD_UPDATE_CHANNELS_TRANSACTIONS,
+)
+from tc_messageBroker.rabbit_mq.queue import Queue
+from tc_messageBroker.rabbit_mq.event import Event
+from tc_messageBroker.rabbit_mq.status import Status
+
+
+def test_discord_update_channels_tx():
+    tx = DISCORD_UPDATE_CHANNELS_TRANSACTIONS
+
+    assert len(tx) == 2
+    assert tx[0].order == 1
+    assert tx[0].queue == Queue.DISCORD_BOT
+    assert tx[0].event == Event.DISCORD_BOT.FETCH
+    assert tx[0].status == Status.NOT_STARTED
+
+    assert tx[1].order == 2
+    assert tx[1].queue == Queue.DISCORD_ANALYZER
+    assert tx[1].event == Event.DISCORD_ANALYZER.RUN
+    assert tx[1].status == Status.NOT_STARTED
+
+
+def test_discord_scheduled_job_tx():
+    tx = DISCORD_SCHEDULED_JOB_TRANSACTIONS
+
+    assert len(tx) == 1
+    # assert tx[0].order == 1
+    # assert tx[0].queue == Queue.DISCORD_BOT
+    # assert tx[0].event == Event.DISCORD_BOT.FETCH
+    # assert tx[0].status == Status.NOT_STARTED
+
+    assert tx[0].order == 2
+    assert tx[0].queue == Queue.DISCORD_ANALYZER
+    assert tx[0].event == Event.DISCORD_ANALYZER.RUN_ONCE
+    assert tx[0].status == Status.NOT_STARTED

tests/unit/test_saga_base.py

@@ -22,7 +22,7 @@ def test_choreagraphy_sorting_orders_predefined_choreogprahy():
         saga.choreography.transactions
     )
 
-    assert not_started_count == 3
+    assert not_started_count == 2
     order_val = 0
     for tx in tx_sorted:
         if tx.status == Status.NOT_STARTED:
@@ -45,7 +45,7 @@ def test_assert_sorting_status_predefined_choreagprahy():
         saga.choreography.transactions
     )
 
-    assert not_started_count == 3
+    assert not_started_count == 2
 
     condition = False
     for tx in tx_sorted:

tests/unit/test_saga_start.py

@@ -0,0 +1,38 @@
+from tc_messageBroker.rabbit_mq.saga.saga_base import Saga
+from tc_messageBroker.rabbit_mq.saga.transaction_base import ITransaction
+from tc_messageBroker.rabbit_mq.saga.choreography_base import IChoreography
+from datetime import datetime
+from tc_messageBroker.rabbit_mq.status import Status
+
+
+def test_start_tx():
+    choreography = IChoreography(
+        name="choreography_with_random_tx",
+        transactions=[
+            ITransaction("queue", "event", 5, Status.NOT_STARTED),
+            ITransaction("queue1", "event1", 2, Status.NOT_STARTED),
+            ITransaction("queue", "event", 3, Status.SUCCESS),
+            ITransaction("queue", "event", 1, Status.FAILED),
+            ITransaction("queue", "event", 0, Status.CANCELLED),
+        ],
+    )
+
+    saga = Saga(choreography, Status.NOT_STARTED, data=None, created_at=datetime.now())
+
+    def sample_publish(**kwargs):
+        assert kwargs["queue_name"] == choreography.transactions[1].queue
+        assert kwargs["event"] == choreography.transactions[1].event
+        assert kwargs["content"]["uuid"] == saga.uuid
+
+    saga.start(
+        publish_method=sample_publish,
+        mongo_connection="",
+        test_mode=True,
+    )
+
+    assert saga.status == Status.IN_PROGRESS
+    assert saga.choreography.transactions[0].status == Status.NOT_STARTED
+    assert saga.choreography.transactions[1].status == Status.NOT_STARTED
+    assert saga.choreography.transactions[2].status == Status.SUCCESS
+    assert saga.choreography.transactions[3].status == Status.FAILED
+    assert saga.choreography.transactions[4].status == Status.CANCELLED