Parse Switch case/default activities (fixes undercounted inventory & missing lineage edges)

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,25 @@ 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]:
+        """Flat list of all Switch children (every case's activities + defaults).
+
+        Empty for non-Switch activities. Used by inventory/lineage walkers that
+        only need to descend into children, not preserve case grouping.
+        """
+        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

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)
 

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.parser.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
 # ---------------------------------------------------------------------------