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

SIENTIAPDE-1325

Browse Source 
Update GITHUB_BRANCH in values.yaml to feature/SIENTIAPDE-1325 for specific external operation metrics. Refactor shutdown methods in Activities class to ensure proper closure of MongoDB, Redis, and Gates connections. Adjust tests to reflect these changes and enhance repository pattern implementation for MongoDB and Redis activities.
This commit is contained in:
vitor-aignosi
2025-10-29 16:58:22 -03:00
parent 10b3310677
commit 0c16075823
8 changed files with 152 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
scouter/activities/activities.py
View File

@@ -97,4 +97,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)

7
scouter/activities/mongodb.py
View File

@@ -8,17 +8,14 @@ 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.repository.mongodb_repository import MongoDBRepository
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
class MongoDB(SientiaMonitoring):
"""
MongoDB operations for data retrieval and storage.
@@ -136,7 +133,7 @@ class MongoDB(SientiaMonitoring):
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(

8
tests/activities/test_activities.py
View File

@@ -88,8 +88,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 +133,5 @@ def test_shutdown(
mock_postgres_close.assert_called()
mock_mongodb_close.assert_called()
mock_redis_close.assert_called()
mock_gates_close.assert_called()

3
tests/activities/test_gates.py
View File

@@ -16,6 +16,9 @@ def gates_fixture():
notification_handler = MagicMock()
gates = Gates(logger=logger, notification_handler=notification_handler)
gates.send_notification = MagicMock()
gates.logger = logger
gates.notification_handler = notification_handler
gates.pod_id = 'localhost'
return gates

126
tests/activities/test_mongo.py
View File

@@ -5,54 +5,38 @@ 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()
mongo = MongoDB(
connection_string='mongodb://localhost:27017',
database_name='test_db',
logger=MagicMock(),
notification_handler=MagicMock(),
logger=logger,
notification_handler=notification_handler,
)
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,
)
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',
@@ -60,32 +44,32 @@ def mongodb_activity(mock_mongo_client):
notification_handler=MagicMock(),
)
mongo.logger = MagicMock()
mongo.notification_handler = MagicMock()
mongo.pod_id = 'localhost'
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 = [
mongodb_activity.mongodb_repository.find.return_value = [
{
'name': 'test1',
'value': 1,
@@ -103,24 +87,26 @@ 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 = [
mongodb_activity.mongodb_repository.find.return_value = [
{
'name': 'test1',
'value': 1,
@@ -138,34 +124,32 @@ 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')
try:
await mongodb_activity.load_latest_data(

115
tests/activities/test_redis.py
View File

@@ -11,8 +11,8 @@ 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)
activity = Redis(
@@ -31,24 +31,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 +41,52 @@ metadata = {
}
@patch('scouter.activities.redis.RedisRepository')
def test_redis_initialization(mock_redis_repository):
"""Test Redis activity initialization"""
logger = MagicMock()
notification_handler = MagicMock(spec=NotificationHandler)
activity = Redis(
host='localhost',
port=6379,
logger=logger,
notification_handler=notification_handler,
username='test',
password='test',
)
mock_redis_repository.assert_called_once_with(
host='localhost',
port=6379,
username='test',
password='test',
logger=logger,
notification_handler=notification_handler,
)
activity.logger = logger
activity.notification_handler = notification_handler
activity.pod_id = 'localhost'
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.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'
)
assert result is None
@@ -76,11 +95,13 @@ 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.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'
)
assert result == '2023-01-01 12:00:00'
@@ -91,7 +112,7 @@ async def test_get_last_data_timestamp_error(redis_activity):
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.redis_repository.get.side_effect = Exception('test')
try:
await redis_activity.get_last_data_timestamp(test_data)
@@ -122,13 +143,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,13 +170,13 @@ 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 = MagicMock()
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(
redis_activity.redis_repository.set.assert_called_once_with(
'last_data_timestamp:test_pipeline:test_schedule', '2023-01-01 12:00:01', ttl=18000
)
@@ -177,7 +198,7 @@ async def test_put_last_data_timestamp_error(redis_activity):
}
redis_activity.send_notification = MagicMock()
redis_activity.set = MagicMock(side_effect=Exception('test'))
redis_activity.redis_repository.set = MagicMock(side_effect=Exception('test'))
try:
await redis_activity.put_last_data_timestamp(test_data)
@@ -220,8 +241,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 = MagicMock(return_value=None)
redis_activity.redis_repository.set = MagicMock()
# Call the method
result = await redis_activity.group_and_hold_data(test_data)
@@ -236,8 +257,8 @@ 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
redis_activity.redis_repository.set.assert_called_once()
args, kwargs = redis_activity.redis_repository.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
@@ -273,8 +294,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 = MagicMock(return_value=existing_data)
redis_activity.redis_repository.set = MagicMock()
# Call the method
result = await redis_activity.group_and_hold_data(test_data)
@@ -294,8 +315,8 @@ 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
redis_activity.redis_repository.set.assert_called_once()
args, kwargs = redis_activity.redis_repository.set.call_args
assert args[0] == 'held_data_test_workflow_test_schedule'
assert args[1] == {
'sensor1': 25.5,
@@ -329,8 +350,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 = MagicMock(return_value=None)
redis_activity.redis_repository.set = MagicMock()
# Call the method
result = await redis_activity.group_and_hold_data(test_data)
@@ -354,7 +375,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 = MagicMock(return_value=None)
# Call the method
result = await redis_activity.group_and_hold_data(test_data)
@@ -376,7 +397,7 @@ 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.redis_repository.get = MagicMock(side_effect=Exception('test'))
redis_activity.send_notification = MagicMock()
try:
@@ -415,8 +436,8 @@ 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.redis_repository.get = MagicMock(return_value=existing_data)
redis_activity.redis_repository.set = MagicMock(side_effect=Exception('test'))
redis_activity.send_notification = MagicMock()
try:
@@ -429,7 +450,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 = MagicMock()
test_data = {
**metadata,
@@ -454,7 +475,7 @@ async def test_store_data_package(redis_activity):
await redis_activity.store_data_package(test_data)
redis_activity.set.assert_called_once_with(
redis_activity.redis_repository.set.assert_called_once_with(
ANY, {'data': test_data['data'], 'held_data': test_data['held_data']}, ttl=120
)
@@ -462,7 +483,7 @@ async def test_store_data_package(redis_activity):
@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.redis_repository.set = MagicMock(side_effect=ValueError('test'))
redis_activity.send_notification = MagicMock()
test_data = {

12
tests/conftest.py Normal file
View File

@@ -0,0 +1,12 @@
from unittest.mock import patch
from pytest import fixture
@fixture(autouse=True, scope='session')
def sientia_monitoring_fixture():
with patch(
'sientia_do.observability.sientia_monitoring.SientiaMonitoring.__init__'
) as mock_sientia_monitoring:
mock_sientia_monitoring.return_value = None
yield

2
values.yaml
View File

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