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

SIENTIAPDE-1193

Browse Source 
refactor: standardize timestamp handling across MongoDB and SlotManager

- Replaced instances of `now()` with `date_now` for consistent timestamp usage in MongoDB and SlotManager activities.
- Updated test cases to mock the new `now()` function, ensuring accurate timestamp handling in unit tests.
- Adjusted timestamp formatting in various tests to align with the DATETIME_FORMAT_MS_WITH_TZ standard.
This commit is contained in:
vitor-aignosi
2025-08-25 08:30:42 -03:00
parent d7effd44e5
commit 4aa7ea70fa
5 changed files with 39 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
orchestrator/activities/mongo_db.py
View File

@@ -216,7 +216,7 @@ class MongoDB(BaseActivity):
""" """
updated_pipelines = input_data.get("updated_pipelines", []) updated_pipelines = input_data.get("updated_pipelines", [])
metadata = input_data.get("metadata", {}) metadata = input_data.get("metadata", {})
now = now() date_now = now()
collection = self.database["orchestrated_schedules"] collection = self.database["orchestrated_schedules"]
self.info("Updating pipelines timestamps...", metadata=metadata) self.info("Updating pipelines timestamps...", metadata=metadata)
@@ -233,7 +233,7 @@ class MongoDB(BaseActivity):
try: try:
collection.update_many( collection.update_many(
data_filter, data_filter,
{"$set": {"updated_at": now}} {"$set": {"updated_at": date_now}}
) )
success_count += 1 success_count += 1
except Exception as e: except Exception as e:
@@ -267,12 +267,12 @@ class MongoDB(BaseActivity):
success_count = 0 success_count = 0
now = now() date_now = now()
argument = [ argument = [
{"schedule_name": pipeline["schedule_name"], {"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"], "namespace": pipeline["namespace"],
"updated_at": now} "updated_at": date_now}
for pipeline in created_pipelines if pipeline["success"] for pipeline in created_pipelines if pipeline["success"]
] ]
data_filter = argument if argument else {} data_filter = argument if argument else {}

4
orchestrator/activities/slot_manager.py
View File

@@ -382,12 +382,12 @@ class SlotManager(Redis):
self.info("Storing notification cache...", metadata=metadata) self.info("Storing notification cache...", metadata=metadata)
now = now().strftime(DATETIME_FORMAT_MS_WITH_TZ) date_now = now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
for index, row in log_report.iterrows(): for index, row in log_report.iterrows():
status = row['status'] status = row['status']
if status == 'sent': if status == 'sent':
key = f"{row['schedule']}:{row['notification_id']}" key = f"{row['schedule']}:{row['notification_id']}"
self.set(key, now, ttl=sent_ttl) self.set(key, date_now, ttl=sent_ttl)
self.info("Notification cache stored...", metadata=metadata) self.info("Notification cache stored...", metadata=metadata)

40
tests/orchestrator/activities/test_mongo_db.py
View File

@@ -266,8 +266,8 @@ async def test_aggregate_documents_in_mongodb_missing_aggregation(mongo_db):
@mark.asyncio @mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime") @patch("orchestrator.activities.mongo_db.now")
async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db): async def test_update_pipelines_timestamps_success(now_mock, mongo_db):
input_data = {"updated_pipelines": [ input_data = {"updated_pipelines": [
{"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test1", "namespace": "test1", "success": True},
{"schedule_name": "test2", "namespace": "test2", "success": True} {"schedule_name": "test2", "namespace": "test2", "success": True}
@@ -280,13 +280,13 @@ async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db):
{"schedule_name": "test2", "namespace": "test2"} {"schedule_name": "test2", "namespace": "test2"}
]}, ]},
{"$set": { {"$set": {
"updated_at": datetime_mock.now.return_value}} "updated_at": now_mock.return_value}}
) )
@mark.asyncio @mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime") @patch("orchestrator.activities.mongo_db.now")
async def test_update_pipelines_timestamps_failure(datetime_mock, mongo_db): async def test_update_pipelines_timestamps_failure(now_mock, mongo_db):
input_data = { input_data = {
"updated_pipelines": [ "updated_pipelines": [
{"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test1", "namespace": "test1", "success": True},
@@ -314,8 +314,8 @@ async def test_update_pipelines_timestamps_failure(datetime_mock, mongo_db):
@mark.asyncio @mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime") @patch("orchestrator.activities.mongo_db.now")
async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db): async def test_create_pipelines_timestamps_success(now_mock, mongo_db):
input_data = {"created_pipelines": [ input_data = {"created_pipelines": [
{"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test1", "namespace": "test1", "success": True},
{"schedule_name": "test2", "namespace": "test2", "success": True} {"schedule_name": "test2", "namespace": "test2", "success": True}
@@ -325,16 +325,16 @@ async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db):
mongo_db.database["pipelines"].insert_many.assert_called_once_with( mongo_db.database["pipelines"].insert_many.assert_called_once_with(
[ [
{"schedule_name": "test1", "namespace": "test1", {"schedule_name": "test1", "namespace": "test1",
"updated_at": datetime_mock.now.return_value}, "updated_at": now_mock.return_value},
{"schedule_name": "test2", "namespace": "test2", {"schedule_name": "test2", "namespace": "test2",
"updated_at": datetime_mock.now.return_value} "updated_at": now_mock.return_value}
] ]
) )
@mark.asyncio @mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime") @patch("orchestrator.activities.mongo_db.now")
async def test_create_pipelines_timestamps_failure(datetime_mock, mongo_db): async def test_create_pipelines_timestamps_failure(now_mock, mongo_db):
input_data = {"created_pipelines": [ input_data = {"created_pipelines": [
{"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test1", "namespace": "test1", "success": True},
{"schedule_name": "test2", "namespace": "test2", "success": True} {"schedule_name": "test2", "namespace": "test2", "success": True}
@@ -360,8 +360,8 @@ async def test_create_pipelines_timestamps_failure(datetime_mock, mongo_db):
@mark.asyncio @mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime") @patch("orchestrator.activities.mongo_db.now")
async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db): async def test_delete_pipelines_timestamps_success(now_mock, mongo_db):
input_data = {"deleted_pipelines": [ input_data = {"deleted_pipelines": [
{"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test1", "namespace": "test1", "success": True},
{"schedule_name": "test2", "namespace": "test2", "success": True} {"schedule_name": "test2", "namespace": "test2", "success": True}
@@ -377,8 +377,8 @@ async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db):
@mark.asyncio @mark.asyncio
@patch("orchestrator.activities.mongo_db.datetime") @patch("orchestrator.activities.mongo_db.now")
async def test_delete_pipelines_timestamps_failure(datetime_mock, mongo_db): async def test_delete_pipelines_timestamps_failure(now_mock, mongo_db):
input_data = {"deleted_pipelines": [ input_data = {"deleted_pipelines": [
{"schedule_name": "test1", "namespace": "test1", "success": True}, {"schedule_name": "test1", "namespace": "test1", "success": True},
{"schedule_name": "test2", "namespace": "test2", "success": True} {"schedule_name": "test2", "namespace": "test2", "success": True}
@@ -573,14 +573,14 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
'name': 'test1', 'name': 'test1',
'value': 1, 'value': 1,
'timestamp': datetime.strptime( 'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f') '2023-01-01 12:00:00.000000+0000', '%Y-%m-%d %H:%M:%S.%f%z')
} }
] ]
result = await mongo_db.load_latest_data({ result = await mongo_db.load_latest_data({
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection', 'collection_name': 'test_collection',
'last_data_timestamp': '2023-01-01 12:00:00.000000', 'last_data_timestamp': '2023-01-01 12:00:00.000000+0000',
'base_data_filter': { 'base_data_filter': {
'level': 'ERROR' 'level': 'ERROR'
} }
@@ -594,7 +594,7 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
'level': 'ERROR', 'level': 'ERROR',
'timestamp': { 'timestamp': {
'$gt': datetime.strptime( '$gt': datetime.strptime(
'2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f') '2023-01-01 12:00:00.000000+0000', '%Y-%m-%d %H:%M:%S.%f%z')
} }
}, },
{"_id": 0} {"_id": 0}
@@ -603,7 +603,7 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
assert result == [{ assert result == [{
'name': 'test1', 'name': 'test1',
'value': 1, 'value': 1,
'timestamp': '2023-01-01 12:00:00.000000' 'timestamp': '2023-01-01 12:00:00.000000+0000'
}] }]
@@ -620,7 +620,7 @@ async def test_load_latest_data_error(mongo_db):
await mongo_db.load_latest_data({ await mongo_db.load_latest_data({
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection', 'collection_name': 'test_collection',
'last_data_timestamp': '2023-01-01 12:00:00.000000', 'last_data_timestamp': '2023-01-01 12:00:00.000000+0000',
'base_data_filter': { 'base_data_filter': {
'level': 'ERROR' 'level': 'ERROR'
} }

14
tests/orchestrator/activities/test_slot_manager.py
View File

@@ -4,7 +4,7 @@ from pandas import DataFrame
from pytest import mark, fixture from pytest import mark, fixture
from orchestrator.activities.slot_manager import SlotManager from orchestrator.activities.slot_manager import SlotManager
from sientia_do.notifications.models import NotificationLevel from sientia_do.notifications.models import NotificationLevel
from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now
metadata = { metadata = {
"metadata": { "metadata": {
@@ -373,13 +373,13 @@ async def test_put_last_data_timestamp_error(slot_manager):
async def test_filter_notification_alerts(slot_manager): async def test_filter_notification_alerts(slot_manager):
slot_manager.get = MagicMock(side_effect=[ slot_manager.get = MagicMock(side_effect=[
None, None,
(datetime.now() - timedelta(seconds=600) (now() - timedelta(seconds=600)
).strftime(DEFAULT_DATE_FORMAT), ).strftime(DATETIME_FORMAT_MS_WITH_TZ),
datetime.now().strftime(DEFAULT_DATE_FORMAT), now().strftime(DATETIME_FORMAT_MS_WITH_TZ),
None, None,
(datetime.now() - timedelta(seconds=600) (now() - timedelta(seconds=600)
).strftime(DEFAULT_DATE_FORMAT), ).strftime(DATETIME_FORMAT_MS_WITH_TZ),
datetime.now().strftime(DEFAULT_DATE_FORMAT)]) now().strftime(DATETIME_FORMAT_MS_WITH_TZ)])
input_data = { input_data = {
**metadata, **metadata,

7
tests/orchestrator/workflows/subworkflows/test_process_notifications.py
View File

@@ -2,6 +2,7 @@ from unittest.mock import AsyncMock, patch, ANY, call
from pytest import fixture, mark from pytest import fixture, mark
from orchestrator.workflows.subworkflows.process_notifications import ProcessNotifications from orchestrator.workflows.subworkflows.process_notifications import ProcessNotifications
from orchestrator.activities.activities import Activities from orchestrator.activities.activities import Activities
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ
@fixture @fixture
@@ -76,7 +77,11 @@ async def test_run(workflow_mock, process_notifications):
**metadata, **metadata,
'schema': input_data['schema'], 'schema': input_data['schema'],
'table_name': input_data['table_name'], 'table_name': input_data['table_name'],
'data': workflow_mock.execute_local_activity_method.return_value 'data': workflow_mock.execute_local_activity_method.return_value,
'timestamp_conversion': {
'column': 'timestamp',
'format': DATETIME_FORMAT_MS_WITH_TZ
}
}, },
schedule_to_close_timeout=ANY, schedule_to_close_timeout=ANY,
retry_policy=ANY retry_policy=ANY