gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #17 from Aignosi/SIENTIAPDE-1205-alterar-opc-para-assincrono

Browse Source 
SIENTIAPDE-1205: Refactor Ingestor and OPC Manager for Asynchronous Shutdown and Enhanced Testing
This commit is contained in:
Bruno Domingues
2025-08-29 13:59:43 +00:00
committed by GitHub
parent ec3c738519 f92ee20831
commit a384a077a9
12 changed files with 418 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/quality-gate.yml vendored
View File

@@ -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: |

30
ingestor/app.py
View File

@@ -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()

38
ingestor/ingestor.py
View File

@@ -1,3 +1,4 @@
import asyncio
from os import getenv
from copy import deepcopy
from typing import Dict, Any
@@ -78,14 +79,11 @@ 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()
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 +101,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 +151,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 +174,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 +212,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 +220,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 +230,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 +247,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 +308,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 +335,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)

36
ingestor/managers/ingestor_manager.py
View File

@@ -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,9 +94,8 @@ class IngestorManager(BaseActivity):
)
manager.config = server_config
manager.connect()
await manager.connect()
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=self.metadata,
@@ -111,15 +111,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.shutdown()
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 +156,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 +166,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 +191,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 +312,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 +323,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 +378,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 +416,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 +428,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 +452,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 +478,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:

55
ingestor/managers/opc_manager.py
View File

@@ -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,15 @@ 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}")
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 +79,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 +105,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 +119,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 +138,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 +147,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 +170,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 +184,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 +204,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 +224,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 +244,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

11
run_coverage.sh Executable file
View File

@@ -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

18
run_local.sh Executable file
View File

@@ -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

94
tests/unit/managers/test_ingestor_manager.py
View File

@@ -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)

162
tests/unit/managers/test_opc_manager.py
View File

@@ -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, call, 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,32 @@ def test___str__(opc_manager):
)
def test_set_security_success(opc_manager):
opc_manager.set_security()
@mark.asyncio
async def test_shutdown_success(opc_manager):
opc_manager.disconnect = AsyncMock()
assert opc_manager.client.application_uri == opc_manager.server_uri
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()
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 +114,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 +137,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 +164,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 +195,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 +221,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 +250,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 +261,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 +277,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 +293,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 +305,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 +318,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 +328,45 @@ 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()
opc_manager_subscribed.subscriptions["sub1"] = MagicMock(
delete=MagicMock(side_effect=Exception("Test error"))
@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."),
]
)
opc_manager_subscribed.disconnect()
@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=AsyncMock(side_effect=Exception("Test error"))
)
await opc_manager_subscribed.disconnect()
opc_manager_subscribed.subscriptions["sub1"].delete.assert_called_once()
opc_manager_subscribed.client = None
@@ -322,13 +375,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 +392,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 +402,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 +421,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 +443,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",

112
tests/unit/test_app.py
View File

@@ -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,8 +127,8 @@ 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_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(
@@ -128,8 +136,6 @@ 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_ingestor_instance.logger.info.assert_any_call(
"Main loop exit_signaled.")
@@ -137,22 +143,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 +168,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 +179,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 +202,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 +213,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 +232,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 +243,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 +254,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 == 4
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 +268,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 +287,55 @@ 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]
# 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:
await app.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"
)

76
tests/unit/test_ingestor.py
View File

@@ -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,26 @@ 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_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([])
ingestor_manager_started.logger.warning.assert_called_once_with(
"No slots available")
@@ -87,8 +99,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 +110,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 +140,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 +184,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 +207,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 +216,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 +224,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 +241,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 +255,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 +282,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 +296,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(

6
values.yaml
View File

@@ -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"