diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 3e094b4..e53f51a 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -216,7 +216,7 @@ class MongoDB(BaseActivity): """ updated_pipelines = input_data.get("updated_pipelines", []) metadata = input_data.get("metadata", {}) - now = now() + date_now = now() collection = self.database["orchestrated_schedules"] self.info("Updating pipelines timestamps...", metadata=metadata) @@ -233,7 +233,7 @@ class MongoDB(BaseActivity): try: collection.update_many( data_filter, - {"$set": {"updated_at": now}} + {"$set": {"updated_at": date_now}} ) success_count += 1 except Exception as e: @@ -267,12 +267,12 @@ class MongoDB(BaseActivity): success_count = 0 - now = now() + date_now = now() argument = [ {"schedule_name": pipeline["schedule_name"], "namespace": pipeline["namespace"], - "updated_at": now} + "updated_at": date_now} for pipeline in created_pipelines if pipeline["success"] ] data_filter = argument if argument else {} diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index bc68d23..862e258 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -382,12 +382,12 @@ class SlotManager(Redis): 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(): status = row['status'] if status == 'sent': 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) diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 742d42f..2ae912d 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -266,8 +266,8 @@ async def test_aggregate_documents_in_mongodb_missing_aggregation(mongo_db): @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_update_pipelines_timestamps_success(now_mock, mongo_db): input_data = {"updated_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "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"} ]}, {"$set": { - "updated_at": datetime_mock.now.return_value}} + "updated_at": now_mock.return_value}} ) @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_update_pipelines_timestamps_failure(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_update_pipelines_timestamps_failure(now_mock, mongo_db): input_data = { "updated_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "success": True}, @@ -314,8 +314,8 @@ async def test_update_pipelines_timestamps_failure(datetime_mock, mongo_db): @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_create_pipelines_timestamps_success(now_mock, mongo_db): input_data = {"created_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "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( [ {"schedule_name": "test1", "namespace": "test1", - "updated_at": datetime_mock.now.return_value}, + "updated_at": now_mock.return_value}, {"schedule_name": "test2", "namespace": "test2", - "updated_at": datetime_mock.now.return_value} + "updated_at": now_mock.return_value} ] ) @mark.asyncio -@patch("orchestrator.activities.mongo_db.datetime") -async def test_create_pipelines_timestamps_failure(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_create_pipelines_timestamps_failure(now_mock, mongo_db): input_data = {"created_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "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 -@patch("orchestrator.activities.mongo_db.datetime") -async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_delete_pipelines_timestamps_success(now_mock, mongo_db): input_data = {"deleted_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "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 -@patch("orchestrator.activities.mongo_db.datetime") -async def test_delete_pipelines_timestamps_failure(datetime_mock, mongo_db): +@patch("orchestrator.activities.mongo_db.now") +async def test_delete_pipelines_timestamps_failure(now_mock, mongo_db): input_data = {"deleted_pipelines": [ {"schedule_name": "test1", "namespace": "test1", "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', 'value': 1, '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({ 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, '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': { 'level': 'ERROR' } @@ -594,7 +594,7 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongo_db): 'level': 'ERROR', 'timestamp': { '$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} @@ -603,7 +603,7 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongo_db): assert result == [{ 'name': 'test1', '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({ 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, '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': { 'level': 'ERROR' } diff --git a/tests/orchestrator/activities/test_slot_manager.py b/tests/orchestrator/activities/test_slot_manager.py index debf0a7..80cd384 100644 --- a/tests/orchestrator/activities/test_slot_manager.py +++ b/tests/orchestrator/activities/test_slot_manager.py @@ -4,7 +4,7 @@ from pandas import DataFrame from pytest import mark, fixture from orchestrator.activities.slot_manager import SlotManager 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": { @@ -373,13 +373,13 @@ async def test_put_last_data_timestamp_error(slot_manager): async def test_filter_notification_alerts(slot_manager): slot_manager.get = MagicMock(side_effect=[ None, - (datetime.now() - timedelta(seconds=600) - ).strftime(DEFAULT_DATE_FORMAT), - datetime.now().strftime(DEFAULT_DATE_FORMAT), + (now() - timedelta(seconds=600) + ).strftime(DATETIME_FORMAT_MS_WITH_TZ), + now().strftime(DATETIME_FORMAT_MS_WITH_TZ), None, - (datetime.now() - timedelta(seconds=600) - ).strftime(DEFAULT_DATE_FORMAT), - datetime.now().strftime(DEFAULT_DATE_FORMAT)]) + (now() - timedelta(seconds=600) + ).strftime(DATETIME_FORMAT_MS_WITH_TZ), + now().strftime(DATETIME_FORMAT_MS_WITH_TZ)]) input_data = { **metadata, diff --git a/tests/orchestrator/workflows/subworkflows/test_process_notifications.py b/tests/orchestrator/workflows/subworkflows/test_process_notifications.py index df25c84..8a41654 100644 --- a/tests/orchestrator/workflows/subworkflows/test_process_notifications.py +++ b/tests/orchestrator/workflows/subworkflows/test_process_notifications.py @@ -2,6 +2,7 @@ from unittest.mock import AsyncMock, patch, ANY, call from pytest import fixture, mark from orchestrator.workflows.subworkflows.process_notifications import ProcessNotifications from orchestrator.activities.activities import Activities +from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ @fixture @@ -76,7 +77,11 @@ async def test_run(workflow_mock, process_notifications): **metadata, 'schema': input_data['schema'], '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, retry_policy=ANY