Merge pull request #9 from Aignosi/SIENTIAPDE-1151-criar-testes-de-stress

Sientiapde 1151 criar testes de stress

orchestrator/activities/formatters.py

@@ -118,14 +118,33 @@ class Formatters(BaseActivity):
         for i in range(1, number_of_slots):
             slot_config[f"{i}"] = {}
             for tag in tags[last_index:last_index + tags_per_slot]:
-                slot_config = build_tag_config(
+                try:
-                    tag, slot_config.copy(), opc_servers, i)
+                    slot_config = build_tag_config(
+                        tag, slot_config.copy(), opc_servers, i)
+                except ValueError as e:
+                    self.send_notification(
+                        metadata=metadata,
+                        notification_id="ORCHESTRATOR_BUILD_TAG_CONFIG_ERROR",
+                        message=str(e),
+                        block="orchestrator",
+                        level=NotificationLevel.ERROR
+                    )
+
             last_index += tags_per_slot
 
         slot_config[f"{number_of_slots}"] = {}
         for tag in tags[last_index:]:
-            slot_config = build_tag_config(
+            try:
-                tag, slot_config.copy(), opc_servers, number_of_slots)
+                slot_config = build_tag_config(
+                    tag, slot_config.copy(), opc_servers, number_of_slots)
+            except ValueError as e:
+                self.send_notification(
+                    metadata=metadata,
+                    notification_id="ORCHESTRATOR_BUILD_TAG_CONFIG_ERROR",
+                    message=str(e),
+                    block="orchestrator",
+                    level=NotificationLevel.ERROR
+                )
 
         self.info("Processed slots", metadata=metadata)
         self.debug(json.dumps(

orchestrator/utils/orchestrator_functions.py

@@ -146,6 +146,10 @@ def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]:
 def build_tag_config(tag: dict[str, Any], slot_config: dict[str, Any],
                      opc_servers: dict[str, Any], i: int):
     server_id = tag['server_id']
+
+    if server_id not in opc_servers:
+        raise ValueError(f"Server {server_id} not found in opc_servers")
+
     server_name = opc_servers[server_id]['server_name']
     if server_name not in slot_config[f"{i}"]:
         slot_config[f"{i}"][server_name] = {

tests/orchestrator/activities/test_formatters.py

@@ -268,6 +268,70 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
     }
 
 
+@mark.asyncio
+@patch("orchestrator.activities.formatters.gather_read_tags",
+       return_value={
+           "1:test_tag_address": {
+               "server_id": "1",
+               "server_name": "test_server_name",
+               "tag_address": "test_tag_address",
+               "topics": ["raw_test_schedule"]
+           },
+           "2:test_tag_address2": {
+               "server_id": "2",
+               "server_name": "test_server_name2",
+               "tag_address": "test_tag_address2",
+               "topics": ["raw_test_schedule2"]
+           }
+       })
+@patch("orchestrator.activities.formatters.build_tag_config", side_effect=ValueError("test_error"))
+async def test_process_slots_exception(mock_build_tag_config, mock_gather_read_tags, formatters):
+    input_data = {
+        **metadata,
+        "opc_servers": [
+            {
+                "id": "1",
+                "server_name": "test_server_name",
+                "url": "test_url",
+                "uri": "test_uri",
+                "security_spec": {
+                    "test_name": "test_spec"
+                }
+            },
+            {
+                "id": "2",
+                "server_name": "test_server_name2",
+                "url": "test_url2",
+                "uri": "test_uri2"
+            }
+        ],
+        "active_ingestors": [
+            "test_active_ingestor1",
+            "test_active_ingestor2"
+        ],
+        "pipelines": "test_gather_read_tags"
+    }
+
+    await formatters.process_slots(input_data)
+
+    formatters.send_notification.assert_has_calls([
+        call(
+            metadata=metadata['metadata'],
+            notification_id="ORCHESTRATOR_BUILD_TAG_CONFIG_ERROR",
+            message="test_error",
+            block="orchestrator",
+            level=NotificationLevel.ERROR
+        ),
+        call(
+            metadata=metadata['metadata'],
+            notification_id="ORCHESTRATOR_BUILD_TAG_CONFIG_ERROR",
+            message="test_error",
+            block="orchestrator",
+            level=NotificationLevel.ERROR
+        )
+    ])
+
+
 @mark.asyncio
 async def test_format_schedule_config(formatters):
     input_data = {

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -342,3 +342,16 @@ def test_build_tag_config():
         }
     }
     assert result == expected
+
+
+def test_build_tag_config_no_server_id():
+    tag = {
+        "server_id": "1",
+        "tag_address": "test_tag_address"
+    }
+    opc_servers = {}
+
+    try:
+        build_tag_config(tag, {}, opc_servers, 1)
+    except ValueError as e:
+        assert str(e) == "Server 1 not found in opc_servers"

values.yaml

@@ -132,7 +132,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1163-alterar-dinamica-de-notificacoes-para-usar-o-mongodb-ao-inves-do-kafka"
+    value: "SIENTIAPDE-1151-criar-testes-de-stress"
   - name: PYTHON_APP
     value: "orchestrator.worker.worker"