SIENTIAPDE-1205
Enhance unit tests for async shutdown and improve assertions - Updated `test_app.py` to verify multiple calls to `asyncio_sleep` and ensure proper handling of sleep intervals. - Added new tests in `test_ingestor.py` and `test_opc_manager.py` to validate shutdown behavior and error handling for async operations. - Improved assertions in existing tests to enhance reliability and clarity of test outcomes.
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
import json
|
||||
from datetime import datetime
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
from unittest.mock import AsyncMock, MagicMock, call, patch
|
||||
from pytest import fixture, mark
|
||||
from asyncua.crypto.security_policies import SecurityPolicyBasic256
|
||||
import pytest
|
||||
@@ -80,6 +80,25 @@ def test___str__(opc_manager):
|
||||
)
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_shutdown_success(opc_manager):
|
||||
opc_manager.disconnect = AsyncMock()
|
||||
|
||||
await opc_manager.shutdown()
|
||||
|
||||
opc_manager.disconnect.assert_called_once()
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_shutdown_error(opc_manager):
|
||||
opc_manager.disconnect = AsyncMock(side_effect=Exception("Test error"))
|
||||
|
||||
await opc_manager.shutdown()
|
||||
|
||||
opc_manager.logger.error.assert_called_once_with(
|
||||
"Error during cleanup: Test error")
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_set_security_success(opc_manager):
|
||||
await opc_manager.set_security()
|
||||
@@ -326,6 +345,18 @@ async def test_disconnect_success(opc_manager_subscribed):
|
||||
assert opc_manager_subscribed.client is None
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_disconnect_no_client(opc_manager_subscribed):
|
||||
opc_manager_subscribed.client = None
|
||||
assert await opc_manager_subscribed.disconnect() is None
|
||||
|
||||
opc_manager_subscribed.logger.warning.assert_has_calls(
|
||||
[
|
||||
call("Client already disconnected."),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_disconnect_error_unsubscribe(opc_manager_subscribed):
|
||||
opc_manager_subscribed.client = MagicMock(
|
||||
|
||||
Reference in New Issue
Block a user