f6913da932
Enhance test cases in Gates and Redis activities by adding 'name' field to data and updating Redis TTL to 5 hours for better data retention.
483 lines
14 KiB
Python
483 lines
14 KiB
Python
from unittest.mock import MagicMock, patch, ANY
|
|
from datetime import datetime
|
|
import pytest
|
|
import numpy as np
|
|
from pandas import DataFrame
|
|
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
|
|
from sientia_do.notifications.models import NotificationLevel
|
|
from scouter.activities.redis import Redis
|
|
|
|
|
|
@pytest.fixture
|
|
@patch('scouter.activities.redis.RedisBase.__init__')
|
|
def redis_activity(_mock_redis_init):
|
|
logger = MagicMock()
|
|
notification_handler = MagicMock(spec=NotificationHandler)
|
|
activity = Redis(host='localhost', port=6379,
|
|
logger=logger, notification_handler=notification_handler,
|
|
username='test', password='test')
|
|
|
|
activity.redis_client = MagicMock()
|
|
activity.logger = logger
|
|
activity.notification_handler = notification_handler
|
|
return activity
|
|
|
|
|
|
@patch('scouter.activities.redis.RedisBase.__init__')
|
|
def test_redis_initialization(mock_redis_init):
|
|
"""Test Redis activity initialization"""
|
|
logger = MagicMock()
|
|
notification_handler = MagicMock(spec=NotificationHandler)
|
|
Redis(host='localhost', port=6379,
|
|
logger=logger, notification_handler=notification_handler,
|
|
username='test', password='test')
|
|
mock_redis_init.assert_called_once_with(
|
|
ANY,
|
|
'localhost',
|
|
6379,
|
|
'test',
|
|
'test',
|
|
logger,
|
|
notification_handler
|
|
)
|
|
|
|
|
|
metadata = {
|
|
'metadata': {
|
|
'model_id': 'test_model_id',
|
|
'model_name': 'test_model',
|
|
'schedule_name': 'test_schedule',
|
|
'workflow_name': 'scouter'
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_last_data_timestamp_none(redis_activity):
|
|
"""Test get_last_data_timestamp"""
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_pipeline',
|
|
'schedule_name': 'test_schedule'
|
|
}
|
|
|
|
redis_activity.get = MagicMock(return_value=None)
|
|
|
|
result = await redis_activity.get_last_data_timestamp(test_data)
|
|
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_last_data_timestamp_not_none(redis_activity):
|
|
"""Test get_last_data_timestamp"""
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_pipeline',
|
|
'schedule_name': 'test_schedule'
|
|
}
|
|
|
|
redis_activity.get = MagicMock(return_value='2023-01-01 12:00:00')
|
|
|
|
result = await redis_activity.get_last_data_timestamp(test_data)
|
|
|
|
redis_activity.get.assert_called_once_with(
|
|
'last_data_timestamp_test_pipeline_test_schedule'
|
|
)
|
|
|
|
assert result == '2023-01-01 12:00:00'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_last_data_timestamp_error(redis_activity):
|
|
"""Test get_last_data_timestamp error"""
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_pipeline',
|
|
'schedule_name': 'test_schedule'
|
|
}
|
|
|
|
redis_activity.send_notification = MagicMock()
|
|
redis_activity.get = MagicMock(side_effect=Exception('test'))
|
|
|
|
try:
|
|
|
|
await redis_activity.get_last_data_timestamp(test_data)
|
|
|
|
except Exception as e:
|
|
assert str(e) == 'test'
|
|
|
|
redis_activity.send_notification.assert_called_once_with(
|
|
metadata=metadata['metadata'],
|
|
notification_id="REDIS_GET_ERROR",
|
|
message="Error getting last data timestamp: test",
|
|
block="get_last_data_timestamp",
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=ANY
|
|
)
|
|
|
|
else:
|
|
assert False, "Expected exception"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_put_last_data_timestamp_empty_dataframe(redis_activity):
|
|
"""Test put_last_data_timestamp with empty dataframe"""
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_pipeline',
|
|
'schedule_name': 'test_schedule',
|
|
'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records')
|
|
}
|
|
|
|
redis_activity.set = MagicMock()
|
|
|
|
result = await redis_activity.put_last_data_timestamp(test_data)
|
|
|
|
assert result is None
|
|
|
|
redis_activity.set.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_put_last_data_timestamp_not_empty_dataframe(redis_activity):
|
|
"""Test put_last_data_timestamp with not empty dataframe"""
|
|
|
|
data = DataFrame({
|
|
'name': ['sensor1', 'sensor2'],
|
|
'value': [25.5, 30.0],
|
|
'inserted_at': ['2023-01-01 12:00:00', '2023-01-01 12:00:01']
|
|
})
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_pipeline',
|
|
'schedule_name': 'test_schedule',
|
|
'data': data.to_dict('records')
|
|
}
|
|
|
|
redis_activity.set = MagicMock()
|
|
|
|
result = await redis_activity.put_last_data_timestamp(test_data)
|
|
|
|
assert result == '2023-01-01 12:00:01'
|
|
|
|
redis_activity.set.assert_called_once_with(
|
|
'last_data_timestamp_test_pipeline_test_schedule',
|
|
'2023-01-01 12:00:01',
|
|
ttl=18000
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_put_last_data_timestamp_error(redis_activity):
|
|
"""Test put_last_data_timestamp error"""
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_pipeline',
|
|
'schedule_name': 'test_schedule',
|
|
'data': DataFrame({
|
|
'name': ['sensor1', 'sensor2'],
|
|
'value': [25.5, 30.0],
|
|
'inserted_at': ['2023-01-01 12:00:00'] * 2
|
|
}).to_dict('records')
|
|
}
|
|
|
|
redis_activity.send_notification = MagicMock()
|
|
redis_activity.set = MagicMock(side_effect=Exception('test'))
|
|
|
|
try:
|
|
await redis_activity.put_last_data_timestamp(test_data)
|
|
|
|
except Exception as e:
|
|
assert str(e) == 'test'
|
|
|
|
redis_activity.send_notification.assert_called_once_with(
|
|
metadata=metadata['metadata'],
|
|
notification_id="REDIS_SET_ERROR",
|
|
message="Error setting last data timestamp: test",
|
|
block="put_last_data_timestamp",
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=ANY
|
|
)
|
|
|
|
else:
|
|
assert False, "Expected exception"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_group_and_hold_data_new_key(redis_activity):
|
|
"""Test group_and_hold_data with a new key"""
|
|
# Setup
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_pipeline',
|
|
'schedule_name': 'test_schedule',
|
|
'retention_time': 3600,
|
|
'model_id': 1,
|
|
'data': DataFrame({
|
|
'name': ['sensor1', 'sensor2'],
|
|
'value': [25.5, 30.0],
|
|
'timestamp': ['2023-01-01 12:00:00'] * 2
|
|
}).to_dict('records')
|
|
}
|
|
|
|
# Mock get to return None for new key
|
|
redis_activity.get = MagicMock(return_value=None)
|
|
redis_activity.set = MagicMock()
|
|
|
|
# Call the method
|
|
result = await redis_activity.group_and_hold_data(test_data)
|
|
|
|
# Verify the result
|
|
expected_result = {
|
|
'timestamp': {0: '2023-01-01 12:00:00', 1: '2023-01-01 12:00:00'},
|
|
'variable': {0: 'sensor1', 1: 'sensor2'},
|
|
'value': {0: 25.5, 1: 30.0},
|
|
'model_id': {0: 1, 1: 1}
|
|
}
|
|
assert result == expected_result
|
|
|
|
# Verify set was called with correct arguments
|
|
redis_activity.set.assert_called_once()
|
|
args, kwargs = redis_activity.set.call_args
|
|
assert args[0] == 'held_data_test_pipeline_test_schedule'
|
|
assert args[1] == {
|
|
'sensor1': 25.5,
|
|
'sensor2': 30.0,
|
|
'timestamp': '2023-01-01 12:00:00'
|
|
}
|
|
assert kwargs['ttl'] == 3600
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_group_and_hold_data_update_existing(redis_activity):
|
|
"""Test updating existing data with group_and_hold_data"""
|
|
# Setup initial data in Redis
|
|
existing_data = {
|
|
'sensor1': 20.0,
|
|
'sensor2': 28.0,
|
|
'timestamp': '2023-01-01 11:00:00'
|
|
}
|
|
|
|
# New data to update with
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_workflow',
|
|
'schedule_name': 'test_schedule',
|
|
'retention_time': 3600,
|
|
'model_id': 1,
|
|
'data': DataFrame({
|
|
'name': ['sensor1', 'sensor3'],
|
|
'value': [25.5, 42.0],
|
|
'timestamp': ['2023-01-01 12:00:00'] * 2
|
|
}).to_dict('records')
|
|
}
|
|
|
|
# Mock get to return existing data
|
|
redis_activity.get = MagicMock(return_value=existing_data)
|
|
redis_activity.set = MagicMock()
|
|
|
|
# Call the method
|
|
result = await redis_activity.group_and_hold_data(test_data)
|
|
|
|
# Verify the result
|
|
expected_result = {
|
|
'timestamp': {0: '2023-01-01 12:00:00', 1: '2023-01-01 12:00:00', 2: '2023-01-01 12:00:00'},
|
|
'variable': {0: 'sensor1', 1: 'sensor2', 2: 'sensor3'},
|
|
'value': {0: 25.5, 1: 28.0, 2: 42.0},
|
|
'model_id': {0: 1, 1: 1, 2: 1}
|
|
}
|
|
assert result == expected_result
|
|
|
|
# Verify set was called with correct arguments
|
|
redis_activity.set.assert_called_once()
|
|
args, kwargs = redis_activity.set.call_args
|
|
assert args[0] == 'held_data_test_workflow_test_schedule'
|
|
assert args[1] == {
|
|
'sensor1': 25.5,
|
|
'sensor2': 28.0,
|
|
'sensor3': 42.0,
|
|
'timestamp': '2023-01-01 12:00:00'
|
|
}
|
|
assert kwargs['ttl'] == 3600
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_group_and_hold_data_with_none_values(redis_activity):
|
|
"""Test handling of None values in group_and_hold_data"""
|
|
# Setup test data with None values
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_workflow',
|
|
'schedule_name': 'test_schedule',
|
|
'retention_time': 3600,
|
|
'model_id': 1,
|
|
'data': DataFrame({
|
|
'name': ['sensor1', 'sensor2'],
|
|
'value': [None, 30.0],
|
|
'timestamp': [datetime(2023, 1, 1, 12, 0, 0)] * 2
|
|
}).to_dict('records')
|
|
}
|
|
|
|
# Mock get to return None for new key
|
|
redis_activity.get = MagicMock(return_value=None)
|
|
redis_activity.set = MagicMock()
|
|
|
|
# Call the method
|
|
result = await redis_activity.group_and_hold_data(test_data)
|
|
|
|
# Verify None was converted to np.nan and values are as expected
|
|
assert np.isnan(result['value'][0])
|
|
assert result['value'][1] == pytest.approx(30.0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_group_and_hold_data_empty_dataframe(redis_activity):
|
|
"""Test group_and_hold_data with empty DataFrame"""
|
|
# Setup test with empty data
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_workflow',
|
|
'schedule_name': 'test_schedule',
|
|
'retention_time': 3600,
|
|
'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records')
|
|
}
|
|
|
|
redis_activity.get = MagicMock(return_value=None)
|
|
|
|
# Call the method
|
|
result = await redis_activity.group_and_hold_data(test_data)
|
|
|
|
assert result == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_group_and_hold_data_error_get(redis_activity):
|
|
"""Test group_and_hold_data error"""
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_workflow',
|
|
'schedule_name': 'test_schedule',
|
|
'retention_time': 3600,
|
|
'model_id': 1,
|
|
'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records')
|
|
}
|
|
|
|
redis_activity.get = MagicMock(side_effect=Exception('test'))
|
|
redis_activity.send_notification = MagicMock()
|
|
|
|
try:
|
|
await redis_activity.group_and_hold_data(test_data)
|
|
|
|
except Exception as e:
|
|
assert str(e) == 'test'
|
|
|
|
redis_activity.send_notification.assert_called_once_with(
|
|
metadata=metadata['metadata'],
|
|
notification_id="REDIS_GET_ERROR",
|
|
message="Error getting held data: test",
|
|
block="group_and_hold_data",
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=ANY
|
|
)
|
|
|
|
else:
|
|
assert False, "Expected exception"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_group_and_hold_data_error_set(redis_activity):
|
|
"""Test group_and_hold_data error"""
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_workflow',
|
|
'schedule_name': 'test_schedule',
|
|
'retention_time': 3600,
|
|
'model_id': 1,
|
|
'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records')
|
|
}
|
|
|
|
existing_data = {
|
|
'sensor1': 20.0,
|
|
'sensor2': 28.0,
|
|
'timestamp': '2023-01-01 11:00:00'
|
|
}
|
|
|
|
# Mock get to return existing data
|
|
redis_activity.get = MagicMock(return_value=existing_data)
|
|
redis_activity.set = MagicMock(side_effect=Exception('test'))
|
|
redis_activity.send_notification = MagicMock()
|
|
|
|
try:
|
|
await redis_activity.group_and_hold_data(test_data)
|
|
|
|
except Exception as e:
|
|
assert str(e) == 'test'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_store_data_package(redis_activity):
|
|
"""Test store_data_package"""
|
|
redis_activity.set = MagicMock()
|
|
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_workflow',
|
|
'schedule_name': 'test_schedule',
|
|
'held_data': DataFrame({
|
|
'name': ['sensor1', 'sensor2'],
|
|
'value': [25.5, 30.0],
|
|
'timestamp': ['2023-01-01 12:00:00'] * 2
|
|
}).to_dict(),
|
|
'data': DataFrame({
|
|
'name': ['sensor1', 'sensor2'],
|
|
'value': [25.5, 30.0],
|
|
'timestamp': ['2023-01-01 12:00:00'] * 2
|
|
}).to_dict()
|
|
}
|
|
|
|
await redis_activity.store_data_package(test_data)
|
|
|
|
redis_activity.set.assert_called_once_with(
|
|
ANY,
|
|
{
|
|
'data': test_data['data'],
|
|
'held_data': test_data['held_data']
|
|
},
|
|
ttl=120)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_store_data_package_error(redis_activity):
|
|
"""Test store_data_package error"""
|
|
redis_activity.set = MagicMock(side_effect=Exception('test'))
|
|
redis_activity.send_notification = MagicMock()
|
|
|
|
test_data = {
|
|
**metadata,
|
|
'workflow_name': 'test_workflow',
|
|
'schedule_name': 'test_schedule',
|
|
'held_data': DataFrame({
|
|
'name': ['sensor1', 'sensor2'],
|
|
'value': [25.5, 30.0],
|
|
'timestamp': ['2023-01-01 12:00:00'] * 2
|
|
}).to_dict(),
|
|
'data': DataFrame({
|
|
'name': ['sensor1', 'sensor2'],
|
|
'value': [25.5, 30.0],
|
|
'timestamp': ['2023-01-01 12:00:00'] * 2
|
|
}).to_dict()
|
|
}
|
|
|
|
with pytest.raises(Exception):
|
|
await redis_activity.store_data_package(test_data)
|
|
|
|
redis_activity.send_notification.assert_called_once_with(
|
|
metadata=metadata['metadata'],
|
|
notification_id="REDIS_SET_ERROR",
|
|
message="Error setting data package: test",
|
|
block="store_data_package",
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=ANY
|
|
)
|