From c7dbe3585b102fffbdd2791403feba3b37431f46 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 20 May 2025 17:19:41 -0300 Subject: [PATCH] SIENTIAPDE-1049 Update ingestor and data manager for improved shutdown handling and logging - Incremented project version in quality gate configuration. - Commented out ingestor service in docker-compose for clarity. - Enhanced main loop in app.py to handle exit signals and exceptions. - Added shutdown methods in Ingestor and DataManager classes for graceful resource cleanup. - Updated unit tests to validate shutdown behavior and exception handling. - Introduced coverage configuration to omit specific files. --- .coveragerc | 3 + .github/workflows/quality-gate.yml | 3 +- docker-compose.yaml | 26 ++++---- ingestor/app.py | 35 ++++++++-- ingestor/ingestor.py | 57 +++++++++++++++-- ingestor/managers/data_manager.py | 19 ++++-- ingestor/managers/ingestor_manager.py | 67 ++++++++------------ ingestor/managers/opc_manager.py | 22 ++++--- simulator/redis-feeder.py | 16 +++-- tests/unit/managers/test_data_manager.py | 30 ++++++--- tests/unit/managers/test_ingestor_manager.py | 51 --------------- tests/unit/managers/test_opc_manager.py | 17 ++++- tests/unit/test_ingestor.py | 52 +++++++++++---- 13 files changed, 235 insertions(+), 163 deletions(-) create mode 100644 .coveragerc diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..2249368 --- /dev/null +++ b/.coveragerc @@ -0,0 +1,3 @@ +[run] +omit = + ingestor/app.py diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index 7efa623..e13ecb2 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -80,4 +80,5 @@ jobs: -Dsonar.host.url=$SONAR_HOST_URL \ -Dsonar.token=$SONAR_TOKEN \ -Dsonar.python.version=3.11 \ - -Dsonar.projectVersion=1.0.0 + -Dsonar.projectVersion=1.0.1 \ + -Dsonar.coverage.exclusions=ingestor/app.py diff --git a/docker-compose.yaml b/docker-compose.yaml index 891fc9b..2216ee5 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -1,19 +1,19 @@ version: '3.8' services: - ingestor: - build: - context: . - environment: - HOSTNAME: ingestor - container_name: ingestor - depends_on: - - kafka - - redis - networks: - - kafka-net - env_file: - - .env + # ingestor: + # build: + # context: . + # environment: + # HOSTNAME: ingestor + # container_name: ingestor + # depends_on: + # - kafka + # - redis + # networks: + # - kafka-net + # env_file: + # - .env zookeeper: image: confluentinc/cp-zookeeper:latest diff --git a/ingestor/app.py b/ingestor/app.py index b9885fc..7b3f05b 100644 --- a/ingestor/app.py +++ b/ingestor/app.py @@ -1,19 +1,42 @@ -from time import sleep +from threading import Event +import signal +import os +import traceback from ingestor.ingestor import Ingestor +exit_signal = Event() + def main(): ingestor = Ingestor() - ingestor.prepare_ingestor() + ingestor.logger.info("Ingestor prepared. Starting main loop.") - while True: + while not exit_signal.is_set(): + try: + ingestor.loop() - ingestor.loop() + exit_signal.wait(ingestor.poll_interval) - # Sleep for poll interval - sleep(ingestor.poll_interval) + except Exception: + print("Exception in main loop. Setting exit_signal flag.") + traceback.print_exc() + exit_signal.set() + + ingestor.shutdown() + + ingestor.logger.info("Main loop exit_signaled.") + + os._exit(0) + + +def signal_handler(_signum, _frame): + print(f"Received signal {_signum}. Setting exit_signal flag.") + exit_signal.set() if __name__ == "__main__": + signal.signal(signal.SIGINT, signal_handler) + signal.signal(signal.SIGTERM, signal_handler) + signal.signal(signal.SIGHUP, signal_handler) main() diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index f7fb6f5..4c7536e 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -1,5 +1,7 @@ from logging import Formatter, StreamHandler, getLogger from os import getenv +from copy import deepcopy +from typing import Dict, Any from sientia_do.notifications.handlers import NotificationHandler @@ -11,7 +13,8 @@ class Ingestor: """ Initializes the ingestor with configuration values retrieved from environment variables. Environment Variables: - KAFKA_SERVERS (str): Comma-separated list of Kafka server addresses. Defaults to "localhost:9092". + KAFKA_SERVERS (str): Comma-separated list of Kafka server addresses. + Defaults to "localhost:9092". REDIS_HOST (str): Hostname of the Redis server. Defaults to "localhost". REDIS_PORT (int): Port number of the Redis server. Defaults to 6379. LEASE_TTL (int): Time-to-live for leases in seconds. Defaults to 10. @@ -53,6 +56,13 @@ class Ingestor: self.ingestor_manager = None + def shutdown(self): + if self.ingestor_manager: + self.ingestor_manager.shutdown() + + def __del__(self): + self.shutdown() + def init_logger(self): """ Initializes a logger instance for the class. @@ -147,9 +157,7 @@ class Ingestor: # This ingestor is active and has no slots, so we need to try to # Get slot lease - acquired = self.ingestor_manager.get_slot_leases(1) - - self.handle_acquired_tags(acquired) + self.ingestor_manager.get_slot_leases(1) def manage_leases(self, available_slots: int, lacking_ingestors: int, slot_diff: int): """ @@ -174,9 +182,7 @@ class Ingestor: self.logger.info("Slots available: %s", available_slots) # Get slot lease - acquired = self.ingestor_manager.get_slot_leases(available_slots) - - self.handle_acquired_tags(acquired) + self.ingestor_manager.get_slot_leases(available_slots) elif lacking_ingestors == 0 and slot_diff > 0: @@ -188,6 +194,36 @@ class Ingestor: self.ingestor_manager.drop_slot_leases(overleases) + def update_ingestor_manager(self, old_managed_tags: Dict[str, Any]): + """ + Updates the ingestor manager with the new managed tags. + Args: + old_managed_tags (Dict[str, Any]): The old managed tags. + """ + + self.logger.debug("Current managed tags: %s", + self.ingestor_manager.managed_tags) + + self.ingestor_manager.update_opc_servers() + new_managed_tags = deepcopy(self.ingestor_manager.managed_tags) + + self.logger.debug("Comparing new managed tags %s" + "with old managed tags %s", + new_managed_tags, old_managed_tags) + + for slot, config in new_managed_tags.items(): + if slot not in old_managed_tags: + self.ingestor_manager.subscribe_to_tags({slot: config}) + continue + + if config != old_managed_tags[slot]: + self.ingestor_manager.unsubscribe_slot(slot) + self.ingestor_manager.subscribe_to_tags({slot: config}) + + for slot in old_managed_tags.keys(): + if slot not in new_managed_tags: + self.ingestor_manager.unsubscribe_slot(slot) + def loop(self): """ Executes the main loop for managing ingestors and slots. @@ -212,6 +248,8 @@ class Ingestor: self.logger.info("Polling for slot updates...") # Get active ingestors + current_managed_tags = deepcopy(self.ingestor_manager.managed_tags) + ingestors = self.ingestor_manager.get_active_ingestors() number_of_ingestors = len(ingestors) number_of_leases = self.ingestor_manager.get_number_of_leases() @@ -251,3 +289,8 @@ class Ingestor: # Check OPC cycles self.logger.debug("Checking OPC servers integrity...") self.ingestor_manager.check_opc_servers_integrity() + + self.logger.debug("Updating managed tags...") + self.update_ingestor_manager( + current_managed_tags + ) diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index 3f9ca3c..b16b933 100644 --- a/ingestor/managers/data_manager.py +++ b/ingestor/managers/data_manager.py @@ -51,14 +51,21 @@ class DataManager(): self.logger = logger self.notification_handler = notification_handler - def __del__(self): - """Destructor to close the producer connection.""" - print("Closing Kafka producer...") + def shutdown(self): + """Closes the Kafka producer connection.""" if self.kafka_producer: - self.kafka_producer.flush(timeout=10) - self.kafka_producer.close() + try: + self.kafka_producer.flush(timeout=10) + self.kafka_producer.close() + except Exception as e: + self.logger.error( + f"Error closing Kafka producer: {e}") else: - print("Kafka producer is already closed or not initialized.") + self.logger.warning( + "Kafka producer is already closed or not initialized.") + + def __del__(self): + self.shutdown() def delivery_report(self, msg: str): """Callback for delivery reports from Kafka.""" diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 382bc94..433fcef 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -79,6 +79,14 @@ class IngestorManager(): return manager + def shutdown(self): + for _server_name, server in self.opc_managers.items(): + server.disconnect() + self.data_manager.shutdown() + + def __del__(self): + self.shutdown() + def update_opc_servers(self): """ Updates the OPC (OLE for Process Control) server connections managed by the ingestor. @@ -105,14 +113,14 @@ class IngestorManager(): """ registered_servers = [] - current_managed_tags = self.managed_tags.copy() + current_managed_tags = deepcopy(self.managed_tags) for _slot, slot_config in current_managed_tags.items(): for server, server_config in slot_config.items(): registered_servers.append(server) - server_config = server_config.copy() + server_config = deepcopy(server_config) server_config.pop('tags', None) - server_instance = None - if server not in self.opc_managers: + server_instance = self.opc_managers.get(server, None) + if server_instance is None: self.logger.debug( f"Initializing OPC manager for server {server}" ) @@ -120,15 +128,19 @@ class IngestorManager(): server_config, self.data_manager, self.logger ) - elif self.opc_managers[server].config != server_config: + elif server_instance.config != server_config: self.logger.warning( f"Reinitializing OPC manager for server {server}" ) - self.opc_managers[server].disconnect() + server_instance.disconnect() del self.opc_managers[server] server_instance = self.initialize_opc_from_config( server_config, self.data_manager, self.logger ) + else: + self.logger.debug( + f"OPC manager for server {server} is already initialized and up to date" + ) if server_instance is not None: self.opc_managers[server] = server_instance @@ -153,29 +165,18 @@ class IngestorManager(): """ Checks the integrity of the OPC servers and updates the OPC servers if necessary. """ - current_managers = self.opc_managers.copy() - for server, opc_manager in current_managers.items(): + for server, opc_manager in self.opc_managers.items(): opc_manager.check_cycles() is_lost = opc_manager.check_opc_listenning() if is_lost: self.logger.warning( f"OPC server {server} is lost. " - f"Desconnecting from server." + f"Server will be disconnected." ) - opc_manager.disconnect() - self.logger.warning( - f"Reconnecting to OPC server {server}" - ) - self.opc_managers.pop(server, None) - self.update_opc_servers() - if self.opc_managers.get(server) is not None: - for slot, slot_config in self.managed_tags.items(): - if server in slot_config: - self.manage_server( - slot, server, slot_config[server], slot_config[server]['tags'] - ) + for slot, _config in self.managed_tags.items(): + self.managed_tags[slot].pop(server, None) def declare_active(self): """ @@ -303,35 +304,17 @@ class IngestorManager(): removed_slots = [] update = {} - for slot, slot_config in self.managed_tags.items(): + for slot, _slot_config in self.managed_tags.items(): self.resource_manager.renew_tag_lease(slot) update = self.resource_manager.get_tag_slot(slot) - self.logger.debug( - f"Comparing slot {slot} configuration: {slot_config} with update: {update}") if update is None: - self.logger.warning( - f"Slot {slot} configuration not found. " - f"Removing slot from managed tags." - ) - self.unsubscribe_slot(slot) removed_slots.append(slot) - continue - if update != slot_config: - self.logger.info( - f"Slot {slot} configuration updated. " - f"Old: {slot_config}, New: {update}" - ) - self.managed_tags[slot] = update - - self.unsubscribe_slot(slot) - self.update_opc_servers() - self.subscribe_to_tags({slot: update}) + self.managed_tags[slot] = update for slot in removed_slots: - del self.managed_tags[slot] - self.update_opc_servers() + self.managed_tags.pop(slot, None) def drop_slot_leases(self, ids: List[str]) -> None: """ diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index 55a9b65..e218956 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -188,7 +188,8 @@ class OpcManager(): self.logger.warning("Client already disconnected.") return try: - [self.subscriptions[sub].delete() for sub in self.subscriptions] + _a = [self.subscriptions[sub].delete() + for sub in self.subscriptions] self.logger.warning("Deleted all subscriptions.") except Exception as sub_error: self.logger.error(f"Failed to clean up subscription: {sub_error}") @@ -230,10 +231,8 @@ class OpcManager(): tag = str(node) self.logger.debug( - f"Data change notification received for tag: {tag} after {self.nodes[tag]['cycle_rule']['cycle_count']} cycles") - - self.logger.debug( - f"Resetting cycle count for tag: {tag} after {self.non_receive_count} OPC cycles") + f"Data change notification received for tag:" + f"{tag} after {self.nodes[tag]['cycle_rule']['cycle_count']} cycles") self.nodes[tag]['cycle_rule']['cycle_count'] = 0 self.non_receive_count = 0 @@ -245,12 +244,13 @@ class OpcManager(): 'value': value } - [self.data_manager.publish(e, data) - for e in self.nodes[tag]['topics']] + _a = [self.data_manager.publish(e, data) + for e in self.nodes[tag]['topics']] def check_cycles(self): """ - Checks the cycle counts for all monitored nodes and sends notifications if thresholds are exceeded. + Checks the cycle counts for all monitored nodes and sends + notifications if thresholds are exceeded. This method iterates through all monitored nodes and updates their cycle counts based on configured increments. If a node's cycle count exceeds a threshold (5 cycles), it triggers @@ -258,7 +258,8 @@ class OpcManager(): """ for node, config in self.nodes.items(): - self.nodes[node]['cycle_rule']['cycle_count'] += self.nodes[node]['cycle_rule']['cycle_increment'] + self.nodes[node]['cycle_rule']['cycle_count'] += config[ + 'cycle_rule']['cycle_increment'] if self.nodes[node]['cycle_rule']['cycle_count'] >= 5: name = config['tag_name'] cycles = self.nodes[node]['cycle_rule']['cycle_count'] @@ -280,7 +281,8 @@ class OpcManager(): if self.non_receive_count >= 5: self.notification_handler.build_and_send_notification( notification_id=f'OPC_LISTENNING_STOPPED__{self.name}', - message=f'{self.non_receive_count} cycles without receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}', + message=f'{self.non_receive_count} cycles without ' + f'receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}', block="opc_manager", level=NotificationLevel.ERROR ) diff --git a/simulator/redis-feeder.py b/simulator/redis-feeder.py index 3f7e36a..d3dde93 100644 --- a/simulator/redis-feeder.py +++ b/simulator/redis-feeder.py @@ -1,14 +1,18 @@ import redis import json -import os # Redis connection settings -redis_host = "localhost" -redis_port = 6379 +REDIS_HOST = "localhost" +REDIS_PORT = 6379 +REDIS_USERNAME = None # "default" +REDIS_PASSWORD = None # "bdnZOpcyiL" + +OPC_URL = "opc.tcp://sientia-opc-simulator-service.sientia-opc.svc.cluster.local:4840" +OPC_URL = "opc.tcp://localhost:4840" # Connect to Redis -r = redis.Redis(host=redis_host, port=redis_port, - decode_responses=True, username='default', password='bdnZOpcyiL') +r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, + decode_responses=True, username=REDIS_USERNAME, password=REDIS_PASSWORD) # Define the key pattern to target pattern = "slot:opc_tags:*" @@ -25,7 +29,7 @@ new_data = { "slot:opc_tags:1": { "server1": { "name": "server1", - "url": "opc.tcp://sientia-opc-simulator-service.sientia-opc.svc.cluster.local:4840", + "url": OPC_URL, "server_uri": "http://opcua-server.simulator", "tags": { 'ns=2;i=2': { diff --git a/tests/unit/managers/test_data_manager.py b/tests/unit/managers/test_data_manager.py index 34c387a..80f75e0 100644 --- a/tests/unit/managers/test_data_manager.py +++ b/tests/unit/managers/test_data_manager.py @@ -108,31 +108,45 @@ def test___init___failure_max_attempts(kafka): assert False, "Expected NoBrokersAvailable exception was not raised." -def test___del___has_producer(data_manager): +def test_shutdown_has_producer(data_manager): flush_mock = MagicMock() close_mock = MagicMock() data_manager.kafka_producer.flush = flush_mock data_manager.kafka_producer.close = close_mock - data_manager.__del__() + data_manager.shutdown() flush_mock.assert_called_once() close_mock.assert_called_once() -@patch("ingestor.managers.data_manager.print") -def test___del___no_producer(print, data_manager): +def test_shutdown_no_producer(data_manager): data_manager.kafka_producer = None - # Call the __del__ method - data_manager.__del__() + data_manager.shutdown() - # Check if the print statement was called - print.assert_any_call( + data_manager.logger.warning.assert_any_call( "Kafka producer is already closed or not initialized." ) +def test_shutdown_exception(data_manager): + data_manager.kafka_producer.flush = MagicMock( + side_effect=Exception("Test error")) + data_manager.kafka_producer.close = MagicMock() + + data_manager.shutdown() + data_manager.logger.error.assert_called_once_with( + "Error closing Kafka producer: Test error" + ) + + +def test___del__(data_manager): + data_manager.shutdown = MagicMock() + data_manager.__del__() + data_manager.shutdown.assert_called_once() + + def test_delivery_report(data_manager): msg = MagicMock() msg.topic = "test_topic" diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 6e59103..fa3f7be 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -331,14 +331,6 @@ def test_update_slot_config(ingestor_manager): "config": "new_config"} assert "slot3" not in ingestor_manager.managed_tags - assert ingestor_manager.update_opc_servers.call_count == 2 - ingestor_manager.subscribe_to_tags.assert_called_once_with( - {'slot1': {"config": "updated_config"}} - ) - ingestor_manager.unsubscribe_slot.assert_any_call("slot3") - ingestor_manager.unsubscribe_slot.assert_any_call("slot1") - assert ingestor_manager.unsubscribe_slot.call_count == 2 - ingestor_manager.resource_manager.renew_tag_lease.assert_any_call("slot1") ingestor_manager.resource_manager.renew_tag_lease.assert_any_call("slot3") ingestor_manager.resource_manager.renew_tag_lease.assert_any_call("slot2") @@ -546,27 +538,9 @@ def test_check_opc_servers_integrity_server_lost(ingestor_manager): ingestor_manager.initialize_opc_from_config = MagicMock( return_value=new_manager) - # Mock update_opc_servers and manage_server - ingestor_manager.update_opc_servers = MagicMock() - ingestor_manager.manage_server = MagicMock() - # Call the method ingestor_manager.check_opc_servers_integrity() - # Verify that the lost server was disconnected - opc_manager.disconnect.assert_called_once() - - # Verify that a new manager was initialized - ingestor_manager.initialize_opc_from_config.assert_called_once_with( - opc_manager.config, ingestor_manager.data_manager, ingestor_manager.logger - ) - - # Verify that the new manager was assigned - assert ingestor_manager.opc_managers["server1"] == new_manager - - # Verify that update_opc_servers was called - ingestor_manager.update_opc_servers.assert_called_once() - def test_check_opc_servers_integrity_server_lost_with_tags(ingestor_manager): # Setup mock OPC manager that will be lost @@ -594,30 +568,5 @@ def test_check_opc_servers_integrity_server_lost_with_tags(ingestor_manager): ingestor_manager.initialize_opc_from_config = MagicMock( return_value=new_manager) - # Mock update_opc_servers and manage_server - ingestor_manager.update_opc_servers = MagicMock() - ingestor_manager.manage_server = MagicMock() - # Call the method ingestor_manager.check_opc_servers_integrity() - - # Verify that the lost server was disconnected - opc_manager.disconnect.assert_called_once() - - # Verify that a new manager was initialized - ingestor_manager.initialize_opc_from_config.assert_called_once_with( - opc_manager.config, ingestor_manager.data_manager, ingestor_manager.logger - ) - - # Verify that the new manager was assigned - assert ingestor_manager.opc_managers["server1"] == new_manager - - # Verify that update_opc_servers was called - ingestor_manager.update_opc_servers.assert_called_once() - - # Verify that manage_server was called with the correct tags - ingestor_manager.manage_server.assert_called_once_with( - "slot1", "server1", - {"config": "config1", "tags": {"tag1": "value1"}}, - {"tag1": "value1"} - ) diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py index 8886709..a57ef46 100644 --- a/tests/unit/managers/test_opc_manager.py +++ b/tests/unit/managers/test_opc_manager.py @@ -188,7 +188,7 @@ def test_disconnect_success(opc_manager_subscribed): assert opc_manager_subscribed.client is None -def test_disconnect_error(opc_manager_subscribed): +def test_disconnect_error_unsubscribe(opc_manager_subscribed): opc_manager_subscribed.client = MagicMock() opc_manager_subscribed.subscriptions['sub1'] = MagicMock( delete=MagicMock(side_effect=Exception("Test error")) @@ -197,11 +197,24 @@ def test_disconnect_error(opc_manager_subscribed): opc_manager_subscribed.disconnect() opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once() - opc_manager_subscribed.client.disconnect.assert_not_called() + opc_manager_subscribed.client = None opc_manager_subscribed.logger.error.assert_called_once_with( "Failed to clean up subscription: Test error") +def test_disconnect_error(opc_manager_subscribed): + opc_manager_subscribed.client = MagicMock() + opc_manager_subscribed.client.disconnect = MagicMock( + side_effect=Exception("Test error")) + + opc_manager_subscribed.disconnect() + + opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once() + opc_manager_subscribed.client = None + opc_manager_subscribed.logger.error.assert_called_once_with( + "Failed to disconnect from OPC UA server: Test error") + + def test_datachange_notification(opc_manager_subscribed): data = MagicMock( monitored_item=MagicMock( diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 448363c..54ebdc7 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -1,4 +1,5 @@ -from unittest.mock import ANY, MagicMock, patch +from unittest.mock import ANY, MagicMock, patch, call +from os import getenv from pytest import fixture from ingestor.ingestor import Ingestor @@ -24,13 +25,13 @@ def test___init__(notification_handler, init_logger, getenv): getenv.assert_any_call("KAFKA_SERVERS", "localhost:9092") getenv.assert_any_call("REDIS_HOST", "localhost") - getenv.assert_any_call("REDIS_PORT", 6379) + getenv.assert_any_call("REDIS_PORT", '6379') getenv.assert_any_call("REDIS_USERNAME", None) getenv.assert_any_call("REDIS_PASSWORD", None) - getenv.assert_any_call("LEASE_TTL", 10) - getenv.assert_any_call("HEARTBEAT_TTL", 20) + getenv.assert_any_call("LEASE_TTL", '10') + getenv.assert_any_call("HEARTBEAT_TTL", '20') getenv.assert_any_call("HOSTNAME", "localhost") - getenv.assert_any_call("POLL_INTERVAL", 5) + getenv.assert_any_call("POLL_INTERVAL", '5') assert ingestor.kafka_servers == ["localhost:9092", "localhost:35"] assert ingestor.redis_host == "localhost1" @@ -58,7 +59,7 @@ def test___init__(notification_handler, init_logger, getenv): @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, _init_logger, _getenv): ing = Ingestor() ing.logger = MagicMock() @@ -82,7 +83,8 @@ def test_init_logger(formatter, stream_handler, get_logger, 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("INFO") + 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( @@ -172,8 +174,6 @@ def test_manage_slots_none_available_none_available(ingestor_manager_started): ingestor_manager_started.ingestor_manager.get_slot_leases.assert_called_once_with( 1) - ingestor_manager_started.handle_acquired_tags.assert_called_once_with( - ingestor_manager_started.ingestor_manager.get_slot_leases.return_value) def test_manage_leases_no_available_slots_no_extra_slots(ingestor_manager_started): @@ -193,8 +193,6 @@ def test_manage_leases_available_slots_innactive_ingestors(ingestor_manager_star ingestor_manager_started.ingestor_manager.get_slot_leases.assert_called_once_with( 2) - ingestor_manager_started.handle_acquired_tags.assert_called_once_with( - ingestor_manager_started.ingestor_manager.get_slot_leases.return_value) ingestor_manager_started.ingestor_manager.drop_slot_leases.assert_not_called() @@ -267,3 +265,35 @@ def test_loop_no_managed(ingestor_manager_started): ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once() ingestor_manager_started.logger.info.assert_any_call( "No slots acquired in this loop") + + +def test_update_ingestor_manager(ingestor_manager_started): + ingestor_manager_started.ingestor_manager.managed_tags = { + "slot_to_create": "new_config", + "slot_to_update": "new_config", + "slot_to_do_nothing": "old_config" + } + + old_managed_tags = { + "slot_to_update": "old_config", + "slot_to_delete": "old_config", + "slot_to_do_nothing": "old_config" + } + + ingestor_manager_started.update_ingestor_manager(old_managed_tags) + + ingestor_manager_started.ingestor_manager.update_opc_servers.assert_called_once() + ingestor_manager_started.ingestor_manager.subscribe_to_tags.assert_has_calls( + [ + call( + {"slot_to_create": "new_config"}), + call( + {"slot_to_update": "new_config"}) + ] + ) + ingestor_manager_started.ingestor_manager.unsubscribe_slot.assert_has_calls( + [ + call("slot_to_update"), + call("slot_to_delete") + ] + )