diff --git a/requirements.txt b/requirements.txt index f9857c6..3010834 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,4 +4,5 @@ sqlalchemy redis pymongo git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.6.1 -prometheus-client \ No newline at end of file +prometheus-client +pycurl \ No newline at end of file diff --git a/scouter/activities/activities.py b/scouter/activities/activities.py index 2f3f8f9..0cc3f50 100644 --- a/scouter/activities/activities.py +++ b/scouter/activities/activities.py @@ -9,12 +9,13 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.observability.logger import Logger from sientia_do.temporal.activities.postgres import Postgres + from scouter.activities.api import API from scouter.activities.gates import Gates from scouter.activities.mongodb import MongoDB from scouter.activities.redis import Redis -class Activities(Postgres, Redis, Gates, MongoDB): +class Activities(Postgres, Redis, Gates, MongoDB, API): """ Unified activities class that combines multiple data processing services. @@ -24,6 +25,7 @@ class Activities(Postgres, Redis, Gates, MongoDB): - Redis operations for caching and temporary storage - Data quality gates and filtering - MongoDB operations for data retrieval + - PI Web API operations for external data ingestion - Notification handling and logging The class implements the multiple inheritance pattern to provide a unified @@ -35,6 +37,7 @@ class Activities(Postgres, Redis, Gates, MongoDB): postgres_config: dict[str, Any], redis_config: dict[str, Any], mongodb_config: dict[str, Any], + api_config: dict[str, Any], logger: Logger, notification_handler: NotificationHandler, ): @@ -48,6 +51,8 @@ class Activities(Postgres, Redis, Gates, MongoDB): Required fields: host, port, username, password mongodb_config (dict[str, Any]): MongoDB connection configuration. Required fields: connection_string, database_name + api_config (dict[str, Any]): PI Web API configuration. + Required fields: base_url, auth_type, auth_token logger (Logger): Logger instance for application logging notification_handler (NotificationHandler): Handler for system notifications """ @@ -100,6 +105,17 @@ class Activities(Postgres, Redis, Gates, MongoDB): metrics_controller=metrics_controller, ) + # Initialize API + API.__init__( + self, + base_url=api_config['base_url'], + auth_type=api_config['auth_type'], + auth_token=api_config['auth_token'], + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + self.pod_id = getenv('HOSTNAME', 'localhost') def shutdown(self): @@ -113,3 +129,4 @@ class Activities(Postgres, Redis, Gates, MongoDB): MongoDB.close(self) Redis.close(self) Gates.close(self) + API.close(self) diff --git a/scouter/activities/api.py b/scouter/activities/api.py new file mode 100644 index 0000000..f588736 --- /dev/null +++ b/scouter/activities/api.py @@ -0,0 +1,133 @@ +from temporalio import activity, workflow + +with workflow.unsafe.imports_passed_through(): + import traceback + from typing import Any + + from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.notifications.models import NotificationLevel + from sientia_do.observability.logger import Logger + from sientia_do.observability.metrics_controller import MetricsController + from sientia_do.observability.sientia_monitoring import SientiaMonitoring + + from scouter.utils.clients.pi_web_api_client import PIWebAPIClient + + +class API(SientiaMonitoring): + """ + PI Web API operations for data retrieval. + + This class provides Temporal activities for interacting with the PI Web API + to retrieve tag values and historical data. It implements: + - Tag value retrieval from PI Web API endpoints + - Data quality filtering and validation + - Error handling with notifications + - Metrics collection for monitoring + + The class wraps the PIWebAPIClient to provide Temporal-aware activity methods + that can be used in workflow orchestration. + """ + + def __init__( + self, + base_url: str, + auth_type: str, + auth_token: str, + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + ) -> None: + """ + Initialize API activity with PI Web API client. + + Args: + base_url (str): Base URL of the PI Web API server + auth_type (str): Authentication type ('basic' or 'bearer') + auth_token (str): Authentication token + logger (Logger): Logger instance for operation logging + notification_handler (NotificationHandler): Handler for system notifications + metrics_controller (MetricsController): Controller for metrics collection + """ + SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller) + self.pi_web_api_client = PIWebAPIClient( + base_url=base_url, + auth_config={ + 'type': auth_type, + 'token': auth_token, + }, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + def close(self) -> None: + """ + Close the PI Web API client and shutdown monitoring services. + """ + self.pi_web_api_client.close() + SientiaMonitoring.shutdown(self) + + @activity.defn(name='get_tag_values') + async def get_tag_values(self, input_data: dict[str, Any]) -> list[dict]: + """ + Retrieve tag values from PI Web API for specified WebIds. + + This activity fetches historical or real-time data from the PI Web API + for a set of configured tags. It returns the data as a list of dictionaries + suitable for further processing in the workflow. + + Args: + input_data (dict[str, Any]): Activity input parameters. + Required fields: + - metadata (dict[str, Any]): Workflow execution metadata + - endpoint (str): PI Web API endpoint path + - web_ids (dict[str, str | None]): Tag names mapped to WebIds + - period (dict[str, str]): Time period with 'start_time' field + - api_timeout (int): Request timeout in seconds + - max_count (int, optional): Maximum data points per tag. Defaults to 1 + + Returns: + list[dict]: List of data records, each containing: + - timestamp: Data point timestamp + - name: Tag name + - value: Numeric value + - tag: WebId + + Raises: + PIMSRequestError: If API request fails + Exception: If data retrieval or processing fails + """ + metadata = input_data['metadata'] + endpoint = input_data['endpoint'] + web_ids = input_data['web_ids'] + period = input_data['period'] + max_count = input_data.get('max_count', 1) + api_timeout = input_data['api_timeout'] + + self.info(f'Getting tag values from {endpoint}', metadata=metadata) + self.debug(f'Web IDs: {web_ids}', metadata=metadata) + try: + latest_values = await self.pi_web_api_client.get_latest_values_df( + endpoint=endpoint, + web_ids=web_ids, + start_time=period, + max_count=max_count, + metadata=metadata, + timeout=api_timeout, + ) + + except Exception as e: + await self.send_notification_async( + metadata=metadata, + notification_id='PI_WEB_API_REQUEST_ERROR', + message=f'Error getting tag values from PI Web API: {e}', + block='get_tag_values', + level=NotificationLevel.ERROR, + attachment_content=traceback.format_exc(), + ) + raise e + + self.debug(f'Latest values: {latest_values}', metadata=metadata) + self.info(f'Gathered {len(latest_values)} tag values', metadata=metadata) + + return latest_values.to_dict(orient='records') diff --git a/scouter/metrics.py b/scouter/metrics.py index b25fa46..3ef2288 100644 --- a/scouter/metrics.py +++ b/scouter/metrics.py @@ -1,4 +1,5 @@ -from prometheus_client import Counter, Gauge +from prometheus_client import Counter, Gauge, Histogram +from sientia_do.observability.metrics import CORE_LABELS as SIENTIA_CORE_LABELS # Application health and status metrics APP_UP = Gauge( @@ -23,3 +24,23 @@ TAG_CHANGES_MONITOR = Gauge( 'Current value change of each tag', [*CORE_LABELS, 'tag_name'], ) + + +# Generic REST client metrics +GENERIC_REST_CLIENT_LAG = Histogram( + 'scouter_generic_rest_client_lag', + 'Lag time for a request to a generic REST client', + SIENTIA_CORE_LABELS, +) + +GENERIC_REST_READ_COUNT = Counter( + 'scouter_generic_rest_client_read_count', + 'Number of reads from a generic REST client', + SIENTIA_CORE_LABELS, +) + +GENERIC_REST_READ_ERROR_COUNT = Counter( + 'scouter_generic_rest_client_read_error_count', + 'Number of read errors from a generic REST client', + SIENTIA_CORE_LABELS, +) diff --git a/scouter/utils/clients/pi_web_api_client.py b/scouter/utils/clients/pi_web_api_client.py new file mode 100644 index 0000000..0453a52 --- /dev/null +++ b/scouter/utils/clients/pi_web_api_client.py @@ -0,0 +1,335 @@ +import io +import json +import time +import warnings +from typing import Any +from urllib.parse import urlencode + +import pandas as pd +import pycurl +from sientia_do.notifications.handlers import NotificationHandler +from sientia_do.observability.logger import Logger +from sientia_do.observability.metrics_controller import MetricsController +from sientia_do.observability.sientia_monitoring import SientiaMonitoring + +from scouter import metrics + +warnings.simplefilter('ignore') # Ignore warnings such as 'verify=False' + + +class PIMSRequestError(Exception): + """Generic error for failed requests to PI Web API using pycurl.""" + + pass + + +class PIWebAPIClient(SientiaMonitoring): + """ + Client for interacting with the PI Web API. + + This class provides a robust interface for querying historical and real-time + data from OSIsoft PI systems through the PI Web API. It implements: + - Asynchronous HTTP requests using pycurl + - Authentication support (Basic and Bearer) + - Automatic data normalization and timestamp handling + - Comprehensive error handling and monitoring + - Metrics collection for observability + + The client is designed for high-performance data retrieval with proper + connection management and error recovery mechanisms. + """ + + def __init__( + self, + base_url: str, + auth_config: dict[str, Any], + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + headers_config: dict[str, Any] | None = None, + ) -> None: + """ + Initialize the PI Web API client with connection parameters. + + Args: + base_url (str): Base URL of the PI Web API server + auth_config (dict[str, Any]): Authentication configuration. + Required fields: + - type (str): Authentication type ('basic' or 'bearer') + - token (str): Authentication token + logger (Logger): Logger instance for operation logging + notification_handler (NotificationHandler): Handler for system notifications + metrics_controller (MetricsController): Controller for metrics collection + headers_config (dict[str, Any], optional): HTTP headers configuration. + Default headers include content-type, accept, and x-requested-with + max_concurrency (int, optional): Maximum number of concurrent requests. Defaults to 8 + """ + if headers_config is None: + headers_config = { + 'Content-Type': 'application/json', + 'Accept': 'application/json', + 'x-requested-with': 'XMLHttpRequest', + } + SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller) + + self.base_url = base_url.rstrip('/') + + # auth_config spec: + # 'type': 'basic' or 'bearer', + # 'token': 'token', + self.auth_config = auth_config + self.auth_config['type'] = self.auth_config['type'].lower() + self.headers: dict[str, str] = headers_config + + self.authenticate() + + def close(self) -> None: + """ + Close the client and shutdown monitoring services. + """ + SientiaMonitoring.shutdown(self) + + def _to_clean_timestamp(self, series: pd.Series) -> pd.Series: + """ + Convert a Series of timestamps to datetime, UTC, and round to the nearest second. + + Args: + series (pd.Series): Series containing timestamp values + + Returns: + pd.Series: Cleaned timestamp series in UTC, floored to seconds + """ + series = pd.to_datetime(series, utc=True, errors='coerce') + return series.dt.floor('s') + + def _extract_numeric(self, value: Any) -> float | None: + """ + Normalize a value (potentially nested) to float. + + This method handles PI Web API response values that may be nested + in dictionaries or other structures, extracting the numeric value. + + Args: + value (Any): Value to extract and normalize + + Returns: + float | None: Numeric value as float, or None if conversion fails + """ + if isinstance(value, dict): + value = value.get('Value', value) + return pd.to_numeric(value, errors='coerce') + + def authenticate(self): + """ + Configure authentication headers based on auth_config. + + This method sets up the Authorization header using either Basic or Bearer + authentication based on the configured authentication type. + + Raises: + ValueError: If authentication type is not 'basic' or 'bearer' + """ + self.logger.info(f'Authenticating with {self.auth_config["type"]} authentication') + + if self.auth_config['type'] == 'basic': + self.headers['Authorization'] = f'Basic {self.auth_config["token"]}' + elif self.auth_config['type'] == 'bearer': + self.headers['Authorization'] = f'Bearer {self.auth_config["token"]}' + else: + raise ValueError(f'Invalid authentication type: {self.auth_config["type"]}') + + async def _curl_get_json( + self, + url: str, + params: list[tuple[str, str]] | None = None, + timeout: int = 30, + verify: bool = True, + metadata: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """ + Perform a GET request using pycurl and return the decoded JSON response. + + This method executes an asynchronous HTTP GET request with proper error handling, + metrics collection, and timeout management. It automatically tracks request + latency and emits monitoring metrics. + + Args: + url (str): Target URL for the GET request + params (list[tuple[str, str]], optional): Query parameters as list of tuples. + Each tuple contains (parameter_name, parameter_value) + timeout (int, optional): Request timeout in seconds. Defaults to 30 + verify (bool, optional): Verify SSL certificates. Defaults to True + metadata (dict[str, Any], optional): Workflow execution metadata for tracking + + Returns: + dict[str, Any]: Parsed JSON response body + + Raises: + PIMSRequestError: If HTTP error, connection error, or JSON parsing error occurs + """ + if metadata is None: + metadata = {} + + buffer = io.BytesIO() + c = pycurl.Curl() + + core_labels = self.get_core_labels(metadata=metadata, operation_type='get_json') + + try: + if params: + query_string = urlencode(params, doseq=True) + full_url = f'{url}?{query_string}' + else: + full_url = url + + c.setopt(pycurl.URL, full_url.encode('utf-8')) + c.setopt(pycurl.WRITEDATA, buffer) + + # Configure HTTP headers + header_list = [f'{k}: {v}' for k, v in self.headers.items()] + if header_list: + c.setopt(pycurl.HTTPHEADER, header_list) + + # Set request timeout + c.setopt(pycurl.TIMEOUT, timeout) + + # Configure SSL verification + if not verify: + c.setopt(pycurl.SSL_VERIFYPEER, 0) + c.setopt(pycurl.SSL_VERIFYHOST, 0) + + start_time = time.time() + try: + c.perform() + except Exception as e: + await self.emit_metric( + metric_object=metrics.GENERIC_REST_READ_ERROR_COUNT, + tags=core_labels, + ) + raise e + await self.observe_lag( + start_time=start_time, + metric_object=metrics.GENERIC_REST_CLIENT_LAG, + tags=core_labels, + ) + await self.emit_metric( + metric_object=metrics.GENERIC_REST_READ_COUNT, + tags=core_labels, + ) + + status_code = c.getinfo(pycurl.RESPONSE_CODE) + body = buffer.getvalue().decode('utf-8', errors='replace') + + if status_code >= 400: + await self.emit_metric( + metric_object=metrics.GENERIC_REST_READ_ERROR_COUNT, + tags=core_labels, + ) + raise PIMSRequestError(f"HTTP {status_code} calling '{full_url}': {body[:200]}") + + try: + return json.loads(body) + except json.JSONDecodeError as e: + raise PIMSRequestError( + f"Error decoding JSON response from '{full_url}': {e}; body: {body[:200]}" + ) from e + + except pycurl.error as e: + raise PIMSRequestError(f"Connection error calling '{url}': {e}") from e + finally: + c.close() + + async def get_latest_values_df( + self, + web_ids: dict[str, dict[str, str]], + endpoint: str, + timeout: int = 30, + start_time: str = '*-1d', + end_time: str = '*', + max_count: int | None = 1, + metadata: dict[str, Any] | None = None, + ) -> pd.DataFrame: + """ + Retrieve historical values for multiple WebIds using PI Web API streamsets. + + This method queries the PI Web API's /streamsets/recorded endpoint to fetch + historical data for multiple tags simultaneously. It returns a normalized + DataFrame with timestamps, tag names, values, and WebIds. + + Args: + web_ids (dict[str, str | None]): Dictionary mapping tag names to their WebIds. + None values are filtered out before querying + endpoint (str): PI Web API endpoint path (e.g., '/streamsets/recorded') + timeout (int, optional): Request timeout in seconds. Defaults to 30 + start_time (str, optional): Start time in PI Web API format (e.g., "*-50d"). + Defaults to "*-1d" (1 day ago) + end_time (str, optional): End time in PI Web API format (e.g., "*"). + Defaults to "*" (current time) + max_count (int, optional): Maximum number of data points per series. + Defaults to 1. If None, maxCount parameter is not sent + metadata (dict[str, Any], optional): Workflow execution metadata for tracking + + Returns: + pd.DataFrame: DataFrame with columns: + - timestamp: Cleaned timestamp (UTC, floored to seconds) + - name: Tag name + - value: Numeric value (normalized) + - tag: WebId of the tag + Returns empty DataFrame if no data is found + + Raises: + PIMSRequestError: If API request fails or returns invalid data + """ + if metadata is None: + metadata = {} + + url = f'{self.base_url}{endpoint}' + + # Build query parameters with WebIds (filtering out None values) + params: list[tuple[str, str]] = [('webid', web_id['webid']) for web_id in web_ids.values()] + params.extend( + [ + ('startTime', start_time), + ('endtime', end_time), + ('selectedFields', 'Items.Name;Items.Items.Timestamp;Items.Items.Value'), + ] + ) + + params.append(('maxCount', str(max_count))) + + data = await self._curl_get_json( + url=url, + params=params, + timeout=timeout, + verify=False, + metadata=metadata, + ) + + raw_data = data.get('Items', []) + + records = [] + for entry in raw_data: + tag_name = entry.get('Name') + series_items = entry.get('Items', []) + for it in series_items: + if isinstance(it, dict) and 'Timestamp' in it and 'Value' in it: + ts = it.get('Timestamp') + val = it.get('Value') + web_id = web_ids.get(tag_name) + if ts is not None: + records.append( + { + 'timestamp': ts, + 'name': tag_name, + 'value': self._extract_numeric(val), + 'tag': web_id, + } + ) + + if not records: + return pd.DataFrame() + + df = pd.DataFrame.from_records(records) + df['timestamp'] = self._to_clean_timestamp(df['timestamp']) + + return df diff --git a/scouter/utils/connectors_config.py b/scouter/utils/connectors_config.py index 4264aa1..abb1879 100644 --- a/scouter/utils/connectors_config.py +++ b/scouter/utils/connectors_config.py @@ -83,6 +83,23 @@ def build_mongodb_config() -> dict[str, Any]: } +def build_api_config() -> dict[str, Any]: + """ + Build API connection configuration from environment variables. + + Returns: + dict[str, Any]: API configuration dictionary with keys: + - base_url: API base URL (default: https://pi.example.com) + - auth_type: API authentication type (default: basic) + - auth_token: API authentication token (default: None) + """ + return { + 'base_url': getenv('API_BASE_URL', 'https://pi.example.com'), + 'auth_type': getenv('API_AUTH_TYPE', 'basic'), + 'auth_token': getenv('API_AUTH_TOKEN', None), + } + + def build_druid_config() -> dict[str, Any]: """ Build Apache Druid connection configuration from environment variables. diff --git a/scouter/worker/worker.py b/scouter/worker/worker.py index cd333f4..9329bbf 100644 --- a/scouter/worker/worker.py +++ b/scouter/worker/worker.py @@ -15,6 +15,7 @@ with workflow.unsafe.imports_passed_through(): from scouter import metrics from scouter.activities.activities import Activities from scouter.utils.connectors_config import ( + build_api_config, build_mongodb_config, build_postgres_config, build_redis_config, @@ -103,6 +104,7 @@ async def main(): postgres_config=build_postgres_config(), redis_config=build_redis_config(), mongodb_config=build_mongodb_config(), + api_config=build_api_config(), ) logger.custom_info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata) diff --git a/scouter/workflow/pi_web_api_scouter.py b/scouter/workflow/pi_web_api_scouter.py new file mode 100644 index 0000000..313f42c --- /dev/null +++ b/scouter/workflow/pi_web_api_scouter.py @@ -0,0 +1,101 @@ +from temporalio import workflow + +with workflow.unsafe.imports_passed_through(): + from datetime import timedelta + from typing import Any + + from sientia_do.temporal.policies import retry_policy + + from scouter.activities.activities import Activities + + +@workflow.defn(name='pi_web_api_scouter') +class PIWebAPIScouter: + """ + PI Web API Scouter workflow that orchestrates data ingestion from PI systems. + + This workflow serves as the entry point for PI Web API data processing pipelines. + Unlike the standard Scouter that loads from MongoDB, this workflow directly queries + PI Web API endpoints to retrieve tag values and processes them for downstream use. + + The workflow implements a direct API ingestion pattern with: + - Real-time data retrieval from PI Web API + - Configurable time periods and data point limits + - Error handling and retry policies + - Child workflow orchestration for data processing + - Integration with CoreScouter for standardized processing + """ + + @workflow.run + async def run(self, input_data: dict[str, Any]) -> None: + """ + Execute the PI Web API Scouter workflow. + + This method orchestrates the complete data ingestion process from PI Web API: + 1. Retrieves tag values from PI Web API using configured WebIds + 2. Validates and normalizes the retrieved data + 3. Delegates data processing to the CoreScouter workflow + + Args: + input_data (dict[str, Any]): Configuration and parameters for the workflow execution. + Required fields: + - model_name (str): Name of the data model being processed + - model_id (str): Unique identifier for the data model + - schedule_name (str): Unique identifier for the data collection schedule + - endpoint (str): PI Web API endpoint path (e.g., '/streamsets/recorded') + - web_ids (dict[str, str | None]): Mapping of tag names to WebIds + - period (dict[str, str]): Time period configuration with 'start_time' + - api_timeout (int): Request timeout in seconds for PI Web API calls + - max_count (int, optional): Maximum data points per tag. Defaults to 1 + - trigger_laborious (bool): Flag to enable intensive data processing + - filters (dict[str, str]): Data quality filters configuration + - schema (str): Target database schema for data export + - table_name (str): Target table name for data export + - retention_time (int): Data retention period in Redis (seconds) + - model_tags (dict[str, Any]): Tag-specific configuration including: + - data_range: [min, max] values for data validation + - aggr_function: Aggregation method (avg, mdn, max, min, lts) + - frequency: Data collection frequency in milliseconds + - topics: List of Kafka topics for data routing + + Returns: + None: This workflow doesn't return data, it orchestrates data processing + + Raises: + WorkflowExecutionError: If workflow execution fails + ActivityExecutionError: If any activity fails after retry attempts + PIMSRequestError: If PI Web API request fails + """ + + input_data['workflow_name'] = 'scouter' + + metadata = { + 'metadata': { + 'model_id': input_data['model_id'], + 'model_name': input_data['model_name'], + 'schedule_name': input_data['schedule_name'], + 'workflow_name': input_data['workflow_name'], + } + } + + data = await workflow.execute_local_activity_method( + Activities.get_tag_values, + { + **metadata, + 'endpoint': input_data['endpoint'], + 'web_ids': input_data['model_tags'], + 'period': input_data['period'], + 'api_timeout': input_data['api_timeout'], + 'max_count': input_data.get('max_count', 1), + }, + start_to_close_timeout=timedelta(seconds=60), + retry_policy=retry_policy, + ) + + if not data: + return + + input_data['data'] = data + input_data['metadata'] = metadata + + await workflow.execute_child_workflow('subworkflow.core_scouter', input_data) diff --git a/tests/activities/test_activities.py b/tests/activities/test_activities.py index bc31dee..2d3f5ab 100644 --- a/tests/activities/test_activities.py +++ b/tests/activities/test_activities.py @@ -3,6 +3,7 @@ from unittest.mock import ANY, MagicMock, patch from sientia_do.temporal.activities.postgres import Postgres from scouter.activities.activities import Activities +from scouter.activities.api import API from scouter.activities.gates import Gates from scouter.activities.mongodb import MongoDB from scouter.activities.redis import Redis @@ -12,9 +13,15 @@ from scouter.activities.redis import Redis @patch('scouter.activities.activities.Postgres.__init__') @patch('scouter.activities.activities.Redis.__init__') @patch('scouter.activities.activities.Gates.__init__') +@patch('scouter.activities.activities.API.__init__') @patch('scouter.activities.activities.MetricsController') def test___init__( - mock_metrics_controller, mock_gates_init, mock_redis_init, mock_postgres_init, mock_mongodb_init + mock_metrics_controller, + mock_api_init, + mock_gates_init, + mock_redis_init, + mock_postgres_init, + mock_mongodb_init, ): postgres_config = { 'host': 'localhost', @@ -33,6 +40,12 @@ def test___init__( 'database_name': 'test_database', } + api_config = { + 'base_url': 'https://api.example.com', + 'auth_type': 'bearer', + 'auth_token': 'test_token', + } + logger = MagicMock() notification_handler = MagicMock() @@ -40,6 +53,7 @@ def test___init__( postgres_config=postgres_config, redis_config=redis_config, mongodb_config=mongodb_config, + api_config=api_config, logger=logger, notification_handler=notification_handler, ) @@ -49,6 +63,7 @@ def test___init__( assert isinstance(activities, Redis) assert isinstance(activities, MongoDB) assert isinstance(activities, Gates) + assert isinstance(activities, API) mock_postgres_init.assert_called_once_with( ANY, @@ -91,20 +106,34 @@ def test___init__( metrics_controller=mock_metrics_controller.return_value, ) + mock_api_init.assert_called_once_with( + ANY, + base_url=api_config['base_url'], + auth_type=api_config['auth_type'], + auth_token=api_config['auth_token'], + logger=logger, + notification_handler=notification_handler, + metrics_controller=mock_metrics_controller.return_value, + ) + @patch('scouter.activities.activities.Postgres.__init__') @patch('scouter.activities.activities.Redis.__init__') @patch('scouter.activities.activities.Gates.__init__') @patch('scouter.activities.activities.MongoDB.__init__') +@patch('scouter.activities.activities.API.__init__') @patch('scouter.activities.activities.Postgres.close') @patch('scouter.activities.activities.MongoDB.close') @patch('scouter.activities.activities.Redis.close') @patch('scouter.activities.activities.Gates.close') +@patch('scouter.activities.activities.API.close') def test_shutdown( + mock_api_close, mock_gates_close, mock_redis_close, mock_mongodb_close, mock_postgres_close, + _mock_api_init, _mock_mongodb_init, _mock_gates_init, _mock_redis_init, @@ -127,6 +156,12 @@ def test_shutdown( 'database_name': 'test_database', } + api_config = { + 'base_url': 'https://api.example.com', + 'auth_type': 'bearer', + 'auth_token': 'test_token', + } + logger = MagicMock() notification_handler = MagicMock() @@ -134,6 +169,7 @@ def test_shutdown( postgres_config=postgres_config, redis_config=redis_config, mongodb_config=mongodb_config, + api_config=api_config, logger=logger, notification_handler=notification_handler, ) @@ -144,3 +180,4 @@ def test_shutdown( mock_mongodb_close.assert_called() mock_redis_close.assert_called() mock_gates_close.assert_called() + mock_api_close.assert_called() diff --git a/tests/activities/test_api.py b/tests/activities/test_api.py new file mode 100644 index 0000000..8b49219 --- /dev/null +++ b/tests/activities/test_api.py @@ -0,0 +1,322 @@ +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', '2023-01-01 12:01:00', '2023-01-01 12:02:00'], + 'name': ['tag1', 'tag2', 'tag3'], + 'value': [10.5, 20.3, 30.7], + 'tag': ['webid1', 'webid2', 'webid3'], + } + ) + + 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' + + +@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'], + 'name': ['tag1'], + 'value': [42.0], + 'tag': ['webid1'], + } + ) + + 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', '2023-01-01 12:01:00'], + 'name': ['tag1', 'tag3'], + 'value': [10.5, 30.7], + 'tag': ['webid1', 'webid3'], + } + ) + + 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', '2023-01-01 12:00:00'], + 'name': ['tag1', 'tag2'], + 'value': [10.0, float('nan')], + 'tag': ['webid1', 'webid2'], + } + ) + + 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']) diff --git a/tests/utils/clients/__init__.py b/tests/utils/clients/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/utils/clients/test_pi_web_api_client.py b/tests/utils/clients/test_pi_web_api_client.py new file mode 100644 index 0000000..7c897e3 --- /dev/null +++ b/tests/utils/clients/test_pi_web_api_client.py @@ -0,0 +1,560 @@ +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pandas as pd +import pycurl +import pytest + +from scouter.utils.clients.pi_web_api_client import PIMSRequestError, PIWebAPIClient + + +@pytest.fixture +def mock_logger(): + return MagicMock() + + +@pytest.fixture +def mock_notification_handler(): + return AsyncMock() + + +@pytest.fixture +def mock_metrics_controller(): + return AsyncMock() + + +@pytest.fixture +def auth_config_basic(): + return {'type': 'basic', 'token': 'test_token_123'} + + +@pytest.fixture +def auth_config_bearer(): + return {'type': 'bearer', 'token': 'bearer_token_456'} + + +@pytest.fixture +def pi_client(mock_logger, mock_notification_handler, mock_metrics_controller, auth_config_basic): + return PIWebAPIClient( + base_url='https://pi.example.com', + auth_config=auth_config_basic, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + + +def test_init_with_basic_auth( + mock_logger, mock_notification_handler, mock_metrics_controller, auth_config_basic +): + """Test initialization with basic authentication""" + client = PIWebAPIClient( + base_url='https://pi.example.com/', + auth_config=auth_config_basic, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + + assert client.base_url == 'https://pi.example.com' + assert client.auth_config['type'] == 'basic' + assert client.headers['Authorization'] == 'Basic test_token_123' + assert client.headers['Content-Type'] == 'application/json' + assert client.headers['Accept'] == 'application/json' + mock_logger.info.assert_called_with('Authenticating with basic authentication') + + +def test_init_with_bearer_auth( + mock_logger, mock_notification_handler, mock_metrics_controller, auth_config_bearer +): + """Test initialization with bearer authentication""" + client = PIWebAPIClient( + base_url='https://pi.example.com', + auth_config=auth_config_bearer, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + + assert client.base_url == 'https://pi.example.com' + assert client.auth_config['type'] == 'bearer' + assert client.headers['Authorization'] == 'Bearer bearer_token_456' + mock_logger.info.assert_called_with('Authenticating with bearer authentication') + + +def test_init_with_custom_headers( + mock_logger, mock_notification_handler, mock_metrics_controller, auth_config_basic +): + """Test initialization with custom headers""" + custom_headers = { + 'Content-Type': 'application/xml', + 'Custom-Header': 'custom_value', + } + + client = PIWebAPIClient( + base_url='https://pi.example.com', + auth_config=auth_config_basic, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + headers_config=custom_headers, + ) + + assert client.headers['Content-Type'] == 'application/xml' + assert client.headers['Custom-Header'] == 'custom_value' + assert client.headers['Authorization'] == 'Basic test_token_123' + + +def test_authenticate_invalid_type(mock_logger, mock_notification_handler, mock_metrics_controller): + """Test that invalid authentication type raises ValueError""" + invalid_auth_config = {'type': 'invalid', 'token': 'test_token'} + + with pytest.raises(ValueError) as exc_info: + PIWebAPIClient( + base_url='https://pi.example.com', + auth_config=invalid_auth_config, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + + assert 'Invalid authentication type: invalid' in str(exc_info.value) + + +@patch('scouter.utils.clients.pi_web_api_client.SientiaMonitoring.shutdown') +def test_close(mock_shutdown, pi_client): + """Test close method calls shutdown""" + pi_client.close() + + mock_shutdown.assert_called_once() + + +def test_to_clean_timestamp(pi_client): + """Test timestamp cleaning and normalization""" + timestamps = pd.Series( + [ + '2025-01-15T10:30:45.123456Z', + '2025-01-15T10:30:46.789012Z', + '2025-01-15T10:30:47.999999Z', + ] + ) + + result = pi_client._to_clean_timestamp(timestamps) + + assert isinstance(result, pd.Series) + assert result.dtype == 'datetime64[ns, UTC]' + # Verify microseconds are floored to seconds + assert result[0] == pd.Timestamp('2025-01-15T10:30:45Z') + assert result[1] == pd.Timestamp('2025-01-15T10:30:46Z') + assert result[2] == pd.Timestamp('2025-01-15T10:30:47Z') + + +def test_to_clean_timestamp_with_invalid_values(pi_client): + """Test timestamp cleaning with invalid values returns NaT""" + timestamps = pd.Series(['invalid', 'not_a_date', '2025-01-15T10:30:45Z']) + + result = pi_client._to_clean_timestamp(timestamps) + + assert pd.isna(result[0]) + assert pd.isna(result[1]) + assert result[2] == pd.Timestamp('2025-01-15T10:30:45Z') + + +def test_extract_numeric_with_float(pi_client): + """Test extracting numeric value from float""" + result = pi_client._extract_numeric(42.5) + + assert result == 42.5 + + +def test_extract_numeric_with_int(pi_client): + """Test extracting numeric value from int""" + result = pi_client._extract_numeric(42) + + assert result == 42.0 + + +def test_extract_numeric_with_string(pi_client): + """Test extracting numeric value from string""" + result = pi_client._extract_numeric('123.45') + + assert result == 123.45 + + +def test_extract_numeric_with_dict(pi_client): + """Test extracting numeric value from dictionary""" + result = pi_client._extract_numeric({'Value': 99.9}) + + assert result == 99.9 + + +def test_extract_numeric_with_invalid_value(pi_client): + """Test extracting numeric value from invalid value returns None/NaN""" + result = pi_client._extract_numeric('invalid_number') + + assert pd.isna(result) + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_success(mock_curl_class, pi_client): + """Test successful GET request with JSON response""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + response_data = {'status': 'success', 'data': [1, 2, 3]} + response_json = json.dumps(response_data).encode('utf-8') + + def mock_perform(): + buffer = mock_curl.setopt.call_args_list[1][0][1] + buffer.write(response_json) + + mock_curl.perform.side_effect = mock_perform + mock_curl.getinfo.return_value = 200 + + result = await pi_client._curl_get_json('https://pi.example.com/api/test') + + assert result == response_data + mock_curl.setopt.assert_any_call(pycurl.TIMEOUT, 30) + mock_curl.perform.assert_called_once() + mock_curl.close.assert_called_once() + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_with_params(mock_curl_class, pi_client): + """Test GET request with query parameters""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + response_data = {'result': 'ok'} + response_json = json.dumps(response_data).encode('utf-8') + + def mock_perform(): + buffer = mock_curl.setopt.call_args_list[1][0][1] + buffer.write(response_json) + + mock_curl.perform.side_effect = mock_perform + mock_curl.getinfo.return_value = 200 + + params = [('key1', 'value1'), ('key2', 'value2')] + result = await pi_client._curl_get_json('https://pi.example.com/api', params=params) + + assert result == response_data + # Verify URL includes query parameters + set_url_call = [call for call in mock_curl.setopt.call_args_list if call[0][0] == pycurl.URL][0] + assert b'key1=value1' in set_url_call[0][1] + assert b'key2=value2' in set_url_call[0][1] + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_http_error(mock_curl_class, pi_client): + """Test GET request with HTTP error response""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + error_response = b'{"error": "Not found"}' + + def mock_perform(): + buffer = mock_curl.setopt.call_args_list[1][0][1] + buffer.write(error_response) + + mock_curl.perform.side_effect = mock_perform + mock_curl.getinfo.return_value = 404 + + with pytest.raises(PIMSRequestError) as exc_info: + await pi_client._curl_get_json('https://pi.example.com/api/notfound') + + assert 'HTTP 404' in str(exc_info.value) + mock_curl.close.assert_called_once() + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_connection_error(mock_curl_class, pi_client): + """Test GET request with connection error""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + mock_curl.perform.side_effect = pycurl.error('Connection failed') + + with pytest.raises(PIMSRequestError) as exc_info: + await pi_client._curl_get_json('https://pi.example.com/api/test') + + assert 'Connection error' in str(exc_info.value) + mock_curl.close.assert_called_once() + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_invalid_json(mock_curl_class, pi_client): + """Test GET request with invalid JSON response""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + invalid_json = b'This is not valid JSON' + + def mock_perform(): + buffer = mock_curl.setopt.call_args_list[1][0][1] + buffer.write(invalid_json) + + mock_curl.perform.side_effect = mock_perform + mock_curl.getinfo.return_value = 200 + + with pytest.raises(PIMSRequestError) as exc_info: + await pi_client._curl_get_json('https://pi.example.com/api/test') + + assert 'Error decoding JSON response' in str(exc_info.value) + mock_curl.close.assert_called_once() + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_with_custom_timeout(mock_curl_class, pi_client): + """Test GET request with custom timeout""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + response_data = {'status': 'ok'} + response_json = json.dumps(response_data).encode('utf-8') + + def mock_perform(): + buffer = mock_curl.setopt.call_args_list[1][0][1] + buffer.write(response_json) + + mock_curl.perform.side_effect = mock_perform + mock_curl.getinfo.return_value = 200 + + await pi_client._curl_get_json('https://pi.example.com/api/test', timeout=60) + + mock_curl.setopt.assert_any_call(pycurl.TIMEOUT, 60) + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_without_ssl_verify(mock_curl_class, pi_client): + """Test GET request with SSL verification disabled""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + response_data = {'status': 'ok'} + response_json = json.dumps(response_data).encode('utf-8') + + def mock_perform(): + buffer = mock_curl.setopt.call_args_list[1][0][1] + buffer.write(response_json) + + mock_curl.perform.side_effect = mock_perform + mock_curl.getinfo.return_value = 200 + + await pi_client._curl_get_json('https://pi.example.com/api/test', verify=False) + + mock_curl.setopt.assert_any_call(pycurl.SSL_VERIFYPEER, 0) + mock_curl.setopt.assert_any_call(pycurl.SSL_VERIFYHOST, 0) + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_success(mock_curl_get_json, pi_client): + """Test successful retrieval of latest values""" + mock_curl_get_json.return_value = { + 'Items': [ + { + 'Name': 'tag1', + 'Items': [ + {'Timestamp': '2025-01-15T10:30:00Z', 'Value': 42.5}, + {'Timestamp': '2025-01-15T10:31:00Z', 'Value': 43.0}, + ], + }, + { + 'Name': 'tag2', + 'Items': [ + {'Timestamp': '2025-01-15T10:30:00Z', 'Value': 100.0}, + ], + }, + ] + } + + web_ids = { + 'tag1': {'webid': 'webid1'}, + 'tag2': {'webid': 'webid2'}, + } + + result = await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + start_time='*-1d', + end_time='*', + max_count=10, + ) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 3 + assert list(result.columns) == ['timestamp', 'name', 'value', 'tag'] + assert result['name'].tolist() == ['tag1', 'tag1', 'tag2'] + assert result['value'].tolist() == [42.5, 43.0, 100.0] + + mock_curl_get_json.assert_called_once() + call_args = mock_curl_get_json.call_args + assert call_args[1]['url'] == 'https://pi.example.com/streamsets/recorded' + assert call_args[1]['timeout'] == 30 + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_with_custom_params(mock_curl_get_json, pi_client): + """Test get_latest_values_df with custom parameters""" + mock_curl_get_json.return_value = { + 'Items': [ + { + 'Name': 'tag1', + 'Items': [ + {'Timestamp': '2025-01-15T10:30:00Z', 'Value': 42.5}, + ], + } + ] + } + + web_ids = {'tag1': {'webid': 'webid1'}} + metadata = {'model_id': 'test_model'} + + result = await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + start_time='*-7d', + end_time='*-1d', + max_count=100, + timeout=60, + metadata=metadata, + ) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 1 + + mock_curl_get_json.assert_called_once() + call_args = mock_curl_get_json.call_args + params = call_args[1]['params'] + + # Verify parameters + assert ('startTime', '*-7d') in params + assert ('endtime', '*-1d') in params + assert ('maxCount', '100') in params + assert call_args[1]['timeout'] == 60 + assert call_args[1]['metadata'] == metadata + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_empty_response(mock_curl_get_json, pi_client): + """Test get_latest_values_df with empty response""" + mock_curl_get_json.return_value = {'Items': []} + + web_ids = {'tag1': {'webid': 'webid1'}} + + result = await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + ) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 0 + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_no_items_in_tag(mock_curl_get_json, pi_client): + """Test get_latest_values_df when tag has no items""" + mock_curl_get_json.return_value = { + 'Items': [ + { + 'Name': 'tag1', + 'Items': [], + } + ] + } + + web_ids = {'tag1': {'webid': 'webid1'}} + + result = await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + ) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 0 + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_with_missing_timestamp(mock_curl_get_json, pi_client): + """Test get_latest_values_df filters out items with missing timestamp""" + mock_curl_get_json.return_value = { + 'Items': [ + { + 'Name': 'tag1', + 'Items': [ + {'Timestamp': '2025-01-15T10:30:00Z', 'Value': 42.5}, + {'Value': 43.0}, # Missing Timestamp + {'Timestamp': None, 'Value': 44.0}, # None Timestamp + ], + } + ] + } + + web_ids = {'tag1': {'webid': 'webid1'}} + + result = await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + ) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 1 # Only the first item should be included + assert result['value'].tolist() == [42.5] + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_with_nested_value(mock_curl_get_json, pi_client): + """Test get_latest_values_df with nested value extraction""" + mock_curl_get_json.return_value = { + 'Items': [ + { + 'Name': 'tag1', + 'Items': [ + {'Timestamp': '2025-01-15T10:30:00Z', 'Value': {'Value': 42.5}}, + ], + } + ] + } + + web_ids = {'tag1': {'webid': 'webid1'}} + + result = await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + ) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 1 + assert result['value'].tolist() == [42.5] + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_default_max_count(mock_curl_get_json, pi_client): + """Test get_latest_values_df uses default max_count of 1""" + mock_curl_get_json.return_value = {'Items': []} + + web_ids = {'tag1': {'webid': 'webid1'}} + + await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + ) + + call_args = mock_curl_get_json.call_args + params = call_args[1]['params'] + + assert ('maxCount', '1') in params diff --git a/tests/utils/test_connectors_config.py b/tests/utils/test_connectors_config.py index d57b335..f71a0fc 100644 --- a/tests/utils/test_connectors_config.py +++ b/tests/utils/test_connectors_config.py @@ -4,6 +4,7 @@ from unittest.mock import patch import pytest from scouter.utils.connectors_config import ( + build_api_config, build_druid_config, build_kafka_config, build_mongodb_config, @@ -154,6 +155,38 @@ def test_build_mongodb_config_with_env_vars(): } +@pytest.mark.usefixtures('mock_env_vars') +def test_build_api_config_defaults(): + """Test that build_api_config returns default values when no env vars are set""" + config = build_api_config() + + assert config == { + 'base_url': 'https://pi.example.com', + 'auth_type': 'basic', + 'auth_token': None, + } + + +@pytest.mark.usefixtures('mock_env_vars') +def test_build_api_config_with_env_vars(): + """Test that build_api_config uses env vars when set""" + with patch.dict( + os.environ, + { + 'API_BASE_URL': 'https://api.production.com', + 'API_AUTH_TYPE': 'bearer', + 'API_AUTH_TOKEN': 'secret_token_123', + }, + ): + config = build_api_config() + + assert config == { + 'base_url': 'https://api.production.com', + 'auth_type': 'bearer', + 'auth_token': 'secret_token_123', + } + + def test_build_druid_config_defaults(): """Test that build_druid_config returns default values when no env vars are set""" config = build_druid_config() diff --git a/tests/workflow/test_pi_web_api_scouter.py b/tests/workflow/test_pi_web_api_scouter.py new file mode 100644 index 0000000..5348411 --- /dev/null +++ b/tests/workflow/test_pi_web_api_scouter.py @@ -0,0 +1,126 @@ +from unittest.mock import ANY, AsyncMock, patch + +from pytest import fixture, mark + +from scouter.activities.activities import Activities +from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter + + +@fixture +def pi_web_api_scouter(): + return PIWebAPIScouter() + + +@mark.asyncio +@patch('scouter.workflow.pi_web_api_scouter.workflow', new_callable=AsyncMock) +async def test_pi_web_api_scouter_workflow(mock_workflow, pi_web_api_scouter): + mock_workflow.execute_local_activity_method.return_value = 'test_data' + await pi_web_api_scouter.run( + input_data={ + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'schedule_name': 'test_schedule', + 'endpoint': '/streamsets/recorded', + 'model_tags': {'tag1': 'webid1', 'tag2': 'webid2'}, + 'period': {'start_time': '2025-01-01T00:00:00Z'}, + 'api_timeout': 30, + 'max_count': 10, + 'trigger_laborious': True, + 'filters': {'quality': 'good'}, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'retention_time': 3600, + } + ) + + expected_metadata = { + 'metadata': { + 'model_id': 'test_model_id', + 'model_name': 'test_model', + 'schedule_name': 'test_schedule', + 'workflow_name': 'scouter', + } + } + + mock_workflow.execute_local_activity_method.assert_called_once_with( + Activities.get_tag_values, + { + **expected_metadata, + 'endpoint': '/streamsets/recorded', + 'web_ids': {'tag1': 'webid1', 'tag2': 'webid2'}, + 'period': {'start_time': '2025-01-01T00:00:00Z'}, + 'api_timeout': 30, + 'max_count': 10, + }, + start_to_close_timeout=ANY, + retry_policy=ANY, + ) + + mock_workflow.execute_child_workflow.assert_called_once_with( + 'subworkflow.core_scouter', + { + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'schedule_name': 'test_schedule', + 'endpoint': '/streamsets/recorded', + 'model_tags': {'tag1': 'webid1', 'tag2': 'webid2'}, + 'period': {'start_time': '2025-01-01T00:00:00Z'}, + 'api_timeout': 30, + 'max_count': 10, + 'trigger_laborious': True, + 'filters': {'quality': 'good'}, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'retention_time': 3600, + 'workflow_name': 'scouter', + 'data': 'test_data', + 'metadata': expected_metadata, + }, + ) + + +@mark.asyncio +@patch('scouter.workflow.pi_web_api_scouter.workflow', new_callable=AsyncMock) +async def test_pi_web_api_scouter_workflow_empty(mock_workflow, pi_web_api_scouter): + mock_workflow.execute_local_activity_method.return_value = [] + await pi_web_api_scouter.run( + input_data={ + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'schedule_name': 'test_schedule', + 'endpoint': '/streamsets/recorded', + 'model_tags': {'tag1': 'webid1', 'tag2': 'webid2'}, + 'period': {'start_time': '2025-01-01T00:00:00Z'}, + 'api_timeout': 30, + 'trigger_laborious': True, + 'filters': {'quality': 'good'}, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'retention_time': 3600, + } + ) + + expected_metadata = { + 'metadata': { + 'model_id': 'test_model_id', + 'model_name': 'test_model', + 'schedule_name': 'test_schedule', + 'workflow_name': 'scouter', + } + } + + mock_workflow.execute_local_activity_method.assert_called_once_with( + Activities.get_tag_values, + { + **expected_metadata, + 'endpoint': '/streamsets/recorded', + 'web_ids': {'tag1': 'webid1', 'tag2': 'webid2'}, + 'period': {'start_time': '2025-01-01T00:00:00Z'}, + 'api_timeout': 30, + 'max_count': 1, + }, + start_to_close_timeout=ANY, + retry_policy=ANY, + ) + + mock_workflow.execute_child_workflow.assert_not_called()