PB103 Migration

.changelog/4340.yml

@@ -0,0 +1,4 @@
+changes:
+- description: Added the PB103 to the new validation format. Validate whether there is an unconnected task.
+  type: internal
+pr_number: 4340

demisto_sdk/commands/validate/sdk_validation_config.toml

@@ -29,7 +29,7 @@ select = [
     "IN100", "IN101", "IN102", "IN104", "IN106", "IN107", "IN108", "IN109", "IN110", "IN112", "IN113", "IN114", "IN115", "IN117", "IN118", "IN121", "IN122", "IN123", "IN124", "IN125", "IN126", "IN127", "IN130",
     "IN131", "IN134", "IN135", "IN139", "IN141", "IN142", "IN144", "IN145", "IN146", "IN149", "IN150", "IN151", "IN152", "IN153", "IN154", "IN156", "IN158", "IN159", "IN160",
     "IN161", "IN162",
-    "PB100", "PB101","PB104","PB118", "PB123",
+    "PB100", "PB101","PB103","PB104","PB118", "PB123",
     "DO100", "DO101", "DO102", "DO103", "DO104", "DO105", "DO106",
     "DS100", "DS101", "DS107",
     "SC100", "SC105", "SC106", "SC109",
@@ -52,7 +52,7 @@ select = [
     "IN131", "IN134", "IN135", "IN139", "IN141", "IN142", "IN144", "IN145", "IN146", "IN149", "IN150", "IN151", "IN152", "IN153", "IN154", "IN156", "IN158", "IN159", "IN160",
     "IN161", "IN162",
     "LO107",
-    "PB100", "PB101","PB104", "PB123",
+    "PB100", "PB101","PB103","PB104", "PB123",
     "BA100", "BA101", "BA105", "BA106", "BA110", "BA111", "BA113", "BA116", "BA118", "BA119", "BA126",
     "DS100", "DS101",
     "PA100", "PA101", "PA102", "PA103", "PA104", "PA105", "PA107", "PA108", "PA109", "PA111", "PA113", "PA115", "PA117", "PA118", "PA119", "PA120",

demisto_sdk/commands/validate/tests/PB_validators_test.py

@@ -8,6 +8,10 @@
 from demisto_sdk.commands.validate.validators.PB_validators.PB101_is_playbook_has_unreachable_condition import (
     IsAskConditionHasUnreachableConditionValidator,
 )
+from demisto_sdk.commands.validate.validators.PB_validators.PB103_does_playbook_have_unconnected_tasks import (
+    ERROR_MSG,
+    DoesPlaybookHaveUnconnectedTasks,
+)
 from demisto_sdk.commands.validate.validators.PB_validators.PB104_deprecated_description import (
     DeprecatedDescriptionValidator,
 )
@@ -187,7 +191,6 @@ def test_is_deprecated_with_invalid_description(content_item, expected_result):
 
 
 def test_IsAskConditionHasUnreachableConditionValidator():
-
     playbook = create_playbook_object()
     assert not IsAskConditionHasUnreachableConditionValidator().is_valid([playbook])
     playbook.tasks = {
@@ -206,7 +209,6 @@ def test_IsAskConditionHasUnreachableConditionValidator():
 
 
 def test_IsAskConditionHasUnhandledReplyOptionsValidator():
-
     playbook = create_playbook_object()
     assert not IsAskConditionHasUnhandledReplyOptionsValidator().is_valid([playbook])
     playbook.tasks = {
@@ -222,3 +224,89 @@ def test_IsAskConditionHasUnhandledReplyOptionsValidator():
         )
     }
     assert IsAskConditionHasUnhandledReplyOptionsValidator().is_valid([playbook])
+
+
+def test_does_playbook_have_unconnected_tasks():
+    """
+    Given: A playbook with tasks that are connected to each other.
+    When: Validating the playbook.
+    Then: The playbook is valid.
+    """
+    playbook = create_playbook_object(
+        paths=["starttaskid", "tasks"],
+        values=[
+            "0",
+            {
+                "0": {
+                    "id": "test task",
+                    "type": "regular",
+                    "message": {"replyOptions": ["yes"]},
+                    "nexttasks": {"#none#": ["1"]},
+                    "task": {"id": "27b9c747-b883-4878-8b60-7f352098a63c"},
+                    "taskid": "27b9c747-b883-4878-8b60-7f352098a631",
+                },
+                "1": {
+                    "id": "test task",
+                    "type": "condition",
+                    "message": {"replyOptions": ["yes"]},
+                    "nexttasks": {"no": ["2"]},
+                    "task": {"id": "27b9c747-b883-4878-8b60-7f352098a63c"},
+                    "taskid": "27b9c747-b883-4878-8b60-7f352098a632",
+                },
+            },
+        ],
+    )
+    validation_results = DoesPlaybookHaveUnconnectedTasks().is_valid([playbook])
+    assert len(validation_results) == 0  # No validation results should be returned
+
+
+def test_does_playbook_have_unconnected_tasks_not_valid():
+    """
+    Given: A playbook with tasks that are not connected to the root task.
+    When: Validating the playbook.
+    Then: The playbook is not valid.
+    """
+    playbook = create_playbook_object(
+        paths=["starttaskid", "tasks"],
+        values=[
+            "0",
+            {
+                "0": {
+                    "id": "test task",
+                    "type": "regular",
+                    "message": {"replyOptions": ["yes"]},
+                    "nexttasks": {"#none#": ["1"]},
+                    "task": {"id": "27b9c747-b883-4878-8b60-7f352098a63c"},
+                    "taskid": "27b9c747-b883-4878-8b60-7f352098a631",
+                },
+                "1": {
+                    "id": "test task",
+                    "type": "condition",
+                    "message": {"replyOptions": ["yes"]},
+                    "nexttasks": {"no": ["2"]},
+                    "task": {"id": "27b9c747-b883-4878-8b60-7f352098a63c"},
+                    "taskid": "27b9c747-b883-4878-8b60-7f352098a632",
+                },
+                "3": {
+                    "id": "test task",
+                    "type": "condition",
+                    "message": {"replyOptions": ["yes"]},
+                    "nexttasks": {"no": ["2"]},
+                    "task": {"id": "27b9c747-b883-4878-8b60-7f352098a63c"},
+                    "taskid": "27b9c747-b883-4878-8b60-7f352098a632",
+                },
+                "4": {
+                    "id": "test task",
+                    "type": "condition",
+                    "message": {"replyOptions": ["yes"]},
+                    "nexttasks": {"no": ["2"]},
+                    "task": {"id": "27b9c747-b883-4878-8b60-7f352098a63c"},
+                    "taskid": "27b9c747-b883-4878-8b60-7f352098a632",
+                },
+            },
+        ],
+    )
+    orphan_tasks = ["3", "4"]
+    validation_result = DoesPlaybookHaveUnconnectedTasks().is_valid([playbook])
+    assert validation_result
+    assert validation_result[0].message == ERROR_MSG.format(orphan_tasks=orphan_tasks)

demisto_sdk/commands/validate/validators/PB_validators/PB103_does_playbook_have_unconnected_tasks.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from typing import Any, Iterable, List
+
+from demisto_sdk.commands.content_graph.objects.playbook import Playbook
+from demisto_sdk.commands.validate.validators.base_validator import (
+    BaseValidator,
+    ValidationResult,
+)
+
+ContentTypes = Playbook
+ERROR_MSG = "The following tasks ids have no previous tasks: {orphan_tasks}."
+
+
+class DoesPlaybookHaveUnconnectedTasks(BaseValidator[ContentTypes]):
+    error_code = "PB103"
+    description = "Validate whether there is an unconnected task."
+    rationale = "Make sure there are no unconnected tasks to ensure the playbook will work as expected."
+    error_message = ERROR_MSG
+    related_field = "tasks"
+    is_auto_fixable = False
+
+    @staticmethod
+    def is_unconnected_task(playbook: ContentTypes) -> set[Any]:
+        """Checks whether a playbook has an unconnected task.
+        Args:
+            - playbook (ContentTypes): The playbook to check.
+        Return:
+            - bool. True if the playbook has an unconnected task, False otherwise.
+        """
+        start_task_id = playbook.data.get("starttaskid")
+        tasks = playbook.tasks
+        tasks_bucket = set()
+        next_tasks_bucket = set()
+
+        for task_id, task in tasks.items():
+            if task_id != start_task_id:
+                tasks_bucket.add(task_id)
+            if next_tasks := task.nexttasks:
+                for next_task_ids in next_tasks.values():
+                    if next_task_ids:
+                        next_tasks_bucket.update(next_task_ids)
+        orphan_tasks = tasks_bucket.difference(next_tasks_bucket)
+        return orphan_tasks
+
+    def is_valid(self, content_items: Iterable[ContentTypes]) -> List[ValidationResult]:
+        return [
+            ValidationResult(
+                validator=self,
+                message=self.error_message.format(orphan_tasks=sorted(orphan_tasks)),
+                content_object=content_item,
+            )
+            for content_item in content_items
+            if (orphan_tasks := self.is_unconnected_task(content_item))
+        ]