from unittest.mock import ANY, AsyncMock, MagicMock, patch import pandas as pd import pytest from sientia_do.notifications.models import NotificationLevel from scouter.activities.api import API @pytest.fixture @patch('scouter.activities.api.PIWebAPIClient') def api_activity(mock_pi_web_api_client): """Fixture to create an API activity instance with mocked dependencies.""" logger = MagicMock() notification_handler = MagicMock() metrics_controller = MagicMock() activity = API( base_url='https://pi.example.com', auth_type='basic', auth_token='test_token', logger=logger, notification_handler=notification_handler, metrics_controller=metrics_controller, ) activity.logger = logger activity.notification_handler = notification_handler activity.metrics_controller = metrics_controller activity.pod_id = 'test_pod_id' return activity metadata = { 'metadata': { 'model_id': 'test_model_id', 'model_name': 'test_model', 'schedule_name': 'test_schedule', 'workflow_name': 'pi_web_api_scouter', } } @patch('scouter.activities.api.PIWebAPIClient') def test_api_initialization(mock_pi_web_api_client): """Test API activity initialization.""" logger = MagicMock() notification_handler = MagicMock() metrics_controller = MagicMock() activity = API( base_url='https://pi.example.com', auth_type='basic', auth_token='test_token', logger=logger, notification_handler=notification_handler, metrics_controller=metrics_controller, ) mock_pi_web_api_client.assert_called_once_with( base_url='https://pi.example.com', auth_config={ 'type': 'basic', 'token': 'test_token', }, logger=logger, notification_handler=notification_handler, metrics_controller=metrics_controller, ) assert activity.pi_web_api_client is not None @patch('scouter.activities.api.SientiaMonitoring') def test_close(mock_sientia_monitoring, api_activity): """Test close method.""" api_activity.close() api_activity.pi_web_api_client.close.assert_called_once() mock_sientia_monitoring.shutdown.assert_called_once() @pytest.mark.asyncio async def test_get_tag_values_success(api_activity): """Test get_tag_values with successful data retrieval.""" # Setup test data test_data = { **metadata, 'endpoint': '/streamsets/recorded', 'web_ids': { 'tag1': { 'webid': 'webid1', 'aggr_function': 'avg', 'data_range': [0, 100], }, 'tag2': { 'webid': 'webid2', 'aggr_function': 'avg', 'data_range': [0, 100], }, 'tag3': { 'webid': 'webid3', 'aggr_function': 'avg', 'data_range': [0, 100], }, }, 'period': '*-1d', 'max_count': 10, 'api_timeout': 30, } # Mock DataFrame response mock_df = pd.DataFrame( { 'timestamp': [ '2023-01-01 12:00:00+0000', '2023-01-01 12:01:00+0000', '2023-01-01 12:02:00+0000', ], 'name': ['tag1', 'tag2', 'tag3'], 'value': [10.5, 20.3, 30.7], 'tag': ['webid1', 'webid2', 'webid3'], } ) mock_df['timestamp'] = pd.to_datetime(mock_df['timestamp']) api_activity.pi_web_api_client.get_latest_values_df = AsyncMock(return_value=mock_df) # Execute result = await api_activity.get_tag_values(test_data) # Verify api_activity.pi_web_api_client.get_latest_values_df.assert_called_once_with( endpoint='/streamsets/recorded', web_ids={ 'tag1': {'webid': 'webid1', 'aggr_function': 'avg', 'data_range': [0, 100]}, 'tag2': {'webid': 'webid2', 'aggr_function': 'avg', 'data_range': [0, 100]}, 'tag3': {'webid': 'webid3', 'aggr_function': 'avg', 'data_range': [0, 100]}, }, start_time='*-1d', max_count=10, metadata=metadata['metadata'], timeout=30, ) assert len(result) == 3 assert result[0]['name'] == 'tag1' assert result[0]['value'] == 10.5 assert result[1]['name'] == 'tag2' assert result[2]['name'] == 'tag3' assert result[0]['timestamp'] == '2023-01-01 12:00:00+0000' assert result[1]['timestamp'] == '2023-01-01 12:01:00+0000' assert result[2]['timestamp'] == '2023-01-01 12:02:00+0000' @pytest.mark.asyncio async def test_get_tag_values_with_default_max_count(api_activity): """Test get_tag_values with default max_count value.""" # Setup test data without max_count test_data = { **metadata, 'endpoint': '/streamsets/recorded', 'web_ids': { 'tag1': { 'webid': 'webid1', 'aggr_function': 'avg', 'data_range': [0, 100], } }, 'period': '*-1h', 'api_timeout': 15, } mock_df = pd.DataFrame( { 'timestamp': ['2023-01-01 12:00:00+0000'], 'name': ['tag1'], 'value': [42.0], 'tag': ['webid1'], } ) mock_df['timestamp'] = pd.to_datetime(mock_df['timestamp']) api_activity.pi_web_api_client.get_latest_values_df = AsyncMock(return_value=mock_df) # Execute result = await api_activity.get_tag_values(test_data) # Verify default max_count is 1 api_activity.pi_web_api_client.get_latest_values_df.assert_called_once_with( endpoint='/streamsets/recorded', web_ids={'tag1': {'webid': 'webid1', 'aggr_function': 'avg', 'data_range': [0, 100]}}, start_time='*-1h', max_count=1, metadata=metadata['metadata'], timeout=15, ) assert len(result) == 1 @pytest.mark.asyncio async def test_get_tag_values_with_none_webids(api_activity): """Test get_tag_values with some None WebIds.""" # Setup test data with None values test_data = { **metadata, 'endpoint': '/streamsets/recorded', 'web_ids': { 'tag1': { 'webid': 'webid1', 'aggr_function': 'avg', 'data_range': [0, 100], }, 'tag2': None, 'tag3': { 'webid': 'webid3', 'aggr_function': 'max', 'data_range': [0, 200], }, }, 'period': '*-1h', 'max_count': 5, 'api_timeout': 20, } mock_df = pd.DataFrame( { 'timestamp': ['2023-01-01 12:00:00+0000', '2023-01-01 12:01:00+0000'], 'name': ['tag1', 'tag3'], 'value': [10.5, 30.7], 'tag': ['webid1', 'webid3'], } ) mock_df['timestamp'] = pd.to_datetime(mock_df['timestamp']) api_activity.pi_web_api_client.get_latest_values_df = AsyncMock(return_value=mock_df) # Execute result = await api_activity.get_tag_values(test_data) # Verify - should only query non-None WebIds assert len(result) == 2 assert all(r['name'] in ['tag1', 'tag3'] for r in result) @pytest.mark.asyncio async def test_get_tag_values_api_error(api_activity): """Test get_tag_values when PI Web API client raises an error and sends notification.""" # Setup test data test_data = { **metadata, 'endpoint': '/streamsets/recorded', 'web_ids': { 'tag1': { 'webid': 'webid1', 'aggr_function': 'avg', 'data_range': [0, 100], } }, 'period': '*-1d', 'max_count': 1, 'api_timeout': 30, } # Mock API error api_activity.pi_web_api_client.get_latest_values_df = AsyncMock( side_effect=Exception('PI Web API connection error') ) api_activity.send_notification_async = AsyncMock() # Execute and verify exception is raised with pytest.raises(Exception) as exc_info: await api_activity.get_tag_values(test_data) assert str(exc_info.value) == 'PI Web API connection error' # Verify notification was sent api_activity.send_notification_async.assert_called_once_with( metadata=metadata['metadata'], notification_id='PI_WEB_API_REQUEST_ERROR', message='Error getting tag values from PI Web API: PI Web API connection error', block='get_tag_values', level=NotificationLevel.ERROR, attachment_content=ANY, ) @pytest.mark.asyncio async def test_get_tag_values_with_nan_values(api_activity): """Test get_tag_values handling NaN values in the DataFrame.""" # Setup test data test_data = { **metadata, 'endpoint': '/streamsets/recorded', 'web_ids': { 'tag1': { 'webid': 'webid1', 'aggr_function': 'avg', 'data_range': [0, 100], }, 'tag2': { 'webid': 'webid2', 'aggr_function': 'avg', 'data_range': [0, 100], }, }, 'period': '*-1d', 'max_count': 1, 'api_timeout': 30, } # Mock DataFrame with NaN values mock_df = pd.DataFrame( { 'timestamp': ['2023-01-01 12:00:00+0000', '2023-01-01 12:00:00+0000'], 'name': ['tag1', 'tag2'], 'value': [10.0, float('nan')], 'tag': ['webid1', 'webid2'], } ) mock_df['timestamp'] = pd.to_datetime(mock_df['timestamp']) api_activity.pi_web_api_client.get_latest_values_df = AsyncMock(return_value=mock_df) # Execute result = await api_activity.get_tag_values(test_data) # Verify assert len(result) == 2 assert result[0]['value'] == 10.0 # NaN should be preserved in the result assert pd.isna(result[1]['value'])