SIENTIAPDE-1222

SIENTIAPDE-1222
Enhance tests for email and MongoDB activities

- Added a test for sending emails when no SMTP server is configured, ensuring proper handling of such cases.
- Updated MongoDB tests to include timestamp fields in the results, improving data accuracy in document retrieval.
- Modified connector configuration tests to reflect changes in SMTP server settings.

tests/orchestrator/activities/test_email.py

@@ -282,6 +282,19 @@ def test_try_send_email_reconnect_quit_failure(smtp, email):
         assert False, "Expected exception"
 
 
+@mark.asyncio
+async def test_send_email_without_smtp_server(email):
+    email.smtp_server = None
+    input_data = {
+        **metadata,
+        "receiver_groups": {},
+        "mail_type": "test_TYPE"
+    }
+    response = await email.send_email(input_data)
+
+    assert response == {}
+
+
 @mark.asyncio
 @patch('orchestrator.activities.email.MIMEText')
 @patch('orchestrator.activities.email.MIMEMultipart')

tests/orchestrator/activities/test_mongo_db.py

@@ -107,19 +107,30 @@ async def test_find_documents_in_mongodb_success(mongo_db):
     }}
     mock_collection = MagicMock()
     mock_collection.find.return_value = [
-        {"_id": "12345", "name": "test1"},
+        {
-        {"_id": "67890", "name": "test2"}
+            "_id": "12345",
+            "name": "test1",
+            "timestamp": datetime.strptime(
+                "2023-01-01 12:00:00.000000+0000", DATETIME_FORMAT_MS_WITH_TZ)},
+        {
+            "_id": "67890",
+            "name": "test2",
+            "timestamp": datetime.strptime(
+                "2023-01-01 12:00:00.000000+0000", DATETIME_FORMAT_MS_WITH_TZ)}
     ]
     mongo_db.database.__getitem__.return_value = mock_collection
 
     result = await mongo_db.find_documents_in_mongodb(
         {
-            "query": input_data
+            "query": input_data,
+            "timestamp_fields": ["timestamp"]
         })
 
     assert len(result) == 2
-    assert result[0] == {"name": "test1"}
+    assert result[0] == {"name": "test1",
-    assert result[1] == {"name": "test2"}
+                         "timestamp": "2023-01-01 12:00:00.000000+0000"}
+    assert result[1] == {"name": "test2",
+                         "timestamp": "2023-01-01 12:00:00.000000+0000"}
     mock_collection.find.assert_called_once_with(
         {"name": {"$exists": True}}, {"_id": 0}
     )
@@ -186,19 +197,30 @@ async def test_aggregate_documents_in_mongodb_success(mongo_db):
     ]}
     mock_collection = MagicMock()
     mock_collection.aggregate.return_value = [
-        {"_id": "asdad", "name": "test1"},
+        {
-        {"_id": "adzx", "name": "test2"}
+            "_id": "asdad",
+            "name": "test1",
+            "timestamp": datetime.strptime(
+                "2023-01-01 12:00:00.000000+0000", DATETIME_FORMAT_MS_WITH_TZ)},
+        {
+            "_id": "adzx",
+            "name": "test2",
+            "timestamp": datetime.strptime(
+                "2023-01-01 12:00:00.000000+0000", DATETIME_FORMAT_MS_WITH_TZ)}
     ]
     mongo_db.database.__getitem__.return_value = mock_collection
 
     result = await mongo_db.aggregate_documents_in_mongodb(
         {
-            "query": input_data
+            "query": input_data,
+            "timestamp_fields": ["timestamp"]
         })
 
     assert len(result) == 2
-    assert result[0] == {"name": "test1"}
+    assert result[0] == {"name": "test1",
-    assert result[1] == {"name": "test2"}
+                         "timestamp": "2023-01-01 12:00:00.000000+0000"}
+    assert result[1] == {"name": "test2",
+                         "timestamp": "2023-01-01 12:00:00.000000+0000"}
     expected_pipeline = input_data["aggregation"]
     expected_pipeline.append({"$project": {"_id": 0}})
 

tests/orchestrator/utils/test_connectors_config.py

@@ -127,7 +127,7 @@ def test_build_email_config_with_defaults():
     assert build_email_config() == {
         'sender_email': 'sientia-alerts@aignosi.com',
         'sender_password': 'sientia',
-        'smtp_server': 'smtp.gmail.com',
+        'smtp_server': None,
         'smtp_port': 587
     }
 

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -16,8 +16,11 @@ def test_common_config():
         "workflow_type": "scouter",
         "schedule_name": "test_schedule",
         "model_id": "test_model_id",
-        "models": {
+        "model": {
-            "name": "test_model_name"
+            "name": "test_model_name",
+            "model_config": {
+                "test_config": "test_config"
+            }
         }
     }
     result = common_config(config)
@@ -27,7 +30,10 @@ def test_common_config():
         "frequency": "1m",
         "max_retry_policy": 1,
         "model_id": "test_model_id",
-        "model_name": "test_model_name"
+        "model_name": "test_model_name",
+        "model_config": {
+            "test_config": "test_config"
+        }
     }
     assert result == expected
 
@@ -37,8 +43,11 @@ def test_minimal_retrain():
         "workflow_type": "minimal_retrain",
         "schedule_name": "test_schedule",
         "model_id": "test_model_id",
-        "models": {
+        "model": {
-            "name": "test_model_name"
+            "name": "test_model_name",
+            "model_config": {
+                "test_config": "test_config"
+            }
         },
         "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;",
         "datetime_columns": ["timestamp"]
@@ -51,6 +60,9 @@ def test_minimal_retrain():
         "max_retry_policy": 1,
         "model_id": "test_model_id",
         "model_name": "test_model_name",
+        "model_config": {
+            "test_config": "test_config"
+        },
         "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;",
         "schema": "sientia_data",
         "table_name": "log_retrain",
@@ -64,8 +76,11 @@ def test_scouter():
         "workflow_type": "scouter",
         "schedule_name": "test_schedule",
         "model_id": "test_model_id",
-        "models": {
+        "model": {
-            "name": "test_model_name"
+            "name": "test_model_name",
+            "model_config": {
+                "test_config": "test_config"
+            }
         },
         "filters": [
             {
@@ -90,6 +105,9 @@ def test_scouter():
         "max_retry_policy": 1,
         "model_id": "test_model_id",
         "model_name": "test_model_name",
+        "model_config": {
+            "test_config": "test_config"
+        },
         "topic": "raw_test_schedule",
         "trigger_laborious": False,
         "filters": {
@@ -162,8 +180,11 @@ def test_predictions_batch(mock_process_path_priority,
         "schedule_name": "test_schedule",
         "workflow_type": "predictions_batch",
         "model_id": "test_model_id",
-        "models": {
+        "model": {
-            "name": "test_model_name"
+            "name": "test_model_name",
+            "model_config": {
+                "test_config": "test_config"
+            }
         },
         "query": "test_query",
         "write_tags": [
@@ -236,6 +257,9 @@ def test_predictions_batch(mock_process_path_priority,
         "max_retry_policy": 1,
         "model_id": "test_model_id",
         "model_name": "test_model_name",
+        "model_config": {
+            "test_config": "test_config"
+        },
         "query": "test_query",
         "schema": "sientia_data",
         "table_name": "predictions",

tests/orchestrator/workflows/subworkflows/test_process_notifications.py

@@ -70,7 +70,10 @@ async def test_run(workflow_mock, process_notifications):
             },
             schedule_to_close_timeout=ANY,
             retry_policy=ANY
-        ),
+        )]
+    )
+
+    workflow_mock.execute_activity_method.assert_has_calls([
         call(
             Activities.export_data_to_postgres,
             {
@@ -87,3 +90,48 @@ async def test_run(workflow_mock, process_notifications):
             retry_policy=ANY
         )
     ])
+
+
+@mark.asyncio
+@patch("orchestrator.workflows.subworkflows.process_notifications.workflow", new_callable=AsyncMock)
+async def test_run_send_email_return_empty(workflow_mock, process_notifications):
+    input_data = {
+        'metadata': metadata,
+        'notification_package': ["content"],
+        'mail_type': 'test_mail_type',
+        'schema': 'test_schema',
+        'table_name': 'test_table_name',
+    }
+
+    workflow_mock.execute_activity_method.return_value = []
+
+    response = await process_notifications.run(input_data)
+
+    workflow_mock.execute_local_activity_method.assert_has_calls([
+        call(
+            Activities.build_email_html,
+            {
+                **metadata,
+                'receiver_groups': input_data['notification_package'],
+                'mail_type': input_data['mail_type']
+            },
+            schedule_to_close_timeout=ANY,
+            retry_policy=ANY
+        )
+    ])
+
+    workflow_mock.execute_activity_method.assert_has_calls([
+        call(
+            Activities.send_email,
+            {
+                **metadata,
+                'receiver_groups': workflow_mock.execute_local_activity_method.return_value,
+                'mail_type': input_data['mail_type']
+            },
+            schedule_to_close_timeout=ANY,
+            retry_policy=ANY
+        )]
+    )
+
+    assert workflow_mock.execute_activity_method.call_count == 1
+    assert workflow_mock.execute_local_activity_method.call_count == 1

tests/orchestrator/workflows/test_orchestrator.py

@@ -34,8 +34,9 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.aggregate_documents_in_mongodb,
             {
+                **metadata,
                 "query": input_data["pipelines_query"],
-                **metadata
+                "timestamp_fields": ["updated_at"]
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -46,8 +47,8 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.find_documents_in_mongodb,
             {
-                "query": input_data["opc_servers_query"],
+                **metadata,
-                **metadata
+                "query": input_data["opc_servers_query"]
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -58,10 +59,11 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.find_documents_in_mongodb,
             {
+                **metadata,
                 "query": {
                     "collection": "orchestrated_schedules"
                 },
-                **metadata
+                "timestamp_fields": ["updated_at"]
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -94,8 +96,8 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.format_schedule_config,
             {
-                'schedule_config': workflow_mock.start_local_activity_method.return_value,
+                **metadata,
-                **metadata
+                'schedule_config': workflow_mock.start_local_activity_method.return_value
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -106,8 +108,8 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.process_schedules,
             {
-                'pipelines': workflow_mock.start_local_activity_method.return_value,
+                **metadata,
-                **metadata
+                'pipelines': workflow_mock.start_local_activity_method.return_value
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -118,10 +120,10 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.process_slots,
             {
+                **metadata,
                 'opc_servers': workflow_mock.start_local_activity_method.return_value,
                 'active_ingestors': workflow_mock.start_local_activity_method.return_value,
                 'pipelines': workflow_mock.start_local_activity_method.return_value,
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -132,9 +134,9 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.create_schedule_config,
             {
+                **metadata,
                 'current_schedule_config': workflow_mock.start_local_activity_method.return_value,
-                'schedule_config': workflow_mock.start_local_activity_method.return_value,
+                'schedule_config': workflow_mock.start_local_activity_method.return_value
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -145,9 +147,9 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.create_slot_config,
             {
+                **metadata,
                 'current_slot_config': workflow_mock.start_local_activity_method.return_value,
-                'slot_config': workflow_mock.start_local_activity_method.return_value,
+                'slot_config': workflow_mock.start_local_activity_method.return_value
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -158,8 +160,8 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.normalize_schedules,
             {
-                'orchestrated_schedules': workflow_mock.start_local_activity_method.return_value,
+                **metadata,
-                **metadata
+                'orchestrated_schedules': workflow_mock.start_local_activity_method.return_value
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -182,9 +184,9 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.delete_slots,
             {
+                **metadata,
                 'to_delete':
-                    workflow_mock.start_local_activity_method.return_value['to_delete'],
+                    workflow_mock.start_local_activity_method.return_value['to_delete']
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -195,9 +197,9 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.update_slots,
             {
+                **metadata,
                 'to_insert':
-                    workflow_mock.start_local_activity_method.return_value['to_insert'],
+                    workflow_mock.start_local_activity_method.return_value['to_insert']
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -208,9 +210,9 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.delete_schedules,
             {
+                **metadata,
                 'schedules':
-                    workflow_mock.start_local_activity_method.return_value['to_delete'],
+                    workflow_mock.start_local_activity_method.return_value['to_delete']
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -221,9 +223,9 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.create_schedules,
             {
+                **metadata,
                 'schedules':
-                    workflow_mock.start_local_activity_method.return_value['to_create'],
+                    workflow_mock.start_local_activity_method.return_value['to_create']
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -234,9 +236,9 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.update_schedules,
             {
+                **metadata,
                 'schedules':
-                    workflow_mock.start_local_activity_method.return_value['to_update'],
+                    workflow_mock.start_local_activity_method.return_value['to_update']
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -247,10 +249,10 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.report_schedule_orchestration,
             {
+                **metadata,
                 'created_schedules': workflow_mock.start_activity_method.return_value,
                 'updated_schedules': workflow_mock.start_activity_method.return_value,
                 'deleted_schedules': workflow_mock.start_activity_method.return_value,
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -261,9 +263,9 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.report_slot_orchestration,
             {
+                **metadata,
                 'inserted_slots': workflow_mock.start_activity_method.return_value,
                 'deleted_slots': workflow_mock.start_activity_method.return_value,
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -274,8 +276,8 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.update_pipelines_timestamps,
             {
+                **metadata,
                 'updated_pipelines': workflow_mock.start_activity_method.return_value,
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -286,8 +288,8 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.delete_pipelines_timestamps,
             {
+                **metadata,
                 'deleted_pipelines': workflow_mock.start_activity_method.return_value,
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -298,8 +300,8 @@ async def test_run(workflow_mock, orchestrator):
         call(
             Activities.create_pipelines_timestamps,
             {
+                **metadata,
                 'created_pipelines': workflow_mock.start_activity_method.return_value,
-                **metadata
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY