feat: function_call, function_argument and expression_parsing

src/python/data_factory_testing_framework/functions/function_argument.py

@@ -0,0 +1,3 @@
+class FunctionArgument:
+    def __init__(self, expression: str):
+        self.expression = expression.strip('\n').strip(' ')

src/python/data_factory_testing_framework/functions/function_call.py

@@ -0,0 +1,9 @@
+from typing import List
+
+from data_factory_testing_framework.functions.function_argument import FunctionArgument
+
+
+class FunctionCall:
+    def __init__(self, name: str, arguments: List):
+        self.name = name
+        self.arguments = arguments

src/python/data_factory_testing_framework/functions/function_parser.py

@@ -0,0 +1,52 @@
+import re
+
+from data_factory_testing_framework.functions.function_argument import FunctionArgument
+from data_factory_testing_framework.functions.function_call import FunctionCall
+
+extract_func_regex = r'^@?{?([^()]+?)\((.*)\)}?$'
+
+
+def parse_expression(expression: str):
+    match = re.match(extract_func_regex, expression, re.DOTALL)
+    if not match:
+        return FunctionArgument(expression)
+
+    function_name = match.group(1).strip()
+    if function_name in ["variables", "activity", "pipeline", "item"]:
+        return FunctionArgument(expression)
+
+    function_arguments_expression = match.group(2)
+
+    start = 0
+    in_quotes = False
+    in_parenthesis = 0
+    arguments = []
+    for i in range(len(function_arguments_expression)):
+        current_char = function_arguments_expression[i]
+        next_char = function_arguments_expression[i + 1] if i < len(function_arguments_expression) - 1 else '\0'
+        if current_char == ',' and not in_quotes and in_parenthesis == 0:
+            arguments.append(function_arguments_expression[start:i].replace("''", "'"))
+            start = i + 1
+            continue
+
+        # Skip escaped single quotes
+        if in_quotes and current_char == "'" and next_char == "'":
+            i += 1
+            continue
+
+        if current_char == "'":
+            in_quotes = not in_quotes
+
+        if current_char == '(':
+            in_parenthesis += 1
+
+        if current_char == ')':
+            in_parenthesis -= 1
+
+        if i == len(function_arguments_expression) - 1:
+            arguments.append(function_arguments_expression[start:i + 1].replace("''", "'"))
+
+    return FunctionCall(
+        function_name,
+        list(map(parse_expression, [x.strip() for x in arguments]))
+    )

src/python/tests/functions/test_function_parser.py

@@ -0,0 +1,49 @@
+import pytest
+
+from data_factory_testing_framework.functions.function_call import FunctionCall
+from data_factory_testing_framework.functions.function_parser import parse_expression
+from data_factory_testing_framework.models.state.pipeline_run_state import PipelineRunState
+
+
+def test_parse_expression_with_nested_function_and_single_quote():
+    # Arrange
+    state = PipelineRunState()
+    raw_expression = "concat('https://example.com/jobs/', '123''', concat('&', 'abc,'))"
+
+    # Act
+    expression = parse_expression(raw_expression)
+
+    # Assert
+    function = expression
+    assert isinstance(expression, FunctionCall)
+    assert function is not None
+    assert function.name == "concat"
+    assert len(function.arguments) == 3
+    assert function.arguments[0].expression == "'https://example.com/jobs/'"
+    assert function.arguments[1].expression == "'123''"
+
+    inner_function = function.arguments[2]
+    assert isinstance(inner_function, FunctionCall)
+    assert inner_function.name == "concat"
+    assert len(inner_function.arguments) == 2
+    assert inner_function.arguments[0].expression == "'&'"
+    assert inner_function.arguments[1].expression == "'abc,'"
+
+
+def test_parse_expression_with_adf_native_functions():
+    # Arrange
+    state = PipelineRunState()
+    raw_expression = "concat('https://example.com/jobs/', '123''', variables('abc'), pipeline().parameters.abc, activity('abc').output.abc)"
+
+    # Act
+    expression = parse_expression(raw_expression)
+
+    # Assert
+    function = expression
+    assert function.name == "concat"
+    assert len(function.arguments) == 5
+    assert function.arguments[0].expression == "'https://example.com/jobs/'"
+    assert function.arguments[1].expression == "'123''"
+    assert function.arguments[2].expression == "variables('abc')"
+    assert function.arguments[3].expression == "pipeline().parameters.abc"
+    assert function.arguments[4].expression == "activity('abc').output.abc"