feat: until activity

src/python/data_factory_testing_framework/models/activities/control_activities/until_activity.py

@@ -0,0 +1,24 @@
+from typing import Callable, Generator
+
+from data_factory_testing_framework.generated.models import ForEachActivity, Activity, ControlActivity, \
+    UntilActivity
+from data_factory_testing_framework.models.state.pipeline_run_state import PipelineRunState
+
+
+class UntilActivity:
+
+    def evaluate(self: UntilActivity, state: PipelineRunState):
+        self.expression.evaluate(state)
+
+        return super(ControlActivity, self).evaluate(state)
+
+    def evaluate_control_activity_iterations(self: UntilActivity, state: PipelineRunState, evaluate_activities: Callable[[PipelineRunState], Generator[Activity, None, None]]):
+        while True:
+            scoped_state = state.create_iteration_scope(None)
+            for activity in evaluate_activities(self.activities, scoped_state):
+                yield activity
+
+            state.add_scoped_activity_results_from_scoped_state(scoped_state)
+
+            if self.expression.evaluate(state):
+                break

src/python/data_factory_testing_framework/models/patch_models.py

@@ -7,6 +7,7 @@
 from data_factory_testing_framework.models.activities.control_activities.for_each_activity import ForEachActivity
 from data_factory_testing_framework.models.activities.control_activities.if_condition_activity import \
     IfConditionActivity
+from data_factory_testing_framework.models.activities.control_activities.until_activity import UntilActivity
 from data_factory_testing_framework.models.expression import Expression
 
 
@@ -18,6 +19,7 @@ def patch_models():
     patch_model(_models.ForEachActivity, ForEachActivity)
     patch_model(_models.Expression, Expression)
     patch_model(_models.IfConditionActivity, IfConditionActivity)
+    patch_model(_models.UntilActivity, UntilActivity)
 
 
 def patch_model(main_class, partial_class):

src/python/tests/models/activities/control_activities/test_until_activity.py

@@ -0,0 +1,45 @@
+from data_factory_testing_framework.generated.models import ForEachActivity, Expression, ExpressionType, \
+    SetVariableActivity, DataFactoryElement, UntilActivity
+from data_factory_testing_framework.models.base.pipeline_run_variable import PipelineRunVariable
+from data_factory_testing_framework.models.state.pipeline_run_state import PipelineRunState
+from data_factory_testing_framework.models.test_framework import TestFramework
+
+
+class TestUntilActivity:
+
+    def test_when_evaluate_until_activity_should_repeat_until_expression_is_true(self):
+        # Arrange
+        test_framework = TestFramework()
+        until_activity = UntilActivity(name="UntilActivity",
+                                       expression=Expression(type=ExpressionType.EXPRESSION, value="@equals(1, 1)"),
+                                       activities=[
+                                           SetVariableActivity(name="setVariable", variable_name="variable",
+                                                               value=DataFactoryElement[str]("'1'"),
+                                                               depends_on=[])
+                                       ],
+                                       depends_on=[])
+
+        state = PipelineRunState()
+        state.variables.append(PipelineRunVariable("variable", ""))
+
+        # Act
+        until_activity.expression.evaluate = lambda state: False
+        activities = test_framework.evaluate_activity(until_activity, state)
+
+        # Assert
+        set_variable_activity = next(activities)
+        assert set_variable_activity is not None
+        assert set_variable_activity.name == "setVariable"
+
+        set_variable_activity = next(activities)
+        assert set_variable_activity is not None
+        assert set_variable_activity.name == "setVariable"
+
+        until_activity.expression.evaluate = lambda state: True
+
+        # Assert that there are no more activities
+        try:
+            next(activities)
+            assert False  # This line should not be reached, an exception should be raised
+        except StopIteration:
+            pass