gitea_admin/sientia-dataops-scouter_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #24 from Aignosi/feature/SIENTIAPDE-1325-adicionar-metricas-especificas-de-operacoes-externas

Browse Source 
SIENTIAPDE-1325: Refactor Activities, Gates, MongoDB, and Redis for Improved Monitoring, Asynchronous Notifications, and Repository Pattern Implementation
This commit is contained in:
vitor-aignosi
2025-11-10 10:12:16 -03:00
committed by GitHub
parent 7265819ae0 1b7391e8f1
commit 5db04c7e5b
19 changed files with 512 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
.env.example
View File

@@ -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

1
.github/workflows/release.yml vendored
View File

@@ -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:

133
init_port_forward.sh Executable file
View File

@@ -0,0 +1,133 @@
#!/bin/bash
# Usage:
# 1) Edit the PORT_FORWARDS list below with entries of:
# <namespace> <service_name> <local_port> <service_port>
# 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

6
pyproject.toml
View File

@@ -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/*",

2
requirements.txt
View File

@@ -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

19
scouter/activities/activities.py
View File

@@ -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)

37
scouter/activities/gates.py
View File

@@ -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'])

103
scouter/activities/mongodb.py
View File

@@ -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}',

56
scouter/activities/redis.py
View File

@@ -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}',

2
scouter/metrics.py
View File

@@ -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(

6
scouter/worker/worker.py
View File

@@ -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()

2
scouter/workflow/scouter.py
View File

@@ -95,7 +95,7 @@ class Scouter:
retry_policy=retry_policy,
)
if data == {}:
if not data:
return
await workflow.execute_activity_method(

21
tests/activities/test_activities.py
View File

@@ -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()

54
tests/activities/test_gates.py
View File

@@ -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',
),
],

169
tests/activities/test_mongo.py
View File

@@ -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',

182
tests/activities/test_redis.py
View File

@@ -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',

4
tests/test_metrics.py
View File

@@ -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',
}

5
tests/utils/test_connectors_config.py
View File

@@ -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 == {

9
values.yaml
View File

@@ -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