Merge pull request #10 from Aignosi/SIENTIAPDE-1163-alterar-dinamica-de-notificacoes-para-usar-o-mongodb-ao-inves-do-kafka

Sientiapde 1163 alterar dinamica de notificacoes para usar o mongodb ao inves do kafka

.gitignore

@@ -40,3 +40,5 @@ htmlcov/
 
 # git keys
 git_key*
+
+git_log

README.md

@@ -93,3 +93,10 @@ The application can be deployed using the following command:
 ```bash
 helm upgrade --install sientia-dataops-laborious sientia/sientia-module -n sientia --create-namespace -f ./values.yaml
 ```
+
+#PR shortcut
+```
+git log origin/main..HEAD --no-merges > git_log
+```
+Prompt:
+Write a summary of PR changes in markdown. Be objective and direct. Write to file

laborious/activities/activities.py

@@ -2,7 +2,7 @@ from temporalio import activity, workflow
 
 with workflow.unsafe.imports_passed_through():
     from sientia_do.temporal.activities.postgres import Postgres
-    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.temporal.utils.logger import Logger
     from laborious.activities.mlflow import MLFlow
     from laborious.activities.gates import Gates

laborious/activities/gates.py

@@ -3,7 +3,7 @@ from temporalio import activity, workflow
 
 with workflow.unsafe.imports_passed_through():
     import traceback
-    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.notifications.models import NotificationLevel
     from sientia_do.temporal.activities.base import BaseActivity
     from sientia_do.temporal.utils.logger import Logger

laborious/activities/mlflow.py

@@ -5,7 +5,7 @@ from temporalio import activity, workflow
 
 with workflow.unsafe.imports_passed_through():
     from sientia_do.temporal.activities.base import BaseActivity
-    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.temporal.utils.logger import Logger
     from laborious.utils.repository.model_repository import MLFlowRepository
     from typing import Any

laborious/activities/opc.py

@@ -2,7 +2,7 @@ from temporalio import activity, workflow
 
 
 with workflow.unsafe.imports_passed_through():
-    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.notifications.models import NotificationLevel
     from sientia_do.temporal.activities.base import BaseActivity
     from sientia_do.temporal.utils.logger import Logger
@@ -22,7 +22,7 @@ class OPC(BaseActivity):
         self.notification_handler = notification_handler
         self.opc_servers = opc_servers
 
-        self.opc_repository = {}
+        self.opc_repository: dict[str, OpcRepository] = {}
         for id, server in opc_servers.items():
             self.opc_repository[id] = OpcRepository(
                 id=server['id'],
@@ -35,8 +35,22 @@ class OPC(BaseActivity):
                 notification_handler=self.notification_handler,
                 reconnection_interval=server['reconnection_interval'],
             )
-            self.opc_repository[id].connect()
+            is_connected, error_data = self.opc_repository[id].connect()
-
+            if not is_connected:
+                self.send_notification(
+                    metadata={
+                        'model_id': '-',
+                        'model_name': '-',
+                        'workflow_name': '-',
+                        'schedule_name': 'INITIALIZATION'
+                    },
+                    notification_id=error_data['notification_id'],
+                    message=error_data['message'],
+                    block=error_data['block'],
+                    level=error_data.get('level', NotificationLevel.ERROR),
+                    attachment_content=error_data.get(
+                        'attachment_content', None)
+                )
         BaseActivity.__init__(self, logger, notification_handler)
 
     def write_data(self, server_id: str, tag: str, data: Any,
@@ -56,8 +70,20 @@ class OPC(BaseActivity):
         """
 
         try:
-            return self.opc_repository[server_id].write_data(
+            is_success, error_data = self.opc_repository[server_id].write_data(
                 tag, data, data_type, self.logger, metadata)
+            if not is_success:
+                self.send_notification(
+                    metadata=metadata,
+                    notification_id=error_data['notification_id'],
+                    message=error_data['message'],
+                    block=error_data['block'],
+                    level=error_data.get('level', NotificationLevel.ERROR),
+                    attachment_content=error_data.get(
+                        'attachment_content', None)
+                )
+                return False
+            return True
         except Exception as e:
             trace = traceback.format_exc()
             self.send_notification(

laborious/utils/connectors_config.py

@@ -44,3 +44,15 @@ def build_opc_config():
             'reconnection_interval': int(getenv('OPC_RECONNECTION_INTERVAL', '120'))
         }
     }
+
+
+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')
+    }

laborious/utils/repository/opc_repository.py

@@ -6,7 +6,7 @@ from asyncua.sync import Client
 from asyncua.crypto.security_policies import SecurityPolicyBasic256
 from asyncua.ua import DataValue, Variant, VariantType, DateTime
 from regex import F
-from sientia_do.notifications.handlers import NotificationHandler
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
 from sientia_do.notifications.models import NotificationLevel
 from sientia_do.temporal.utils.logger import Logger
 
@@ -91,7 +91,7 @@ class OpcRepository():
         self.client.secure_channel_timeout = 10000000
         self.client.session_timeout = 10000000
 
-    def connect(self):
+    def connect(self) -> tuple[bool, dict[str, Any]]:
         """
         Establishes a connection to the OPC server.
         This method initializes the OPC client using the provided URL and
@@ -107,7 +107,7 @@ class OpcRepository():
         self.logger.info(f'Starting connection to OPC server {self.id}...')
         return self.try_connect()
 
-    def try_connect(self):
+    def try_connect(self) -> tuple[bool, dict[str, Any]]:
         """
         Tries to connect to the OPC server.
 
@@ -118,18 +118,18 @@ class OpcRepository():
         try:
             self.last_reconnection_time = datetime.now()
             self.client.connect()
-            return True
+            return True, {}
         except Exception as e:
             trace = traceback.format_exc()
-            self.notification_handler.build_and_send_notification(
-                notification_id=f"OPC_CONNECTION_ERROR_{self.id}",
-                message=f"Failed to connect to OPC server: {e}",
-                block="opc_repository",
-                level=NotificationLevel.ERROR,
-                attachment_content=trace
-            )
             self.logger.error(trace)
-            return False
+
+            return False, {
+                "notification_id": f"OPC_CONNECTION_ERROR_{self.id}",
+                "message": f"Failed to connect to OPC server: {e}",
+                "block": "opc_repository",
+                "level": NotificationLevel.ERROR,
+                "attachment_content": trace
+            }
 
     def disconnect(self):
         """
@@ -153,7 +153,7 @@ class OpcRepository():
         except Exception as e:
             self.logger.error(f"Error in destructor: {e}")
 
-    def validate_connection(self):
+    def validate_connection(self) -> tuple[bool, dict[str, Any]]:
         """
         Validates the connection to the OPC server.
         If the connection is not established, it attempts to reconnect.
@@ -193,12 +193,12 @@ class OpcRepository():
                     f"Trying to reconnect to OPC server {self.id}...")
                 return self.connect()
 
-            return False
+            return False, {}
 
-        return True
+        return True, {}
 
     def write_data(self, node: str, value: Any, data_type: str,
-                   logger: Logger, metadata: dict[str, Any]) -> bool:
+                   logger: Logger, metadata: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
         """
         Writes data to the OPC server.
         If the connection is not established, it attempts to reconnect.
@@ -209,31 +209,33 @@ class OpcRepository():
         If the client is not connected, it attempts to reconnect.
         If the client is connected, it returns True.
         """
-        if not self.validate_connection():
+
-            return False
+        is_connected, error = self.validate_connection()
+
+        if not is_connected:
+            return False, error
+
         try:
             node = self.client.get_node(node)
         except Exception as e:
             trace = traceback.format_exc()
-            self.notification_handler.build_and_send_notification(
-                notification_id=f"OPC_WRITE_GET_NODE_ERROR_{self.id}",
-                message=f"Failed to get node from OPC server: {e} | metadata: {metadata}",
-                block="opc_repository",
-                level=NotificationLevel.ERROR,
-                attachment_content=trace
-            )
             logger.custom_error(trace, metadata.get('schedule_name', 'N/A'))
             self.error_count += 1
-            return False
+            return False, {
+                "notification_id": f"OPC_WRITE_GET_NODE_ERROR_{self.id}",
+                "message": f"Failed to get node from OPC server: {e} | metadata: {metadata}",
+                "block": "opc_repository",
+                "level": NotificationLevel.ERROR,
+                "attachment_content": trace
+            }
 
         if data_type not in data_type_map:
-            self.notification_handler.build_and_send_notification(
+            return False, {
-                notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}",
+                "notification_id": f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}",
-                message=f"Unsupported data type: {data_type} | metadata: {metadata}",
+                "message": f"Unsupported data type: {data_type} | metadata: {metadata}",
-                block="opc_repository",
+                "block": "opc_repository",
-                level=NotificationLevel.ERROR
+                "level": NotificationLevel.ERROR
-            )
+            }
-            return False
 
         data = data_type_map[data_type]['converter'](value)
         logger.custom_info(
@@ -256,16 +258,15 @@ class OpcRepository():
             node.write_value(ua_data)
         except Exception as e:
             trace = traceback.format_exc()
-            self.notification_handler.build_and_send_notification(
-                notification_id=f"OPC_WRITE_DATA_ERROR_{self.id}",
-                message=f"Failed to write data to OPC server: {e} | metadata: {metadata}",
-                block="opc_repository",
-                level=NotificationLevel.ERROR,
-                attachment_content=trace
-            )
             logger.custom_error(trace, metadata.get('schedule_name', 'N/A'))
             self.error_count += 1
-            return False
+            return False, {
+                "notification_id": f"OPC_WRITE_DATA_ERROR_{self.id}",
+                "message": f"Failed to write data to OPC server: {e} | metadata: {metadata}",
+                "block": "opc_repository",
+                "level": NotificationLevel.ERROR,
+                "attachment_content": trace
+            }
         self.error_count = 0
 
-        return True
+        return True, {}

laborious/worker/worker.py

@@ -14,9 +14,10 @@ with workflow.unsafe.imports_passed_through():
     from laborious.utils.connectors_config import (
         build_postgres_config,
         build_mlflow_config,
-        build_opc_config
+        build_opc_config,
+        build_mongodb_config
     )
-    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.temporal.utils.logger import get_logger
 
 
@@ -28,10 +29,12 @@ async def main():
 
     logger.info('Starting Notification Handler...')
 
+    mongo_config = build_mongodb_config()
     notification_handler = NotificationHandler(
-        servers=os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'http://localhost:9092'),
+        connection_string=mongo_config['connection_string'],
+        database=mongo_config['database_name'],
         logger=logger,
-        project_name=os.getenv('PROJECT_NAME', 'laborious'),
+        project_name=os.getenv('PROJECT_NAME', 'laborious')
     )
 
     logger.info('Starting Activities...')

laborious/workflows/sub_workflows/prediction_process.py

@@ -64,9 +64,6 @@ class PredictionProcess():
             'path_priority': input_data['path_priority']
         }
 
-        print(f"Metadata e input atualizadas {gate_input}")
-        print(f"Metadata: {metadata}")
-
         path_flag, confidence, comment = await workflow.execute_local_activity_method(
             Activities.input_gate,
             gate_input,

requirements.txt

@@ -3,5 +3,5 @@ psycopg2-binary
 sqlalchemy
 asyncua
 redis
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.2.1
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.3
 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.1

tests/laborious/activities/test_opc.py

@@ -16,11 +16,28 @@ metadata = {
 
 
 @patch("laborious.activities.opc.OpcRepository")
-def test___init__(mock_opc_repository):
+@patch("laborious.activities.opc.OPC.send_notification")
+def test___init__(mock_send_notification, mock_opc_repository):
     mock_logger = MagicMock()
-    server1 = MagicMock()
+    server1 = MagicMock(
-    server2 = MagicMock()
+        connect=MagicMock(return_value=(True, {})),
-    mock_opc_repository.side_effect = [server1, server2]
+        write_data=MagicMock(return_value=(True, {}))
+    )
+    server2 = MagicMock(
+        connect=MagicMock(return_value=(True, {})),
+        write_data=MagicMock(return_value=(True, {}))
+    )
+    server3 = MagicMock(
+        connect=MagicMock(return_value=(False, {
+            'notification_id': 'OPC_CONNECTION_ERROR_server3',
+            'message': 'Failed to connect to OPC server: Test error',
+            'block': 'opc_repository',
+            'level': NotificationLevel.ERROR,
+            'attachment_content': 'Test error'
+        })),
+        write_data=MagicMock(return_value=(True, {}))
+    )
+    mock_opc_repository.side_effect = [server1, server2, server3]
     mock_notification_handler = MagicMock()
     servers = {
         'server1': {
@@ -40,6 +57,15 @@ def test___init__(mock_opc_repository):
             'private_key_path': '',
             'server_cert_path': '',
             'reconnection_interval': 60,
+        },
+        'server3': {
+            'id': 'server3',
+            'url': 'http://localhost:8080',
+            'server_uri': 'opc.tcp://localhost:4840',
+            'cert_path': '',
+            'private_key_path': '',
+            'server_cert_path': '',
+            'reconnection_interval': 60,
         }
     }
     opc = OPC(
@@ -84,6 +110,22 @@ def test___init__(mock_opc_repository):
     server1.connect.assert_called_once()
     server2.connect.assert_called_once()
 
+    mock_send_notification.assert_has_calls([
+        call(
+            metadata={
+                'model_id': '-',
+                'model_name': '-',
+                'workflow_name': '-',
+                'schedule_name': 'INITIALIZATION'
+            },
+            notification_id="OPC_CONNECTION_ERROR_server3",
+            message="Failed to connect to OPC server: Test error",
+            block="opc_repository",
+            level=NotificationLevel.ERROR,
+            attachment_content=ANY
+        )
+    ])
+
 
 @fixture
 @patch("laborious.activities.opc.OpcRepository")
@@ -99,8 +141,12 @@ def opc(mock_opc_repository):
             'reconnection_interval': 60,
         }
     }
-    mock_opc_repository.write_data = MagicMock(
+
-        return_value=True
+    mock_opc_repository.return_value.write_data = MagicMock(
+        return_value=(True, {})
+    )
+    mock_opc_repository.return_value.connect = MagicMock(
+        return_value=(True, {})
     )
     opc = OPC(
         opc_servers=servers,
@@ -128,6 +174,28 @@ def test_write_data_success(opc, tag, data_type, data):
         tag, data, data_type, opc.logger, metadata)
 
 
+def test_write_data_failed(opc):
+    opc.opc_repository['server1'].write_data.return_value = (False, {
+        'notification_id': 'OPC_WRITE_DATA_ERROR_server1',
+        'message': 'Failed to write data to OPC server: Test error',
+        'block': 'opc_repository',
+        'level': NotificationLevel.ERROR,
+        'attachment_content': 'Test error'
+    })
+
+    assert opc.write_data(server_id='server1', tag='tag1', data=50,
+                          data_type='int', tag_type='prediction', metadata=metadata) is False
+
+    opc.send_notification.assert_called_once_with(
+        metadata=metadata,
+        notification_id="OPC_WRITE_DATA_ERROR_server1",
+        message="Failed to write data to OPC server: Test error",
+        block="opc_repository",
+        level=NotificationLevel.ERROR,
+        attachment_content=ANY
+    )
+
+
 def test_write_data_exception(opc):
     opc.opc_repository['server1'].write_data.side_effect = Exception(
         "Test error")

tests/laborious/utils/repository/test_opc_repository.py

@@ -115,17 +115,15 @@ def test_try_connect_fail(opc_repository):
     opc_repository.client = MagicMock()
     opc_repository.client.connect.side_effect = Exception("Test error")
 
-    opc_repository.try_connect()
+    is_connected, error_data = opc_repository.try_connect()
 
     opc_repository.client.connect.assert_called_once()
-    assert opc_repository.last_reconnection_time is not None
+    assert is_connected is False
-    opc_repository.notification_handler.build_and_send_notification.assert_called_once_with(
+    assert error_data['notification_id'] == f"OPC_CONNECTION_ERROR_{opc_repository.id}"
-        notification_id=f"OPC_CONNECTION_ERROR_{opc_repository.id}",
+    assert error_data['message'] == "Failed to connect to OPC server: Test error"
-        message="Failed to connect to OPC server: Test error",
+    assert error_data['block'] == "opc_repository"
-        block="opc_repository",
+    assert error_data['level'] == NotificationLevel.ERROR
-        level=NotificationLevel.ERROR,
+    assert error_data['attachment_content'] is not None
-        attachment_content=ANY
-    )
 
 
 def test_disconnect(opc_repository, mock_client):
@@ -184,7 +182,7 @@ def test_validate_connection_lost_not_time_to_reconect(_mock_datetime, opc_repos
 
     response = opc_repository.validate_connection()
     opc_repository.try_connect.assert_not_called()
-    assert response is False
+    assert response == (False, {})
 
 
 @patch('laborious.utils.repository.opc_repository.hasattr', return_value=True)
@@ -207,11 +205,11 @@ def test_validate_connection_failed(opc_repository):
     opc_repository.error_count = 0
 
     output = opc_repository.validate_connection()
-    assert output is True
+    assert output == (True, {})
 
 
 def test_write_data_validate_connection_do_nothing(opc_repository):
-    opc_repository.validate_connection = MagicMock(return_value=True)
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
     opc_repository.client = MagicMock()
     opc_repository.write_data("ns=2;s=TestNode", 42.0,
                               "float", opc_repository.logger, metadata)
@@ -220,7 +218,7 @@ def test_write_data_validate_connection_do_nothing(opc_repository):
 
 
 def test_write_data_validate_connection_failed(opc_repository):
-    opc_repository.validate_connection = MagicMock(return_value=False)
+    opc_repository.validate_connection = MagicMock(return_value=(False, {}))
     opc_repository.client = MagicMock()
     opc_repository.error_count = 0
     opc_repository.write_data("ns=2;s=TestNode", 42.0,
@@ -230,45 +228,43 @@ def test_write_data_validate_connection_failed(opc_repository):
 
 
 def test_write_data_get_node_failed(opc_repository):
-    opc_repository.validate_connection = MagicMock(return_value=True)
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
     opc_repository.client = MagicMock()
     opc_repository.error_count = 0
     opc_repository.client.get_node.side_effect = Exception("Test error")
-    opc_repository.write_data("ns=2;s=TestNode", 42.0,
+    is_success, error_data = opc_repository.write_data("ns=2;s=TestNode", 42.0,
-                              "float", opc_repository.logger, metadata)
+                                                       "float", opc_repository.logger, metadata)
     opc_repository.validate_connection.assert_called_once()
     opc_repository.client.get_node.assert_called_once_with("ns=2;s=TestNode")
-    opc_repository.notification_handler.build_and_send_notification.assert_called_once_with(
+    assert is_success is False
-        notification_id=f"OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}",
+    assert error_data['notification_id'] == f"OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}"
-        message="Failed to get node from OPC server: Test error | metadata: {'metadata': {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}}",
+    assert error_data['message'] == "Failed to get node from OPC server: Test error | metadata: {'metadata': {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}}"
-        block="opc_repository",
+    assert error_data['block'] == "opc_repository"
-        level=NotificationLevel.ERROR,
+    assert error_data['level'] == NotificationLevel.ERROR
-        attachment_content=ANY
+    assert error_data['attachment_content'] is not None
-    )
-    assert opc_repository.error_count == 1
 
 
 def test_write_data_invalid_data_type(opc_repository, mock_client):
-    opc_repository.validate_connection = MagicMock(return_value=True)
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
     opc_repository.client = mock_client
     mock_node = MagicMock()
     mock_client.get_node.return_value = mock_node
 
-    opc_repository.write_data("ns=2;s=TestNode", 42.0,
+    is_success, error_data = opc_repository.write_data("ns=2;s=TestNode", 42.0,
-                              "invalid_type", opc_repository.logger, metadata)
+                                                       "invalid_type", opc_repository.logger, metadata)
     opc_repository.validate_connection.assert_called_once()
     mock_client.get_node.assert_called_once_with("ns=2;s=TestNode")
 
-    opc_repository.notification_handler.build_and_send_notification.assert_called_once_with(
+    assert is_success is False
-        notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.id}",
+    assert error_data['notification_id'] == f"OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.id}"
-        message="Unsupported data type: invalid_type | metadata: {'metadata': {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}}",
+    assert error_data['message'] == "Unsupported data type: invalid_type | metadata: {'metadata': {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}}"
-        block="opc_repository",
+    assert error_data['block'] == "opc_repository"
-        level=NotificationLevel.ERROR
+    assert error_data['level'] == NotificationLevel.ERROR
-    )
+    assert error_data.get('attachment_content') is None
 
 
 def test_write_data(opc_repository, mock_client):
-    opc_repository.validate_connection = MagicMock(return_value=True)
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
     opc_repository.client = mock_client
     mock_node = MagicMock()
     mock_client.get_node.return_value = mock_node
@@ -281,22 +277,20 @@ def test_write_data(opc_repository, mock_client):
 
 
 def test_write_data_write_value_failed(opc_repository, mock_client):
-    opc_repository.validate_connection = MagicMock(return_value=True)
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
     opc_repository.client = mock_client
     mock_node = MagicMock()
     opc_repository.error_count = 0
     mock_client.get_node.return_value = mock_node
     mock_node.write_value.side_effect = Exception("Test error")
-    opc_repository.write_data("ns=2;s=TestNode", 42.0,
+    is_success, error_data = opc_repository.write_data("ns=2;s=TestNode", 42.0,
-                              "float", opc_repository.logger, metadata)
+                                                       "float", opc_repository.logger, metadata)
     opc_repository.validate_connection.assert_called_once()
     mock_client.get_node.assert_called_once_with("ns=2;s=TestNode")
     mock_node.write_value.assert_called_once()
-    opc_repository.notification_handler.build_and_send_notification.assert_called_once_with(
+    assert is_success is False
-        notification_id=f"OPC_WRITE_DATA_ERROR_{opc_repository.id}",
+    assert error_data['notification_id'] == f"OPC_WRITE_DATA_ERROR_{opc_repository.id}"
-        message="Failed to write data to OPC server: Test error | metadata: {'metadata': {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}}",
+    assert error_data['message'] == "Failed to write data to OPC server: Test error | metadata: {'metadata': {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}}"
-        block="opc_repository",
+    assert error_data['block'] == "opc_repository"
-        level=NotificationLevel.ERROR,
+    assert error_data['level'] == NotificationLevel.ERROR
-        attachment_content=ANY
+    assert error_data['attachment_content'] is not None
-    )
-    assert opc_repository.error_count == 1

tests/laborious/utils/test_connectors_config.py

@@ -1,7 +1,8 @@
 from os import environ
 from laborious.utils.connectors_config import (build_mlflow_config,
                                                build_opc_config,
-                                               build_postgres_config)
+                                               build_postgres_config,
+                                               build_mongodb_config)
 
 
 def test_build_mlflow_config_with_env_vars():
@@ -131,3 +132,27 @@ def test_build_postgres_config_with_defaults():
     assert config['dbname'] == 'sientia'
     assert config['min_connections'] == 5
     assert config['max_connections'] == 20
+
+
+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'
+    }

values.yaml

@@ -123,7 +123,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1148-separar-scouter-laborious-e-orchestrator-por-namespaces"
+    value: "SIENTIAPDE-1163-alterar-dinamica-de-notificacoes-para-usar-o-mongodb-ao-inves-do-kafka"
   - name: PYTHON_APP
     value: "laborious.worker.worker"
 
@@ -170,6 +170,15 @@ env:
   - name: TEMPORAL_NAMESPACE
     value: "laborious"
 
+  - 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"
+
 ssh:
   enabled: true
   secretName: git-ssh-key-sientia-laborious-worker