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=None ) @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') } redis_activity.get = MagicMock(return_value=None) 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 )