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

SIENTIAPDE-1325

Browse Source 
Refactor initialization of Activities, Gates, and MongoDB classes for improved readability by using multi-line argument formatting. Update related tests to match the new initialization style.
This commit is contained in:
vitor-aignosi
2025-10-30 16:50:39 -03:00
parent d4a74ec5eb
commit 8350a3a0c3
9 changed files with 71 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
tests/activities/test_redis.py
View File

@@ -80,7 +80,7 @@ async def test_get_last_data_timestamp_none(redis_activity):
'schedule_name': 'test_schedule',
}
redis_activity.redis_repository.get = AsyncMock(return_value = None)
redis_activity.redis_repository.get = AsyncMock(return_value=None)
result = await redis_activity.get_last_data_timestamp(test_data)
@@ -96,7 +96,7 @@ 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.redis_repository.get = AsyncMock(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)
@@ -259,9 +259,12 @@ async def test_group_and_hold_data_new_key(redis_activity):
# Verify set was called with correct arguments
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
'held_data_test_pipeline_test_schedule',
{'sensor1': 25.5, 'sensor2': 30.0, 'timestamp': '2023-01-01 12:00:00'},
ttl=3600,
)
@pytest.mark.asyncio
async def test_group_and_hold_data_update_existing_fill_missing(redis_activity):
"""Test updating existing data with group_and_hold_data"""
@@ -314,11 +317,18 @@ async def test_group_and_hold_data_update_existing_fill_missing(redis_activity):
# Verify set was called with correct arguments
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
'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,
)
@pytest.mark.asyncio
async def test_group_and_hold_data_with_none_values(redis_activity):
"""Test handling of None values in group_and_hold_data"""