diff --git a/.gitignore b/.gitignore index f9bab0b..9035481 100644 --- a/.gitignore +++ b/.gitignore @@ -39,4 +39,6 @@ htmlcov/ .coverage # git keys -git_key* \ No newline at end of file +git_key* + +git_log \ No newline at end of file diff --git a/README.md b/README.md index d69a75c..51598d0 100644 --- a/README.md +++ b/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 \ No newline at end of file diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py index 2156009..3e47efe 100644 --- a/laborious/activities/activities.py +++ b/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 diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 10d3f64..2a54052 100644 --- a/laborious/activities/gates.py +++ b/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 diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index 4b8c129..bc0113e 100644 --- a/laborious/activities/mlflow.py +++ b/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 diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 6b62e22..c99a54d 100644 --- a/laborious/activities/opc.py +++ b/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( diff --git a/laborious/utils/connectors_config.py b/laborious/utils/connectors_config.py index 49cce2b..5158f05 100644 --- a/laborious/utils/connectors_config.py +++ b/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') + } diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index f0a768e..031e171 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/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( - notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}", - message=f"Unsupported data type: {data_type} | metadata: {metadata}", - block="opc_repository", - level=NotificationLevel.ERROR - ) - return False + return False, { + "notification_id": f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}", + "message": f"Unsupported data type: {data_type} | metadata: {metadata}", + "block": "opc_repository", + "level": NotificationLevel.ERROR + } 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, {} diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index d2879e3..005bb18 100644 --- a/laborious/worker/worker.py +++ b/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...') diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index 10f820c..e20f7ab 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/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, diff --git a/requirements.txt b/requirements.txt index 5dfca57..8b1535e 100644 --- a/requirements.txt +++ b/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 diff --git a/tests/laborious/activities/test_opc.py b/tests/laborious/activities/test_opc.py index ebce052..6540a16 100644 --- a/tests/laborious/activities/test_opc.py +++ b/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() - server2 = MagicMock() - mock_opc_repository.side_effect = [server1, server2] + server1 = MagicMock( + connect=MagicMock(return_value=(True, {})), + 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") diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py index 1b661d2..c399139 100644 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ b/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 - opc_repository.notification_handler.build_and_send_notification.assert_called_once_with( - notification_id=f"OPC_CONNECTION_ERROR_{opc_repository.id}", - message="Failed to connect to OPC server: Test error", - block="opc_repository", - level=NotificationLevel.ERROR, - attachment_content=ANY - ) + assert is_connected is False + assert error_data['notification_id'] == f"OPC_CONNECTION_ERROR_{opc_repository.id}" + assert error_data['message'] == "Failed to connect to OPC server: Test error" + assert error_data['block'] == "opc_repository" + assert error_data['level'] == NotificationLevel.ERROR + assert error_data['attachment_content'] is not None 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, - "float", opc_repository.logger, metadata) + is_success, error_data = opc_repository.write_data("ns=2;s=TestNode", 42.0, + "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( - 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'}}", - block="opc_repository", - level=NotificationLevel.ERROR, - attachment_content=ANY - ) - assert opc_repository.error_count == 1 + assert is_success is False + assert error_data['notification_id'] == f"OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}" + 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'}}" + assert error_data['block'] == "opc_repository" + assert error_data['level'] == NotificationLevel.ERROR + assert error_data['attachment_content'] is not None 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, - "invalid_type", opc_repository.logger, metadata) + is_success, error_data = opc_repository.write_data("ns=2;s=TestNode", 42.0, + "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( - 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'}}", - block="opc_repository", - level=NotificationLevel.ERROR - ) + assert is_success is False + assert error_data['notification_id'] == f"OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.id}" + 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'}}" + assert error_data['block'] == "opc_repository" + 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, - "float", opc_repository.logger, metadata) + is_success, error_data = opc_repository.write_data("ns=2;s=TestNode", 42.0, + "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( - 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'}}", - block="opc_repository", - level=NotificationLevel.ERROR, - attachment_content=ANY - ) - assert opc_repository.error_count == 1 + assert is_success is False + assert error_data['notification_id'] == f"OPC_WRITE_DATA_ERROR_{opc_repository.id}" + 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'}}" + assert error_data['block'] == "opc_repository" + assert error_data['level'] == NotificationLevel.ERROR + assert error_data['attachment_content'] is not None diff --git a/tests/laborious/utils/test_connectors_config.py b/tests/laborious/utils/test_connectors_config.py index a7b36aa..acb5c21 100644 --- a/tests/laborious/utils/test_connectors_config.py +++ b/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' + } diff --git a/values.yaml b/values.yaml index 61c5ea7..5d86e41 100644 --- a/values.yaml +++ b/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