test: circular dependency between activities in pipeline

src/python/data_factory_testing_framework/exceptions/pipeline_activities_circular_dependency_error.py

@@ -0,0 +1,4 @@
+class PipelineActivitiesCircularDependencyError(Exception):
+    def __init__(self) -> None:
+        """Exception for circular dependencies in pipeline activities."""
+        super().__init__("Circular dependency detected in pipeline activities")

src/python/data_factory_testing_framework/test_framework.py

@@ -1,5 +1,8 @@
 from typing import List
 
+from data_factory_testing_framework.exceptions.pipeline_activities_circular_dependency_error import (
+    PipelineActivitiesCircularDependencyError,
+)
 from data_factory_testing_framework.generated.models import (
     Activity,
     ControlActivity,
@@ -10,6 +13,7 @@
     UntilActivity,
 )
 from data_factory_testing_framework.models import DataFactoryRepositoryFactory
+from data_factory_testing_framework.models.repositories.data_factory_repository import DataFactoryRepository
 from data_factory_testing_framework.state import PipelineRunState, RunParameter
 
 
@@ -22,9 +26,11 @@ def __init__(self, data_factory_folder_path: str = None, should_evaluate_child_p
             The repository attribute will be populated with the data factory entities if provided.
             should_evaluate_child_pipelines: optional boolean indicating whether child pipelines should be evaluated. Defaults to False.
         """
-        self.repository = data_factory_folder_path is not None and DataFactoryRepositoryFactory.parse_from_folder(
-            data_factory_folder_path,
-        )
+        if data_factory_folder_path is not None:
+            self.repository = DataFactoryRepositoryFactory.parse_from_folder(data_factory_folder_path)
+        else:
+            self.repository = DataFactoryRepository([])
+
         self.should_evaluate_child_pipelines = should_evaluate_child_pipelines
 
     def evaluate_activity(self, activity: Activity, state: PipelineRunState) -> List[Activity]:
@@ -109,9 +115,7 @@ def evaluate_activities(self, activities: List[Activity], state: PipelineRunStat
                             yield child_activity
 
             if not any_activity_evaluated:
-                raise Exception(
-                    "Validate that there are no circular dependencies or whether activity results were not set correctly.",
-                )
+                raise PipelineActivitiesCircularDependencyError()
 
     @staticmethod
     def _is_iteration_activity(activity: Activity) -> bool:

src/python/tests/test_test_framework.py

@@ -0,0 +1,44 @@
+import pytest
+
+from data_factory_testing_framework import TestFramework
+from data_factory_testing_framework.exceptions.pipeline_activities_circular_dependency_error import (
+    PipelineActivitiesCircularDependencyError,
+)
+from data_factory_testing_framework.generated.data_factory_element import DataFactoryElement
+from data_factory_testing_framework.generated.models import ActivityDependency, PipelineResource, SetVariableActivity
+
+
+def test_circular_dependency_between_activities_should_throw_error() -> None:
+    # Arrange
+    test_framework = TestFramework()
+    pipeline = PipelineResource(
+        name="main",
+        parameters={},
+        activities=[
+            SetVariableActivity(
+                name="setVariable1",
+                variable_name="variable",
+                value=DataFactoryElement[str]("'1'"),
+                depends_on=[
+                    ActivityDependency(
+                        activity="setVariable2",
+                        dependency_conditions=["Succeeded"]),
+                ],
+            ),
+            SetVariableActivity(
+                name="setVariable2",
+                variable_name="variable",
+                value=DataFactoryElement[str]("'1'"),
+                depends_on=[
+                    ActivityDependency(
+                        activity="setVariable1",
+                        dependency_conditions=["Succeeded"]),
+                ],
+            ),
+        ],
+    )
+    test_framework.repository.pipelines.append(pipeline)
+
+    # Act & Assert
+    with pytest.raises(PipelineActivitiesCircularDependencyError):
+        next(test_framework.evaluate_pipeline(pipeline, []))