feat: setVariable activity

src/python/data_factory_testing_framework/exceptions/variable_being_evaluated_does_not_exist_error.py

@@ -0,0 +1,6 @@
+class VariableBeingEvaluatedDoesNotExistError(Exception):
+    def __init__(self, variable_name):
+        super().__init__(f"Variable being evaluated does not exist: {variable_name}")
+
+
+

src/python/data_factory_testing_framework/generated/models/data_factory_element.py

@@ -10,5 +10,5 @@ class DataFactoryElement(Generic[T]):
     def __init__(self, expression: str):
         self.expression = expression
 
-    def evaluate(self):
-        self.value = "test2"  # TODO: Evaluate expression
+    def evaluate(self, state):
+        return self.expression

src/python/data_factory_testing_framework/models/activities/base/activity.py

@@ -7,11 +7,11 @@ class Activity:
     status: DependencyCondition
 
     def evaluate(self, state: PipelineRunState) -> Activity:
-        self.evaluate_expressions(self)
+        self.evaluate_expressions(self, state)
         self.status = DependencyCondition.Succeeded
         return self
 
-    def evaluate_expressions(self, obj: Any, visited: List[Any] = None):
+    def evaluate_expressions(self, obj: Any, state: PipelineRunState, visited: List[Any] = None):
         if visited is None:
             visited = []
 
@@ -24,9 +24,9 @@ def evaluate_expressions(self, obj: Any, visited: List[Any] = None):
         for attribute_name in attribute_names:
             attribute = getattr(obj, attribute_name)
             if data_factory_element := isinstance(attribute, DataFactoryElement) and attribute:
-                data_factory_element.evaluate()
+                data_factory_element.evaluate(state)
             else:
-                self.evaluate_expressions(attribute, visited)
+                self.evaluate_expressions(attribute, state, visited)
 
     def get_scoped_activity_result_by_name(self, name: str, state: PipelineRunState):
         return next((activity_result for activity_result in state.scoped_pipeline_activity_results if activity_result.name == name), None)

src/python/data_factory_testing_framework/models/activities/set_variable_activity.py

@@ -0,0 +1,16 @@
+from data_factory_testing_framework.generated.models import SetVariableActivity, ControlActivity
+from data_factory_testing_framework.models.state.pipeline_run_state import PipelineRunState
+
+
+class SetVariableActivity:
+
+    def evaluate(self: SetVariableActivity, state: PipelineRunState):
+        super(ControlActivity, self).evaluate(state)
+
+        if self.variable_name == "pipelineReturnValue":
+            return self
+
+        state.set_variable(self.variable_name, self.value.evaluate(state))
+
+        return self
+

src/python/data_factory_testing_framework/models/patch_models.py

@@ -8,6 +8,7 @@
 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.activities.set_variable_activity import SetVariableActivity
 from data_factory_testing_framework.models.expression import Expression
 
 
@@ -20,6 +21,7 @@ def patch_models():
     patch_model(_models.Expression, Expression)
     patch_model(_models.IfConditionActivity, IfConditionActivity)
     patch_model(_models.UntilActivity, UntilActivity)
+    patch_model(_models.SetVariableActivity, SetVariableActivity)
 
 
 def patch_model(main_class, partial_class):

src/python/data_factory_testing_framework/models/state/pipeline_run_state.py

@@ -1,5 +1,7 @@
 from typing import List, Dict
 
+from data_factory_testing_framework.exceptions.variable_being_evaluated_does_not_exist_error import \
+    VariableBeingEvaluatedDoesNotExistError
 from data_factory_testing_framework.generated.models import VariableSpecification
 from data_factory_testing_framework.models.base.run_parameter import RunParameter
 from data_factory_testing_framework.models.base.pipeline_run_variable import PipelineRunVariable
@@ -26,3 +28,11 @@ def create_iteration_scope(self, iteration_item: str):
 
     def add_scoped_activity_results_from_scoped_state(self, scoped_state):
         self.pipeline_activity_results.extend(scoped_state.pipeline_activity_results)
+
+    def set_variable(self, variable_name: str, value):
+        for variable in self.variables:
+            if variable.name == variable_name:
+                variable.value = value
+                return
+
+        raise VariableBeingEvaluatedDoesNotExistError(variable_name)

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

@@ -33,13 +33,13 @@ def test_when_evaluated_should_evaluate_correct_child_activities(self, expressio
                 SetVariableActivity(
                     name="setVariableActivity1",
                     variable_name="variable",
-                    value="dummy")
+                    value=DataFactoryElement("dummy"))
             ],
             if_false_activities=[
                 SetVariableActivity(
                     name="setVariableActivity2",
                     variable_name="variable",
-                    value="dummy")
+                    value=DataFactoryElement("dummy"))
             ]
         )
 

src/python/tests/models/activities/test_set_variable_activity.py

@@ -0,0 +1,48 @@
+import pytest
+
+from data_factory_testing_framework.exceptions.variable_being_evaluated_does_not_exist_error import \
+    VariableBeingEvaluatedDoesNotExistError
+from data_factory_testing_framework.generated.models import SetVariableActivity, DataFactoryElement
+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 TestSetVariableActivity:
+
+    def test_when_string_variable_evaluated_then_state_variable_should_be_set(self):
+        # Arrange
+        test_framework = TestFramework()
+        variable_name = "TestVariable"
+        variable = PipelineRunVariable[str](variable_name, "")
+        set_variable_activity = SetVariableActivity(
+            name="SetVariableActivity",
+            variable_name=variable_name,
+            value=DataFactoryElement("TestValue")
+        )
+        state = PipelineRunState()
+        state.variables.append(variable)
+
+        # Act
+        set_variable_activity.evaluate(state)
+
+        # Assert
+        assert variable.value == "TestValue"
+
+    def test_when_unknown_variable_evaluated_then_should_raise_exception(self):
+        # Arrange
+        test_framework = TestFramework()
+        variable_name = "TestVariable"
+        set_variable_activity = SetVariableActivity(
+            name="SetVariableActivity",
+            variable_name=variable_name,
+            value=DataFactoryElement("TestValue")
+        )
+        state = PipelineRunState()
+
+        # Act
+        with pytest.raises(VariableBeingEvaluatedDoesNotExistError) as exception_info:
+            set_variable_activity.evaluate(state)
+
+        # Assert
+        assert exception_info.value.args[0] == "Variable being evaluated does not exist: TestVariable"