Merge pull request #3 from Aignosi/SIENTIAPDE-1107-testar-o-uso-do-mongo-db-no-lugar-do-couchbase

Sientiapde 1107 testar o uso do mongo db no lugar do couchbase

docker-compose.yml

@@ -117,6 +117,19 @@ services:
     networks:
       - sientia-network
 
+  mongodb:
+    image: mongo:7.0
+    container_name: mongodb
+    ports:
+      - "27017:27017"
+    environment:
+      MONGO_INITDB_ROOT_USERNAME: sientia
+      MONGO_INITDB_ROOT_PASSWORD: sientia
+    volumes:
+      - mongodb_data:/data/db
+    networks:
+      - sientia-network
+
 
 networks:
   sientia-network:
@@ -130,4 +143,6 @@ volumes:
   redis_data:
     driver: local
   kafka_data:
+    driver: local
+  mongodb_data:
     driver: local

input_sample.json

@@ -1,5 +1,37 @@
+[
 {
     "schedule_name": "orchestrator-test",
     "pipelines_query": "SELECT pipelines.*, models FROM `pipelines` JOIN `models` ON KEYS pipelines.model_id;",
     "opc_servers_query": "select META().id, opc_servers.* from `opc_servers`;"
-}
+},
+{
+    "schedule_name": "orchestrator-test",
+    "pipelines_query": {
+        "collection": "pipelines",
+        "aggregation": [
+            {
+                "$lookup": {
+                    "from": "models",
+                    "localField": "model_id",
+                    "foreignField": "id",
+                    "as": "model_docs"
+                }
+            },
+            {
+                "$addFields": {
+                    "models": { "$arrayElemAt": ["$model_docs", 0] }
+                }
+            },
+            {
+                "$project": {
+                "model_docs": 0
+                }  
+            }
+        ]
+    },
+    "opc_servers_query": {
+        "collection": "opc-servers",
+        "filters": {}
+    }
+}
+]

orchestrator/activities/activities.py

@@ -1,6 +1,8 @@
 from temporalio import activity, workflow
 from temporalio.client import Client
 
+from orchestrator.activities.mongo_db import MongoDB
+
 with workflow.unsafe.imports_passed_through():
     from orchestrator.activities.couchbase import Couchbase
     from orchestrator.activities.temporal_manager import TemporalManager
@@ -11,21 +13,23 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.notifications.handlers import NotificationHandler
 
 
-class Activities(Couchbase, TemporalManager, SlotManager, Formatters):
+class Activities(  # Couchbase,
+        TemporalManager, SlotManager, Formatters, MongoDB):
 
     def __init__(self,
                  temporal_client: Client,
-                 couchbase_config: dict[str, Any],
+                 # couchbase_config: dict[str, Any],
                  redis_config: dict[str, Any],
+                 mongodb_config: dict[str, Any],
                  logger: Logger,
                  notification_handler: NotificationHandler):
 
         # Initialize parent classes
-        Couchbase.__init__(self, connection_string=couchbase_config['connection_string'],
+        # Couchbase.__init__(self, connection_string=couchbase_config['connection_string'],
-                           username=couchbase_config['username'],
+        #                    username=couchbase_config['username'],
-                           password=couchbase_config['password'],
+        #                    password=couchbase_config['password'],
-                           logger=logger,
+        #                    logger=logger,
-                           notification_handler=notification_handler)
+        #                    notification_handler=notification_handler)
 
         TemporalManager.__init__(self,
                                  temporal_client=temporal_client,
@@ -44,9 +48,15 @@ class Activities(Couchbase, TemporalManager, SlotManager, Formatters):
                             logger=logger,
                             notification_handler=notification_handler)
 
+        MongoDB.__init__(self,
+                         connection_string=mongodb_config['connection_string'],
+                         database_name=mongodb_config['database_name'],
+                         logger=logger,
+                         notification_handler=notification_handler)
+
     @activity.defn(name="prepare_activity")
     async def prepare_activity(self, input_data: dict[str, Any]):
         await super().prepare_activity(input_data)
 
     def shutdown(self):
-        Couchbase.shutdown(self)
+        MongoDB.shutdown(self)

orchestrator/activities/mongo_db.py

@@ -0,0 +1,186 @@
+from temporalio import workflow, activity
+
+with workflow.unsafe.imports_passed_through():
+    from typing import Any
+    import traceback
+    from logging import Logger
+    from pymongo import MongoClient
+    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.notifications.models import NotificationLevel
+    from sientia_do.temporal.activities.base import BaseActivity
+
+
+def clear_mongo_id(docs: list) -> list:
+    """
+    Remove the MongoDB internal `_id` field from the document.
+
+    Args:
+        docs (list): The document to clear.
+
+    Returns:
+        list: The documents without the `_id` field.
+    """
+    for doc in docs:
+        if isinstance(doc, list):
+            clear_mongo_id(doc)
+
+        elif isinstance(doc, dict):
+            if "_id" in doc:
+                del doc["_id"]
+
+            for key, value in doc.items():
+                if isinstance(value, list):
+                    clear_mongo_id(value)
+                elif isinstance(value, dict):
+                    clear_mongo_id([value])
+
+    return docs
+
+
+class MongoDB(BaseActivity):
+    def __init__(self, connection_string: str, database_name: str,
+                 logger: Logger,
+                 notification_handler: NotificationHandler):
+        self.connection_string = connection_string
+        self.database_name = database_name
+
+        self.client = MongoClient(
+            self.connection_string, serverSelectionTimeoutMS=5000)
+        self.client.server_info()  # Trigger an exception if connection fails
+
+        self.database = self.client[self.database_name]
+
+        # Initialize MongoDB client here (omitted for brevity)
+        logger.info("MongoDB connection initialized")
+
+        BaseActivity.__init__(self,
+                              logger=logger,
+                              notification_handler=notification_handler)
+
+    def shutdown(self):
+        """
+        Close the MongoDB client connection.
+        """
+        try:
+            if self.client:
+                self.logger.info("Closing MongoDB connection...")
+                self.client.close()
+                self.logger.info("MongoDB connection closed successfully")
+        except Exception as e:
+            self.logger.error(f"Failed to close MongoDB connection: {e}")
+
+    def __del__(self):
+        """
+        Destructor to ensure MongoDB client is closed when the object is deleted.
+        """
+        self.shutdown()
+
+    @activity.defn(name="find_documents_in_mongodb",)
+    async def find_documents_in_mongodb(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
+        """
+        Find documents in a MongoDB collection based on the provided query parameters.
+
+        Args:
+            - input_data (dict): Input data containing query parameters. Contains:
+                - query (dict): Query parameters to filter documents.
+
+        Returns:
+            list[dict]: List of documents matching the query.
+        """
+
+        query = input_data.get("query", {})
+
+        collection_name = query.get("collection")
+        if not collection_name:
+            raise ValueError("Collection name must be provided in the query.")
+
+        filters = query.get("filters", {})
+
+        self.logger.info(
+            f"Loading documents from collection '{collection_name}' with filters: {filters}")
+
+        try:
+            collection = self.database[collection_name]
+
+            documents = list(collection.find(filters, {"_id": 0}))
+
+            documents = clear_mongo_id(documents)
+
+            self.logger.info(
+                f"Loaded {len(documents)} documents from collection '{collection_name}'")
+
+            self.logger.debug(
+                f"Documents loaded: {documents}"
+            )
+
+            return documents
+
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.notification_handler.build_and_send_notification(
+                notification_id="MONGODB_QUERY_ERROR",
+                message=f"Failed to execute MongoDB query: {e}",
+                block="load_query_from_mongodb",
+                level=NotificationLevel.ERROR,
+                attachment_content=trace
+            )
+            self.logger.error(trace)
+
+            raise e
+
+    @activity.defn(name="aggregate_documents_in_mongodb")
+    async def aggregate_documents_in_mongodb(self,
+                                             input_data: dict[str, Any]) -> list[dict[str, Any]]:
+        """
+        Aggregate documents in a MongoDB collection based on the provided aggregation pipeline.
+
+        Args:
+            - input_data (dict): Input data containing aggregation parameters. Contains:
+                - query (dict): Query parameters to filter documents.
+
+        Returns:
+            list[dict]: List of aggregated documents.
+        """
+
+        query = input_data.get("query", {})
+
+        collection_name = query.get("collection")
+        if not collection_name:
+            raise ValueError("Collection name must be provided in the query.")
+        aggregation = query.get("aggregation")
+        if not aggregation:
+            raise ValueError("Aggregation must be provided.")
+        aggregation.append({"$project": {"_id": 0}})
+
+        self.logger.info(
+            f"Aggregating documents from collection '{collection_name}' with aggregation: {aggregation}")
+
+        try:
+            collection = self.database[collection_name]
+
+            aggregated_documents = list(
+                collection.aggregate(aggregation))
+
+            aggregated_documents = clear_mongo_id(aggregated_documents)
+
+            self.logger.info(
+                f"Aggregated {len(aggregated_documents)} documents from collection '{collection_name}'")
+
+            self.logger.debug(
+                f"Aggregation result: {aggregated_documents}"
+            )
+
+            return aggregated_documents
+
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.notification_handler.build_and_send_notification(
+                notification_id="MONGODB_AGGREGATION_ERROR",
+                message=f"Failed to execute MongoDB aggregation: {e}",
+                block="aggregate_documents_in_mongodb",
+                level=NotificationLevel.ERROR,
+                attachment_content=trace
+            )
+            self.logger.error(trace)
+
+            raise e

orchestrator/utils/connectors_config.py

@@ -10,6 +10,18 @@ def build_redis_config():
     }
 
 
+def build_mongodb_config():
+    username = getenv('MONGODB_USERNAME', 'sientia')
+    password = getenv('MONGODB_PASSWORD', 'sientia')
+    uri = getenv('MONGODB_URL', 'localhost:27017')
+
+    connection_string = f'mongodb://{username}:{password}@{uri}'
+    return {
+        'connection_string': connection_string,
+        'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia')
+    }
+
+
 def build_couchbase_config():
     return {
         'connection_string': getenv('COUCHBASE_CONNECTION_STRING', 'couchbase://localhost'),

orchestrator/worker/worker.py

@@ -8,8 +8,9 @@ with workflow.unsafe.imports_passed_through():
     from orchestrator.workflows.orchestrator import Orchestrator
     from orchestrator.activities.activities import Activities
     from orchestrator.utils.connectors_config import (
-        build_couchbase_config,
+        # build_couchbase_config,
         build_redis_config,
+        build_mongodb_config
     )
     from sientia_do.notifications.handlers import NotificationHandler
     from sientia_do.temporal.utils.logger import get_logger
@@ -40,8 +41,9 @@ async def main():
 
     activities = Activities(
         temporal_client=temporal_client,
-        couchbase_config=build_couchbase_config(),
+        # couchbase_config=build_couchbase_config(),
         redis_config=build_redis_config(),
+        mongodb_config=build_mongodb_config(),
         logger=logger,
         notification_handler=notification_handler
     )
@@ -62,7 +64,10 @@ async def main():
                 activities.update_slots,
                 activities.delete_slots,
                 # Couchbase
-                activities.load_query_from_couchbase,
+                # activities.load_query_from_couchbase,
+                # MongoDB
+                activities.aggregate_documents_in_mongodb,
+                activities.find_documents_in_mongodb,
                 # Temporal
                 activities.load_schedule,
                 activities.create_schedules,

orchestrator/workflows/orchestrator.py

@@ -26,8 +26,17 @@ class Orchestrator:
             start_to_close_timeout=timedelta(seconds=60)
         )
 
+        # pipeline_config_handler = workflow.execute_local_activity_method(
+        #     Activities.load_query_from_couchbase,
+        #     {
+        #         'query': input_data['pipelines_query']
+        #     },
+        #     retry_policy=retry_policy,
+        #     start_to_close_timeout=timedelta(seconds=60)
+        # )
+
         pipeline_config_handler = workflow.execute_local_activity_method(
-            Activities.load_query_from_couchbase,
+            Activities.aggregate_documents_in_mongodb,
             {
                 'query': input_data['pipelines_query']
             },
@@ -35,8 +44,17 @@ class Orchestrator:
             start_to_close_timeout=timedelta(seconds=60)
         )
 
+        # opc_servers_handler = workflow.execute_local_activity_method(
+        #     Activities.load_query_from_couchbase,
+        #     {
+        #         'query': input_data['opc_servers_query']
+        #     },
+        #     retry_policy=retry_policy,
+        #     start_to_close_timeout=timedelta(seconds=60)
+        # )
+
         opc_servers_handler = workflow.execute_local_activity_method(
-            Activities.load_query_from_couchbase,
+            Activities.find_documents_in_mongodb,
             {
                 'query': input_data['opc_servers_query']
             },

requirements.txt

@@ -3,4 +3,5 @@ psycopg2-binary
 sqlalchemy
 redis
 couchbase
+pymongo
 git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.14

tests/orchestrator/activities/test_activities.py

@@ -1,18 +1,21 @@
 from unittest.mock import patch, MagicMock, ANY
 from pytest import mark
+from orchestrator.activities import mongo_db
 from orchestrator.activities.activities import Activities
-from orchestrator.activities.couchbase import Couchbase
+from orchestrator.activities.mongo_db import MongoDB
 from orchestrator.activities.temporal_manager import TemporalManager
 from orchestrator.activities.slot_manager import SlotManager
 from orchestrator.activities.formatters import Formatters
 
 
 @patch('orchestrator.activities.couchbase.Couchbase.__init__')
+@patch('orchestrator.activities.mongo_db.MongoDB.__init__')
 @patch('orchestrator.activities.temporal_manager.TemporalManager.__init__')
 @patch('orchestrator.activities.slot_manager.SlotManager.__init__')
 @patch('orchestrator.activities.formatters.Formatters.__init__')
 def test___init__(mock_formatters_init, mock_slot_manager_init,
                   mock_temporal_manager_init,
+                  mock_mongodb_init,
                   mock_couchbase_init):
 
     couchbase_config = {
@@ -21,6 +24,11 @@ def test___init__(mock_formatters_init, mock_slot_manager_init,
         'password': 'password'
     }
 
+    mongo_db_config = {
+        'connection_string': 'mongodb://localhost:27017',
+        'database_name': 'test_db'
+    }
+
     redis_config = {
         'host': 'localhost',
         'port': 6379,
@@ -34,14 +42,16 @@ def test___init__(mock_formatters_init, mock_slot_manager_init,
 
     activities = Activities(
         temporal_client=temporal_client,
-        couchbase_config=couchbase_config,
+        # couchbase_config=couchbase_config,
         redis_config=redis_config,
+        mongodb_config=mongo_db_config,
         logger=logger,
         notification_handler=notification_handler
     )
 
     assert isinstance(activities, Activities)
-    assert isinstance(activities, Couchbase)
+    # assert isinstance(activities, Couchbase)
+    assert isinstance(activities, MongoDB)
     assert isinstance(activities, TemporalManager)
     assert isinstance(activities, SlotManager)
     assert isinstance(activities, Formatters)
@@ -56,11 +66,19 @@ def test___init__(mock_formatters_init, mock_slot_manager_init,
         notification_handler=notification_handler
     )
 
-    mock_couchbase_init.assert_called_once_with(
+    # mock_couchbase_init.assert_called_once_with(
+    #     ANY,
+    #     connection_string=couchbase_config['connection_string'],
+    #     username=couchbase_config['username'],
+    #     password=couchbase_config['password'],
+    #     logger=logger,
+    #     notification_handler=notification_handler
+    # )
+
+    mock_mongodb_init.assert_called_once_with(
         ANY,
-        connection_string=couchbase_config['connection_string'],
+        connection_string='mongodb://localhost:27017',
-        username=couchbase_config['username'],
+        database_name='test_db',
-        password=couchbase_config['password'],
         logger=logger,
         notification_handler=notification_handler
     )
@@ -80,7 +98,7 @@ def test___init__(mock_formatters_init, mock_slot_manager_init,
 
 
 @mark.asyncio
-@patch('orchestrator.activities.couchbase.Cluster')
+@patch('orchestrator.activities.mongo_db.MongoClient')
 async def test_prepare_activity(_mock_cluster):
     couchbase_config = {
         'connection_string': 'couchbase://localhost',
@@ -88,6 +106,11 @@ async def test_prepare_activity(_mock_cluster):
         'password': 'password'
     }
 
+    mongo_db_config = {
+        'connection_string': 'mongodb://localhost:27017',
+        'database_name': 'test_db'
+    }
+
     redis_config = {
         'host': 'localhost',
         'port': 6379,
@@ -101,8 +124,9 @@ async def test_prepare_activity(_mock_cluster):
 
     activities = Activities(
         temporal_client=temporal_client,
-        couchbase_config=couchbase_config,
+        # couchbase_config=couchbase_config,
         redis_config=redis_config,
+        mongodb_config=mongo_db_config,
         logger=logger,
         notification_handler=notification_handler
     )

tests/orchestrator/activities/test_couchbase.py

@@ -16,6 +16,18 @@ def couchbase(_cluster_mock):
     )
 
 
+def test_shutdown_success(couchbase):
+    couchbase.shutdown()
+    couchbase.cluster.close.assert_called_once()
+
+
+def test_shutdown_failure(couchbase):
+    couchbase.cluster.close.side_effect = Exception("Test error")
+    couchbase.shutdown()
+    couchbase.logger.error.assert_called_once_with(
+        "Failed to close Couchbase connection: %s", couchbase.cluster.close.side_effect)
+
+
 @mark.asyncio
 async def test_load_query_from_couchbase_success(couchbase):
     couchbase.cluster.query.return_value.rows.return_value = [

tests/orchestrator/activities/test_mongo_db.py

@@ -0,0 +1,246 @@
+from unittest.mock import MagicMock, patch, ANY
+from pytest import fixture, mark
+from orchestrator.activities.mongo_db import clear_mongo_id
+from orchestrator.activities.mongo_db import MongoDB
+from sientia_do.notifications.models import NotificationLevel
+
+
+def test_clear_mongo_id():
+    input_data = [
+        [
+            {
+                "name": "test",
+                "_id": "12345",
+            }
+        ],
+        {
+            "name": "test",
+            "_id": "12345",
+            "nested": {
+                "_id": "67890",
+                "value": [1, 2, 3],
+                "list": [{"_id": "abcde", "item": "value"}]
+            },
+            "nested_list": [
+                {"_id": "fghij", "item": "value1"},
+                {"_id": "klmno", "item": "value2"}
+            ]
+        }
+    ]
+
+    output = clear_mongo_id(input_data)
+
+    assert output == [
+        [
+            {"name": "test"}
+        ],
+        {
+            "name": "test",
+            "nested": {
+                "value": [1, 2, 3],
+                "list": [{"item": "value"}]
+            },
+            "nested_list": [
+                {"item": "value1"},
+                {"item": "value2"}
+            ]
+        }]
+
+
+@fixture
+@patch("orchestrator.activities.mongo_db.MongoClient")
+def mongo_db(mongo_mock):
+    return (
+        MongoDB(
+            connection_string="mongodb://localhost:27017",
+            database_name="test_db",
+            logger=MagicMock(),
+            notification_handler=MagicMock()
+        )
+    )
+
+
+@patch("orchestrator.activities.mongo_db.MongoClient")
+def test___init__(mongo_mock):
+    mongo_db = MongoDB(
+        connection_string="mongodb://localhost:27017",
+        database_name="test_db",
+        logger=MagicMock(),
+        notification_handler=MagicMock()
+    )
+    assert mongo_db.connection_string == "mongodb://localhost:27017"
+    assert mongo_db.database_name == "test_db"
+    mongo_mock.assert_called_once_with(
+        "mongodb://localhost:27017", serverSelectionTimeoutMS=5000
+    )
+    mongo_db.client.server_info.assert_called_once()
+    mongo_db.client.__getitem__.assert_called_once_with("test_db")
+
+
+def test_shutdown_success(mongo_db):
+    mongo_db.shutdown()
+    mongo_db.client.close.assert_called_once()
+    mongo_db.logger.info.assert_any_call("Closing MongoDB connection...")
+    mongo_db.logger.info.assert_any_call(
+        "MongoDB connection closed successfully")
+
+
+def test_shutdown_failure(mongo_db):
+    mongo_db.client.close.side_effect = Exception("Close failed")
+    mongo_db.shutdown()
+    mongo_db.logger.error.assert_called_once_with(
+        "Failed to close MongoDB connection: Close failed"
+    )
+
+
+@mark.asyncio
+async def test_find_documents_in_mongodb_success(mongo_db):
+    input_data = {"collection": "test_collection", "filters": {
+        "name": {"$exists": True}
+    }}
+    mock_collection = MagicMock()
+    mock_collection.find.return_value = [
+        {"_id": "12345", "name": "test1"},
+        {"_id": "67890", "name": "test2"}
+    ]
+    mongo_db.database.__getitem__.return_value = mock_collection
+
+    result = await mongo_db.find_documents_in_mongodb(
+        {
+            "query": input_data
+        })
+
+    assert len(result) == 2
+    assert result[0] == {"name": "test1"}
+    assert result[1] == {"name": "test2"}
+    mock_collection.find.assert_called_once_with(
+        {"name": {"$exists": True}}, {"_id": 0}
+    )
+
+
+@mark.asyncio
+async def test_find_documents_in_mongodb_failure(mongo_db):
+    input_data = {"collection": "test_collection", "filters": {
+        "name": {"$exists": True}
+    }}
+    mongo_db.database.__getitem__.return_value = MagicMock(
+        find=MagicMock(side_effect=Exception("Error"))
+    )
+
+    try:
+        await mongo_db.find_documents_in_mongodb(
+            {
+                "query": input_data
+            })
+    except Exception as e:
+        assert str(e) == "Error"
+        mongo_db.notification_handler.build_and_send_notification.assert_called_once_with(
+            notification_id="MONGODB_QUERY_ERROR",
+            message="Failed to execute MongoDB query: Error",
+            level=NotificationLevel.ERROR,
+            block="load_query_from_mongodb",
+            attachment_content=ANY
+        )
+
+    else:
+        assert False, "Expected an exception to be raised"
+
+
+@mark.asyncio
+async def test_find_documents_in_mongodb_missing_collection(mongo_db):
+    input_data = {"query": {"filters": {}}}
+
+    try:
+        await mongo_db.find_documents_in_mongodb(input_data)
+
+    except ValueError as e:
+        assert str(e) == "Collection name must be provided in the query."
+
+    else:
+        assert False, "Expected a ValueError to be raised"
+
+
+@mark.asyncio
+async def test_aggregate_documents_in_mongodb_success(mongo_db):
+    input_data = {"collection": "test_collection", "aggregation": [
+        {"$match": {"name": {"$exists": True}}},
+        {"$project": {"name": 1}}
+    ]}
+    mock_collection = MagicMock()
+    mock_collection.aggregate.return_value = [
+        {"_id": "asdad", "name": "test1"},
+        {"_id": "adzx", "name": "test2"}
+    ]
+    mongo_db.database.__getitem__.return_value = mock_collection
+
+    result = await mongo_db.aggregate_documents_in_mongodb(
+        {
+            "query": input_data
+        })
+
+    assert len(result) == 2
+    assert result[0] == {"name": "test1"}
+    assert result[1] == {"name": "test2"}
+    expected_pipeline = input_data["aggregation"]
+    expected_pipeline.append({"$project": {"_id": 0}})
+
+    mock_collection.aggregate.assert_called_once_with(
+        expected_pipeline
+    )
+
+
+@mark.asyncio
+async def test_aggregate_documents_in_mongodb_failure(mongo_db):
+    input_data = {"collection": "test_collection", "aggregation": [
+        {"$match": {"name": {"$exists": True}}},
+        {"$project": {"name": 1}}
+    ]}
+    mongo_db.database.__getitem__.return_value = MagicMock(
+        aggregate=MagicMock(side_effect=Exception("Error"))
+    )
+
+    try:
+        await mongo_db.aggregate_documents_in_mongodb(
+            {
+                "query": input_data
+            })
+    except Exception as e:
+        assert str(e) == "Error"
+        mongo_db.notification_handler.build_and_send_notification.assert_called_once_with(
+            notification_id="MONGODB_AGGREGATION_ERROR",
+            message="Failed to execute MongoDB aggregation: Error",
+            level=NotificationLevel.ERROR,
+            block="aggregate_documents_in_mongodb",
+            attachment_content=ANY
+        )
+
+    else:
+        assert False, "Expected an exception to be raised"
+
+
+@mark.asyncio
+async def test_aggregate_documents_in_mongodb_missing_collection(mongo_db):
+    input_data = {"query": {"aggregation": []}}
+
+    try:
+        await mongo_db.aggregate_documents_in_mongodb(input_data)
+
+    except ValueError as e:
+        assert str(e) == "Collection name must be provided in the query."
+
+    else:
+        assert False, "Expected a ValueError to be raised"
+
+
+@mark.asyncio
+async def test_aggregate_documents_in_mongodb_missing_aggregation(mongo_db):
+    input_data = {"query": {"collection": "test_collection"}}
+
+    try:
+        await mongo_db.aggregate_documents_in_mongodb(input_data)
+
+    except ValueError as e:
+        assert str(e) == "Aggregation must be provided."
+
+    else:
+        assert False, "Expected a ValueError to be raised"

tests/orchestrator/utils/test_connectors_config.py

@@ -1,6 +1,7 @@
 from os import environ
 from orchestrator.utils.connectors_config import (build_redis_config,
-                                                  build_couchbase_config)
+                                                  build_couchbase_config,
+                                                  build_mongodb_config)
 
 
 def test_build_redis_config_with_env_vars():
@@ -49,3 +50,27 @@ def test_build_couchbase_config_with_defaults():
         'username': 'sientia',
         'password': 'sientia'
     }
+
+
+def test_build_mongo_db_config_with_env_vars():
+    environ['MONGODB_USERNAME'] = 'sientia1'
+    environ['MONGODB_PASSWORD'] = 'sientia1'
+    environ['MONGODB_URL'] = 'localhost:27018'
+    environ['MONGODB_DATABASE_NAME'] = 'test_db'
+
+    assert build_mongodb_config() == {
+        'connection_string': 'mongodb://sientia1:sientia1@localhost:27018',
+        'database_name': 'test_db'
+    }
+
+
+def test_build_mongo_db_config_with_defaults():
+    environ.pop('MONGODB_USERNAME', None)
+    environ.pop('MONGODB_PASSWORD', None)
+    environ.pop('MONGODB_DATABASE_NAME', None)
+    environ.pop('MONGODB_URL', None)
+
+    assert build_mongodb_config() == {
+        'connection_string': 'mongodb://sientia:sientia@localhost:27017',
+        'database_name': 'sientia'
+    }

tests/orchestrator/workflows/test_orchestrator.py

@@ -36,7 +36,7 @@ async def test_run(workflow_mock, orchestrator):
 
     workflow_mock.execute_local_activity_method.assert_has_calls([
         call(
-            Activities.load_query_from_couchbase,
+            Activities.aggregate_documents_in_mongodb,
             {
                 "query": input_data["pipelines_query"]
             },
@@ -47,7 +47,7 @@ async def test_run(workflow_mock, orchestrator):
 
     workflow_mock.execute_local_activity_method.assert_has_calls([
         call(
-            Activities.load_query_from_couchbase,
+            Activities.find_documents_in_mongodb,
             {
                 "query": input_data["opc_servers_query"]
             },

values.yaml

@@ -11,7 +11,7 @@ image:
   # This sets the pull policy for images.
   pullPolicy: Always
   # Overrides the image tag whose default is the chart appVersion.
-  tag: "0.1.0"
+  tag: "0.2.0"
 
 # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
 imagePullSecrets:
@@ -123,7 +123,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
   - name: GITHUB_BRANCH
-    value: "main"
+    value: "SIENTIAPDE-1107-testar-o-uso-do-mongo-db-no-lugar-do-couchbase"
   - name: PYTHON_APP
     value: "orchestrator.worker.worker"
 
@@ -150,6 +150,15 @@ env:
   - name: COUCHBASE_PASSWORD
     value: "sientia"
 
+  - name: MONGODB_USERNAME
+    value: "root"
+  - name: MONGODB_PASSWORD
+    value: "wKZDbMNU1c"
+  - name: MONGODB_URL
+    value: "my-release-mongodb.mongodb.svc.cluster.local:27017"
+  - name: MONGODB_DATABASE
+    value: "sientia"
+
   - name: KAFKA_BOOTSTRAP_SERVERS
     value: "kafka.kafka.svc.cluster.local:9092"
 
@@ -171,7 +180,7 @@ ssh:
 
 # kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp
 
-# helm upgrade --install sientia-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.1.0-uat
+# helm upgrade --install sientia-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.2.0-uat
 
 # kubectl create secret generic git-ssh-key-sientia-orchestrator-worker \
 #   --namespace sientia \