diff --git a/src/flowx/models/adf_ast.py b/src/flowx/models/adf_ast.py index a91ef30..da34823 100644 --- a/src/flowx/models/adf_ast.py +++ b/src/flowx/models/adf_ast.py @@ -136,6 +136,11 @@ class AdfActivity: if_true_activities: Activities to run when an IfCondition evaluates to true. if_false_activities: Activities to run when an IfCondition evaluates to false. activities: Child activities for ForEach / Until containers. + switch_cases: Per-case child activities for a Switch, keyed by case value + (``cases[].value`` -> that case's ``activities``). Preserves case + grouping; use :func:`switch_child_activities` for a flat view. + switch_default_activities: A Switch's ``defaultActivities`` (the branch + taken when no case matches). """ name: str @@ -149,9 +154,30 @@ class AdfActivity: if_true_activities: list[AdfActivity] | None = None if_false_activities: list[AdfActivity] | None = None activities: list[AdfActivity] | None = None # ForEach, Until + switch_cases: dict[str, list[AdfActivity]] | None = None # Switch cases[].value -> activities + switch_default_activities: list[AdfActivity] | None = None # Switch defaultActivities # Original ADF/ARM activity JSON, retained so agentic handlers can translate from the source. raw: dict[str, Any] | None = None + def switch_child_activities(self) -> list[AdfActivity]: + """Gets a flat list of all child activities of a Switch activity. + Includes every case's activities and defaults. + + Note: + Empty for non-Switch activities. Used by inventory/lineage walkers that + only need to descend into children, not preserve case grouping. + + Returns: + A flat list of all child activities. + """ + children: list[AdfActivity] = [] + if self.switch_cases: + for case_activities in self.switch_cases.values(): + children.extend(case_activities) + if self.switch_default_activities: + children.extend(self.switch_default_activities) + return children + # --------------------------------------------------------------------------- # Pipeline node diff --git a/src/flowx/models/ir.py b/src/flowx/models/ir.py index c193356..600e8c7 100644 --- a/src/flowx/models/ir.py +++ b/src/flowx/models/ir.py @@ -621,6 +621,9 @@ class Pipeline: audit: Source-audit counts and transformation ledger. bundle_variables: DAB bundle-variable declarations (name -> ``{"description", "default"}``) for factory globals hoisted under the ``bundle_variable`` resolution policy. + reconciliation_status: Source-audit result for this pipeline. + migration_status: Whether the pipeline is included or explicitly excluded. + audit: Source-audit counts and transformation ledger. """ name: str diff --git a/src/flowx/sources/adf/loader.py b/src/flowx/sources/adf/loader.py index af9917a..5fcc5d6 100644 --- a/src/flowx/sources/adf/loader.py +++ b/src/flowx/sources/adf/loader.py @@ -386,6 +386,8 @@ def parse_activity(data: dict[str, Any]) -> AdfActivity: if_true_activities: list[AdfActivity] | None = None if_false_activities: list[AdfActivity] | None = None child_activities: list[AdfActivity] | None = None + switch_cases: dict[str, list[AdfActivity]] | None = None + switch_default_activities: list[AdfActivity] | None = None if type_properties: raw_if_true = type_properties.get("ifTrueActivities") @@ -397,6 +399,22 @@ def parse_activity(data: dict[str, Any]) -> AdfActivity: raw_children = type_properties.get("activities") if raw_children: child_activities = [parse_activity(raw_activity) for raw_activity in raw_children] + # Switch: cases[].activities (keyed by case value) + defaultActivities. Without + # this, a Switch's nested activities never enter the AST, so inventory counts and + # lineage edges silently drop everything inside a Switch case. + raw_cases = type_properties.get("cases") + if raw_cases: + parsed_cases: dict[str, list[AdfActivity]] = {} + for raw_case in raw_cases: + case_value = str(raw_case.get("value", "")) + case_activities = [parse_activity(raw_activity) for raw_activity in raw_case.get("activities", [])] + if case_activities: + parsed_cases[case_value] = case_activities + if parsed_cases: + switch_cases = parsed_cases + raw_default = type_properties.get("defaultActivities") + if raw_default: + switch_default_activities = [parse_activity(raw_activity) for raw_activity in raw_default] return AdfActivity( name=name, @@ -410,6 +428,8 @@ def parse_activity(data: dict[str, Any]) -> AdfActivity: if_true_activities=if_true_activities, if_false_activities=if_false_activities, activities=child_activities, + switch_cases=switch_cases, + switch_default_activities=switch_default_activities, raw=data, ) @@ -764,6 +784,9 @@ def _classify_activities( _classify_activities(pipeline_name, activity.if_false_activities, items) if activity.activities: _classify_activities(pipeline_name, activity.activities, items) + switch_children = activity.switch_child_activities() + if switch_children: + _classify_activities(pipeline_name, switch_children, items) # --------------------------------------------------------------------------- @@ -819,7 +842,12 @@ def _walk_activities(activities: list[AdfActivity]): """Yields every activity in *activities*, descending into container children.""" for activity in activities: yield activity - for child in (activity.if_true_activities, activity.if_false_activities, activity.activities): + for child in ( + activity.if_true_activities, + activity.if_false_activities, + activity.activities, + activity.switch_child_activities(), + ): if child: yield from _walk_activities(child) diff --git a/tests/unit/test_adf_loader.py b/tests/unit/test_adf_loader.py index 769b1fc..e673051 100644 --- a/tests/unit/test_adf_loader.py +++ b/tests/unit/test_adf_loader.py @@ -262,6 +262,97 @@ def test_parse_pipeline_annotations_and_folder(self): assert pipeline.folder == "ETL/Ingestion" +# --------------------------------------------------------------------------- +# Switch case / default nesting (regression: activities inside a Switch were +# invisible to the parser, so inventory counts and lineage walks dropped them) +# --------------------------------------------------------------------------- + + +def _switch_pipeline_json() -> dict: + """A pipeline whose Switch nests activities in cases and the default branch, + including an ExecutePipeline (a lineage edge) hidden inside a case.""" + return { + "name": "switch_pipeline", + "properties": { + "activities": [ + {"name": "SetVar", "type": "SetVariable", "typeProperties": {}}, + { + "name": "RouteByEnv", + "type": "Switch", + "typeProperties": { + "on": {"value": "@variables('env')", "type": "Expression"}, + "cases": [ + { + "value": "prod", + "activities": [ + {"name": "ProdNotebook", "type": "DatabricksNotebook", "typeProperties": {}}, + { + "name": "RunProdChild", + "type": "ExecutePipeline", + "typeProperties": {"pipeline": {"referenceName": "child_prod"}}, + }, + ], + }, + { + "value": "dev", + "activities": [ + {"name": "DevNotebook", "type": "DatabricksNotebook", "typeProperties": {}}, + ], + }, + ], + "defaultActivities": [ + {"name": "UnknownEnvNotebook", "type": "DatabricksNotebook", "typeProperties": {}}, + ], + }, + }, + ], + }, + } + + +class TestSwitchNesting: + def test_parse_populates_switch_cases_and_default(self): + """parse_activity captures cases[].activities (keyed by value) + defaultActivities.""" + pipeline = _parse_pipeline_json(_switch_pipeline_json()) + switch = next(a for a in pipeline.activities if a.type == "Switch") + assert switch.switch_cases is not None + assert set(switch.switch_cases) == {"prod", "dev"} + assert [a.name for a in switch.switch_cases["prod"]] == ["ProdNotebook", "RunProdChild"] + assert [a.name for a in switch.switch_cases["dev"]] == ["DevNotebook"] + assert switch.switch_default_activities is not None + assert [a.name for a in switch.switch_default_activities] == ["UnknownEnvNotebook"] + + def test_switch_child_activities_flattens_all_branches(self): + """The helper returns every case activity plus the defaults, flat.""" + pipeline = _parse_pipeline_json(_switch_pipeline_json()) + switch = next(a for a in pipeline.activities if a.type == "Switch") + names = {a.name for a in switch.switch_child_activities()} + assert names == {"ProdNotebook", "RunProdChild", "DevNotebook", "UnknownEnvNotebook"} + # A non-Switch activity yields an empty list, never raises. + setvar = next(a for a in pipeline.activities if a.type == "SetVariable") + assert setvar.switch_child_activities() == [] + + def test_inventory_counts_switch_nested_activities(self): + """build_inventory descends into Switch cases: the 4 nested activities are + counted, not just the 2 top-level ones (SetVar + Switch).""" + defs = AdfDefinitions(pipelines=[_parse_pipeline_json(_switch_pipeline_json())]) + inv = build_inventory(defs) + names = [i.activity_name for i in inv.items if i.pipeline_name == "switch_pipeline"] + # 2 top-level + 3 case activities + 1 default = 6 + assert len(names) == 6 + for expected in ("SetVar", "RouteByEnv", "ProdNotebook", "RunProdChild", "DevNotebook", "UnknownEnvNotebook"): + assert expected in names + + def test_lineage_walk_reaches_executepipeline_inside_switch(self): + """The ExecutePipeline nested in a Switch case is reachable by the activity + walker -- i.e. its control edge is no longer dropped.""" + from flowx.sources.adf.loader import _walk_activities + + pipeline = _parse_pipeline_json(_switch_pipeline_json()) + walked = {a.name for a in _walk_activities(pipeline.activities)} + assert "RunProdChild" in walked + + # --------------------------------------------------------------------------- # ARM template normalization # ---------------------------------------------------------------------------