SIENTIAPDE-1150

SIENTIAPDE-1150: enhance MongoDB activities to filter successful pipelines; update orchestration workflow to include timestamp management for updated, created, and deleted pipelines; adjust tests to reflect new success criteria

orchestrator/activities/mongo_db.py

@@ -202,7 +202,7 @@ class MongoDB(BaseActivity):
             {"$or": [
                 {"schedule_name": pipeline["schedule_name"],
                     "namespace": pipeline["namespace"]}
-                for pipeline in updated_pipelines
+                for pipeline in updated_pipelines if pipeline["success"]
             ]},
             {"$set": {"updated_at": now}}
         )
@@ -222,7 +222,7 @@ class MongoDB(BaseActivity):
             "schedule_name": pipeline["schedule_name"],
             "namespace": pipeline["namespace"],
             "updated_at": now
-        } for pipeline in created_pipelines])
+        } for pipeline in created_pipelines if pipeline["success"]])
 
     @activity.defn(name="delete_pipelines_timestamps")
     async def delete_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
@@ -237,6 +237,6 @@ class MongoDB(BaseActivity):
             "$or": [
                 {"schedule_name": pipeline["schedule_name"],
                     "namespace": pipeline["namespace"]}
-                for pipeline in deleted_pipelines
+                for pipeline in deleted_pipelines if pipeline["success"]
             ]
         })

orchestrator/workflows/orchestrator.py

@@ -199,5 +199,35 @@ class Orchestrator:
             start_to_close_timeout=timedelta(seconds=60)
         )
 
+        update_pipelines_timestamps_handler = workflow.execute_activity_method(
+            Activities.update_pipelines_timestamps,
+            {
+                'updated_pipelines': schedule_update_report
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        create_pipelines_timestamps_handler = workflow.execute_activity_method(
+            Activities.create_pipelines_timestamps,
+            {
+                'created_pipelines': schedule_insertion_report
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        delete_pipelines_timestamps_handler = workflow.execute_activity_method(
+            Activities.delete_pipelines_timestamps,
+            {
+                'deleted_pipelines': schedule_deletion_report
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
         await schedule_report_handler
         await slot_report_handler
+        await update_pipelines_timestamps_handler
+        await create_pipelines_timestamps_handler
+        await delete_pipelines_timestamps_handler

tests/orchestrator/activities/test_mongo_db.py

@@ -250,8 +250,8 @@ async def test_aggregate_documents_in_mongodb_missing_aggregation(mongo_db):
 @patch("orchestrator.activities.mongo_db.datetime")
 async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db):
     input_data = {"updated_pipelines": [
-        {"schedule_name": "test1", "namespace": "test1"},
+        {"schedule_name": "test1", "namespace": "test1", "success": True},
-        {"schedule_name": "test2", "namespace": "test2"}
+        {"schedule_name": "test2", "namespace": "test2", "success": True}
     ]}
     mongo_db.database["pipelines"].update_many.return_value = MagicMock()
     await mongo_db.update_pipelines_timestamps(input_data)
@@ -269,8 +269,8 @@ async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db):
 @patch("orchestrator.activities.mongo_db.datetime")
 async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db):
     input_data = {"created_pipelines": [
-        {"schedule_name": "test1", "namespace": "test1"},
+        {"schedule_name": "test1", "namespace": "test1", "success": True},
-        {"schedule_name": "test2", "namespace": "test2"}
+        {"schedule_name": "test2", "namespace": "test2", "success": True}
     ]}
     mongo_db.database["pipelines"].insert_many.return_value = MagicMock()
     await mongo_db.create_pipelines_timestamps(input_data)
@@ -288,8 +288,8 @@ async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db):
 @patch("orchestrator.activities.mongo_db.datetime")
 async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db):
     input_data = {"deleted_pipelines": [
-        {"schedule_name": "test1", "namespace": "test1"},
+        {"schedule_name": "test1", "namespace": "test1", "success": True},
-        {"schedule_name": "test2", "namespace": "test2"}
+        {"schedule_name": "test2", "namespace": "test2", "success": True}
     ]}
     mongo_db.database["pipelines"].delete_many.return_value = MagicMock()
     await mongo_db.delete_pipelines_timestamps(input_data)

values.yaml

@@ -123,7 +123,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1148-separar-scouter-laborious-e-orchestrator-por-namespaces"
+    value: "SIENTIAPDE-1150-ajustar-orquestrador-para-manter-uma-store-de-detalhes-dos-schedules"
   - name: PYTHON_APP
     value: "orchestrator.worker.worker"