341 lines
9.9 KiB
Python
341 lines
9.9 KiB
Python
from unittest.mock import ANY, MagicMock, Mock, 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,
|
|
headers_config={
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json',
|
|
'x-requested-with': 'piwebapistreams',
|
|
'User-Agent': 'Aig-Scouter-Agent/1.0',
|
|
},
|
|
)
|
|
|
|
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()
|
|
|
|
|
|
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 = Mock(return_value=mock_df)
|
|
|
|
# Execute
|
|
result = 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',
|
|
end_time='*',
|
|
max_count=10,
|
|
metadata=metadata['metadata'],
|
|
request_timeout=30,
|
|
)
|
|
|
|
assert len(result) == 3
|
|
assert result[0]['name'] == 'tag1'
|
|
assert result[0]['value'] == pytest.approx(10.5)
|
|
assert result[1]['name'] == 'tag2'
|
|
assert result[2]['name'] == 'tag3'
|
|
assert result[0]['timestamp'] == '2023-01-01 12:02:00+0000'
|
|
assert result[1]['timestamp'] == '2023-01-01 12:02:00+0000'
|
|
assert result[2]['timestamp'] == '2023-01-01 12:02:00+0000'
|
|
|
|
|
|
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 = Mock(return_value=mock_df)
|
|
|
|
# Execute
|
|
result = 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',
|
|
end_time='*',
|
|
max_count=1,
|
|
metadata=metadata['metadata'],
|
|
request_timeout=15,
|
|
)
|
|
|
|
assert len(result) == 1
|
|
|
|
|
|
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 = Mock(return_value=mock_df)
|
|
|
|
# Execute
|
|
result = 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)
|
|
|
|
|
|
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 = Mock(
|
|
side_effect=Exception('PI Web API connection error')
|
|
)
|
|
api_activity.send_notification = MagicMock()
|
|
|
|
# Execute and verify exception is raised
|
|
with pytest.raises(Exception) as exc_info:
|
|
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.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,
|
|
)
|
|
|
|
|
|
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 = Mock(return_value=mock_df)
|
|
|
|
# Execute
|
|
result = api_activity.get_tag_values(test_data)
|
|
|
|
# Verify
|
|
assert len(result) == 2
|
|
assert result[0]['value'] == pytest.approx(10.0)
|
|
# NaN should be preserved in the result
|
|
assert pd.isna(result[1]['value'])
|