feat: allow null parameters, system variables and a bunch of fixes

src/azure_data_factory_testing_framework/functions/_expression_transformer.py

@@ -2,6 +2,7 @@
 from typing import Callable, Optional
 
 from lark import Discard, Token, Transformer
+from lxml.etree import _Element
 
 from azure_data_factory_testing_framework.exceptions.activity_not_found_error import ActivityNotFoundError
 from azure_data_factory_testing_framework.exceptions.activity_output_field_not_found_error import (
@@ -193,17 +194,37 @@ def expression_item_reference(self, value: list[Token, str, int, float, bool]) -
             raise StateIterationItemNotSetError()
         return item
 
+    def expression_system_variable_reference(
+        self, value: list[Token, str, int, float, bool]
+    ) -> [str, int, float, bool]:
+        if not (isinstance(value[0], Token) and value[0].type == "EXPRESSION_SYSTEM_VARIABLE_NAME"):
+            raise ExpressionEvaluationError(
+                'System variable reference requires Token "EXPRESSION_SYSTEM_VARIABLE_NAME"'
+            )
+
+        system_variable_name: Token = value[0]
+
+        system_variable_parameters: list[RunParameter] = list(
+            filter(lambda p: p.type == RunParameterType.System, self.state.parameters)
+        )
+
+        system_parameters = list(filter(lambda p: p.name == system_variable_name, system_variable_parameters))
+        if len(system_parameters) == 0:
+            raise ExpressionParameterNotFoundError(system_variable_name)
+
+        return system_parameters[0].value
+
     def expression_function_parameters(self, values: list[Token, str, int, float, bool]) -> list:
-        if not all(type(value) in [str, int, float, bool, list] for value in values):
-            raise ExpressionEvaluationError("Function parameters should be string, int, float, bool or list")
+        if not all(type(value) in [str, int, float, bool, list, _Element] or value is None for value in values):
+            raise ExpressionEvaluationError("Function parameters should be string, int, float, bool, list or _Element")
         return values
 
     def expression_parameter(self, values: list[Token, str, int, float, bool, list]) -> str:
         if len(values) != 1:
             raise ExpressionEvaluationError("Function parameter must have only one value")
         parameter = values[0]
-        if type(parameter) not in [str, int, float, bool, list]:
-            raise ExpressionEvaluationError("Function parameters should be string, int, float, bool or list")
+        if type(parameter) not in [str, int, float, bool, list, _Element, None] and parameter is not None:
+            raise ExpressionEvaluationError("Function parameters should be string, int, float, bool, list or _Element")
         return parameter
 
     def expression_evaluation(self, values: list[Token, str, int, float, bool, list]) -> [str, int, float, bool]:
@@ -230,7 +251,7 @@ def expression_array_indices(self, values: list[Token, str, int, float, bool]) -
 
     def expression_function_call(self, values: list[Token, str, int, float, bool]) -> [str, int, float, bool]:
         fn = values[0]
-        fn_parameters = values[1]
+        fn_parameters = values[1] if values[1] is not None else []
         function: Callable = FunctionsRepository.functions.get(fn.value)
 
         pos_or_keyword_parameters = []

src/azure_data_factory_testing_framework/functions/expression_evaluator.py

@@ -12,7 +12,7 @@
 class ExpressionEvaluator:
     def __init__(self) -> None:
         """Evaluator for the expression language."""
-        literal_grammer = """
+        literal_grammar = """
 
             // literal rules
             ?literal_start: literal_evaluation
@@ -34,7 +34,7 @@ def __init__(self) -> None:
             LITERAL_NULL: NULL
         """
 
-        expression_grammer = f"""
+        expression_grammar = f"""
             // TODO: add support for array index
             ?expression_start: expression_evaluation
             expression_evaluation: expression_call [expression_array_indices]
@@ -45,6 +45,7 @@ def __init__(self) -> None:
                                     | expression_dataset_reference
                                     | expression_linked_service_reference
                                     | expression_item_reference
+                                    | expression_system_variable_reference
             expression_array_indices: [EXPRESSION_ARRAY_INDEX]*
 
             // reference rules:
@@ -54,9 +55,10 @@ def __init__(self) -> None:
             expression_dataset_reference: "dataset" "(" EXPRESSION_DATASET_NAME ")"
             expression_linked_service_reference: "linkedService" "(" EXPRESSION_LINKED_SERVICE_NAME ")"
             expression_item_reference: "item()"
+            expression_system_variable_reference: "pipeline" "()" "." EXPRESSION_SYSTEM_VARIABLE_NAME
 
             // function call rules
-            expression_function_call: EXPRESSION_FUNCTION_NAME  "(" expression_function_parameters ")"
+            expression_function_call: EXPRESSION_FUNCTION_NAME  "(" [expression_function_parameters] ")"
             expression_function_parameters: expression_parameter ("," expression_parameter )*
             expression_parameter: EXPRESSION_WS* (EXPRESSION_NULL | EXPRESSION_INTEGER | EXPRESSION_FLOAT | EXPRESSION_BOOLEAN | EXPRESSION_STRING | expression_start) EXPRESSION_WS*
             
@@ -67,6 +69,7 @@ def __init__(self) -> None:
             EXPRESSION_ACTIVITY_NAME: "'" /[^']*/ "'"
             EXPRESSION_DATASET_NAME: "'" /[^']*/ "'"
             EXPRESSION_LINKED_SERVICE_NAME: "'" /[^']*/ "'"
+            EXPRESSION_SYSTEM_VARIABLE_NAME: /[a-zA-Z0-9_]+/
             EXPRESSION_FUNCTION_NAME: {self._supported_functions()}
             EXPRESSION_NULL: NULL
             EXPRESSION_STRING: SINGLE_QUOTED_STRING
@@ -93,7 +96,7 @@ def __init__(self) -> None:
             %import common.WS
         """
 
-        grammer = base_grammar + literal_grammer + expression_grammer
+        grammer = base_grammar + literal_grammar + expression_grammar
         self.lark_parser = Lark(grammer, start="start")
 
     def _supported_functions(self) -> str:

src/azure_data_factory_testing_framework/functions/functions_conversion_implementation.py

@@ -45,10 +45,10 @@ def bool_(value: Any) -> bool:  # noqa: ANN401
 def coalesce(*objects: Any) -> Any:  # noqa: ANN401
     """Return the first non-null value from one or more parameters.
 
-    Empty strings, empty arrays, and empty objects are not null.
+    Returns the first non-null (or non-empty for string) expression
     """
     for obj in objects:
-        if obj is not None:
+        if obj is not None and obj != "":
             return obj
 
     return None

src/azure_data_factory_testing_framework/models/activities/activity.py

@@ -84,6 +84,9 @@ def _evaluate_expressions(
             if not attribute.startswith("_") and not callable(getattr(obj, attribute))
         ]
         for attribute_name in attribute_names:
+            if "activities" in attribute_name:
+                continue
+
             attribute = getattr(obj, attribute_name)
             if attribute is None:
                 continue
@@ -93,6 +96,9 @@ def _evaluate_expressions(
         # Dictionary
         if isinstance(obj, dict):
             for key in obj.keys():
+                if "activities" in key:
+                    continue
+
                 self._evaluate_expressions(obj[key], state, visited, types_to_ignore)
 
         # List

src/azure_data_factory_testing_framework/models/activities/execute_pipeline_activity.py

@@ -24,8 +24,8 @@ def __init__(self, **kwargs: Any) -> None:  # noqa: ANN401
     def get_child_run_parameters(self, state: PipelineRunState) -> List[RunParameter]:
         child_parameters = []
         for parameter in state.parameters:
-            if parameter.type == RunParameterType.Global:
-                child_parameters.append(RunParameter(RunParameterType.Global, parameter.name, parameter.value))
+            if parameter.type == RunParameterType.Global or parameter.type == RunParameterType.System:
+                child_parameters.append(RunParameter(parameter.type, parameter.name, parameter.value))
 
         for parameter_name, parameter_value in self.parameters.items():
             parameter_value = (

src/azure_data_factory_testing_framework/models/activities/until_activity.py

@@ -4,6 +4,7 @@
 from azure_data_factory_testing_framework.models.activities.control_activity import ControlActivity
 from azure_data_factory_testing_framework.models.data_factory_element import DataFactoryElement
 from azure_data_factory_testing_framework.state import PipelineRunState
+from azure_data_factory_testing_framework.state.dependency_condition import DependencyCondition
 
 
 class UntilActivity(ControlActivity):
@@ -26,10 +27,7 @@ def __init__(
         self.activities = activities
 
     def evaluate(self, state: PipelineRunState) -> "UntilActivity":
-        self.expression.evaluate(state)
-
-        super(ControlActivity, self).evaluate(state)
-
+        # Explicitly not evaluate here, but in the evaluate_control_activities method after the first iteration
         return self
 
     def evaluate_control_activities(
@@ -45,4 +43,6 @@ def evaluate_control_activities(
             state.add_scoped_activity_results_from_scoped_state(scoped_state)
 
             if self.expression.evaluate(state):
+                state.add_activity_result(self.name, DependencyCondition.Succeeded)
+                self.set_result(DependencyCondition.Succeeded)
                 break

src/azure_data_factory_testing_framework/state/run_parameter_type.py

@@ -8,6 +8,7 @@ class RunParameterType(str, Enum, metaclass=CaseInsensitiveEnumMeta):
     Global = "Global"
     Dataset = "Dataset"
     LinkedService = "LinkedService"
+    System = "System"
 
     def __str__(self) -> str:
         """Get the string representation of the enum.

tests/unit/functions/test_expression_evaluator.py

@@ -21,6 +21,7 @@
 from azure_data_factory_testing_framework.state.pipeline_run_variable import PipelineRunVariable
 from azure_data_factory_testing_framework.state.run_parameter import RunParameter
 from azure_data_factory_testing_framework.state.run_parameter_type import RunParameterType
+from freezegun import freeze_time
 from lark import Token, Tree
 from pytest import param as p
 
@@ -606,8 +607,21 @@ def test_parse(expression: str, expected: Tree[Token]) -> None:
             0.016666666666666666,
             id="activity_reference_with_nested_property_and_array_index",
         ),
+        p(
+            "@utcNow()",
+            PipelineRunState(),
+            "2021-11-24T12:11:49.753132Z",
+            id="function_call_with_zero_parameters",
+        ),
+        p(
+            "@coalesce(null)",
+            PipelineRunState(),
+            None,
+            id="function_call_with_null_parameter",
+        ),
     ],
 )
+@freeze_time("2021-11-24 12:11:49.753132")
 def test_evaluate(expression: str, state: PipelineRunState, expected: Union[str, int, bool, float]) -> None:
     # Arrange
     evaluator = ExpressionEvaluator()
@@ -715,3 +729,34 @@ def test_evaluate_raises_exception_when_state_iteration_item_not_set() -> None:
 
     # Assert
     assert str(exinfo.value) == "Iteration item not set."
+
+
+def test_evaluate_system_variable() -> None:
+    # Arrange
+    expression = "@pipeline().RunId"
+    evaluator = ExpressionEvaluator()
+    state = PipelineRunState(
+        parameters=[
+            RunParameter(RunParameterType.System, "RunId", "123"),
+        ]
+    )
+
+    # Act
+    actual = evaluator.evaluate(expression, state)
+
+    # Assert
+    assert actual == "123"
+
+
+def test_evaluate_system_variable_raises_exception_when_parameter_not_set() -> None:
+    # Arrange
+    expression = "@pipeline().RunId"
+    evaluator = ExpressionEvaluator()
+    state = PipelineRunState()
+
+    # Act
+    with pytest.raises(ExpressionParameterNotFoundError) as exinfo:
+        evaluator.evaluate(expression, state)
+
+    # Assert
+    assert str(exinfo.value) == "Parameter 'RunId' not found"