diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 18dc49c..9614302 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -70,10 +70,10 @@ class Ingestor: ) self.metadata = { - 'schedule_name': 'OPC_INGESTOR', - 'trigger': 'INGESTOR', + 'model_id': '-', 'model_name': '-', - 'model_id': '-' + 'workflow_name': 'OPC_INGESTOR', + 'schema_name': 'OPC_INGESTOR', } self.ingestor_manager = None diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index 9fd50f6..68dbe06 100644 --- a/ingestor/managers/data_manager.py +++ b/ingestor/managers/data_manager.py @@ -87,6 +87,8 @@ class DataManager(BaseActivity): self.mongo_client = MongoClient(self.connection_string) self.mongo_client.server_info() + self.metadata = metadata + self.mongo_db = self.mongo_client[self.database] logger.info( diff --git a/tests/unit/managers/test_data_manager.py b/tests/unit/managers/test_data_manager.py index da78e26..ed157cc 100644 --- a/tests/unit/managers/test_data_manager.py +++ b/tests/unit/managers/test_data_manager.py @@ -4,20 +4,35 @@ from kafka.errors import NoBrokersAvailable from sientia_do.notifications.models import NotificationLevel from ingestor.managers.data_manager import DataManager +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + @fixture @patch("ingestor.managers.data_manager.KafkaProducer") @patch("ingestor.managers.data_manager.MongoClient") def data_manager(mongo, kafka): - return DataManager( + + data_manager = DataManager( kafka_servers="localhost:9092", mongo_connection_string="mongodb://localhost:27017", mongo_database="sientia", export_to_kafka=True, logger=MagicMock(), - notification_handler=MagicMock() + notification_handler=MagicMock(), + metadata=metadata["metadata"], ) + data_manager.send_notification = MagicMock() + + return data_manager + @patch("ingestor.managers.data_manager.KafkaProducer") @patch("ingestor.managers.data_manager.MongoClient") @@ -25,6 +40,7 @@ def test___init___success(mongo, kafka): logger_mock = MagicMock() data_manager = DataManager( + metadata=metadata["metadata"], kafka_servers="localhost:9092", mongo_connection_string="mongodb://localhost:27017", mongo_database="sientia", @@ -61,7 +77,8 @@ def test___init___second_attempt(mongo, kafka): mongo_database="sientia", export_to_kafka=True, logger=logger_mock, - notification_handler=MagicMock() + notification_handler=MagicMock(), + metadata=metadata["metadata"], ) kafka.assert_any_call( @@ -99,7 +116,8 @@ def test___init___failure_max_attempts(mongo, kafka): mongo_database="sientia", export_to_kafka=True, logger=logger_mock, - notification_handler=MagicMock() + notification_handler=MagicMock(), + metadata=metadata["metadata"], ) except NoBrokersAvailable as e: assert str( @@ -255,12 +273,13 @@ def test_publish_error(traceback, data_manager): ) # Check if the error was logged - data_manager.notification_handler.build_and_send_notification.assert_called_once_with( + data_manager.send_notification.assert_called_once_with( notification_id=f"KAFKA_PRODUCER_ERROR_{topic}", message=f"Error publishing message to topic {topic}: Test error", block="kafka_producer", level=NotificationLevel.ERROR, - attachment_content=traceback.format_exc.return_value + attachment_content=traceback.format_exc.return_value, + metadata=metadata["metadata"] ) @@ -271,10 +290,11 @@ def test_publish_error_mongo(data_manager): data_manager.publish("test_topic", {"key": "value"}) - data_manager.notification_handler.build_and_send_notification.assert_called_once_with( + data_manager.send_notification.assert_called_once_with( notification_id="MONGO_PRODUCER_ERROR_test_topic", message="Error inserting message to MongoDB: Test error", block="mongo_producer", level=NotificationLevel.ERROR, - attachment_content=ANY + attachment_content=ANY, + metadata=metadata["metadata"] ) diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 9d27c55..f6ace9f 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -3,12 +3,21 @@ from pytest import fixture from sientia_do.notifications.models import NotificationLevel from ingestor.managers.ingestor_manager import IngestorManager +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + @fixture @patch('ingestor.managers.ingestor_manager.DataManager') @patch('ingestor.managers.ingestor_manager.ResourceManager') def ingestor_manager(data_manager_mock, resource_manager_mock): - return IngestorManager( + ingestor = IngestorManager( kafka_servers="localhost:9092", redis_data={ "host": "localhost", @@ -22,9 +31,14 @@ def ingestor_manager(data_manager_mock, resource_manager_mock): mongo_database="sientia", export_to_kafka=False, logger=MagicMock(), - notification_handler=MagicMock() + notification_handler=MagicMock(), + metadata=metadata["metadata"], ) + ingestor.send_notification = MagicMock() + + return ingestor + @patch('ingestor.managers.ingestor_manager.OpcManager') @patch('ingestor.managers.ingestor_manager.DataManager') @@ -46,15 +60,32 @@ def test___init__(notification_handler_mock, resource_manager_mock, data_manager mongo_database="sientia", export_to_kafka=False, logger=MagicMock(), - notification_handler=MagicMock() + notification_handler=MagicMock(), + metadata=metadata["metadata"], ) opc_manager_mock.assert_not_called() data_manager_mock.assert_called_once_with( - "localhost:9092", "mongodb://localhost:27017", "sientia", False, - ingestor.logger, ingestor.notification_handler) + kafka_servers="localhost:9092", + mongo_connection_string="mongodb://localhost:27017", + mongo_database="sientia", + export_to_kafka=False, + metadata=metadata["metadata"], + logger=ingestor.logger, + notification_handler=ingestor.notification_handler, + ) resource_manager_mock.assert_called_once_with( - "localhost", 6379, 60, 60, "test_pod", None, None) + host="localhost", + port=6379, + lease_ttl=60, + heartbeat_ttl=60, + pod_id="test_pod", + metadata=metadata["metadata"], + logger=ingestor.logger, + notification_handler=ingestor.notification_handler, + username=None, + password=None, + ) assert ingestor.poll_interval == 5 assert ingestor.managed_tags == {} assert ingestor.opc_servers == {} @@ -77,14 +108,20 @@ def test_initialize_opc_from_config(opc_manager, ingestor_manager): opc_manager.return_value = MagicMock() result = ingestor_manager.initialize_opc_from_config( - server_config, ingestor_manager.data_manager, ingestor_manager.logger) + server_config) opc_manager.assert_called_once_with( - server_config['name'], server_config['url'], ingestor_manager.data_manager, ingestor_manager.logger, - server_config['server_uri'], ingestor_manager.notification_handler, - server_config['pod_id'], - server_config['cert_path'], server_config['private_key_path'], - server_config['server_cert_path'] + name=server_config['name'], + url=server_config['url'], + data_manager=ingestor_manager.data_manager, + logger=ingestor_manager.logger, + server_uri=server_config['server_uri'], + notification_handler=ingestor_manager.notification_handler, + pod_id=server_config['pod_id'], + cert_path=server_config['cert_path'], + private_key_path=server_config['private_key_path'], + server_cert_path=server_config['server_cert_path'], + metadata=metadata["metadata"], ) assert result == opc_manager.return_value @@ -107,12 +144,13 @@ def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, inges opc_manager.side_effect = Exception("Initialization error") result = ingestor_manager.initialize_opc_from_config( - server_config, ingestor_manager.data_manager, ingestor_manager.logger) + server_config) assert result is None traceback_mock.format_exc.assert_called_once() - ingestor_manager.notification_handler.build_and_send_notification.assert_called_once_with( + ingestor_manager.send_notification.assert_called_once_with( + metadata=metadata["metadata"], notification_id=f'OPC_CONNECTION_ERROR_{server_config["name"]}', message='Error initializing OPC manager: Initialization error', block="opc_manager", @@ -131,7 +169,7 @@ def test_update_opc_servers(metrics, opc_manager, ingestor_manager): manager3 = MagicMock( config={"config": "config3"}) - def mock_initialize_from_config(config, data_manager, logger): + def mock_initialize_from_config(config): if config == {"config": "config1"}: return manager1 elif config == {"config": "config2"}: @@ -169,11 +207,11 @@ def test_update_opc_servers(metrics, opc_manager, ingestor_manager): assert len(ingestor_manager.opc_managers) == 3 ingestor_manager.initialize_opc_from_config.assert_any_call( - {"config": "config1"}, ingestor_manager.data_manager, ingestor_manager.logger) + {"config": "config1"}) ingestor_manager.initialize_opc_from_config.assert_any_call( - {"config": "config2"}, ingestor_manager.data_manager, ingestor_manager.logger) + {"config": "config2"}) ingestor_manager.initialize_opc_from_config.assert_any_call( - {"config": "config5"}, ingestor_manager.data_manager, ingestor_manager.logger) + {"config": "config5"}) assert ingestor_manager.initialize_opc_from_config.call_count == 3 assert ingestor_manager.opc_managers['server1'].config == { @@ -470,7 +508,8 @@ def test_manage_server_subscribe_failure(traceback_mock, ingestor_manager): traceback_mock.format_exc.assert_called_once() - ingestor_manager.notification_handler.build_and_send_notification.assert_called_once_with( + ingestor_manager.send_notification.assert_called_once_with( + metadata=metadata["metadata"], notification_id='OPC_SUBSCRIPTION_ERROR_slot1:server1', message='Failed to subscribe to tags from slot1:server1\n{\'tags\': \'config1\'}: Subscription error', block="opc_manager", diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py index f9be3dc..386f2ae 100644 --- a/tests/unit/managers/test_opc_manager.py +++ b/tests/unit/managers/test_opc_manager.py @@ -33,6 +33,15 @@ tags = { }, } +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + @fixture def raw_opc_manager(): @@ -44,6 +53,7 @@ def raw_opc_manager(): "opc.tcp://localhost:4840", MagicMock(), "localhost", + metadata=metadata["metadata"], ) @@ -53,6 +63,7 @@ def opc_manager(raw_opc_manager): raw_opc_manager.cert_path = "cert.pem" raw_opc_manager.private_key_path = "private_key.pem" raw_opc_manager.server_cert_path = "server_cert.pem" + raw_opc_manager.send_notification = MagicMock() return raw_opc_manager @@ -414,7 +425,6 @@ def test_check_cycles_no_notification(opc_manager): "cycle_rule": {"cycle_increment": 1.0, "cycle_count": 3.0}, } } - opc_manager.notification_handler.build_and_send_notification = MagicMock() opc_manager.check_cycles() @@ -422,7 +432,7 @@ def test_check_cycles_no_notification(opc_manager): assert opc_manager.nodes["ns=3;i=1001"]["cycle_rule"][ "cycle_count" ] == pytest.approx(4.0) - opc_manager.notification_handler.build_and_send_notification.assert_not_called() + opc_manager.send_notification.assert_not_called() def test_check_cycles_triggers_notification(opc_manager): @@ -441,11 +451,12 @@ def test_check_cycles_triggers_notification(opc_manager): assert opc_manager.nodes["ns=3;i=1001"]["cycle_rule"][ "cycle_count" ] == pytest.approx(5.5) - opc_manager.notification_handler.build_and_send_notification.assert_called_once_with( + opc_manager.send_notification.assert_called_once_with( notification_id="TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED", message="5.5 cycles without receive from ns=3;i=1001:Counter", block="opc_manager", level=NotificationLevel.WARNING, + metadata=metadata["metadata"], ) @@ -473,16 +484,16 @@ def test_check_opc_listenning_no_notification(metrics, opc_manager): def test_check_opc_listenning_warning_notification(opc_manager): opc_manager.non_receive_count = 4 - opc_manager.notification_handler.build_and_send_notification = MagicMock() result = opc_manager.check_opc_listenning() assert opc_manager.non_receive_count == 5 - opc_manager.notification_handler.build_and_send_notification.assert_called_once_with( + opc_manager.send_notification.assert_called_once_with( notification_id=f"OPC_LISTENNING_STOPPED__{opc_manager.name}", message=f"5 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}", block="opc_manager", level=NotificationLevel.ERROR, + metadata=metadata["metadata"], ) assert result is False @@ -498,14 +509,15 @@ def test_check_opc_listenning_error_notification_and_retry(metrics, opc_manager) assert opc_manager.non_receive_count == 15 # Should be called twice: once for 5, once for 15 - assert opc_manager.notification_handler.build_and_send_notification.call_count == 2 - calls = opc_manager.notification_handler.build_and_send_notification.call_args_list + assert opc_manager.send_notification.call_count == 2 + calls = opc_manager.send_notification.call_args_list # First call: 5 cycles warning assert calls[0].kwargs == dict( notification_id=f"OPC_LISTENNING_STOPPED__{opc_manager.name}", message=f"15 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}", block="opc_manager", level=NotificationLevel.ERROR, + metadata=metadata["metadata"], ) # Second call: 15 cycles retry assert calls[1].kwargs == dict( @@ -513,6 +525,7 @@ def test_check_opc_listenning_error_notification_and_retry(metrics, opc_manager) message=f"Retrying to connect to server {opc_manager.name}", block="opc_manager", level=NotificationLevel.ERROR, + metadata=metadata["metadata"], ) assert result is True @@ -540,6 +553,7 @@ def test_init_metrics_calls_correct_metric_methods(mock_metrics): server_uri="opc.tcp://init.test:4840/uri", notification_handler=MagicMock(), pod_id="init_pod_localhost", + metadata=metadata["metadata"], ) mock_metrics.OPC_CONNECTION_STATUS.labels.assert_called_once_with( diff --git a/tests/unit/managers/test_resource_manager.py b/tests/unit/managers/test_resource_manager.py index 19ca732..0e528a2 100644 --- a/tests/unit/managers/test_resource_manager.py +++ b/tests/unit/managers/test_resource_manager.py @@ -1,13 +1,36 @@ from unittest.mock import MagicMock, patch from pytest import fixture, raises +from sientia_do.notifications.models import NotificationLevel from ingestor.managers.resource_manager import ResourceManager +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + @fixture @patch("ingestor.managers.resource_manager.Redis") def resource_manager(redis): - return ResourceManager("localhost", 6379, 10, 10, "pod_id") + resource_manager = ResourceManager( + host="localhost", + port=6379, + lease_ttl=10, + heartbeat_ttl=10, + pod_id="pod_id", + metadata=metadata["metadata"], + logger=MagicMock(), + notification_handler=MagicMock(), + ) + + resource_manager.send_notification = MagicMock() + + return resource_manager def test_get_success(resource_manager): @@ -54,7 +77,8 @@ def test_renew_tag_lease_success(resource_manager): result = resource_manager.renew_tag_lease("tag_id") assert result is True resource_manager.redis.get.assert_called_once_with("lease:opc_tags:tag_id") - resource_manager.redis.expire.assert_called_once_with("lease:opc_tags:tag_id", 10) + resource_manager.redis.expire.assert_called_once_with( + "lease:opc_tags:tag_id", 10) def test_renew_tag_lease_failure(resource_manager): @@ -69,7 +93,8 @@ def test_renew_tag_lease_failure(resource_manager): def test_drop_tag_lease(resource_manager): resource_manager.redis.delete.return_value = True resource_manager.drop_tag_lease("tag_id") - resource_manager.redis.delete.assert_called_once_with("lease:opc_tags:tag_id") + resource_manager.redis.delete.assert_called_once_with( + "lease:opc_tags:tag_id") def test_get_all_ingestors(resource_manager): @@ -106,7 +131,16 @@ def test_init_connection_failure(monkeypatch): mock_metrics.labels.return_value = mock_status with raises(Exception, match="Connection failed"): - ResourceManager("localhost", 6379, 10, 10, "pod_id") + ResourceManager( + host="localhost", + port=6379, + lease_ttl=10, + heartbeat_ttl=10, + pod_id="pod_id", + metadata=metadata["metadata"], + logger=MagicMock(), + notification_handler=MagicMock(), + ) mock_metrics.labels.assert_called_once_with(pod_id="pod_id") mock_status.set.assert_called_once_with(0) @@ -118,7 +152,8 @@ def test_init_ping_failure(monkeypatch): mock_redis_instance.ping.side_effect = Exception("Ping failed") mock_redis_class = MagicMock(return_value=mock_redis_instance) - monkeypatch.setattr("ingestor.managers.resource_manager.Redis", mock_redis_class) + monkeypatch.setattr( + "ingestor.managers.resource_manager.Redis", mock_redis_class) # Test that the exception is raised and metrics are set properly with patch("ingestor.metrics.REDIS_CONNECTION_STATUS") as mock_metrics: @@ -126,7 +161,16 @@ def test_init_ping_failure(monkeypatch): mock_metrics.labels.return_value = mock_status with raises(Exception, match="Ping failed"): - ResourceManager("localhost", 6379, 10, 10, "pod_id") + ResourceManager( + host="localhost", + port=6379, + lease_ttl=10, + heartbeat_ttl=10, + pod_id="pod_id", + metadata=metadata["metadata"], + logger=MagicMock(), + notification_handler=MagicMock(), + ) mock_metrics.labels.assert_called_once_with(pod_id="pod_id") mock_status.set.assert_called_once_with(0) @@ -172,19 +216,23 @@ def test_execute_redis_op_exception(resource_manager): with patch("ingestor.managers.resource_manager.time", return_value=100): with patch("ingestor.metrics.REDIS_OPERATIONS_ERRORS") as mock_errors: - with patch("builtins.print") as mock_print: - mock_errors_labels = MagicMock() - mock_errors.labels.return_value = mock_errors_labels + mock_errors_labels = MagicMock() + mock_errors.labels.return_value = mock_errors_labels - # Execute the operation and expect an exception - with raises(Exception, match="Operation failed"): - resource_manager._execute_redis_op("test_op", mock_func, "arg1") + # Execute the operation and expect an exception + with raises(Exception, match="Operation failed"): + resource_manager._execute_redis_op( + "test_op", mock_func, "arg1") - # Verify metrics and error handling - mock_errors.labels.assert_called_once_with( - pod_id="pod_id", operation="test_op" - ) - mock_errors_labels.inc.assert_called_once() - mock_print.assert_called_once_with( - "Error in Redis operation 'test_op': Operation failed" - ) + # Verify metrics and error handling + mock_errors.labels.assert_called_once_with( + pod_id="pod_id", operation="test_op" + ) + mock_errors_labels.inc.assert_called_once() + resource_manager.send_notification.assert_called_once_with( + metadata=metadata["metadata"], + notification_id="REDIS_OPERATION_ERROR_test_op", + message="Error in Redis operation 'test_op': Operation failed", + block="redis_manager", + level=NotificationLevel.ERROR, + ) diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 189c0a6..2f0a42f 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -6,9 +6,8 @@ from ingestor.ingestor import Ingestor @patch("ingestor.ingestor.getenv") -@patch("ingestor.ingestor.Ingestor.init_logger") @patch("ingestor.ingestor.NotificationHandler") -def test___init__(notification_handler, init_logger, getenv): +def test___init__(notification_handler, getenv): getenv.side_effect = [ "localhost:9092,localhost:35", # KAFKA_SERVERS "true", # EXPORT_TO_KAFKA @@ -20,7 +19,7 @@ def test___init__(notification_handler, init_logger, getenv): '200', # HEARTBEAT_TTL "localhost1", # HOSTNAME '50', # POLL_INTERVAL - "mongodb://localhost:27017", # MONGODB_URL + "localhost:27017", # MONGODB_URL "sientia", # MONGODB_USERNAME "sientia", # MONGODB_PASSWORD "sientia" # MONGODB_DATABASE @@ -47,10 +46,16 @@ def test___init__(notification_handler, init_logger, getenv): assert ingestor.heartbeat_ttl == 200 assert ingestor.pod_id == "localhost1" assert ingestor.poll_interval == 50 + assert ingestor.metadata == { + "model_id": "-", + "model_name": "-", + "workflow_name": "OPC_INGESTOR", + "schema_name": "OPC_INGESTOR", + } - init_logger.assert_called_once() notification_handler.assert_called_once_with( - servers=["localhost:9092", "localhost:35"], + connection_string="mongodb://sientia:sientia@localhost:27017", + database="sientia", logger=ingestor.logger, project_name="OPC_INGESTOR" ) @@ -58,9 +63,8 @@ def test___init__(notification_handler, init_logger, getenv): @fixture @patch("ingestor.ingestor.getenv") -@patch("ingestor.ingestor.Ingestor.init_logger") @patch("ingestor.ingestor.NotificationHandler") -def ingestor(_notification_handler, _init_logger, _getenv): +def ingestor(_notification_handler, _getenv): ing = Ingestor() ing.logger = MagicMock() @@ -73,25 +77,6 @@ def ingestor_manager_started(ingestor): return ingestor -@patch("ingestor.ingestor.getLogger") -@patch("ingestor.ingestor.StreamHandler") -@patch("ingestor.ingestor.Formatter") -def test_init_logger(formatter, stream_handler, get_logger, ingestor): - ingestor.logger = None - ingestor.init_logger() - - get_logger.assert_called_once_with('ingestor.ingestor') - stream_handler.assert_called_once() - formatter.assert_called_once_with( - '%(asctime)s - %(name)s - %(levelname)s - %(message)s') - ingestor.logger.setLevel.assert_called_once_with( - getenv("LOG_LEVEL", "INFO")) - ingestor.logger.addHandler.assert_called_once_with( - stream_handler.return_value) - stream_handler.return_value.setFormatter.assert_called_once_with( - formatter.return_value) - - def test_handle_acquired_tags_not_acquired(ingestor_manager_started): ingestor_manager_started.handle_acquired_tags([]) @@ -118,22 +103,23 @@ def test_prepare_ingestor(ingestor_manager_mock, ingestor): ingestor.prepare_ingestor() ingestor_manager_mock.assert_called_once_with( - ingestor.kafka_servers, - { + kafka_servers=ingestor.kafka_servers, + redis_data={ "host": ingestor.redis_host, "port": ingestor.redis_port, "username": ingestor.redis_username, "password": ingestor.redis_password }, - ingestor.lease_ttl, - ingestor.heartbeat_ttl, - ingestor.pod_id, - ingestor.poll_interval, - ingestor.mongo_connection_string, - ingestor.mongo_database, - ingestor.logger, - ingestor.notification_handler, - ingestor.export_to_kafka + lease_ttl=ingestor.lease_ttl, + heartbeat_ttl=ingestor.heartbeat_ttl, + pod_id=ingestor.pod_id, + poll_interval=ingestor.poll_interval, + mongo_connection_string=ingestor.mongo_connection_string, + mongo_database=ingestor.mongo_database, + metadata=ingestor.metadata, + logger=ingestor.logger, + notification_handler=ingestor.notification_handler, + export_to_kafka=ingestor.export_to_kafka, ) ingestor_manager.declare_active.assert_called_once() ingestor_manager.get_slot_leases.assert_called_once()