diff --git a/.env.example b/.env.example index 28055e7..a4ba811 100644 --- a/.env.example +++ b/.env.example @@ -6,14 +6,18 @@ POSTGRES_DB=sientia POSTGRES_MIN_CONNECTIONS=5 POSTGRES_MAX_CONNECTIONS=20 -KAFKA_BOOTSTRAP_SERVERS=localhost:9092 -KAFKA_POLLING_TIME=1000 +MONGODB_USERNAME="root" +MONGODB_PASSWORD="password" +MONGODB_URL="my-release-mongodb.mongodb.svc.cluster.local:27017" +MONGODB_DATABASE="sientia" REDIS_HOST=localhost REDIS_PORT=6379 +REDIS_USERNAME="user" +REDIS_PASSWORD="pass" TEMPORAL_HOST=localhost:7233 -TEMPORAL_NAMESPACE=default +TEMPORAL_NAMESPACE=scouter LOG_LEVEL=INFO diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index cba1cce..4c48fb6 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -8,6 +8,7 @@ on: jobs: release: + if: github.event.pull_request.merged == true uses: Aignosi/github_workflow_templates/.github/workflows/dataops-module-release.yml@main permissions: write-all with: diff --git a/init_port_forward.sh b/init_port_forward.sh new file mode 100755 index 0000000..e9c0e3d --- /dev/null +++ b/init_port_forward.sh @@ -0,0 +1,133 @@ +#!/bin/bash + +# Usage: +# 1) Edit the PORT_FORWARDS list below with entries of: +# +# 2) Run: ./init_port_forward.sh +# +# The script will start all port-forwards in the background and keep running +# until interrupted (Ctrl+C). On exit, it will clean up started port-forward processes. + +set -euo pipefail + +# Define your namespace/service/port combinations here +# Example entries: +# "default my-service 8080 80" +# "observability grafana 3000 3000" +PORT_FORWARDS=( + "mongodb my-release-mongodb 27017 27017" + "paradedb paradedb-rw 5432 5432" + "redis redis-master 6379 6379" + "temporal temporal-frontend 7233 7233" +) + +if [ ${#PORT_FORWARDS[@]} -eq 0 ]; then + echo "No port-forward entries defined. Edit PORT_FORWARDS in $(basename "$0")." + exit 1 +fi + +PIDS=() + +cleanup() { + echo "\nStopping port-forward processes..." + for pid in "${PIDS[@]}"; do + if kill -0 "$pid" >/dev/null 2>&1; then + kill "$pid" >/dev/null 2>&1 || true + fi + done +} + +trap cleanup EXIT INT TERM + +timestamp() { date '+%Y-%m-%d %H:%M:%S'; } + +# Allow overriding kubectl binary if needed +KUBECTL=${KUBECTL:-kubectl} + +is_port_free() { + local port="$1" + # Consider port free if nothing is listening locally on it + if command -v ss >/dev/null 2>&1; then + ! ss -ltn | awk '{print $4}' | grep -E "(^|:|\\])${port}$" >/dev/null 2>&1 + else + if command -v lsof >/dev/null 2>&1; then + ! lsof -tiTCP:"${port}" -sTCP:LISTEN >/dev/null 2>&1 + else + # Fallback: attempt to open a TCP connection; expect failure when nothing is listening + ! (exec 3<>"/dev/tcp/127.0.0.1/${port}") 2>/dev/null + fi + fi +} + +free_port_if_stuck() { + local port="$1" + # Try multiple tools to free a stuck listener (often old kubectl PF) + if command -v lsof >/dev/null 2>&1; then + local pids + pids=$(lsof -tiTCP:"${port}" -sTCP:LISTEN 2>/dev/null || true) + if [ -n "${pids}" ]; then + echo "[$(timestamp)] Found listeners on ${port}: ${pids}; terminating" + kill ${pids} 2>/dev/null || true + sleep 0.5 + fi + fi + if ! is_port_free "${port}"; then + if command -v fuser >/dev/null 2>&1; then + echo "[$(timestamp)] Forcing free of ${port} via fuser" + fuser -k "${port}/tcp" 2>/dev/null || true + sleep 0.5 + fi + fi +} + +run_port_forward() { + local namespace="$1" + local service_name="$2" + local local_port="$3" + local service_port="$4" + + # simple and robust supervisor loop with gentle backoff on failures + local delay=2 + local max_delay=20 + while true; do + free_port_if_stuck "${local_port}" + if ! is_port_free "${local_port}"; then + echo "[$(timestamp)] ns=${namespace} svc=${service_name} ${local_port}:${service_port} -> local port busy, retrying in 3s" + sleep 3 + continue + fi + + echo "[$(timestamp)] Starting port-forward: ns=${namespace} svc=${service_name} ${local_port}:${service_port}" + ${KUBECTL} -n "${namespace}" port-forward "svc/${service_name}" "${local_port}:${service_port}" \ + --address=127.0.0.1 --pod-running-timeout=2m --request-timeout=0 + rc=$? + + # If kubectl exits (e.g., connection reset by peer), wait a bit and retry + echo "[$(timestamp)] Port-forward exited (rc=${rc}): ns=${namespace} svc=${service_name} ${local_port}:${service_port}" + sleep "${delay}" + # Exponential backoff up to max_delay + if [ ${delay} -lt ${max_delay} ]; then + delay=$(( delay * 2 )) + if [ ${delay} -gt ${max_delay} ]; then + delay=${max_delay} + fi + fi + done +} + +ONLY_SERVICE_NAME="${ONLY_SERVICE_NAME:-}" + +for entry in "${PORT_FORWARDS[@]}"; do + read -r NAMESPACE SERVICE_NAME LOCAL_PORT SERVICE_PORT <<< "$entry" + if [ -n "${ONLY_SERVICE_NAME}" ] && [ "${SERVICE_NAME}" != "${ONLY_SERVICE_NAME}" ]; then + continue + fi + run_port_forward "${NAMESPACE}" "${SERVICE_NAME}" "${LOCAL_PORT}" "${SERVICE_PORT}" & + PIDS+=("$!") +done + +echo "All port-forwards started: ${#PIDS[@]} process(es). Press Ctrl+C to stop." + +# Do not exit the script if one port-forward fails; they self-restart +set +e +wait diff --git a/pyproject.toml b/pyproject.toml index 517fead..14d6e25 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -114,10 +114,6 @@ python_functions = ["test_*"] addopts = [ "-v", "--strict-markers", - "--cov=model_manager", - "--cov-report=term-missing", - "--cov-report=html", - "--cov-report=xml", ] markers = [ "asyncio: marks tests as async", @@ -126,7 +122,7 @@ markers = [ ] [tool.coverage.run] -source = ["model_manager"] +source = ["scouter"] omit = [ "*/tests/*", "*/venv/*", diff --git a/requirements.txt b/requirements.txt index 8589717..1dc88da 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,6 +5,6 @@ asyncua redis aiokafka pymongo -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.7 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3 pydruid[pandas] prometheus-client \ No newline at end of file diff --git a/scouter/activities/activities.py b/scouter/activities/activities.py index 79ddfe1..2f3f8f9 100644 --- a/scouter/activities/activities.py +++ b/scouter/activities/activities.py @@ -1,3 +1,4 @@ +from sientia_do.observability.metrics_controller import MetricsController from temporalio import workflow with workflow.unsafe.imports_passed_through(): @@ -50,6 +51,10 @@ class Activities(Postgres, Redis, Gates, MongoDB): logger (Logger): Logger instance for application logging notification_handler (NotificationHandler): Handler for system notifications """ + metrics_controller = MetricsController( + logger=logger, + ) + # Initialize Postgres Postgres.__init__( self, @@ -62,6 +67,7 @@ class Activities(Postgres, Redis, Gates, MongoDB): max_connections=postgres_config['max_connections'], logger=logger, notification_handler=notification_handler, + metrics_controller=metrics_controller, ) # Initialize Redis @@ -73,10 +79,16 @@ class Activities(Postgres, Redis, Gates, MongoDB): notification_handler=notification_handler, username=redis_config['username'], password=redis_config['password'], + metrics_controller=metrics_controller, ) # Initialize Gates - Gates.__init__(self, logger=logger, notification_handler=notification_handler) + Gates.__init__( + self, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) # Initialize MongoDB MongoDB.__init__( @@ -85,6 +97,7 @@ class Activities(Postgres, Redis, Gates, MongoDB): database_name=mongodb_config['database_name'], logger=logger, notification_handler=notification_handler, + metrics_controller=metrics_controller, ) self.pod_id = getenv('HOSTNAME', 'localhost') @@ -97,4 +110,6 @@ class Activities(Postgres, Redis, Gates, MongoDB): to prevent connection leaks and ensure graceful application termination. """ Postgres.close(self) - MongoDB.shutdown(self) + MongoDB.close(self) + Redis.close(self) + Gates.close(self) diff --git a/scouter/activities/gates.py b/scouter/activities/gates.py index ca9e2ce..0c2df2e 100644 --- a/scouter/activities/gates.py +++ b/scouter/activities/gates.py @@ -1,5 +1,6 @@ from collections.abc import Hashable +from sientia_do.observability.metrics_controller import MetricsController from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): @@ -10,7 +11,7 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger - from sientia_do.temporal.activities.base import BaseActivity + from sientia_do.observability.sientia_monitoring import SientiaMonitoring from scouter import metrics from scouter.utils.quality.filters import null_values_filter, out_of_bounds_filter @@ -21,7 +22,7 @@ quality_gate_filters = { } -class Gates(BaseActivity): +class Gates(SientiaMonitoring): """ Data quality gates and filtering operations. @@ -36,17 +37,29 @@ class Gates(BaseActivity): ensure data integrity and enable flexible data processing workflows. """ - def __init__(self, logger: Logger, notification_handler: NotificationHandler): + def __init__( + self, + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + ): """ Initialize the Gates class with logging and notification services. Args: logger (Logger): Logger instance for operation logging notification_handler (NotificationHandler): Handler for system notifications + metrics_controller (MetricsController): Metrics controller instance """ - BaseActivity.__init__(self, logger, notification_handler, set_error_counter=True) + SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller) - def apply_aggregation( + def close(self): + """ + Close the Gates class. + """ + SientiaMonitoring.shutdown(self) + + async def apply_aggregation( self, values: DataFrame, aggr_function: str, metadata: dict[str, Any] ) -> float | None | str: """ @@ -93,7 +106,7 @@ class Gates(BaseActivity): if aggr_function in aggregation_map: return aggregation_map[aggr_function](clean_values) else: - self.send_notification( + await self.send_notification_async( metadata=metadata, notification_id='AGGREGATION_ISSUES', message=f'Invalid aggregation function: {aggr_function}', @@ -152,7 +165,7 @@ class Gates(BaseActivity): # Get the latest timestamp (last row since data is sorted) latest_timestamp = group['timestamp'].iloc[-1] - aggr_value = self.apply_aggregation(group, aggr_function, metadata) + aggr_value = await self.apply_aggregation(group, aggr_function, metadata) if aggr_value == 'continue': continue @@ -184,7 +197,7 @@ class Gates(BaseActivity): except Exception as e: trace = traceback.format_exc() - self.send_notification( + await self.send_notification_async( metadata=metadata, notification_id='AGGREGATION_ISSUES', message=f'Error aggregating data: {e}', @@ -242,7 +255,7 @@ class Gates(BaseActivity): except Exception as e: trace = traceback.format_exc() - self.send_notification( + await self.send_notification_async( metadata=metadata, notification_id='DATA_QUALITY_GATE_ISSUES', message=f'Error applying filter {filter_name}: {e}', @@ -260,7 +273,7 @@ class Gates(BaseActivity): message = f'{len(filtered_data)} rows has quality issues: {filter_name}: {policy}' attachment = filtered_data.to_string() - self.send_notification( + await self.send_notification_async( metadata=metadata, notification_id=f'DATA_QUALITY_GATE_ISSUES__{filter_name}', message=message, @@ -291,7 +304,7 @@ class Gates(BaseActivity): metrics.LABORIOUS_DATA_WRITTEN_COUNT.labels( pod_id=self.pod_id, model_name=metadata['model_name'], - pipeline_name=metadata['workflow_name'], + workflow_name=metadata['workflow_name'], ).inc() # Register metrics @@ -299,7 +312,7 @@ class Gates(BaseActivity): metrics.TAG_CHANGES_MONITOR.labels( pod_id=self.pod_id, model_name=metadata['model_name'], - pipeline_name=metadata['workflow_name'], + workflow_name=metadata['workflow_name'], tag_name=row['variable'], ).set(row['value']) diff --git a/scouter/activities/mongodb.py b/scouter/activities/mongodb.py index a20e356..55d269e 100644 --- a/scouter/activities/mongodb.py +++ b/scouter/activities/mongodb.py @@ -1,6 +1,6 @@ -from collections.abc import Hashable from datetime import UTC +from sientia_do.observability.metrics_controller import MetricsController from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): @@ -8,50 +8,15 @@ with workflow.unsafe.imports_passed_through(): from datetime import datetime from typing import Any - from pandas import DataFrame - from pymongo import MongoClient from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger - from sientia_do.temporal.activities.base import BaseActivity + from sientia_do.observability.sientia_monitoring import SientiaMonitoring + from sientia_do.repository.mongodb_repository import MongoDBRepository from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ -def clear_mongo_id(docs: list) -> list: - """ - Remove MongoDB internal `_id` fields from documents. - - This utility function recursively removes the MongoDB `_id` field from - documents and nested structures. It's used to clean data before - processing or export operations. - - Args: - docs (list): List of documents to clean - - Returns: - list: Documents with `_id` fields removed - - Note: - This function modifies the input list in-place and returns the same reference - """ - for doc in docs: - if isinstance(doc, list): - clear_mongo_id(doc) - - elif isinstance(doc, dict): - if '_id' in doc: - del doc['_id'] - - for _key, value in doc.items(): - if isinstance(value, list): - clear_mongo_id(value) - elif isinstance(value, dict): - clear_mongo_id([value]) - - return docs - - -class MongoDB(BaseActivity): +class MongoDB(SientiaMonitoring): """ MongoDB operations for data retrieval and storage. @@ -72,6 +37,7 @@ class MongoDB(BaseActivity): database_name: str, logger: Logger, notification_handler: NotificationHandler, + metrics_controller: MetricsController, ): """ Initialize MongoDB connection and services. @@ -85,37 +51,28 @@ class MongoDB(BaseActivity): Raises: ConnectionError: If MongoDB connection fails """ - self.connection_string = connection_string - self.database_name = database_name - self.client: MongoClient = MongoClient( - self.connection_string, serverSelectionTimeoutMS=5000 - ) - self.client.server_info() # Trigger an exception if connection fails - - self.database = self.client[self.database_name] - - # Initialize MongoDB client here (omitted for brevity) - logger.info('MongoDB connection initialized') - - BaseActivity.__init__( - self, logger=logger, notification_handler=notification_handler, set_error_counter=True + self.mongodb_repository = MongoDBRepository( + connection_string=connection_string, + database_name=database_name, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, ) - def shutdown(self): - """ - Gracefully close MongoDB client connection. + SientiaMonitoring.__init__( + self, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) - This method ensures proper cleanup of MongoDB connections to prevent - connection leaks and ensure graceful application termination. + def close(self): """ - try: - if self.client: - self.logger.info('Closing MongoDB connection...') - self.client.close() - self.logger.info('MongoDB connection closed successfully') - except Exception as e: - self.logger.error(f'Failed to close MongoDB connection: {e}') + Close the MongoDB connection. + """ + self.mongodb_repository.close() + SientiaMonitoring.shutdown(self) def __del__(self): """ @@ -124,10 +81,10 @@ class MongoDB(BaseActivity): This destructor ensures that MongoDB connections are properly closed when the object is garbage collected, preventing resource leaks. """ - self.shutdown() + self.close() @activity.defn(name='load_latest_data') - async def load_latest_data(self, input_data: dict[str, Any]) -> dict[Hashable, Any]: + async def load_latest_data(self, input_data: dict[str, Any]) -> list[dict[str, Any]]: """ Load the latest data from MongoDB collection since a specified timestamp. @@ -166,9 +123,11 @@ class MongoDB(BaseActivity): self.debug(f'Data filter: {data_filter}', metadata=metadata) - data = list(self.database[collection_name].find(data_filter, {'_id': 0})) - - data = clear_mongo_id(data) + data = await self.mongodb_repository.find( + collection_name=collection_name, + filters=data_filter, + metadata=metadata, + ) self.debug(f'Collected: {data}', metadata=metadata) @@ -181,10 +140,10 @@ class MongoDB(BaseActivity): self.debug(f'Loaded data: {data}', metadata=metadata) - return DataFrame(data).to_dict() + return data except Exception as e: trace = traceback.format_exc() - self.send_notification( + await self.send_notification_async( metadata=metadata, notification_id='MONGO_LOAD_ERROR', message=f'Error loading data from MongoDB: {e}', diff --git a/scouter/activities/redis.py b/scouter/activities/redis.py index 9c76fe4..57083af 100644 --- a/scouter/activities/redis.py +++ b/scouter/activities/redis.py @@ -1,20 +1,21 @@ -from collections.abc import Hashable - +from sientia_do.observability.metrics_controller import MetricsController from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): import traceback + from collections.abc import Hashable from typing import Any from pandas import DataFrame from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger - from sientia_do.temporal.activities.redis_base import Redis as RedisBase + from sientia_do.observability.sientia_monitoring import SientiaMonitoring + from sientia_do.repository.redis_repository import RedisRepository from sientia_do.temporal.constants import DATETIME_FORMAT, now -class Redis(RedisBase): +class Redis(SientiaMonitoring): """ Redis operations for data caching and temporary storage. @@ -37,6 +38,7 @@ class Redis(RedisBase): password: str, logger: Logger, notification_handler: NotificationHandler, + metrics_controller: MetricsController, ): """ Initialize Redis connection and services. @@ -49,7 +51,23 @@ class Redis(RedisBase): logger (Logger): Logger instance for operation logging notification_handler (NotificationHandler): Handler for system notifications """ - RedisBase.__init__(self, host, port, username, password, logger, notification_handler) + SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller) + self.redis_repository = RedisRepository( + host=host, + port=port, + username=username, + password=password, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + def close(self): + """ + Close the Redis connection. + """ + self.redis_repository.close() + SientiaMonitoring.shutdown(self) @activity.defn(name='get_last_data_timestamp') async def get_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None: @@ -76,12 +94,12 @@ class Redis(RedisBase): metadata = input_data['metadata'] key = f'last_data_timestamp:{input_data["workflow_name"]}:{input_data["schedule_name"]}' - self.info(f'Getting last data timestamp for {key}') + self.info(f'Getting last data timestamp for {key}', metadata=metadata) try: - data_hold = self.get(key) + data_hold = await self.redis_repository.get(key, metadata=metadata) except Exception as e: - self.send_notification( + await self.send_notification_async( metadata=metadata, notification_id='REDIS_GET_ERROR', message=f'Error getting last data timestamp: {e}', @@ -124,7 +142,7 @@ class Redis(RedisBase): metadata = input_data['metadata'] key = f'last_data_timestamp:{input_data["workflow_name"]}:{input_data["schedule_name"]}' - self.info(f'Putting last data timestamp for {key}') + self.info(f'Putting last data timestamp for {key}', metadata=metadata) data = DataFrame(input_data['data']) @@ -137,9 +155,11 @@ class Redis(RedisBase): self.info(f'Last collected timestamp to insert: {last_data_timestamp}', metadata=metadata) try: - self.set(key, last_data_timestamp, ttl=60 * 60 * 5) + await self.redis_repository.set( + key, last_data_timestamp, ttl=60 * 60 * 5, metadata=metadata + ) except Exception as e: - self.send_notification( + await self.send_notification_async( metadata=metadata, notification_id='REDIS_SET_ERROR', message=f'Error setting last data timestamp: {e}', @@ -187,12 +207,12 @@ class Redis(RedisBase): key = f'held_data_{input_data["workflow_name"]}_{input_data["schedule_name"]}' - self.info(f'Getting held data for {key}') + self.info(f'Getting held data for {key}', metadata=metadata) try: - data_hold = self.get(key) + data_hold = await self.redis_repository.get(key, metadata=metadata) except Exception as e: - self.send_notification( + await self.send_notification_async( metadata=metadata, notification_id='REDIS_GET_ERROR', message=f'Error getting held data: {e}', @@ -234,7 +254,7 @@ class Redis(RedisBase): data['timestamp'].max() if not data.empty else data_hold['timestamp'] ) - self.set(key, data_hold, ttl=retention_time) + await self.redis_repository.set(key, data_hold, ttl=retention_time, metadata=metadata) data_hold_df = DataFrame(data_hold, index=[0]) data_hold_melted = data_hold_df.melt( @@ -244,7 +264,7 @@ class Redis(RedisBase): data_hold_melted.reset_index(drop=True, inplace=True) except Exception as e: - self.send_notification( + await self.send_notification_async( metadata=metadata, notification_id='REDIS_SET_ERROR', message=f'Error setting held data: {e}', @@ -280,9 +300,9 @@ class Redis(RedisBase): cache = {'data': data.to_dict(), 'held_data': held_data.to_dict()} try: - self.set(key, cache, ttl=120) + await self.redis_repository.set(key, cache, ttl=120, metadata=metadata) except Exception as e: - self.send_notification( + await self.send_notification_async( metadata=metadata, notification_id='REDIS_SET_ERROR', message=f'Error setting data package: {e}', diff --git a/scouter/metrics.py b/scouter/metrics.py index ac249c7..b25fa46 100644 --- a/scouter/metrics.py +++ b/scouter/metrics.py @@ -8,7 +8,7 @@ APP_UP = Gauge( ) # Core labels for consistent metric labeling -CORE_LABELS = ['pod_id', 'model_name', 'pipeline_name'] +CORE_LABELS = ['pod_id', 'model_name', 'workflow_name'] # Data processing metrics LABORIOUS_DATA_WRITTEN_COUNT = Counter( diff --git a/scouter/worker/worker.py b/scouter/worker/worker.py index c842761..75e6715 100644 --- a/scouter/worker/worker.py +++ b/scouter/worker/worker.py @@ -134,10 +134,8 @@ async def main(): try: await asyncio.gather(*handlers) - except BaseException as e: # NOSONAR - logger.custom_error( - 'An unhandled exception occurred: %s', e, exc_info=True, metadata=metadata - ) + except BaseException: # NOSONAR + logger.custom_error('An unhandled exception occurred: %s', metadata=metadata) finally: if notification_handler: notification_handler.shutdown() diff --git a/scouter/workflow/scouter.py b/scouter/workflow/scouter.py index 5888627..52393e2 100644 --- a/scouter/workflow/scouter.py +++ b/scouter/workflow/scouter.py @@ -95,7 +95,7 @@ class Scouter: retry_policy=retry_policy, ) - if data == {}: + if not data: return await workflow.execute_activity_method( diff --git a/tests/activities/test_activities.py b/tests/activities/test_activities.py index 2ec9bb5..bc31dee 100644 --- a/tests/activities/test_activities.py +++ b/tests/activities/test_activities.py @@ -12,7 +12,10 @@ from scouter.activities.redis import Redis @patch('scouter.activities.activities.Postgres.__init__') @patch('scouter.activities.activities.Redis.__init__') @patch('scouter.activities.activities.Gates.__init__') -def test___init__(mock_gates_init, mock_redis_init, mock_postgres_init, mock_mongodb_init): +@patch('scouter.activities.activities.MetricsController') +def test___init__( + mock_metrics_controller, mock_gates_init, mock_redis_init, mock_postgres_init, mock_mongodb_init +): postgres_config = { 'host': 'localhost', 'port': 5432, @@ -58,6 +61,7 @@ def test___init__(mock_gates_init, mock_redis_init, mock_postgres_init, mock_mon max_connections=postgres_config['max_connections'], logger=logger, notification_handler=notification_handler, + metrics_controller=mock_metrics_controller.return_value, ) mock_redis_init.assert_called_once_with( @@ -68,6 +72,7 @@ def test___init__(mock_gates_init, mock_redis_init, mock_postgres_init, mock_mon password=redis_config['password'], logger=logger, notification_handler=notification_handler, + metrics_controller=mock_metrics_controller.return_value, ) mock_mongodb_init.assert_called_once_with( @@ -76,10 +81,14 @@ def test___init__(mock_gates_init, mock_redis_init, mock_postgres_init, mock_mon database_name=mongodb_config['database_name'], logger=logger, notification_handler=notification_handler, + metrics_controller=mock_metrics_controller.return_value, ) mock_gates_init.assert_called_once_with( - ANY, logger=logger, notification_handler=notification_handler + ANY, + logger=logger, + notification_handler=notification_handler, + metrics_controller=mock_metrics_controller.return_value, ) @@ -88,8 +97,12 @@ def test___init__(mock_gates_init, mock_redis_init, mock_postgres_init, mock_mon @patch('scouter.activities.activities.Gates.__init__') @patch('scouter.activities.activities.MongoDB.__init__') @patch('scouter.activities.activities.Postgres.close') -@patch('scouter.activities.activities.MongoDB.shutdown') +@patch('scouter.activities.activities.MongoDB.close') +@patch('scouter.activities.activities.Redis.close') +@patch('scouter.activities.activities.Gates.close') def test_shutdown( + mock_gates_close, + mock_redis_close, mock_mongodb_close, mock_postgres_close, _mock_mongodb_init, @@ -129,3 +142,5 @@ def test_shutdown( mock_postgres_close.assert_called() mock_mongodb_close.assert_called() + mock_redis_close.assert_called() + mock_gates_close.assert_called() diff --git a/tests/activities/test_gates.py b/tests/activities/test_gates.py index c916ace..32d173e 100644 --- a/tests/activities/test_gates.py +++ b/tests/activities/test_gates.py @@ -1,5 +1,5 @@ from typing import Any -from unittest.mock import ANY, MagicMock, Mock, call, patch +from unittest.mock import ANY, AsyncMock, MagicMock, Mock, call, patch import numpy as np import pandas as pd @@ -14,8 +14,20 @@ def gates_fixture(): """Fixture to create a Gates instance with mocked dependencies.""" logger = Mock() notification_handler = MagicMock() - gates = Gates(logger=logger, notification_handler=notification_handler) + metrics_controller = MagicMock() + gates = Gates( + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) gates.send_notification = MagicMock() + gates.send_notification_async = AsyncMock() + gates.emit_metric = AsyncMock() + + gates.logger = logger + gates.notification_handler = notification_handler + gates.metrics_controller = metrics_controller + gates.pod_id = 'localhost' return gates @@ -29,6 +41,13 @@ metadata = { } +@patch('scouter.activities.gates.SientiaMonitoring') +def test_close(mock_sientia_monitoring, gates_fixture): + """Test close method.""" + gates_fixture.close() + mock_sientia_monitoring.shutdown.assert_called_once() + + @pytest.mark.asyncio async def test_data_quality_gate_with_null_values_filter_discard(gates_fixture): """Test data_quality_gate with NULL_VALUES_FILTER and DISCARD policy.""" @@ -55,7 +74,7 @@ async def test_data_quality_gate_with_null_values_filter_discard(gates_fixture): # Verify assert len(result['tag']) == 2 assert 'tag2' not in result['tag'] - gates_fixture.send_notification.assert_called_once() + gates_fixture.send_notification_async.assert_called_once() @pytest.mark.asyncio @@ -88,7 +107,7 @@ async def test_data_quality_gate_with_out_of_bounds_filter_keep(gates_fixture): # Verify data is kept but notification is sent assert len(result['tag']) == 3 # All rows kept - gates_fixture.send_notification.assert_called_once() + gates_fixture.send_notification_async.assert_called_once() @pytest.mark.asyncio @@ -125,7 +144,7 @@ async def test_data_quality_gate_with_multiple_filters(gates_fixture): 'timestamp': {0: '2023-01-01', 3: '2023-01-04'}, } # Should be called twice (once for each filter) - assert gates_fixture.send_notification.call_count == 2 + assert gates_fixture.send_notification_async.call_count == 2 @pytest.mark.asyncio @@ -173,8 +192,8 @@ async def test_data_quality_gate_with_filter_error(gates_fixture): # Verify error notification is sent and data is unchanged assert len(result['tag']) == 1 - gates_fixture.send_notification.assert_called_once() - call_args = gates_fixture.send_notification.call_args[1] + gates_fixture.send_notification_async.assert_called_once() + call_args = gates_fixture.send_notification_async.call_args[1] assert call_args['notification_id'] == 'DATA_QUALITY_GATE_ISSUES' assert call_args['level'] == NotificationLevel.ERROR assert 'Filter error' in call_args['message'] @@ -237,16 +256,17 @@ async def test_data_quality_gate_with_no_filters(gates_fixture): (pd.DataFrame({'value': [1.0, 2.0]}), 'invalid', 'continue'), ], ) -def test_apply_aggregation(gates_fixture, group_data, aggr_function, expected_result): +@pytest.mark.asyncio +async def test_apply_aggregation(gates_fixture, group_data, aggr_function, expected_result): """Test apply_aggregation method with various scenarios.""" - result = gates_fixture.apply_aggregation(group_data, aggr_function, metadata) + result = await gates_fixture.apply_aggregation(group_data, aggr_function, metadata) assert result == expected_result # Check notification was sent for invalid function if aggr_function == 'invalid': - gates_fixture.send_notification.assert_called_once() + gates_fixture.send_notification_async.assert_called_once() else: - gates_fixture.send_notification.assert_not_called() + gates_fixture.send_notification_async.assert_not_called() @pytest.mark.asyncio @@ -288,7 +308,7 @@ async def test_aggregate_data(gates_fixture): @pytest.mark.asyncio async def test_aggregate_data_with_continue(gates_fixture): - gates_fixture.apply_aggregation = MagicMock(return_value='continue') + gates_fixture.apply_aggregation = AsyncMock(return_value='continue') input_data = { 'data': [ @@ -315,7 +335,7 @@ async def test_aggregate_data_with_continue(gates_fixture): # Verify assert result == expected_result - gates_fixture.send_notification.assert_not_called() + gates_fixture.send_notification_async.assert_not_called() @pytest.mark.asyncio @@ -343,7 +363,7 @@ async def test_aggregate_data_raise_exception(gates_fixture): await gates_fixture.aggregate_data(input_data) except Exception as e: assert str(e) == 'Test exception' - gates_fixture.send_notification.assert_called_once_with( + gates_fixture.send_notification_async.assert_called_once_with( metadata=metadata['metadata'], notification_id='AGGREGATION_ISSUES', message='Error aggregating data: Test exception', @@ -370,7 +390,7 @@ async def test_write_metrics(mock_metrics, gates_fixture): mock_metrics.LABORIOUS_DATA_WRITTEN_COUNT.labels.assert_called_once_with( pod_id=gates_fixture.pod_id, model_name=metadata['metadata']['model_name'], - pipeline_name=metadata['metadata']['workflow_name'], + workflow_name=metadata['metadata']['workflow_name'], ) mock_metrics.LABORIOUS_DATA_WRITTEN_COUNT.labels.return_value.inc.assert_called_once() @@ -387,13 +407,13 @@ async def test_write_metrics(mock_metrics, gates_fixture): call( pod_id=gates_fixture.pod_id, model_name=metadata['metadata']['model_name'], - pipeline_name=metadata['metadata']['workflow_name'], + workflow_name=metadata['metadata']['workflow_name'], tag_name='tag1', ), call( pod_id=gates_fixture.pod_id, model_name=metadata['metadata']['model_name'], - pipeline_name=metadata['metadata']['workflow_name'], + workflow_name=metadata['metadata']['workflow_name'], tag_name='tag2', ), ], diff --git a/tests/activities/test_mongo.py b/tests/activities/test_mongo.py index 0d62e0a..d0b00d7 100644 --- a/tests/activities/test_mongo.py +++ b/tests/activities/test_mongo.py @@ -1,99 +1,84 @@ from datetime import datetime -from unittest.mock import ANY, MagicMock, patch +from unittest.mock import ANY, AsyncMock, MagicMock, patch from pytest import fixture, mark from sientia_do.notifications.models import NotificationLevel from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ -from scouter.activities.mongodb import MongoDB, clear_mongo_id +from scouter.activities.mongodb import MongoDB -def test_clear_mongo_id(): - """Test clear_mongo_id""" - data = [ - {'_id': '1', 'name': 'test1'}, - {'_id': '2', 'name': [{'_id': '3', 'name': 'test3'}]}, - {'_id': '4', 'name': {'_id': '5', 'name': 'test2'}}, - [{'_id': '6', 'name': 'test2'}], - ] - - result = clear_mongo_id(data) - - assert result == [ - {'name': 'test1'}, - {'name': [{'name': 'test3'}]}, - {'name': {'name': 'test2'}}, - [{'name': 'test2'}], - ] - - -@patch('scouter.activities.mongodb.MongoClient') -def test_mongodb___init__(mock_mongo_client): +@patch('scouter.activities.mongodb.MongoDBRepository') +def test_mongodb___init__(mock_mongodb_repository): """Test MongoDB __init__""" + + logger = MagicMock() + notification_handler = MagicMock() + metrics_controller = MagicMock() mongo = MongoDB( connection_string='mongodb://localhost:27017', database_name='test_db', - logger=MagicMock(), - notification_handler=MagicMock(), + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, ) - mock_mongo_client.assert_called_once_with( - 'mongodb://localhost:27017', serverSelectionTimeoutMS=5000 + mock_mongodb_repository.assert_called_once_with( + connection_string='mongodb://localhost:27017', + database_name='test_db', + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, ) - mock_mongo_client.return_value.server_info.assert_called_once() - - mock_mongo_client.return_value.__getitem__.assert_called_once_with('test_db') - - assert mongo.client is not None - assert mongo.database is not None + assert mongo.mongodb_repository is not None @fixture -@patch('scouter.activities.mongodb.MongoClient') -def mongodb_activity(mock_mongo_client): +@patch('scouter.activities.mongodb.MongoDBRepository') +def mongodb_activity(mock_mongodb_repository): """Test MongoDB activity""" + mongo = MongoDB( connection_string='mongodb://localhost:27017', database_name='test_db', logger=MagicMock(), notification_handler=MagicMock(), + metrics_controller=AsyncMock(), ) - return mongo -def test_shutdown_success(mongodb_activity): - """Test shutdown""" - mongodb_activity.shutdown() +def test_close(mongodb_activity): + """Test close""" + mongodb_activity.close() - mongodb_activity.client.close.assert_called_once() + mongodb_activity.mongodb_repository.close.assert_called_once() -def test_shutdown_error(mongodb_activity): - """Test shutdown""" - mongodb_activity.client.close = MagicMock(side_effect=Exception('test')) +def test_del(mongodb_activity): + mongodb_activity.close = MagicMock() - mongodb_activity.shutdown() + mongodb_activity.__del__() - mongodb_activity.client.close.assert_called_once() + mongodb_activity.close.assert_called_once() @mark.asyncio async def test_load_latest_data_none_last_data_timestamp(mongodb_activity): """Test load_latest_data""" - collection = MagicMock() - mongodb_activity.database.__getitem__.return_value = collection - collection.find.return_value = [ - { - 'name': 'test1', - 'value': 1, - 'inserted_at': datetime.strptime( - '2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ - ), - } - ] + mongodb_activity.mongodb_repository.find = AsyncMock( + return_value=[ + { + 'name': 'test1', + 'value': 1, + 'inserted_at': datetime.strptime( + '2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ + ), + } + ] + ) result = await mongodb_activity.load_latest_data( { @@ -103,32 +88,36 @@ async def test_load_latest_data_none_last_data_timestamp(mongodb_activity): } ) - mongodb_activity.database.__getitem__.assert_called_once_with('test_collection') + mongodb_activity.mongodb_repository.find.assert_called_once_with( + collection_name='test_collection', + filters={}, + metadata={'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, + ) - collection.find.assert_called_once_with({}, {'_id': 0}) - - assert result == { - 'name': {0: 'test1'}, - 'value': {0: 1}, - 'inserted_at': {0: '2023-01-01 12:00:00.000000+0000'}, - } + assert result == [ + { + 'name': 'test1', + 'value': 1, + 'inserted_at': '2023-01-01 12:00:00.000000+0000', + } + ] @mark.asyncio async def test_load_latest_data_not_none_last_data_timestamp(mongodb_activity): """Test load_latest_data""" - collection = MagicMock() - mongodb_activity.database.__getitem__.return_value = collection - collection.find.return_value = [ - { - 'name': 'test1', - 'value': 1, - 'inserted_at': datetime.strptime( - '2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ - ), - } - ] + mongodb_activity.mongodb_repository.find = AsyncMock( + return_value=[ + { + 'name': 'test1', + 'value': 1, + 'inserted_at': datetime.strptime( + '2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ + ), + } + ] + ) result = await mongodb_activity.load_latest_data( { @@ -138,34 +127,34 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongodb_activity): } ) - mongodb_activity.database.__getitem__.assert_called_once_with('test_collection') - - collection.find.assert_called_once_with( - { + mongodb_activity.mongodb_repository.find.assert_called_once_with( + collection_name='test_collection', + filters={ 'inserted_at': { '$gt': datetime.strptime( '2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ ) } }, - {'_id': 0}, + metadata={'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, ) - assert result == { - 'name': {0: 'test1'}, - 'value': {0: 1}, - 'inserted_at': {0: '2023-01-01 12:00:00.000000+0000'}, - } + assert result == [ + { + 'name': 'test1', + 'value': 1, + 'inserted_at': '2023-01-01 12:00:00.000000+0000', + } + ] @mark.asyncio async def test_load_latest_data_error(mongodb_activity): """Test load_latest_data""" - collection = MagicMock() + mongodb_activity.mongodb_repository.find.side_effect = Exception('test') mongodb_activity.send_notification = MagicMock() - mongodb_activity.database.__getitem__.return_value = collection - - collection.find.side_effect = Exception('test') + mongodb_activity.send_notification_async = AsyncMock() + mongodb_activity.emit_metric = AsyncMock() try: await mongodb_activity.load_latest_data( @@ -178,7 +167,7 @@ async def test_load_latest_data_error(mongodb_activity): except Exception as e: assert str(e) == 'test' - mongodb_activity.send_notification.assert_called_once_with( + mongodb_activity.send_notification_async.assert_called_once_with( metadata={'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, notification_id='MONGO_LOAD_ERROR', message='Error loading data from MongoDB: test', diff --git a/tests/activities/test_redis.py b/tests/activities/test_redis.py index b220424..9944951 100644 --- a/tests/activities/test_redis.py +++ b/tests/activities/test_redis.py @@ -1,5 +1,5 @@ from datetime import datetime -from unittest.mock import ANY, MagicMock, patch +from unittest.mock import ANY, AsyncMock, MagicMock, patch import numpy as np import pytest @@ -11,10 +11,11 @@ from scouter.activities.redis import Redis @pytest.fixture -@patch('scouter.activities.redis.RedisBase.__init__') -def redis_activity(_mock_redis_init): +@patch('scouter.activities.redis.RedisRepository') +def redis_activity(mock_redis_repository): logger = MagicMock() notification_handler = MagicMock(spec=NotificationHandler) + metrics_controller = MagicMock() activity = Redis( host='localhost', port=6379, @@ -22,6 +23,7 @@ def redis_activity(_mock_redis_init): notification_handler=notification_handler, username='test', password='test', + metrics_controller=metrics_controller, ) activity.redis_client = MagicMock() @@ -31,24 +33,6 @@ def redis_activity(_mock_redis_init): 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', @@ -59,15 +43,60 @@ metadata = { } +@patch('scouter.activities.redis.SientiaMonitoring') +def test_close(mock_sientia_monitoring, redis_activity): + """Test close method.""" + redis_activity.close() + redis_activity.redis_repository.close.assert_called_once() + mock_sientia_monitoring.shutdown.assert_called_once() + + +@patch('scouter.activities.redis.RedisRepository') +def test_redis_initialization(mock_redis_repository): + """Test Redis activity initialization""" + logger = MagicMock() + notification_handler = MagicMock(spec=NotificationHandler) + metrics_controller = MagicMock() + activity = Redis( + host='localhost', + port=6379, + logger=logger, + notification_handler=notification_handler, + username='test', + password='test', + metrics_controller=metrics_controller, + ) + mock_redis_repository.assert_called_once_with( + host='localhost', + port=6379, + username='test', + password='test', + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + assert activity.redis_repository is not None + + @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'} + test_data = { + **metadata, + 'workflow_name': 'test_pipeline', + 'schedule_name': 'test_schedule', + } - redis_activity.get = MagicMock(return_value=None) + redis_activity.redis_repository.get = AsyncMock(return_value=None) result = await redis_activity.get_last_data_timestamp(test_data) + redis_activity.redis_repository.get.assert_called_once_with( + 'last_data_timestamp:test_pipeline:test_schedule', + metadata=metadata['metadata'], + ) + assert result is None @@ -76,11 +105,14 @@ 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') + redis_activity.redis_repository.get = AsyncMock(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') + redis_activity.redis_repository.get.assert_called_once_with( + 'last_data_timestamp:test_pipeline:test_schedule', + metadata=metadata['metadata'], + ) assert result == '2023-01-01 12:00:00' @@ -90,8 +122,8 @@ 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')) + redis_activity.send_notification_async = AsyncMock() + redis_activity.redis_repository.get.side_effect = Exception('test') try: await redis_activity.get_last_data_timestamp(test_data) @@ -99,7 +131,7 @@ async def test_get_last_data_timestamp_error(redis_activity): except Exception as e: assert str(e) == 'test' - redis_activity.send_notification.assert_called_once_with( + redis_activity.send_notification_async.assert_called_once_with( metadata=metadata['metadata'], notification_id='REDIS_GET_ERROR', message='Error getting last data timestamp: test', @@ -122,13 +154,13 @@ async def test_put_last_data_timestamp_empty_dataframe(redis_activity): 'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records'), } - redis_activity.set = MagicMock() + redis_activity.redis_repository.set = MagicMock() result = await redis_activity.put_last_data_timestamp(test_data) assert result is None - redis_activity.set.assert_not_called() + redis_activity.redis_repository.set.assert_not_called() @pytest.mark.asyncio @@ -149,14 +181,17 @@ async def test_put_last_data_timestamp_not_empty_dataframe(redis_activity): 'data': data.to_dict('records'), } - redis_activity.set = MagicMock() + redis_activity.redis_repository.set = AsyncMock() 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 + redis_activity.redis_repository.set.assert_called_once_with( + 'last_data_timestamp:test_pipeline:test_schedule', + '2023-01-01 12:00:01', + ttl=18000, + metadata=metadata['metadata'], ) @@ -176,8 +211,8 @@ async def test_put_last_data_timestamp_error(redis_activity): ).to_dict('records'), } - redis_activity.send_notification = MagicMock() - redis_activity.set = MagicMock(side_effect=Exception('test')) + redis_activity.send_notification_async = AsyncMock() + redis_activity.redis_repository.set = AsyncMock(side_effect=Exception('test')) try: await redis_activity.put_last_data_timestamp(test_data) @@ -185,7 +220,7 @@ async def test_put_last_data_timestamp_error(redis_activity): except Exception as e: assert str(e) == 'test' - redis_activity.send_notification.assert_called_once_with( + redis_activity.send_notification_async.assert_called_once_with( metadata=metadata['metadata'], notification_id='REDIS_SET_ERROR', message='Error setting last data timestamp: test', @@ -220,8 +255,8 @@ async def test_group_and_hold_data_new_key(redis_activity): } # Mock get to return None for new key - redis_activity.get = MagicMock(return_value=None) - redis_activity.set = MagicMock() + redis_activity.redis_repository.get = AsyncMock(return_value=None) + redis_activity.redis_repository.set = AsyncMock() # Call the method result = await redis_activity.group_and_hold_data(test_data) @@ -236,11 +271,12 @@ async def test_group_and_hold_data_new_key(redis_activity): 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 + redis_activity.redis_repository.set.assert_called_once_with( + 'held_data_test_pipeline_test_schedule', + {'sensor1': 25.5, 'sensor2': 30.0, 'timestamp': '2023-01-01 12:00:00'}, + ttl=3600, + metadata=metadata['metadata'], + ) @pytest.mark.asyncio @@ -273,8 +309,8 @@ async def test_group_and_hold_data_update_existing_fill_missing(redis_activity): } # Mock get to return existing data - redis_activity.get = MagicMock(return_value=existing_data) - redis_activity.set = MagicMock() + redis_activity.redis_repository.get = AsyncMock(return_value=existing_data) + redis_activity.redis_repository.set = AsyncMock() # Call the method result = await redis_activity.group_and_hold_data(test_data) @@ -294,17 +330,18 @@ async def test_group_and_hold_data_update_existing_fill_missing(redis_activity): 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, - 'sensor4': None, - 'timestamp': '2023-01-01 12:00:00', - } - assert kwargs['ttl'] == 3600 + redis_activity.redis_repository.set.assert_called_once_with( + 'held_data_test_workflow_test_schedule', + { + 'sensor1': 25.5, + 'sensor2': 28.0, + 'sensor3': 42.0, + 'sensor4': None, + 'timestamp': '2023-01-01 12:00:00', + }, + ttl=3600, + metadata=metadata['metadata'], + ) @pytest.mark.asyncio @@ -329,8 +366,8 @@ async def test_group_and_hold_data_with_none_values(redis_activity): } # Mock get to return None for new key - redis_activity.get = MagicMock(return_value=None) - redis_activity.set = MagicMock() + redis_activity.redis_repository.get = AsyncMock(return_value=None) + redis_activity.redis_repository.set = AsyncMock() # Call the method result = await redis_activity.group_and_hold_data(test_data) @@ -354,7 +391,7 @@ async def test_group_and_hold_data_empty_dataframe(redis_activity): 'fill_missing_tags': False, } - redis_activity.get = MagicMock(return_value=None) + redis_activity.redis_repository.get = AsyncMock(return_value=None) # Call the method result = await redis_activity.group_and_hold_data(test_data) @@ -376,8 +413,8 @@ async def test_group_and_hold_data_error_get(redis_activity): 'fill_missing_tags': False, } - redis_activity.get = MagicMock(side_effect=Exception('test')) - redis_activity.send_notification = MagicMock() + redis_activity.redis_repository.get = AsyncMock(side_effect=Exception('test')) + redis_activity.send_notification_async = AsyncMock() try: await redis_activity.group_and_hold_data(test_data) @@ -385,7 +422,7 @@ async def test_group_and_hold_data_error_get(redis_activity): except Exception as e: assert str(e) == 'test' - redis_activity.send_notification.assert_called_once_with( + redis_activity.send_notification_async.assert_called_once_with( metadata=metadata['metadata'], notification_id='REDIS_GET_ERROR', message='Error getting held data: test', @@ -415,9 +452,9 @@ async def test_group_and_hold_data_error_set(redis_activity): 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() + redis_activity.redis_repository.get = AsyncMock(return_value=existing_data) + redis_activity.redis_repository.set = AsyncMock(side_effect=Exception('test')) + redis_activity.send_notification_async = AsyncMock() try: await redis_activity.group_and_hold_data(test_data) @@ -429,7 +466,7 @@ async def test_group_and_hold_data_error_set(redis_activity): @pytest.mark.asyncio async def test_store_data_package(redis_activity): """Test store_data_package""" - redis_activity.set = MagicMock() + redis_activity.redis_repository.set = AsyncMock() test_data = { **metadata, @@ -454,16 +491,19 @@ async def test_store_data_package(redis_activity): 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 + redis_activity.redis_repository.set.assert_called_once_with( + ANY, + {'data': test_data['data'], 'held_data': test_data['held_data']}, + ttl=120, + metadata=metadata['metadata'], ) @pytest.mark.asyncio async def test_store_data_package_error(redis_activity): """Test store_data_package error""" - redis_activity.set = MagicMock(side_effect=ValueError('test')) - redis_activity.send_notification = MagicMock() + redis_activity.redis_repository.set = AsyncMock(side_effect=ValueError('test')) + redis_activity.send_notification_async = AsyncMock() test_data = { **metadata, @@ -489,7 +529,7 @@ async def test_store_data_package_error(redis_activity): with pytest.raises(ValueError): await redis_activity.store_data_package(test_data) - redis_activity.send_notification.assert_called_once_with( + redis_activity.send_notification_async.assert_called_once_with( metadata=metadata['metadata'], notification_id='REDIS_SET_ERROR', message='Error setting data package: test', diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 79a96d9..be7b4f3 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -15,7 +15,7 @@ def test_scouter_laborious_data_written_count(): assert set(metrics.LABORIOUS_DATA_WRITTEN_COUNT._labelnames) == { 'pod_id', 'model_name', - 'pipeline_name', + 'workflow_name', } @@ -27,6 +27,6 @@ def test_scouter_tag_changes_monitor(): assert set(metrics.TAG_CHANGES_MONITOR._labelnames) == { 'pod_id', 'model_name', - 'pipeline_name', + 'workflow_name', 'tag_name', } diff --git a/tests/utils/test_connectors_config.py b/tests/utils/test_connectors_config.py index 61d2b0a..d57b335 100644 --- a/tests/utils/test_connectors_config.py +++ b/tests/utils/test_connectors_config.py @@ -122,6 +122,11 @@ def test_build_redis_config_with_env_vars(): def test_build_mongodb_config_defaults(): """Test that build_mongodb_config returns default values when no env vars are set""" + os.environ['MONGODB_URL'] = 'localhost:27017' + os.environ['MONGODB_DATABASE_NAME'] = 'sientia' + os.environ['MONGODB_USERNAME'] = 'sientia' + os.environ['MONGODB_PASSWORD'] = 'sientia' + config = build_mongodb_config() assert config == { diff --git a/values.yaml b/values.yaml index 7371af9..5692f2b 100644 --- a/values.yaml +++ b/values.yaml @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.4.9" + tag: "0.5.0" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: @@ -150,7 +150,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git" - name: GITHUB_BRANCH - value: "fix/SIENTIAPDE-1314-ajustes-nas-camadas-de-monitoramento-do-sientia" + value: "feature/SIENTIAPDE-1325-adicionar-metricas-especificas-de-operacoes-externas" - name: PYTHON_APP value: "scouter.worker.worker" @@ -170,11 +170,6 @@ env: - name: POSTGRES_MAX_CONNECTIONS value: "40" - - name: KAFKA_BOOTSTRAP_SERVERS - value: "kafka.kafka.svc.cluster.local:9092" - - name: KAFKA_POLLING_TIME - value: "10000" - - name: REDIS_HOST value: "redis-master.redis.svc.cluster.local" - name: REDIS_PORT