From f55604c7db24b258fa5c17d702329cbd03cdcd47 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 27 Aug 2025 17:00:06 -0300 Subject: [PATCH] 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(