test(integration): let run.py select cells by pytest node id (#44319)

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>

tests/integration/README.md

@@ -14,7 +14,7 @@ Reuse the existing canned provider handlers through `_support/upstream.py`. It r
 
 The CircleCI workflow starts its own database and Redis, restricts test-phase egress to its owned services and writes JUnit plus an executed-node manifest. Missing setup, failed cleanup or a selected test with neither a passed call nor a skip fail qualification. Skipped nodes are listed under `skipped` in `execution.json`, so the skip reasons double as the open bug list. Existing GitHub Actions jobs do not own these tests
 
-There is no per-node manifest. The runner fails only when pytest fails, when collection errors, or when a selected file collects zero tests. Older tests still carry `@pytest.mark.covers(...)` decorators; the marker stays registered so they collect, but the IDs are not checked against anything and new tests should not use it. The GitHub Actions coverage census reads the `GROUPS` literal in `run.py` and treats every `tests/integration/<directory>/test_*.py` file in a scheduled group as owned by CircleCI
+There is no per-node manifest. A positional argument is a file of the group or a pytest node id inside one (`path::test[param]`), so one cell of a parametrized file can run alone. The runner fails only when pytest fails, when collection errors, or when a selected file collects zero tests. Older tests still carry `@pytest.mark.covers(...)` decorators; the marker stays registered so they collect, but the IDs are not checked against anything and new tests should not use it. The GitHub Actions coverage census reads the `GROUPS` literal in `run.py` and treats every `tests/integration/<directory>/test_*.py` file in a scheduled group as owned by CircleCI
 
 Provider sentinels currently use the controlled server, not live recordings. The provider shard also runs the existing strict replay controls for changed requests, exhausted interactions, leftover interactions and no provider connection. Future recorded scenarios must use that replay-only implementation; missing recordings cannot fall back to a real provider. The observation endpoint is destructive and the current selection runs serially against one owned upstream
 

tests/integration/run.py

@@ -5,6 +5,7 @@ import json
 import os
 import subprocess
 import sys
+from dataclasses import dataclass
 from pathlib import Path
 from types import MappingProxyType
 from typing import Final
@@ -24,6 +25,29 @@ GROUPS: Final = MappingProxyType(
 )
 
 
+@dataclass(frozen=True, slots=True)
+class Selection:
+    nodes: tuple[str, ...]
+    foreign: tuple[str, ...]
+
+
+def file_of(node: str) -> str:
+    return node.split("::", 1)[0]
+
+
+def select(requested: tuple[str, ...], group_files: tuple[str, ...]) -> Selection:
+    members: Final = frozenset(group_files)
+    return Selection(
+        nodes=requested or group_files,
+        foreign=tuple(sorted({node for node in requested if file_of(node) not in members})),
+    )
+
+
+def uncollected(nodes: tuple[str, ...], collected: frozenset[str]) -> tuple[str, ...]:
+    collected_files: Final = frozenset(file_of(node) for node in collected)
+    return tuple(node for node in nodes if file_of(node) not in collected_files)
+
+
 def main() -> int:
     parser: Final = argparse.ArgumentParser()
     parser.add_argument("group", choices=tuple(GROUPS))
@@ -32,7 +56,7 @@ def main() -> int:
     parser.add_argument("--order-seed", type=int, default=int(os.environ.get("INTEGRATION_ORDER_SEED", "0")))
     parser.add_argument("--workers", type=int, default=int(os.environ.get("INTEGRATION_WORKERS", "1")))
     parser.add_argument("--list", action="store_true", help="print the group's test files and exit")
-    parser.add_argument("files", nargs="*", help="run only these files of the group")
+    parser.add_argument("files", nargs="*", help="run only these files, or pytest node ids inside them, of the group")
     options: Final = parser.parse_intermixed_args()
     root: Final = Path(__file__).resolve().parents[2]
     group_files: Final = tuple(
@@ -43,11 +67,10 @@ def main() -> int:
     if options.list:
         print("\n".join(group_files))
         return 0
-    foreign: Final = sorted(set(options.files) - set(group_files))
-    if foreign:
-        parser.error(f"Not in the {options.group} group: {', '.join(foreign)}")
-    selected: Final = tuple(options.files) or group_files
-    if not selected:
+    selection: Final = select(tuple(options.files), group_files)
+    if selection.foreign:
+        parser.error(f"Not in the {options.group} group: {', '.join(selection.foreign)}")
+    if not selection.nodes:
         parser.error(f"No integration test files selected for {options.group}")
     output: Final = options.results.resolve()
     output.mkdir(parents=True, exist_ok=True)
@@ -62,7 +85,7 @@ def main() -> int:
             sys.executable,
             "-m",
             "pytest",
-            *selected,
+            *selection.nodes,
             "-vv",
             "-rs",
             "--strict-markers",
@@ -86,8 +109,7 @@ def main() -> int:
     if result != 0:
         return result
     evidence: Final = json.loads((output / "execution.json").read_text())
-    collected_files: Final = {node.split("::", 1)[0] for node in evidence["collected"]}
-    empty: Final = tuple(path for path in selected if path not in collected_files)
+    empty: Final = uncollected(selection.nodes, frozenset(evidence["collected"]))
     if empty:
         sys.stderr.write(f"Selected integration files collected zero tests: {', '.join(empty)}\n")
         return 1

tests/unit/test_integration_run.py

@@ -0,0 +1,36 @@
+from typing import Final
+
+from tests.integration.run import select, uncollected
+
+_GROUP: Final = (
+    "tests/integration/cost_calculation/test_cost_tracking.py",
+    "tests/integration/cost_calculation/test_rollups.py",
+)
+_CELL: Final = (
+    "tests/integration/cost_calculation/test_cost_tracking.py"
+    "::test_case_bills_expected_cost[perplexity/pplx-decider-v1-27b-decisions]"
+)
+
+
+def test_a_node_id_inside_a_group_file_is_selected_as_written() -> None:
+    selection: Final = select((_CELL,), _GROUP)
+    assert selection.nodes == (_CELL,)
+    assert selection.foreign == ()
+
+
+def test_a_node_id_outside_the_group_is_foreign_by_its_file() -> None:
+    foreign: Final = "tests/integration/providers/test_decisions_wire.py::test_key_checks_match_chat"
+    assert select((foreign, _CELL), _GROUP).foreign == (foreign,)
+
+
+def test_no_request_selects_every_group_file() -> None:
+    assert select((), _GROUP).nodes == _GROUP
+
+
+def test_a_node_id_whose_file_collected_tests_is_not_empty() -> None:
+    collected: Final = frozenset({_CELL, "tests/integration/cost_calculation/test_cost_tracking.py::test_other"})
+    assert uncollected((_CELL,), collected) == ()
+
+
+def test_a_selected_file_that_collected_nothing_is_reported() -> None:
+    assert uncollected(_GROUP, frozenset({_CELL})) == ("tests/integration/cost_calculation/test_rollups.py",)