test(integration): report uncollected shard files as collection errors instead of crashing xdist (#47075)

The integration conftest raised UsageError from pytest_collection_modifyitems when a
shard-selected file collected nothing. Under xdist that ran on the workers after the
unfiltered item list was already published, so the worker ended its session with tests
still queued and the controller hit xdist's crashitem assertion, naming an unrelated
queued test and hiding the real ImportError.

A selected file that collects nothing now produces a failed collection report instead.
Files whose own collection already failed keep pytest's real error. Serial runs stop with
"Interrupted: N errors during collection", xdist runs report the error and exit 1, and
both still fail the shard.

tests/integration/conftest.py

@@ -38,6 +38,7 @@ REPORTS: Final = pytest.StashKey[list[pytest.TestReport]]()
 WORKER_INVENTORY: Final = TypeAdapter(tuple[str, ...])
 SHARD_PATH: Final[TypeAdapter[Path | None]] = TypeAdapter(Path | None)
 SHARD_FILES: Final = pytest.StashKey[frozenset[str] | None]()
+FAILED_COLLECTIONS: Final = pytest.StashKey[frozenset[str]]()
 GROUP_SUFFIX: Final = re.compile(r"@integration_fixture_[0-9a-f]{64}$")
 INTEGER_OPTION: Final = TypeAdapter(int)
 STRING_ENVIRONMENT: Final = TypeAdapter(dict[str, str])
@@ -104,6 +105,12 @@ class IntegrationReportPlugin:
     def pytest_runtest_logreport(self, report: pytest.TestReport) -> None:
         self.config.stash[REPORTS].append(report)
 
+    def pytest_collectreport(self, report: pytest.CollectReport) -> None:
+        if report.failed:
+            self.config.stash[FAILED_COLLECTIONS] = self.config.stash.get(FAILED_COLLECTIONS, frozenset()) | {
+                report.nodeid.split("::", 1)[0]
+            }
+
     @pytest.hookimpl(optionalhook=True)
     def pytest_xdist_node_collection_finished(self, node: object, ids: Sequence[str]) -> None:
         self.config.stash[COLLECTED] = tuple(_case_identity(nodeid) for nodeid in ids if _owned(nodeid))
@@ -224,8 +231,12 @@ def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item
         raise pytest.UsageError("Parallel integration runs require CircleCI's shard file selection")
     if files is not None:
         available: Final = frozenset(item.nodeid.split("::", 1)[0] for item in owned)
-        if files - available:
-            raise pytest.UsageError(f"Integration shard selected uncollected files: {sorted(files - available)}")
+        for file in sorted(files - available - config.stash.get(FAILED_COLLECTIONS, frozenset())):
+            config.hook.pytest_collectreport(
+                report=pytest.CollectReport(
+                    file, "failed", f"Integration shard selected {file}, which has no tests", []
+                )
+            )
         selected: Final = tuple(item for item in items if item not in owned or item.nodeid.split("::", 1)[0] in files)
         deselected: Final = tuple(item for item in owned if item.nodeid.split("::", 1)[0] not in files)
         config.hook.pytest_deselected(items=deselected)

tests/integration/database/test_harness_lifecycle.py

@@ -784,9 +784,16 @@ def test_worker_crash_keeps_the_original_junit_identity_and_an_incomplete_execut
         assert cases[0].find("error") is not None
 
 
-@pytest.mark.parametrize("assignment", ("absent", "missing", "foreign"))
+@pytest.mark.parametrize(
+    ("assignment", "exit_code"),
+    (
+        ("absent", pytest.ExitCode.USAGE_ERROR),
+        ("missing", pytest.ExitCode.USAGE_ERROR),
+        ("foreign", pytest.ExitCode.INTERRUPTED),
+    ),
+)
 def test_invalid_native_selections_fail_before_any_case_runs(
-    tmp_path: Path, assignment: Literal["absent", "missing", "foreign"]
+    tmp_path: Path, assignment: Literal["absent", "missing", "foreign"], exit_code: pytest.ExitCode
 ) -> None:
     root: Final = Path(__file__).resolve().parents[3]
     events: Final = tmp_path / "executed.txt"
@@ -823,7 +830,7 @@ def test_invalid_native_selections_fail_before_any_case_runs(
             timeout=15,
             check=False,
         )
-        assert result.returncode == pytest.ExitCode.USAGE_ERROR, result.stdout + result.stderr
+        assert result.returncode == exit_code, result.stdout + result.stderr
         assert not events.exists()
 
 

tests/unit/integration_support/test_sharding.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import os
+import subprocess
+import sys
 from collections.abc import Mapping
 from dataclasses import dataclass
 from pathlib import Path
@@ -13,6 +16,21 @@ from tests.integration.conftest import COLLECTED, IntegrationReportPlugin, _shar
 from tests.integration.conftest import INVENTORY as WORKER_INVENTORY
 
 INVENTORY: Final = ("test_owned.py::test_first", "test_owned.py::test_skipped", "test_owned.py::test_last")
+INTEGRATION_PACKAGE: Final = Path(__file__).resolve().parents[2] / "integration"
+HEALTHY: Final = "tests/integration/providers/test_healthy.py"
+BROKEN: Final = "tests/integration/providers/test_broken.py"
+RENAMED: Final = "tests/integration/providers/test_renamed.py"
+CLASS_ERROR: Final = "tests/integration/providers/test_class_error.py"
+MODULES: Final = MappingProxyType(
+    {
+        HEALTHY: "def test_one():\n    pass\n\n\ndef test_two():\n    pass\n",
+        BROKEN: "from tests.integration._support.process import _removed_helper\n\n\ndef test_never():\n    pass\n",
+        CLASS_ERROR: (
+            "import pytest\n\n\nclass TestBroken:\n"
+            "    @pytest.mark.parametrize('unused', [1])\n    def test_case(self):\n        pass\n"
+        ),
+    }
+)
 
 
 def _execution(
@@ -111,3 +129,88 @@ def test_worker_results_retain_the_full_inventory_and_reject_disagreement() -> N
     incomplete: Final = WorkerResult(MappingProxyType({"integration_inventory": INVENTORY[:1]}))
     with pytest.raises(pytest.UsageError, match="inventories disagree"):
         plugin.pytest_testnodedown(incomplete, None)
+
+
+XDIST: Final = ("-n", "2", "--dist=loadgroup")
+SERIAL: Final = ("-p", "no:xdist")
+
+
+def _run_shard(
+    root: Path, selected: tuple[str, ...], distribution: tuple[str, ...] = XDIST
+) -> subprocess.CompletedProcess[str]:
+    package: Final = root / "tests" / "integration"
+    (package / "providers").mkdir(parents=True)
+    (root / "pytest.ini").write_text("[pytest]\n")
+    (root / "tests" / "__init__.py").write_text("")
+    for name in ("__init__.py", "conftest.py", "run.py", "_support"):
+        (package / name).symlink_to(INTEGRATION_PACKAGE / name)
+    for file in selected:
+        if file in MODULES:
+            (root / file).write_text(MODULES[file])
+    (root / "shard.txt").write_text("".join(f"{file}\n" for file in selected))
+    return subprocess.run(
+        (
+            sys.executable,
+            "-P",
+            "-m",
+            "pytest",
+            "tests/integration/providers",
+            "-p",
+            "no:cacheprovider",
+            *distribution,
+            "--integration-shard-count=2",
+            "--integration-shard-index=0",
+            "--integration-shard-files",
+            "shard.txt",
+        ),
+        cwd=root,
+        env={
+            "PATH": os.environ.get("PATH", ""),
+            "HOME": str(root),
+            "PYTHONPATH": os.pathsep.join((str(root), str(root / "tests"))),
+        },
+        capture_output=True,
+        text=True,
+        timeout=180,
+        check=False,
+    )
+
+
+@pytest.mark.parametrize(
+    ("distribution", "exit_code", "outcome"),
+    (
+        (XDIST, pytest.ExitCode.TESTS_FAILED, "2 passed, 1 error"),
+        (SERIAL, pytest.ExitCode.INTERRUPTED, "Interrupted: 1 error during collection"),
+    ),
+    ids=("xdist", "serial"),
+)
+def test_a_broken_import_in_a_shard_is_a_collection_error_not_a_worker_crash(
+    tmp_path: Path, distribution: tuple[str, ...], exit_code: pytest.ExitCode, outcome: str
+) -> None:
+    result: Final = _run_shard(tmp_path, (HEALTHY, BROKEN), distribution)
+    assert result.returncode == exit_code, result.stdout + result.stderr
+    assert outcome in result.stdout
+    assert result.stdout.count(f"ERROR collecting {BROKEN}") == 1
+    assert "cannot import name '_removed_helper'" in result.stdout
+    assert "INTERNALERROR" not in result.stdout + result.stderr
+
+
+def test_a_shard_file_that_collects_no_tests_still_fails_as_a_collection_error(tmp_path: Path) -> None:
+    result: Final = _run_shard(tmp_path, (HEALTHY, RENAMED))
+    assert result.returncode == pytest.ExitCode.TESTS_FAILED, result.stdout + result.stderr
+    assert result.stdout.count(f"ERROR collecting {RENAMED}") == 1
+    assert "2 passed" in result.stdout
+    assert "INTERNALERROR" not in result.stdout + result.stderr
+
+
+def test_a_class_level_collection_error_is_not_also_reported_as_a_file_without_tests(tmp_path: Path) -> None:
+    result: Final = _run_shard(tmp_path, (HEALTHY, CLASS_ERROR), SERIAL)
+    assert result.returncode == pytest.ExitCode.INTERRUPTED, result.stdout + result.stderr
+    assert f"ERROR {CLASS_ERROR}::TestBroken" in result.stdout
+    assert "which has no tests" not in result.stdout
+
+
+def test_a_healthy_shard_runs_every_selected_test(tmp_path: Path) -> None:
+    result: Final = _run_shard(tmp_path, (HEALTHY,))
+    assert result.returncode == pytest.ExitCode.OK, result.stdout + result.stderr
+    assert "2 passed" in result.stdout