test: split card-dag tests out of test_graph_inference

Bundle the three card-related tests into test_card_dag.py:
- test_transform_flow_graph_supports_explicit_endpoints
- test_transform_flow_graph_keeps_legacy_start_end_detection
- test_default_card_includes_custom_graph_endpoints

All three target the cards' graph data layer (transform_flow_graph
and the DAG component render path). Keeping them next to the
graph-execution tests in test_graph_inference.py blurred file
boundaries and meant test_graph_inference.py had to import
DefaultCardJSON, transform_flow_graph, and a four-line importlib
helper just for the one render test.

To support the render test cleanly, hoist the FlowSpec class loader
into conftest.py as _load_flow_class plus a custom_named_card_flow
fixture. DefaultCardJSON.render() does getattr(self.flow, step_name)
and inspect.getsource(...) to build the Task Code panel and crashes
if flow is None, so the test still needs the actual class object;
moving the loader into conftest gets the importlib dance out of the
test files.

No behavior change.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: talsperre

test/unit/graph_inference/conftest.py

@@ -1,10 +1,28 @@
 import os
+from importlib.util import module_from_spec, spec_from_file_location
+
 import pytest
 from metaflow import Runner, Flow
 
 FLOWS_DIR = os.path.join(os.path.dirname(__file__), "flows")
 
 
+def _load_flow_class(flow_file, flow_class_name):
+    """Import a FlowSpec subclass from a file in FLOWS_DIR.
+
+    Loads the module by file path so FLOWS_DIR is not added to sys.path.
+    Used by fixtures that need to pass `flow=` to a Card constructor;
+    `DefaultCardJSON` does `getattr(self.flow, step_name)` and
+    `inspect.getsource(...)` to build the Task Code panel and crashes if
+    `flow` is None.
+    """
+    spec = spec_from_file_location(flow_class_name, os.path.join(FLOWS_DIR, flow_file))
+    module = module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return getattr(module, flow_class_name)
+
+
 def create_flow_fixture(flow_name, flow_file):
     """Factory function to create flow fixtures."""
 
@@ -60,3 +78,11 @@ def flow_fixture(request):
         "single_step_bare_flow.py",
     )
 )
+
+
+@pytest.fixture(scope="session")
+def custom_named_card_flow():
+    """Instance of CustomNamedCardFlow for tests that pass flow= to a Card."""
+    return _load_flow_class("custom_named_card_flow.py", "CustomNamedCardFlow")(
+        use_cli=False
+    )

test/unit/graph_inference/test_card_dag.py

@@ -0,0 +1,95 @@
+"""
+Tests for the cards' DAG-related graph data layer.
+
+Covers:
+- `transform_flow_graph` - the function that normalizes the legacy
+  flat-step-dict and the new `{steps, start_step, end_step}` shapes
+  into the structure the DAG card component renders against.
+- The end-to-end render path: a Run with custom-named entry/terminal
+  steps produces a DAG card whose `start_step` / `end_step` /
+  `steps` reflect the user's actual step names rather than the
+  legacy hardcoded "start" / "end".
+"""
+
+import json
+
+from metaflow.plugins.cards.card_modules.basic import (
+    DefaultCardJSON,
+    transform_flow_graph,
+)
+
+
+def _find_components_by_type(node, component_type):
+    if isinstance(node, dict):
+        if node.get("type") == component_type:
+            yield node
+        for value in node.values():
+            yield from _find_components_by_type(value, component_type)
+    elif isinstance(node, list):
+        for item in node:
+            yield from _find_components_by_type(item, component_type)
+
+
+# ---------------------------------------------------------------------------
+# transform_flow_graph: shape-detection unit tests
+# ---------------------------------------------------------------------------
+
+
+def test_transform_flow_graph_supports_explicit_endpoints():
+    graph = {
+        "start_step": "begin",
+        "end_step": "finish",
+        "steps": {
+            "begin": {"type": "start", "next": ["middle"], "doc": "begin"},
+            "middle": {"type": "linear", "next": ["finish"], "doc": "middle"},
+            "finish": {"type": "end", "next": [], "doc": "finish"},
+        },
+    }
+
+    transformed = transform_flow_graph(graph)
+
+    assert transformed["start_step"] == "begin"
+    assert transformed["end_step"] == "finish"
+    assert set(transformed["steps"]) == {"begin", "middle", "finish"}
+    assert transformed["steps"]["begin"]["type"] == "start"
+    assert transformed["steps"]["middle"]["box_next"] is False
+    assert transformed["steps"]["finish"]["type"] == "end"
+
+
+def test_transform_flow_graph_keeps_legacy_start_end_detection():
+    graph = {
+        "start": {"type": "start", "next": ["end"], "doc": ""},
+        "end": {"type": "end", "next": [], "doc": ""},
+    }
+
+    transformed = transform_flow_graph(graph)
+
+    assert transformed["start_step"] == "start"
+    assert transformed["end_step"] == "end"
+    assert set(transformed["steps"]) == {"start", "end"}
+
+
+# ---------------------------------------------------------------------------
+# DefaultCardJSON: end-to-end render with custom-named endpoints
+# ---------------------------------------------------------------------------
+
+
+def test_default_card_includes_custom_graph_endpoints(
+    custom_named_card_run, custom_named_card_flow
+):
+    graph = custom_named_card_run["_parameters"].task["_graph_info"].data
+    card_data = json.loads(
+        DefaultCardJSON(graph=graph, flow=custom_named_card_flow).render(
+            custom_named_card_run["begin"].task
+        )
+    )
+
+    dag_components = list(_find_components_by_type(card_data["components"], "dag"))
+    assert len(dag_components) == 1
+
+    dag_data = dag_components[0]["data"]
+    assert dag_data["start_step"] == "begin"
+    assert dag_data["end_step"] == "finish"
+    assert set(dag_data["steps"]) == {"begin", "middle", "finish"}
+    assert "start" not in dag_data["steps"]
+    assert "end" not in dag_data["steps"]

test/unit/graph_inference/test_graph_inference.py

@@ -9,37 +9,7 @@
 - Single-step flows execute end-to-end
 """
 
-import json
-from importlib.util import module_from_spec, spec_from_file_location
-from pathlib import Path
-
-import pytest
 from metaflow.events import Trigger
-from metaflow.plugins.cards.card_modules.basic import (
-    DefaultCardJSON,
-    transform_flow_graph,
-)
-
-FLOWS_DIR = Path(__file__).resolve().parent / "flows"
-
-
-def _find_components_by_type(node, component_type):
-    if isinstance(node, dict):
-        if node.get("type") == component_type:
-            yield node
-        for value in node.values():
-            yield from _find_components_by_type(value, component_type)
-    elif isinstance(node, list):
-        for item in node:
-            yield from _find_components_by_type(item, component_type)
-
-
-def _load_flow(flow_file, flow_class_name):
-    spec = spec_from_file_location(flow_class_name, FLOWS_DIR / flow_file)
-    module = module_from_spec(spec)
-    assert spec.loader is not None
-    spec.loader.exec_module(module)
-    return getattr(module, flow_class_name)(use_cli=False)
 
 
 # ---------------------------------------------------------------------------
@@ -227,70 +197,6 @@ def test_trigger_from_runs_uses_custom_terminal_step(custom_named_run):
     assert trigger.run.pathspec == custom_named_run.pathspec
 
 
-# ---------------------------------------------------------------------------
-# Card graph transform
-# ---------------------------------------------------------------------------
-
-
-def test_transform_flow_graph_supports_explicit_endpoints():
-    graph = {
-        "start_step": "begin",
-        "end_step": "finish",
-        "steps": {
-            "begin": {"type": "start", "next": ["middle"], "doc": "begin"},
-            "middle": {"type": "linear", "next": ["finish"], "doc": "middle"},
-            "finish": {"type": "end", "next": [], "doc": "finish"},
-        },
-    }
-
-    transformed = transform_flow_graph(graph)
-
-    assert transformed["start_step"] == "begin"
-    assert transformed["end_step"] == "finish"
-    assert set(transformed["steps"]) == {"begin", "middle", "finish"}
-    assert transformed["steps"]["begin"]["type"] == "start"
-    assert transformed["steps"]["middle"]["box_next"] is False
-    assert transformed["steps"]["finish"]["type"] == "end"
-
-
-def test_transform_flow_graph_keeps_legacy_start_end_detection():
-    graph = {
-        "start": {"type": "start", "next": ["end"], "doc": ""},
-        "end": {"type": "end", "next": [], "doc": ""},
-    }
-
-    transformed = transform_flow_graph(graph)
-
-    assert transformed["start_step"] == "start"
-    assert transformed["end_step"] == "end"
-    assert set(transformed["steps"]) == {"start", "end"}
-
-
-# ---------------------------------------------------------------------------
-# Cards integration
-# ---------------------------------------------------------------------------
-
-
-def test_default_card_includes_custom_graph_endpoints(custom_named_card_run):
-    flow = _load_flow("custom_named_card_flow.py", "CustomNamedCardFlow")
-    graph = custom_named_card_run["_parameters"].task["_graph_info"].data
-    card_data = json.loads(
-        DefaultCardJSON(graph=graph, flow=flow).render(
-            custom_named_card_run["begin"].task
-        )
-    )
-
-    dag_components = list(_find_components_by_type(card_data["components"], "dag"))
-    assert len(dag_components) == 1
-
-    dag_data = dag_components[0]["data"]
-    assert dag_data["start_step"] == "begin"
-    assert dag_data["end_step"] == "finish"
-    assert set(dag_data["steps"]) == {"begin", "middle", "finish"}
-    assert "start" not in dag_data["steps"]
-    assert "end" not in dag_data["steps"]
-
-
 # ---------------------------------------------------------------------------
 # Composition: single-step flows with Config, stacked decorators, FlowMutator
 # ---------------------------------------------------------------------------