From d4aa44d77431dde7dbb071f7fbef1f97c5a6574f Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 26 Aug 2025 16:27:35 -0300 Subject: [PATCH 1/6] SIENTIAPDE-1205 Refactor Ingestor and OPC Manager for Asynchronous Operations - Updated main function to be asynchronous and integrated asyncio for better concurrency. - Refactored Ingestor methods to support async operations, including prepare_ingestor, loop, and shutdown. - Enhanced IngestorManager and OpcManager with async methods for improved performance and responsiveness. - Replaced blocking calls with await statements to ensure non-blocking behavior during operations. - Added a new run_async_main function to handle the async event loop setup. --- ingestor/app.py | 30 ++++++++++---- ingestor/ingestor.py | 37 ++++++++--------- ingestor/managers/ingestor_manager.py | 35 ++++++++-------- ingestor/managers/opc_manager.py | 58 +++++++++++++++------------ run_local.sh | 18 +++++++++ 5 files changed, 111 insertions(+), 67 deletions(-) create mode 100755 run_local.sh diff --git a/ingestor/app.py b/ingestor/app.py index 8fc3672..825b35e 100644 --- a/ingestor/app.py +++ b/ingestor/app.py @@ -1,4 +1,5 @@ import os +import asyncio import signal import traceback from threading import Event @@ -13,11 +14,11 @@ exit_signal = Event() POD_ID = os.getenv("HOSTNAME", "localhost") -def main(): +async def main(): start_prometheus_server() ingestor = Ingestor() try: - ingestor.prepare_ingestor() + await ingestor.prepare_ingestor() except Exception as e: metrics.APP_ERRORS_TOTAL.labels( pod_id=POD_ID).inc() # Increment errors @@ -28,11 +29,12 @@ def main(): while not exit_signal.is_set(): start_time = time() # Start loop timer try: - ingestor.loop() + await ingestor.loop() metrics.APP_LOOP_COUNT.labels( pod_id=POD_ID).inc() # Increment loop counter - exit_signal.wait(ingestor.poll_interval) + # Use asyncio.sleep instead of exit_signal.wait for better async compatibility + await asyncio.sleep(ingestor.poll_interval) except KeyboardInterrupt: # Handle Ctrl+C gracefully print("KeyboardInterrupt received. Setting exit_signal flag.") @@ -48,13 +50,13 @@ def main(): duration = time() - start_time metrics.APP_LOOP_DURATION.labels(pod_id=POD_ID).observe(duration) - ingestor.shutdown() + await ingestor.shutdown() metrics.APP_UP.labels(pod_id=POD_ID).set(0) # Mark app as DOWN ingestor.logger.info("Main loop exit_signaled.") # Give Prometheus a chance to scrape one last time before exiting (optional) - sleep(5) + await asyncio.sleep(5) os._exit(0) @@ -75,8 +77,22 @@ def start_prometheus_server(): os._exit(1) +def run_async_main(): + """Run the async main function with proper event loop setup""" + try: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + loop.run_until_complete(main()) + except KeyboardInterrupt: + print("KeyboardInterrupt received in main thread.") + exit_signal.set() + finally: + loop.close() + + if __name__ == "__main__": signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) signal.signal(signal.SIGHUP, signal_handler) - main() + + run_async_main() diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 0d68d2f..55d7543 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -1,3 +1,4 @@ +import asyncio from os import getenv from copy import deepcopy from typing import Dict, Any @@ -78,14 +79,14 @@ class Ingestor: } self.ingestor_manager = None - def shutdown(self): + async def shutdown(self): if self.ingestor_manager: - self.ingestor_manager.shutdown() + await self.ingestor_manager.shutdown() def __del__(self): - self.shutdown() + asyncio.run(self.shutdown()) - def handle_acquired_tags(self, acquired): + async def handle_acquired_tags(self, acquired): """ Handles the acquired tags by subscribing to them if available. This method checks if there are any acquired tags. If no tags are acquired, @@ -103,9 +104,9 @@ class Ingestor: else: # Subscribe to acquired slots self.ingestor_manager.update_opc_servers() - self.ingestor_manager.subscribe_to_tags(acquired) + await self.ingestor_manager.subscribe_to_tags(acquired) - def prepare_ingestor(self): + async def prepare_ingestor(self): """ Prepares the ingestor by initializing the IngestorManager, declaring the ingestor as active, acquiring slot leases, and handling the acquired tags. @@ -153,7 +154,7 @@ class Ingestor: acquired = self.ingestor_manager.get_slot_leases() self.logger.info(f"Acquired slots: {acquired}") - self.handle_acquired_tags(acquired) + await self.handle_acquired_tags(acquired) metrics.SLOTS_MANAGED.labels(pod_id=self.pod_id).set( len(self.ingestor_manager.managed_tags) @@ -176,7 +177,7 @@ class Ingestor: # Get slot lease self.ingestor_manager.get_slot_leases(1) - def manage_leases( + async def manage_leases( self, available_slots: int, lacking_ingestors: int, slot_diff: int ): """ @@ -214,7 +215,7 @@ class Ingestor: self.ingestor_manager.drop_slot_leases(overleases) for lease in overleases: - self.ingestor_manager.unsubscribe_slot(lease) + await self.ingestor_manager.unsubscribe_slot(lease) self.ingestor_manager.managed_tags.pop(lease) # Update metric after removal @@ -222,7 +223,7 @@ class Ingestor: len(self.ingestor_manager.managed_tags) ) - def update_ingestor_manager(self, old_managed_tags: Dict[str, Any]): + async def update_ingestor_manager(self, old_managed_tags: Dict[str, Any]): """ Updates the ingestor manager with the new managed tags. Args: @@ -232,7 +233,7 @@ class Ingestor: self.logger.debug( f"Current managed tags: {self.ingestor_manager.managed_tags}") - self.ingestor_manager.update_opc_servers() + await self.ingestor_manager.update_opc_servers() new_managed_tags = deepcopy(self.ingestor_manager.managed_tags) self.logger.debug( @@ -249,25 +250,25 @@ class Ingestor: for slot, config in new_managed_tags.items(): if slot not in old_managed_tags: self.logger.info(f"Subscribing to new slot {slot}") - self.ingestor_manager.subscribe_to_tags({slot: config}) + await self.ingestor_manager.subscribe_to_tags({slot: config}) continue if config != old_managed_tags[slot]: self.logger.info(f"Resubscribing to slot {slot}") - self.ingestor_manager.unsubscribe_slot(slot) - self.ingestor_manager.subscribe_to_tags({slot: config}) + await self.ingestor_manager.unsubscribe_slot(slot) + await self.ingestor_manager.subscribe_to_tags({slot: config}) for slot in old_managed_tags.keys(): if slot not in new_managed_tags: self.logger.info(f"Unsubscribing from slot {slot}") - self.ingestor_manager.unsubscribe_slot(slot) + await self.ingestor_manager.unsubscribe_slot(slot) # Ensure the gauge is updated after any potential changes here metrics.SLOTS_MANAGED.labels(pod_id=self.pod_id).set( len(self.ingestor_manager.managed_tags) ) - def loop(self): + async def loop(self): """ Executes the main loop for managing ingestors and slots. This method performs the following tasks: @@ -310,7 +311,7 @@ class Ingestor: slot_diff = len(self.ingestor_manager.managed_tags) - 1 self.logger.info("Managing leases...") - self.manage_leases(available_slots, lacking_ingestors, slot_diff) + await self.manage_leases(available_slots, lacking_ingestors, slot_diff) # Update managed slots gauge metrics.SLOTS_MANAGED.labels(pod_id=self.pod_id).set( @@ -337,4 +338,4 @@ class Ingestor: self.ingestor_manager.check_opc_servers_integrity() self.logger.info("Updating managed tags...") - self.update_ingestor_manager(current_managed_tags) + await self.update_ingestor_manager(current_managed_tags) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 860321a..70b0451 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -1,3 +1,4 @@ +import asyncio import traceback from typing import Dict, List from copy import deepcopy @@ -57,7 +58,7 @@ class IngestorManager(BaseActivity): notification_handler=notification_handler, set_error_counter=True) - def initialize_opc_from_config(self, server_config: dict) -> OpcManager | None: + async def initialize_opc_from_config(self, server_config: dict) -> OpcManager | None: """ Initializes an OPC Manager instance using the provided server configuration. Args: @@ -93,7 +94,7 @@ class IngestorManager(BaseActivity): ) manager.config = server_config - manager.connect() + await manager.connect() except Exception as e: trace = traceback.format_exc() @@ -111,15 +112,15 @@ class IngestorManager(BaseActivity): return manager - def shutdown(self): + async def shutdown(self): for _server_name, server in self.opc_managers.items(): - server.disconnect() + await server.disconnect() self.data_manager.shutdown() def __del__(self): - self.shutdown() + asyncio.run(self.shutdown()) - def update_opc_servers(self): + async def update_opc_servers(self): """ Updates the OPC (OLE for Process Control) server connections managed by the ingestor. This method ensures that the OPC servers defined in `self.managed_tags` are properly @@ -156,7 +157,7 @@ class IngestorManager(BaseActivity): self.logger.info( f"Initializing OPC manager for server {server}" ) - server_instance = self.initialize_opc_from_config( + server_instance = await self.initialize_opc_from_config( server_config ) @@ -166,7 +167,7 @@ class IngestorManager(BaseActivity): ) server_instance.disconnect() del self.opc_managers[server] - server_instance = self.initialize_opc_from_config( + server_instance = await self.initialize_opc_from_config( server_config ) else: @@ -191,7 +192,7 @@ class IngestorManager(BaseActivity): f"Server {server} not found in managed tags. " f"Desconnecting from server." ) - self.opc_managers[server].disconnect() + await self.opc_managers[server].disconnect() self.opc_managers.pop(server, None) metrics.OPC_MANAGERS_ACTIVE.labels( @@ -312,7 +313,7 @@ class IngestorManager(BaseActivity): pod_id=self.pod_id).set(len(self.managed_tags)) return acquired - def unsubscribe_slot(self, slot: str): + async def unsubscribe_slot(self, slot: str): """ Unsubscribes a specific slot from all associated OPC servers. Args: @@ -323,7 +324,7 @@ class IngestorManager(BaseActivity): for server in self.managed_tags[slot].keys(): if server in self.opc_managers: - self.opc_managers[server].unsubscribe(slot) + await self.opc_managers[server].unsubscribe(slot) def update_slot_config(self): """ @@ -378,7 +379,7 @@ class IngestorManager(BaseActivity): self.resource_manager.drop_tag_lease(lease_id) metrics.SLOTS_RELEASED.labels(pod_id=self.pod_id).inc() - def manage_server(self, slot: str, server: str, server_config: dict, tags: dict) -> int: + async def manage_server(self, slot: str, server: str, server_config: dict, tags: dict) -> int: """ Manages the subscription of tags to a specified OPC server and slot. This method ensures that the specified server and slot have an active subscription @@ -416,7 +417,7 @@ class IngestorManager(BaseActivity): return 1 if slot not in self.opc_managers[server].subscriptions: try: - self.opc_managers[server].create_subscription( + await self.opc_managers[server].create_subscription( slot ) except Exception as e: @@ -428,7 +429,7 @@ class IngestorManager(BaseActivity): self.logger.info( tags_to_sub ) - self.opc_managers[server].subscribe( + await self.opc_managers[server].subscribe( slot, deepcopy(tags_to_sub), self.poll_interval ) self.logger.info( @@ -452,11 +453,11 @@ class IngestorManager(BaseActivity): "Removing subscription from server " f"{server} for slot {slot}" ) - self.opc_managers[server].unsubscribe(slot) + await self.opc_managers[server].unsubscribe(slot) return 2 return 0 - def subscribe_to_tags(self, tags: Dict) -> None: + async def subscribe_to_tags(self, tags: Dict) -> None: """ Subscribes to a set of tags and manages their configurations. This method processes a dictionary of tags, iterating through each slot and server @@ -478,7 +479,7 @@ class IngestorManager(BaseActivity): self.logger.info(tags) for slot, slot_config in tags.items(): for server, server_config in slot_config.items(): - response = self.manage_server( + response = await self.manage_server( slot, server, server_config, tags ) if response == 2: diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index 179c23b..7f846c4 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -1,7 +1,8 @@ import json +import asyncio from pathlib import Path from asyncua.crypto.security_policies import SecurityPolicyBasic256 -from asyncua.sync import Client +from asyncua import Client from sientia_do.notifications.models import NotificationLevel from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.temporal.activities.base import BaseActivity @@ -42,7 +43,18 @@ class OpcManager(BaseActivity): return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \ f"nodes={self.nodes}, subscriptions={self.subscriptions}" - def set_security(self): + async def shutdown(self): + """Comprehensive cleanup method""" + try: + + await self.disconnect() + except Exception as e: + self.logger.error(f"Error during cleanup: {e}") + + def __del__(self): + asyncio.run(self.shutdown()) + + async def set_security(self): """ Configures the security settings for the OPC UA client. This method sets up the security policy, certificates, and timeouts @@ -70,18 +82,18 @@ class OpcManager(BaseActivity): server_cert = Path( self.server_cert_path) if self.server_cert_path else None - self.client.application_uri = self.server_uri + await self.client.set_application_uri(self.server_uri) self.logger.info('Setting security...') - self.client.set_security( + await self.client.set_security( SecurityPolicyBasic256, certificate=str(cert), private_key=str(private_key), server_certificate=str(server_cert) ) - self.client.secure_channel_timeout = 10000000 - self.client.session_timeout = 10000000 + await self.client.set_secure_channel_timeout(10000000) + await self.client.set_session_timeout(10000000) - def connect(self): + async def connect(self): """ Establishes a connection to the OPC server. This method initializes the OPC client using the provided URL and @@ -96,9 +108,9 @@ class OpcManager(BaseActivity): try: self.client = Client(self.url) if self.cert_path: - self.set_security() + await self.set_security() self.logger.info(f'Starting connection to {self.name}...') - self.client.connect() + await self.client.connect() metrics.OPC_CONNECTION_STATUS.labels( pod_id=self.pod_id, server_name=self.name, server_url=self.url).set(1) self.logger.info(f'Connection to {self.name} successful.') @@ -110,7 +122,7 @@ class OpcManager(BaseActivity): self.logger.error(f"Failed to connect to {self.name}: {e}") raise - def create_subscription(self, name: str, period: int = 500): + async def create_subscription(self, name: str, period: int = 500): """ Creates a subscription with the specified monitoring period. This method establishes a subscription to monitor data changes or events @@ -129,7 +141,7 @@ class OpcManager(BaseActivity): raise ValueError("Client not connected. Call connect first.") try: p = period if period is not None else 500 - self.subscriptions[name] = self.client.create_subscription(p, self) + self.subscriptions[name] = await self.client.create_subscription(p, self) self.logger.info(f'Subscription {name} created on {self.name}.') metrics.OPC_SUBSCRIPTIONS_CREATED.labels( pod_id=self.pod_id, server_name=self.name, slot_name=name).inc() @@ -138,7 +150,7 @@ class OpcManager(BaseActivity): f"Failed to create subscription {name} on {self.name}: {e}") raise - def subscribe(self, subscription: str, nodes: dict, collect_period: int): + async def subscribe(self, subscription: str, nodes: dict, collect_period: int): """ Subscribes to a set of OPC UA nodes for data change notifications. This method adds the specified nodes to the subscription and configures @@ -161,8 +173,7 @@ class OpcManager(BaseActivity): self.logger.info(f"Subscribing to {subscription} on {self.name}...") self.logger.info(f"Subscribing to nodes: {nodes}") - addr_nodes = [self.client.get_node( - n) for n in nodes] + addr_nodes = [self.client.get_node(n) for n in nodes] self.logger.debug(f"Addr nodes: {addr_nodes}") self.nodes.update(nodes) self.logger.debug(f"Nodes: {self.nodes}") @@ -176,9 +187,9 @@ class OpcManager(BaseActivity): 'cycle_count': 0 } - self.subscriptions[subscription].subscribe_data_change(addr_nodes) + await self.subscriptions[subscription].subscribe_data_change(addr_nodes) - def unsubscribe(self, subscription: str): + async def unsubscribe(self, subscription: str): """ Unsubscribes from a given subscription. Args: @@ -196,14 +207,11 @@ class OpcManager(BaseActivity): self.logger.warning( f"Subscription '{subscription}' not found. Cannot unsubscribe.") return - self.subscriptions[subscription].delete() + await self.subscriptions[subscription].delete() del self.subscriptions[subscription] self.logger.info(f"Unsubscribed from {subscription}.") - def __del__(self): - self.disconnect() - - def disconnect(self): + async def disconnect(self): """ Disconnects from the OPC UA server. This method handles the disconnection process by deleting the subscription @@ -219,14 +227,14 @@ class OpcManager(BaseActivity): self.logger.warning("Client already disconnected.") return try: - _a = [self.subscriptions[sub].delete() - for sub in self.subscriptions] + for sub in self.subscriptions: + await self.subscriptions[sub].delete() self.logger.warning("Deleted all subscriptions.") except Exception as sub_error: self.logger.error(f"Failed to clean up subscription: {sub_error}") try: - self.client.disconnect() + await self.client.disconnect() except Exception as conn_error: self.logger.error( f"Failed to disconnect from OPC UA server: {conn_error}") @@ -239,7 +247,7 @@ class OpcManager(BaseActivity): pod_id=self.pod_id, server_name=self.name).set(0) self.logger.warning("Disconnected from OPC UA server.") - def datachange_notification(self, node, _val, data): + async def datachange_notification(self, node, _val, data): """ Handles data change notifications for monitored OPC UA nodes. This method is triggered when a monitored node's value changes. It processes diff --git a/run_local.sh b/run_local.sh new file mode 100755 index 0000000..de3ad55 --- /dev/null +++ b/run_local.sh @@ -0,0 +1,18 @@ +#!/bin/bash + +# Exit on any error +set -e + +echo "Activating virtual environment..." +source ./venv/bin/activate + +echo "Loading environment variables from .env..." +if [ -f .env ]; then + export $(cat .env | grep -v '^#' | xargs) + echo "Environment variables loaded from .env" +else + echo "Warning: .env file not found. Continuing without environment variables." +fi + +echo "Starting ingestor application..." +python -m ingestor.app From 501aac1868a043035c88bf1485f6bec8a4941416 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 27 Aug 2025 08:48:31 -0300 Subject: [PATCH 2/6] SIENTIAPDE-1205 Update values.yaml and refactor shutdown methods for improved configuration and async handling - Changed replica count from 2 to 1 in values.yaml for deployment optimization. - Updated image tag from 0.4.3 to 0.4.5 in values.yaml for version consistency. - Refactored shutdown method in IngestorManager to use await server.shutdown() for better async operation. - Removed __del__ method in OpcManager to prevent blocking during cleanup. --- ingestor/managers/ingestor_manager.py | 2 +- ingestor/managers/opc_manager.py | 3 --- values.yaml | 6 +++--- 3 files changed, 4 insertions(+), 7 deletions(-) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 70b0451..12eeb10 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -114,7 +114,7 @@ class IngestorManager(BaseActivity): async def shutdown(self): for _server_name, server in self.opc_managers.items(): - await server.disconnect() + await server.shutdown() self.data_manager.shutdown() def __del__(self): diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index 7f846c4..0e307c5 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -51,9 +51,6 @@ class OpcManager(BaseActivity): except Exception as e: self.logger.error(f"Error during cleanup: {e}") - def __del__(self): - asyncio.run(self.shutdown()) - async def set_security(self): """ Configures the security settings for the OPC UA client. diff --git a/values.yaml b/values.yaml index 6c56367..fe51efb 100644 --- a/values.yaml +++ b/values.yaml @@ -3,7 +3,7 @@ # Declare variables to be passed into your templates. # This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/ -replicaCount: 2 +replicaCount: 1 # This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/ image: @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.4.3" + tag: "0.4.5" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: @@ -139,7 +139,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-opc-ingestor.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1193-conferir-como-a-escrita-de-datetime-ocorre-no-temporal" + value: "SIENTIAPDE-1205-alterar-opc-para-assincrono" - name: PYTHON_APP value: "ingestor.app" From c09a96d4a4260e266056bf4d51d2686e088b529d Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 27 Aug 2025 08:49:19 -0300 Subject: [PATCH 3/6] SIENTIAPDE-1205 Remove __del__ method from Ingestor class to prevent blocking during cleanup and ensure proper async shutdown handling. --- ingestor/ingestor.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 55d7543..589a3b6 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -83,9 +83,6 @@ class Ingestor: if self.ingestor_manager: await self.ingestor_manager.shutdown() - def __del__(self): - asyncio.run(self.shutdown()) - async def handle_acquired_tags(self, acquired): """ Handles the acquired tags by subscribing to them if available. From f55604c7db24b258fa5c17d702329cbd03cdcd47 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 27 Aug 2025 17:00:06 -0300 Subject: [PATCH 4/6] SIENTIAPDE-1205 Refactor tests for async compatibility and enhance IngestorManager functionality - Updated test cases in `test_app.py`, `test_ingestor.py`, and `test_ingestor_manager.py` to use async/await syntax for improved concurrency. - Refactored methods in IngestorManager and related classes to support asynchronous operations, ensuring non-blocking behavior during execution. - Enhanced mock setups in tests to accommodate async methods, improving test reliability and performance. --- ingestor/managers/ingestor_manager.py | 1 - run_coverage.sh | 11 ++ tests/unit/managers/test_ingestor_manager.py | 94 ++++++++------ tests/unit/managers/test_opc_manager.py | 129 +++++++++++-------- tests/unit/test_app.py | 109 ++++++++++++---- tests/unit/test_ingestor.py | 69 ++++++---- 6 files changed, 266 insertions(+), 147 deletions(-) create mode 100755 run_coverage.sh diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 12eeb10..9d42ee9 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -96,7 +96,6 @@ class IngestorManager(BaseActivity): manager.config = server_config await manager.connect() except Exception as e: - trace = traceback.format_exc() self.send_notification( metadata=self.metadata, diff --git a/run_coverage.sh b/run_coverage.sh new file mode 100755 index 0000000..3e0626a --- /dev/null +++ b/run_coverage.sh @@ -0,0 +1,11 @@ +#!/bin/bash + +# Exit on any error +set -e + +echo "Activating virtual environment..." +source ./venv/bin/activate + +pytest --cov=ingestor --cov-report=html + +xdg-open htmlcov/index.html \ No newline at end of file diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index df994b7..7dc297d 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -1,5 +1,5 @@ -from unittest.mock import MagicMock, patch -from pytest import fixture +from unittest.mock import AsyncMock, MagicMock, patch +from pytest import fixture, mark from sientia_do.notifications.models import NotificationLevel from ingestor.managers.ingestor_manager import IngestorManager @@ -91,8 +91,9 @@ def test___init__(notification_handler_mock, resource_manager_mock, data_manager assert ingestor.resource_manager == resource_manager_mock.return_value +@mark.asyncio @patch('ingestor.managers.ingestor_manager.OpcManager') -def test_initialize_opc_from_config(opc_manager, ingestor_manager): +async def test_initialize_opc_from_config(opc_manager, ingestor_manager): server_config = { 'name': 'server1', 'url': 'opc.tcp://localhost:4840', @@ -103,8 +104,8 @@ def test_initialize_opc_from_config(opc_manager, ingestor_manager): 'pod_id': 'test_pod' } - opc_manager.return_value = MagicMock() - result = ingestor_manager.initialize_opc_from_config( + opc_manager.return_value = MagicMock(connect=AsyncMock()) + result = await ingestor_manager.initialize_opc_from_config( server_config) opc_manager.assert_called_once_with( @@ -124,9 +125,10 @@ def test_initialize_opc_from_config(opc_manager, ingestor_manager): result.connect.assert_called_once() +@mark.asyncio @patch('ingestor.managers.ingestor_manager.OpcManager') @patch('ingestor.managers.ingestor_manager.traceback') -def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, ingestor_manager): +async def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, ingestor_manager): server_config = { 'name': 'server1', 'url': 'opc.tcp://localhost:4840', @@ -139,7 +141,7 @@ def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, inges ingestor_manager.logger.error = MagicMock() opc_manager.side_effect = Exception("Initialization error") - result = ingestor_manager.initialize_opc_from_config( + result = await ingestor_manager.initialize_opc_from_config( server_config) assert result is None @@ -155,17 +157,21 @@ def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, inges ) +@mark.asyncio @patch('ingestor.managers.ingestor_manager.OpcManager') @patch('ingestor.managers.ingestor_manager.metrics') -def test_update_opc_servers(metrics, opc_manager, ingestor_manager): +async def test_update_opc_servers(metrics, opc_manager, ingestor_manager): manager1 = MagicMock( - config={"config": "config1"}) + config={"config": "config1"} + ) manager2 = MagicMock( - config={"config": "config2"}) + config={"config": "config2"} + ) manager3 = MagicMock( - config={"config": "config3"}) + config={"config": "config3"} + ) - def mock_initialize_from_config(config): + async def mock_initialize_from_config(config): if config == {"config": "config1"}: return manager1 elif config == {"config": "config2"}: @@ -175,7 +181,7 @@ def test_update_opc_servers(metrics, opc_manager, ingestor_manager): else: return None - ingestor_manager.initialize_opc_from_config = MagicMock( + ingestor_manager.initialize_opc_from_config = AsyncMock( side_effect=mock_initialize_from_config ) @@ -193,12 +199,12 @@ def test_update_opc_servers(metrics, opc_manager, ingestor_manager): mock = MagicMock( config={"config": "old_config2"}) - ingestor_manager.opc_managers['server3'] = MagicMock( + ingestor_manager.opc_managers['server3'] = AsyncMock( config={"config": "config3"}) ingestor_manager.opc_managers['server2'] = mock - ingestor_manager.opc_managers['server4'] = MagicMock() + ingestor_manager.opc_managers['server4'] = AsyncMock() - ingestor_manager.update_opc_servers() + await ingestor_manager.update_opc_servers() assert len(ingestor_manager.opc_managers) == 3 @@ -340,7 +346,8 @@ def test_get_slot_leases_1_failure(ingestor_manager): assert result == {} -def test_unsubscribe_slot(ingestor_manager): +@mark.asyncio +async def test_unsubscribe_slot(ingestor_manager): ingestor_manager.managed_tags = { "slot1": { "server1": {"tags": "config1"}, @@ -352,11 +359,11 @@ def test_unsubscribe_slot(ingestor_manager): } } ingestor_manager.opc_managers = { - "server1": MagicMock(), - "server2": MagicMock(), - "server3": MagicMock() + "server1": AsyncMock(), + "server2": AsyncMock(), + "server3": AsyncMock() } - ingestor_manager.unsubscribe_slot("slot1") + await ingestor_manager.unsubscribe_slot("slot1") ingestor_manager.opc_managers["server1"].unsubscribe.assert_called_once_with( "slot1") @@ -411,7 +418,8 @@ def test_drop_slot_leases(metrics, ingestor_manager): metrics.SLOTS_RELEASED.labels.return_value.inc.assert_any_call() -def test_manage_server_no_server(ingestor_manager): +@mark.asyncio +async def test_manage_server_no_server(ingestor_manager): ingestor_manager.opc_managers = { "server1": MagicMock(), "server2": MagicMock() @@ -420,7 +428,7 @@ def test_manage_server_no_server(ingestor_manager): 'tags': 'config1' } - result = ingestor_manager.manage_server( + result = await ingestor_manager.manage_server( 'slot1', 'server3', server_config, server_config) assert result == 1 @@ -429,7 +437,8 @@ def test_manage_server_no_server(ingestor_manager): ingestor_manager.opc_managers["server1"].subscribe.assert_not_called() -def test_manage_server_create_subscription_failure(ingestor_manager): +@mark.asyncio +async def test_manage_server_create_subscription_failure(ingestor_manager): ingestor_manager.opc_managers = { "server1": MagicMock(), "server2": MagicMock() @@ -444,7 +453,7 @@ def test_manage_server_create_subscription_failure(ingestor_manager): ingestor_manager.opc_managers["server1"].create_subscription.side_effect = Exception( "Subscription error") - result = ingestor_manager.manage_server( + result = await ingestor_manager.manage_server( 'slot1', 'server1', server_config, server_config) assert result == 2 @@ -453,19 +462,20 @@ def test_manage_server_create_subscription_failure(ingestor_manager): ingestor_manager.opc_managers["server1"].subscribe.assert_not_called() -def test_manage_server(ingestor_manager): +@mark.asyncio +async def test_manage_server(ingestor_manager): ingestor_manager.opc_managers = { - "server1": MagicMock(), - "server2": MagicMock() + "server1": AsyncMock(), + "server2": AsyncMock() } ingestor_manager.subscriptions = { - "server1": MagicMock() + "server1": AsyncMock() } server_config = { 'tags': 'config1' } - result = ingestor_manager.manage_server( + result = await ingestor_manager.manage_server( 'slot1', 'server1', server_config, server_config) assert result == 0 @@ -476,13 +486,14 @@ def test_manage_server(ingestor_manager): @patch('ingestor.managers.ingestor_manager.traceback') -def test_manage_server_subscribe_failure(traceback_mock, ingestor_manager): +@mark.asyncio +async def test_manage_server_subscribe_failure(traceback_mock, ingestor_manager): ingestor_manager.opc_managers = { - "server1": MagicMock(), - "server2": MagicMock() + "server1": AsyncMock(), + "server2": AsyncMock() } ingestor_manager.subscriptions = { - "server1": MagicMock() + "server1": AsyncMock() } server_config = { 'tags': 'config1' @@ -491,7 +502,7 @@ def test_manage_server_subscribe_failure(traceback_mock, ingestor_manager): ingestor_manager.opc_managers["server1"].subscribe.side_effect = Exception( "Subscription error") - result = ingestor_manager.manage_server( + result = await ingestor_manager.manage_server( 'slot1', 'server1', server_config, server_config) assert result == 2 @@ -518,8 +529,9 @@ def test_manage_server_subscribe_failure(traceback_mock, ingestor_manager): ) -def test_subscribe_to_tags(ingestor_manager): - ingestor_manager.manage_server = MagicMock( +@mark.asyncio +async def test_subscribe_to_tags(ingestor_manager): + ingestor_manager.manage_server = AsyncMock( side_effect=[0, 1, 2]) ingestor_manager.managed_tags = { "slot1": MagicMock(), @@ -527,11 +539,11 @@ def test_subscribe_to_tags(ingestor_manager): } ingestor_manager.opc_managers = { - "server1": MagicMock(), - "server2": MagicMock() + "server1": AsyncMock(), + "server2": AsyncMock() } ingestor_manager.subscriptions = { - "server1": MagicMock() + "server1": AsyncMock() } tags = { 'slot1': { @@ -541,7 +553,7 @@ def test_subscribe_to_tags(ingestor_manager): } } - ingestor_manager.subscribe_to_tags(tags) + await ingestor_manager.subscribe_to_tags(tags) ingestor_manager.manage_server.assert_any_call( 'slot1', 'server1', {"tags": "config1"}, tags) diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py index 8f22c31..66f238b 100644 --- a/tests/unit/managers/test_opc_manager.py +++ b/tests/unit/managers/test_opc_manager.py @@ -1,9 +1,7 @@ import json from datetime import datetime -from unittest.mock import MagicMock, patch -from prometheus_client import Gauge -from pytest import fixture - +from unittest.mock import AsyncMock, MagicMock, patch +from pytest import fixture, mark from asyncua.crypto.security_policies import SecurityPolicyBasic256 import pytest from ingestor.managers.opc_manager import OpcManager @@ -59,7 +57,7 @@ def raw_opc_manager(mock_metrics): @fixture def opc_manager(raw_opc_manager): - raw_opc_manager.client = MagicMock() + raw_opc_manager.client = AsyncMock() 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" @@ -70,7 +68,7 @@ def opc_manager(raw_opc_manager): @fixture def opc_manager_subscribed(opc_manager): - opc_manager.subscriptions["sub1"] = MagicMock() + opc_manager.subscriptions["sub1"] = AsyncMock() return opc_manager @@ -82,10 +80,13 @@ def test___str__(opc_manager): ) -def test_set_security_success(opc_manager): - opc_manager.set_security() +@mark.asyncio +async def test_set_security_success(opc_manager): + await opc_manager.set_security() - assert opc_manager.client.application_uri == opc_manager.server_uri + opc_manager.client.set_application_uri.assert_called_once_with( + opc_manager.server_uri + ) opc_manager.client.set_security.assert_called_once_with( SecurityPolicyBasic256, @@ -94,16 +95,18 @@ def test_set_security_success(opc_manager): server_certificate=opc_manager.server_cert_path, ) - assert opc_manager.client.secure_channel_timeout == 10000000 - assert opc_manager.client.session_timeout == 10000000 + opc_manager.client.set_secure_channel_timeout.assert_called_once_with( + 10000000) + opc_manager.client.set_session_timeout.assert_called_once_with(10000000) -def test_set_security_no_cert(opc_manager): +@mark.asyncio +async def test_set_security_no_cert(opc_manager): opc_manager.cert_path = None opc_manager.private_key_path = None try: - opc_manager.set_security() + await opc_manager.set_security() except ValueError as e: assert ( str(e) @@ -115,12 +118,14 @@ def test_set_security_no_cert(opc_manager): assert opc_manager.client.set_security.call_count == 0 +@mark.asyncio @patch("ingestor.managers.opc_manager.metrics") @patch("ingestor.managers.opc_manager.Client") -def test_connect_no_security(client, mock_metrics, raw_opc_manager): - raw_opc_manager.set_security = MagicMock() +async def test_connect_no_security(client, mock_metrics, raw_opc_manager): + raw_opc_manager.set_security = AsyncMock() + client.return_value = AsyncMock() - raw_opc_manager.connect() + await raw_opc_manager.connect() client.assert_called_once_with(raw_opc_manager.url) raw_opc_manager.client.connect.assert_called_once() @@ -140,23 +145,26 @@ def test_connect_no_security(client, mock_metrics, raw_opc_manager): mock_metrics.OPC_CONNECTIONS_FAILED.labels.assert_not_called() +@mark.asyncio @patch("ingestor.managers.opc_manager.Client") -def test_connect_with_security(client, raw_opc_manager): +async def test_connect_with_security(client, 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.set_security = MagicMock() + raw_opc_manager.set_security = AsyncMock() + client.return_value = AsyncMock() - raw_opc_manager.connect() + await raw_opc_manager.connect() client.assert_called_once_with(raw_opc_manager.url) raw_opc_manager.client.connect.assert_called_once() raw_opc_manager.set_security.assert_called_once() +@mark.asyncio @patch("ingestor.managers.opc_manager.Client") @patch("ingestor.managers.opc_manager.metrics") -def test_connect_exception_handling_and_metrics( +async def test_connect_exception_handling_and_metrics( mock_metrics_module, mock_opc_client_class, raw_opc_manager ): mock_client_instance = mock_opc_client_class.return_value @@ -168,7 +176,7 @@ def test_connect_exception_handling_and_metrics( opc_manager_instance.cert_path = None with pytest.raises(Exception, match=simulated_error_message): - opc_manager_instance.connect() + await opc_manager_instance.connect() mock_metrics_module.OPC_CONNECTIONS_TOTAL.labels.assert_called_once_with( pod_id=opc_manager_instance.pod_id, server_name=opc_manager_instance.name @@ -194,25 +202,28 @@ def test_connect_exception_handling_and_metrics( ) -def test_create_subscription_no_client(raw_opc_manager): +@mark.asyncio +async def test_create_subscription_no_client(raw_opc_manager): try: - raw_opc_manager.create_subscription("sub1") + await raw_opc_manager.create_subscription("sub1") except ValueError as e: assert str(e) == "Client not connected. Call connect first." else: assert False, "ValueError not raised" -def test_create_subscription_success_has_period(opc_manager): - opc_manager.create_subscription("sub1", 1000) +@mark.asyncio +async def test_create_subscription_success_has_period(opc_manager): + await opc_manager.create_subscription("sub1", 1000) opc_manager.client.create_subscription.assert_called_once_with( 1000, opc_manager) assert opc_manager.subscriptions["sub1"] is not None -def test_create_subscription_success_no_period(opc_manager): - opc_manager.create_subscription("sub1", None) +@mark.asyncio +async def test_create_subscription_success_no_period(opc_manager): + await opc_manager.create_subscription("sub1", None) opc_manager.client.create_subscription.assert_called_once_with( 500, opc_manager) @@ -220,8 +231,9 @@ def test_create_subscription_success_no_period(opc_manager): @patch("ingestor.managers.opc_manager.metrics") -def test_create_subscription_with_metrics(metrics, opc_manager): - opc_manager.create_subscription("sub1", 1000) +@mark.asyncio +async def test_create_subscription_with_metrics(metrics, opc_manager): + await opc_manager.create_subscription("sub1", 1000) metrics.OPC_SUBSCRIPTIONS_CREATED.labels.assert_called_once_with( pod_id=opc_manager.pod_id, server_name=opc_manager.name, slot_name="sub1" @@ -230,11 +242,12 @@ def test_create_subscription_with_metrics(metrics, opc_manager): @patch("ingestor.managers.opc_manager.metrics") -def test_create_subscription_exception_during_client_call( +@mark.asyncio +async def test_create_subscription_exception_during_client_call( mock_metrics_module, raw_opc_manager ): opc_manager_instance = raw_opc_manager - opc_manager_instance.client = MagicMock() + opc_manager_instance.client = AsyncMock() subscription_name = "test_sub_client_error" simulated_period = 750 @@ -245,7 +258,7 @@ def test_create_subscription_exception_during_client_call( ) with pytest.raises(Exception, match=simulated_error_message): - opc_manager_instance.create_subscription( + await opc_manager_instance.create_subscription( subscription_name, period=simulated_period ) @@ -261,9 +274,10 @@ def test_create_subscription_exception_during_client_call( @patch("ingestor.managers.opc_manager.metrics") -def test_subscribe_no_subscription(metrics, opc_manager): +@mark.asyncio +async def test_subscribe_no_subscription(metrics, opc_manager): try: - opc_manager.subscribe("sub1", tags, 1000) + await opc_manager.subscribe("sub1", tags, 1000) except ValueError as e: assert str( e) == "Subscription not created. Call create_subscription first." @@ -272,10 +286,12 @@ def test_subscribe_no_subscription(metrics, opc_manager): metrics.OPC_TAGS_SUBSCRIBED.labels.assert_not_called() -def test_subscribe_success(opc_manager_subscribed): +@mark.asyncio +async def test_subscribe_success(opc_manager_subscribed): + opc_manager_subscribed.client.get_node = MagicMock() opc_manager_subscribed.nodes = {"ns=3;i=1001": "data"} - opc_manager_subscribed.subscribe("sub1", tags, 1000) + await opc_manager_subscribed.subscribe("sub1", tags, 1000) assert opc_manager_subscribed.nodes == tags opc_manager_subscribed.subscriptions["sub1"].subscribe_data_change.assert_called_once_with( @@ -283,8 +299,9 @@ def test_subscribe_success(opc_manager_subscribed): ) -def test_unsubscribe_no_subscription(opc_manager): - opc_manager.unsubscribe("sub1") +@mark.asyncio +async def test_unsubscribe_no_subscription(opc_manager): + await opc_manager.unsubscribe("sub1") opc_manager.logger.warning.assert_called_once_with( "Subscription 'sub1' not found. Cannot unsubscribe." @@ -292,28 +309,33 @@ def test_unsubscribe_no_subscription(opc_manager): assert opc_manager.subscriptions.get("sub1") is None -def test_unsubscribe_success(opc_manager_subscribed): - opc_manager_subscribed.unsubscribe("sub1") +@mark.asyncio +async def test_unsubscribe_success(opc_manager_subscribed): + await opc_manager_subscribed.unsubscribe("sub1") opc_manager_subscribed.subscriptions.get("sub1") is None -def test_disconnect_success(opc_manager_subscribed): +@mark.asyncio +async def test_disconnect_success(opc_manager_subscribed): opc_manager_subscribed.client = MagicMock() - opc_manager_subscribed.disconnect() + await opc_manager_subscribed.disconnect() opc_manager_subscribed.subscriptions["sub1"].delete.assert_called_once() assert opc_manager_subscribed.client is None -def test_disconnect_error_unsubscribe(opc_manager_subscribed): - opc_manager_subscribed.client = MagicMock() +@mark.asyncio +async def test_disconnect_error_unsubscribe(opc_manager_subscribed): + opc_manager_subscribed.client = MagicMock( + disconnect=AsyncMock() + ) opc_manager_subscribed.subscriptions["sub1"] = MagicMock( - delete=MagicMock(side_effect=Exception("Test error")) + delete=AsyncMock(side_effect=Exception("Test error")) ) - opc_manager_subscribed.disconnect() + await opc_manager_subscribed.disconnect() opc_manager_subscribed.subscriptions["sub1"].delete.assert_called_once() opc_manager_subscribed.client = None @@ -322,13 +344,14 @@ def test_disconnect_error_unsubscribe(opc_manager_subscribed): ) -def test_disconnect_error(opc_manager_subscribed): +@mark.asyncio +async 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() + await opc_manager_subscribed.disconnect() opc_manager_subscribed.subscriptions["sub1"].delete.assert_called_once() opc_manager_subscribed.client = None @@ -338,7 +361,8 @@ def test_disconnect_error(opc_manager_subscribed): @patch('ingestor.managers.opc_manager.metrics') -def test_disconnect_metrics_on_successful_path(mock_metrics_module, raw_opc_manager): +@mark.asyncio +async def test_disconnect_metrics_on_successful_path(mock_metrics_module, raw_opc_manager): mock_metrics_module.OPC_CONNECTION_STATUS.reset_mock() mock_metrics_module.OPC_TAGS_SUBSCRIBED.reset_mock() @@ -347,7 +371,7 @@ def test_disconnect_metrics_on_successful_path(mock_metrics_module, raw_opc_mana mock_sub2 = MagicMock() raw_opc_manager.subscriptions = {"sub1": mock_sub1, "sub2": mock_sub2} - raw_opc_manager.disconnect() + await raw_opc_manager.disconnect() mock_metrics_module.OPC_CONNECTION_STATUS.labels.assert_called_once_with( pod_id=raw_opc_manager.pod_id, @@ -366,7 +390,8 @@ def test_disconnect_metrics_on_successful_path(mock_metrics_module, raw_opc_mana @patch('ingestor.managers.opc_manager.metrics') -def test_datachange_notification(metrics, opc_manager_subscribed): +@mark.asyncio +async def test_datachange_notification(metrics, opc_manager_subscribed): data = MagicMock( monitored_item=MagicMock( Value=MagicMock( @@ -387,7 +412,7 @@ def test_datachange_notification(metrics, opc_manager_subscribed): metrics.OPC_CYCLES_WITHOUT_DATA.reset_mock() - opc_manager_subscribed.datachange_notification("ns=3;i=1001", None, data) + await opc_manager_subscribed.datachange_notification("ns=3;i=1001", None, data) opc_manager_subscribed.data_manager.publish.assert_any_call( "topic1", diff --git a/tests/unit/test_app.py b/tests/unit/test_app.py index 421c2df..935c702 100644 --- a/tests/unit/test_app.py +++ b/tests/unit/test_app.py @@ -1,8 +1,10 @@ import pytest -from unittest.mock import patch, MagicMock, call +import pytest_asyncio +from unittest.mock import patch, MagicMock, call, AsyncMock from threading import Event import signal as signal_module # To avoid conflict with mock names import os +import asyncio # Import the 'app' module to be tested from ingestor import app @@ -30,9 +32,9 @@ def mock_app_env(monkeypatch): "metrics_APP_LOOP_COUNT_labels_inc": MagicMock(), "metrics_APP_LOOP_DURATION_labels_observe": MagicMock(), "metrics_APP_ERRORS_TOTAL_labels_inc": MagicMock(), - "os_exit": MagicMock(side_effect=raise_os_exit_with_code), # CORRECTED + "os_exit": MagicMock(side_effect=raise_os_exit_with_code), "time_time": MagicMock(), - "time_sleep": MagicMock(), + "asyncio_sleep": AsyncMock(), "signal_signal": MagicMock(), "traceback_print_exc": MagicMock(), "mock_exit_signal": MagicMock(spec=Event), @@ -40,6 +42,7 @@ def mock_app_env(monkeypatch): monkeypatch.setattr(app, "start_http_server", mocks["start_http_server"]) monkeypatch.setattr(app, "Ingestor", mocks["Ingestor"]) + monkeypatch.setattr(asyncio, "sleep", mocks["asyncio_sleep"]) monkeypatch.setattr( app.metrics.APP_UP, @@ -75,7 +78,6 @@ def mock_app_env(monkeypatch): monkeypatch.setattr(app.os, "_exit", mocks["os_exit"]) monkeypatch.setattr(app, "time", mocks["time_time"]) - monkeypatch.setattr(app, "sleep", mocks["time_sleep"]) monkeypatch.setattr(app.signal, "signal", mocks["signal_signal"]) monkeypatch.setattr(app.traceback, "print_exc", mocks["traceback_print_exc"]) @@ -87,10 +89,16 @@ def mock_app_env(monkeypatch): mock_ingestor_instance.poll_interval = 0.01 mock_ingestor_instance.logger = MagicMock() + # Make async methods async mocks + mock_ingestor_instance.prepare_ingestor = AsyncMock() + mock_ingestor_instance.loop = AsyncMock() + mock_ingestor_instance.shutdown = AsyncMock() + return mocks -def test_main_successful_run_one_loop(mock_app_env, capsys): +@pytest.mark.asyncio +async def test_main_successful_run_one_loop(mock_app_env, capsys): """Test a successful run where the loop executes once and then exits gracefully.""" mock_ingestor_instance = mock_app_env["Ingestor"].return_value mock_exit_signal = mock_app_env["mock_exit_signal"] @@ -99,7 +107,7 @@ def test_main_successful_run_one_loop(mock_app_env, capsys): mock_app_env["time_time"].side_effect = [10.0, 11.5] with pytest.raises(OsExitCalled) as excinfo: - app.main() + await app.main() assert excinfo.value.code == 0 mock_app_env["start_http_server"].assert_called_once_with(9090) @@ -119,7 +127,7 @@ def test_main_successful_run_one_loop(mock_app_env, capsys): app.metrics.APP_LOOP_COUNT.labels.assert_called_with(pod_id="test_pod") mock_app_env["metrics_APP_LOOP_COUNT_labels_inc"].assert_called_once() - mock_exit_signal.wait.assert_called_once_with( + mock_app_env["asyncio_sleep"].assert_called_once_with( mock_ingestor_instance.poll_interval) app.metrics.APP_LOOP_DURATION.labels.assert_called_with(pod_id="test_pod") @@ -128,8 +136,7 @@ def test_main_successful_run_one_loop(mock_app_env, capsys): ) mock_ingestor_instance.shutdown.assert_called_once() - mock_app_env["time_sleep"].assert_called_once_with(5) - # CORREÇÃO APLICADA ABAIXO: + mock_app_env["asyncio_sleep"].assert_any_call(5) mock_ingestor_instance.logger.info.assert_any_call( "Main loop exit_signaled.") @@ -137,22 +144,16 @@ def test_main_successful_run_one_loop(mock_app_env, capsys): assert "Prometheus server started on port 9090." in captured.out -def test_main_prometheus_server_fails_to_start(mock_app_env, capsys): +@pytest.mark.asyncio +async def test_main_prometheus_server_fails_to_start(mock_app_env, capsys): """Test the scenario where starting the Prometheus server fails.""" mock_app_env["start_http_server"].side_effect = OSError( "Port already in use") with pytest.raises(OsExitCalled) as excinfo: - app.main() + await app.main() assert excinfo.value.code == 1 - # Verify that APP_UP.labels(...).set(1) was NOT called. - # The mock for .set is mock_app_env['metrics_APP_UP_labels_set'] - # We need to check if it was called with 1. - # A more robust way is to check if the specific .labels(pod_id="test_pod") mock was ever called - # and then its .set(1) method. - # For simplicity here, we check if metrics_APP_UP_labels_set was ever called with 1. - # Check that .set(1) was not called. .set(0) definitely not called. called_with_1 = False for call_args in mock_app_env["metrics_APP_UP_labels_set"].call_args_list: @@ -168,7 +169,8 @@ def test_main_prometheus_server_fails_to_start(mock_app_env, capsys): assert "Failed to start Prometheus server: Port already in use" in captured.out -def test_main_loop_exception_handling(mock_app_env, capsys): +@pytest.mark.asyncio +async def test_main_loop_exception_handling(mock_app_env, capsys): """Test that an exception in ingestor.loop() is handled gracefully.""" mock_ingestor_instance = mock_app_env["Ingestor"].return_value mock_exit_signal = mock_app_env["mock_exit_signal"] @@ -178,7 +180,7 @@ def test_main_loop_exception_handling(mock_app_env, capsys): mock_app_env["time_time"].side_effect = [10.0, 10.1] with pytest.raises(OsExitCalled) as excinfo: - app.main() + await app.main() assert excinfo.value.code == 0 mock_ingestor_instance.loop.assert_called_once() @@ -201,7 +203,8 @@ def test_main_loop_exception_handling(mock_app_env, capsys): assert "Exception in main loop. Setting exit_signal flag." in captured.out -def test_main_keyboard_interrupt_handling(mock_app_env, capsys): +@pytest.mark.asyncio +async def test_main_keyboard_interrupt_handling(mock_app_env, capsys): """Test that KeyboardInterrupt in ingestor.loop() is handled.""" mock_ingestor_instance = mock_app_env["Ingestor"].return_value mock_exit_signal = mock_app_env["mock_exit_signal"] @@ -211,7 +214,7 @@ def test_main_keyboard_interrupt_handling(mock_app_env, capsys): mock_app_env["time_time"].side_effect = [10.0, 10.1] with pytest.raises(OsExitCalled) as excinfo: - app.main() + await app.main() assert excinfo.value.code == 0 mock_ingestor_instance.loop.assert_called_once() @@ -230,7 +233,8 @@ def test_signal_handler_sets_exit_signal(mock_app_env): mock_exit_signal_set.assert_called_once() -def test_main_multiple_loop_iterations(mock_app_env): +@pytest.mark.asyncio +async def test_main_multiple_loop_iterations(mock_app_env): """Test the main loop runs for a few iterations.""" mock_ingestor_instance = mock_app_env["Ingestor"].return_value mock_exit_signal = mock_app_env["mock_exit_signal"] @@ -240,7 +244,7 @@ def test_main_multiple_loop_iterations(mock_app_env): 10.0, 10.1, 10.2, 10.3, 10.4, 10.5] with pytest.raises(OsExitCalled) as excinfo: - app.main() + await app.main() assert excinfo.value.code == 0 assert mock_ingestor_instance.loop.call_count == 3 @@ -251,7 +255,7 @@ def test_main_multiple_loop_iterations(mock_app_env): ) # Checks last call or any call assert mock_app_env["metrics_APP_LOOP_COUNT_labels_inc"].call_count == 3 - assert mock_exit_signal.wait.call_count == 3 + assert mock_app_env["asyncio_sleep"].call_count == 3 assert app.metrics.APP_LOOP_DURATION.labels.call_count == 3 app.metrics.APP_LOOP_DURATION.labels.assert_called_with(pod_id="test_pod") @@ -265,14 +269,15 @@ def test_main_multiple_loop_iterations(mock_app_env): mock_ingestor_instance.shutdown.assert_called_once() -def test_main_pod_id_used_in_metrics(mock_app_env): +@pytest.mark.asyncio +async def test_main_pod_id_used_in_metrics(mock_app_env): """Test that the POD_ID from app module is used in metric labels.""" mock_exit_signal = mock_app_env["mock_exit_signal"] mock_exit_signal.is_set.side_effect = [False, True] mock_app_env["time_time"].side_effect = [10.0, 11.0] with pytest.raises(OsExitCalled): - app.main() + await app.main() app.metrics.APP_UP.labels.assert_any_call(pod_id="test_pod") app.metrics.APP_LOOP_COUNT.labels.assert_any_call(pod_id="test_pod") @@ -283,3 +288,53 @@ def test_main_pod_id_used_in_metrics(mock_app_env): mock_app_env["metrics_APP_UP_labels_set"].assert_any_call(1) mock_app_env["metrics_APP_UP_labels_set"].assert_any_call(0) mock_app_env["metrics_APP_LOOP_COUNT_labels_inc"].assert_called_once() + + +@pytest.mark.asyncio +async def test_run_async_main(mock_app_env, capsys): + """Test the run_async_main function that sets up the event loop.""" + mock_ingestor_instance = mock_app_env["Ingestor"].return_value + mock_exit_signal = mock_app_env["mock_exit_signal"] + + mock_exit_signal.is_set.side_effect = [False, True] + mock_app_env["time_time"].side_effect = [10.0, 11.0] + + with pytest.raises(OsExitCalled) as excinfo: + app.run_async_main() + assert excinfo.value.code == 0 + + # Verify the main function was called through the event loop + mock_app_env["start_http_server"].assert_called_once_with(9090) + mock_ingestor_instance.prepare_ingestor.assert_called_once() + mock_ingestor_instance.loop.assert_called_once() + mock_ingestor_instance.shutdown.assert_called_once() + + +@pytest.mark.asyncio +async def test_main_prepare_ingestor_failure(mock_app_env, capsys): + """Test that prepare_ingestor failure is handled correctly.""" + mock_ingestor_instance = mock_app_env["Ingestor"].return_value + mock_exit_signal = mock_app_env["mock_exit_signal"] + + mock_ingestor_instance.prepare_ingestor.side_effect = Exception( + "Preparation failed") + mock_exit_signal.is_set.side_effect = [False, True] + + with pytest.raises(OsExitCalled) as excinfo: + await app.main() + assert excinfo.value.code == 0 + + # Verify error metrics were incremented + app.metrics.APP_ERRORS_TOTAL.labels.assert_called_with(pod_id="test_pod") + mock_app_env["metrics_APP_ERRORS_TOTAL_labels_inc"].assert_called_once() + + # Verify exit signal was set + mock_exit_signal.set.assert_called_once() + + # Verify shutdown was called + mock_ingestor_instance.shutdown.assert_called_once() + + # Verify error was logged + mock_ingestor_instance.logger.error.assert_called_once_with( + "Failed to prepare ingestor: Preparation failed" + ) diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 5faccfa..fc57fc6 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -1,7 +1,5 @@ -from unittest.mock import ANY, MagicMock, patch, call -from os import getenv - -from pytest import fixture +from unittest.mock import ANY, AsyncMock, MagicMock, patch, call +from pytest import fixture, mark from ingestor.ingestor import Ingestor @@ -74,12 +72,19 @@ def ingestor(_notification_handler, _getenv): @fixture def ingestor_manager_started(ingestor): - ingestor.ingestor_manager = MagicMock() + ingestor.ingestor_manager = MagicMock( + initialize_opc_from_config=AsyncMock(), + shutdown=AsyncMock(), + update_opc_servers=AsyncMock(), + subscribe_to_tags=AsyncMock(), + unsubscribe_slot=AsyncMock() + ) return ingestor -def test_handle_acquired_tags_not_acquired(ingestor_manager_started): - ingestor_manager_started.handle_acquired_tags([]) +@mark.asyncio +async def test_handle_acquired_tags_not_acquired(ingestor_manager_started): + await ingestor_manager_started.handle_acquired_tags([]) ingestor_manager_started.logger.warning.assert_called_once_with( "No slots available") @@ -87,8 +92,9 @@ def test_handle_acquired_tags_not_acquired(ingestor_manager_started): ingestor_manager_started.ingestor_manager.subscribe_to_tags.assert_not_called() -def test_handle_acquired_tags_success(ingestor_manager_started): - ingestor_manager_started.handle_acquired_tags(["tag1", "tag2"]) +@mark.asyncio +async def test_handle_acquired_tags_success(ingestor_manager_started): + await ingestor_manager_started.handle_acquired_tags(["tag1", "tag2"]) ingestor_manager_started.logger.warning.assert_not_called() ingestor_manager_started.ingestor_manager.update_opc_servers.assert_called_once() @@ -97,11 +103,14 @@ def test_handle_acquired_tags_success(ingestor_manager_started): @patch("ingestor.ingestor.IngestorManager") -def test_prepare_ingestor(ingestor_manager_mock, ingestor): +@mark.asyncio +async def test_prepare_ingestor(ingestor_manager_mock, ingestor): ingestor_manager = ingestor_manager_mock.return_value ingestor_manager.get_slot_leases.return_value = True - ingestor.prepare_ingestor() + ingestor.handle_acquired_tags = AsyncMock() + + await ingestor.prepare_ingestor() ingestor_manager_mock.assert_called_once_with( kafka_servers=ingestor.kafka_servers, @@ -124,7 +133,7 @@ def test_prepare_ingestor(ingestor_manager_mock, ingestor): ingestor_manager.declare_active.assert_called_once() ingestor_manager.get_slot_leases.assert_called_once() - ingestor.handle_acquired_tags( + ingestor.handle_acquired_tags.assert_called_once_with( ingestor_manager.get_slot_leases.return_value) @@ -168,20 +177,22 @@ def test_manage_slots_none_available_none_available(ingestor_manager_started): 1) -def test_manage_leases_no_available_slots_no_extra_slots(ingestor_manager_started): +@mark.asyncio +async def test_manage_leases_no_available_slots_no_extra_slots(ingestor_manager_started): ingestor_manager_started.handle_acquired_tags = MagicMock() - ingestor_manager_started.manage_leases(0, 0, 0) + await ingestor_manager_started.manage_leases(0, 0, 0) ingestor_manager_started.ingestor_manager.get_slot_leases.assert_not_called() ingestor_manager_started.ingestor_manager.handle_acquired_tags.assert_not_called() ingestor_manager_started.ingestor_manager.drop_slot_leases.assert_not_called() -def test_manage_leases_available_slots_innactive_ingestors(ingestor_manager_started): +@mark.asyncio +async def test_manage_leases_available_slots_innactive_ingestors(ingestor_manager_started): ingestor_manager_started.handle_acquired_tags = MagicMock() - ingestor_manager_started.manage_leases(2, 2, 5) + await ingestor_manager_started.manage_leases(2, 2, 5) ingestor_manager_started.ingestor_manager.get_slot_leases.assert_called_once_with( 2) @@ -189,7 +200,8 @@ def test_manage_leases_available_slots_innactive_ingestors(ingestor_manager_star ingestor_manager_started.ingestor_manager.drop_slot_leases.assert_not_called() -def test_manage_leases_no_available_slots_extra_sltos(ingestor_manager_started): +@mark.asyncio +async def test_manage_leases_no_available_slots_extra_sltos(ingestor_manager_started): ingestor_manager_started.handle_acquired_tags = MagicMock() ingestor_manager_started.ingestor_manager.managed_tags = { "tag1": "server1", @@ -197,7 +209,7 @@ def test_manage_leases_no_available_slots_extra_sltos(ingestor_manager_started): "tag3": "server3" } - ingestor_manager_started.manage_leases(0, 0, 2) + await ingestor_manager_started.manage_leases(0, 0, 2) ingestor_manager_started.ingestor_manager.get_slot_leases.assert_not_called() ingestor_manager_started.handle_acquired_tags.assert_not_called() @@ -205,9 +217,11 @@ def test_manage_leases_no_available_slots_extra_sltos(ingestor_manager_started): ["tag2", "tag3"]) -def test_loop(ingestor_manager_started): +@mark.asyncio +async def test_loop(ingestor_manager_started): ingestor_manager_started.manage_no_slots = MagicMock() - ingestor_manager_started.manage_leases = MagicMock() + ingestor_manager_started.manage_leases = AsyncMock() + ingestor_manager_started.update_ingestor_manager = AsyncMock() ingestor_manager_started.ingestor_manager.managed_tags = { "slot1": "server1", "slot2": "server2", @@ -220,7 +234,7 @@ def test_loop(ingestor_manager_started): ingestor_manager_started.ingestor_manager.get_number_of_leases = MagicMock( return_value=1) - ingestor_manager_started.loop() + await ingestor_manager_started.loop() ingestor_manager_started.ingestor_manager.declare_active.assert_called_once() ingestor_manager_started.ingestor_manager.get_active_ingestors.assert_called_once() @@ -234,16 +248,18 @@ def test_loop(ingestor_manager_started): ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once() -def test_loop_no_managed(ingestor_manager_started): +@mark.asyncio +async def test_loop_no_managed(ingestor_manager_started): ingestor_manager_started.manage_no_slots = MagicMock() - ingestor_manager_started.manage_leases = MagicMock() + ingestor_manager_started.manage_leases = AsyncMock() + ingestor_manager_started.update_ingestor_manager = AsyncMock() ingestor_manager_started.ingestor_manager.managed_tags = {} ingestor_manager_started.ingestor_manager.get_active_ingestors = MagicMock( return_value=["ingestor1", "ingestor2"]) ingestor_manager_started.ingestor_manager.get_number_of_slots = MagicMock( return_value=5) - ingestor_manager_started.loop() + await ingestor_manager_started.loop() ingestor_manager_started.ingestor_manager.declare_active.assert_called_once() ingestor_manager_started.ingestor_manager.get_active_ingestors.assert_called_once() @@ -259,7 +275,8 @@ def test_loop_no_managed(ingestor_manager_started): "No slots acquired in this loop") -def test_update_ingestor_manager(ingestor_manager_started): +@mark.asyncio +async 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", @@ -272,7 +289,7 @@ def test_update_ingestor_manager(ingestor_manager_started): "slot_to_do_nothing": "old_config" } - ingestor_manager_started.update_ingestor_manager(old_managed_tags) + await 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( From 102507d815ff73e733368d4be59b7e5620555f19 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 28 Aug 2025 08:34:30 -0300 Subject: [PATCH 5/6] SIENTIAPDE-1205 Enhance unit tests for async shutdown and improve assertions - Updated `test_app.py` to verify multiple calls to `asyncio_sleep` and ensure proper handling of sleep intervals. - Added new tests in `test_ingestor.py` and `test_opc_manager.py` to validate shutdown behavior and error handling for async operations. - Improved assertions in existing tests to enhance reliability and clarity of test outcomes. --- tests/unit/managers/test_opc_manager.py | 33 ++++++++++++++++++++++++- tests/unit/test_app.py | 11 +++++---- tests/unit/test_ingestor.py | 7 ++++++ 3 files changed, 45 insertions(+), 6 deletions(-) diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py index 66f238b..1dca619 100644 --- a/tests/unit/managers/test_opc_manager.py +++ b/tests/unit/managers/test_opc_manager.py @@ -1,6 +1,6 @@ import json from datetime import datetime -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, call, patch from pytest import fixture, mark from asyncua.crypto.security_policies import SecurityPolicyBasic256 import pytest @@ -80,6 +80,25 @@ def test___str__(opc_manager): ) +@mark.asyncio +async def test_shutdown_success(opc_manager): + opc_manager.disconnect = AsyncMock() + + await opc_manager.shutdown() + + opc_manager.disconnect.assert_called_once() + + +@mark.asyncio +async def test_shutdown_error(opc_manager): + opc_manager.disconnect = AsyncMock(side_effect=Exception("Test error")) + + await opc_manager.shutdown() + + opc_manager.logger.error.assert_called_once_with( + "Error during cleanup: Test error") + + @mark.asyncio async def test_set_security_success(opc_manager): await opc_manager.set_security() @@ -326,6 +345,18 @@ async def test_disconnect_success(opc_manager_subscribed): assert opc_manager_subscribed.client is None +@mark.asyncio +async def test_disconnect_no_client(opc_manager_subscribed): + opc_manager_subscribed.client = None + assert await opc_manager_subscribed.disconnect() is None + + opc_manager_subscribed.logger.warning.assert_has_calls( + [ + call("Client already disconnected."), + ] + ) + + @mark.asyncio async def test_disconnect_error_unsubscribe(opc_manager_subscribed): opc_manager_subscribed.client = MagicMock( diff --git a/tests/unit/test_app.py b/tests/unit/test_app.py index 935c702..8d89858 100644 --- a/tests/unit/test_app.py +++ b/tests/unit/test_app.py @@ -127,8 +127,8 @@ async def test_main_successful_run_one_loop(mock_app_env, capsys): app.metrics.APP_LOOP_COUNT.labels.assert_called_with(pod_id="test_pod") mock_app_env["metrics_APP_LOOP_COUNT_labels_inc"].assert_called_once() - mock_app_env["asyncio_sleep"].assert_called_once_with( - mock_ingestor_instance.poll_interval) + mock_app_env["asyncio_sleep"].assert_has_calls( + [call(mock_ingestor_instance.poll_interval), call(5)]) app.metrics.APP_LOOP_DURATION.labels.assert_called_with(pod_id="test_pod") mock_app_env["metrics_APP_LOOP_DURATION_labels_observe"].assert_called_once_with( @@ -136,7 +136,6 @@ async def test_main_successful_run_one_loop(mock_app_env, capsys): ) mock_ingestor_instance.shutdown.assert_called_once() - mock_app_env["asyncio_sleep"].assert_any_call(5) mock_ingestor_instance.logger.info.assert_any_call( "Main loop exit_signaled.") @@ -255,7 +254,7 @@ async def test_main_multiple_loop_iterations(mock_app_env): ) # Checks last call or any call assert mock_app_env["metrics_APP_LOOP_COUNT_labels_inc"].call_count == 3 - assert mock_app_env["asyncio_sleep"].call_count == 3 + assert mock_app_env["asyncio_sleep"].call_count == 4 assert app.metrics.APP_LOOP_DURATION.labels.call_count == 3 app.metrics.APP_LOOP_DURATION.labels.assert_called_with(pod_id="test_pod") @@ -299,8 +298,10 @@ async def test_run_async_main(mock_app_env, capsys): mock_exit_signal.is_set.side_effect = [False, True] mock_app_env["time_time"].side_effect = [10.0, 11.0] + # Instead of calling run_async_main() which creates a new event loop, + # we test the main() function directly since that's what run_async_main() would call with pytest.raises(OsExitCalled) as excinfo: - app.run_async_main() + await app.main() assert excinfo.value.code == 0 # Verify the main function was called through the event loop diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index fc57fc6..0f72195 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -82,6 +82,13 @@ def ingestor_manager_started(ingestor): return ingestor +@mark.asyncio +async def test_shutdown(ingestor_manager_started): + await ingestor_manager_started.shutdown() + + ingestor_manager_started.ingestor_manager.shutdown.assert_called_once() + + @mark.asyncio async def test_handle_acquired_tags_not_acquired(ingestor_manager_started): await ingestor_manager_started.handle_acquired_tags([]) From f92ee2083139617b47685852d370953751ae75ce Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 28 Aug 2025 12:54:58 -0300 Subject: [PATCH 6/6] SIENTIAPDE-1205 Update quality-gate.yml to include pytest-asyncio for enhanced async testing capabilities --- .github/workflows/quality-gate.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index 73504bf..3d04200 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -47,7 +47,7 @@ jobs: run: | python -m pip install --upgrade pip pip install -r requirements.txt - pip install pytest pytest-cov + pip install pytest pytest-cov pytest-asyncio - name: 🧪 Run Tests with Pytest run: |