Skip to content

Commit

Permalink
fix: execute_pipeline_activity.py defaults back to empty parameters i…
Browse files Browse the repository at this point in the history
…f none were supplied
  • Loading branch information
arjendev committed Feb 5, 2024
1 parent 7f6fae5 commit f8cc4b9
Show file tree
Hide file tree
Showing 2 changed files with 50 additions and 1 deletion.
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,9 @@ def __init__(self, **kwargs: Any) -> None: # noqa: ANN401

super(ControlActivity, self).__init__(**kwargs)

self.parameters: dict = self.type_properties["parameters"]
self.parameters: dict = {}
if "parameters" in self.type_properties:
self.parameters = self.type_properties["parameters"]

def get_child_run_parameters(self, state: PipelineRunState) -> List[RunParameter]:
child_parameters = []
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
from data_factory_testing_framework.models.activities.execute_pipeline_activity import ExecutePipelineActivity
from data_factory_testing_framework.models.data_factory_element import DataFactoryElement
from data_factory_testing_framework.state import PipelineRunState, RunParameter, RunParameterType


def test_execute_pipeline_activity_evaluates_parameters() -> None:
# Arrange
execute_pipeline_activity = ExecutePipelineActivity(
name="ExecutePipelineActivity",
typeProperties={
"parameters": {
"url": DataFactoryElement("@pipeline().parameters.param1"),
},
},
depends_on=[],
)
state = PipelineRunState(
parameters=[
RunParameter(name="param1", value="value1", parameter_type=RunParameterType.Pipeline),
],
)

# Act
activity = execute_pipeline_activity.evaluate(state)

# Assert
assert activity is not None
assert activity.name == "ExecutePipelineActivity"
assert activity.parameters["url"].value == "value1"


def test_execute_pipeline_activity_evaluates_no_parameters() -> None:
# Arrange
execute_pipeline_activity = ExecutePipelineActivity(
name="ExecutePipelineActivity",
typeProperties={},
depends_on=[],
)
state = PipelineRunState()

# Act
activity = execute_pipeline_activity.evaluate(state)

# Assert
assert activity is not None
assert activity.name == "ExecutePipelineActivity"
assert activity.parameters == {}

0 comments on commit f8cc4b9

Please sign in to comment.