Keep the release test split's settings out of pytest runs that tests start

.github/scripts/run_live_staging_tests.sh

@@ -6,13 +6,13 @@ set -euo pipefail
 #
 # One pytest worker per test: each test waits on its own requests to the
 # staging candidate and the simulation API, so the step takes about as long as
-# the longest test. tests/unit/test_cloud_run_push_workflow.py checks that -n
-# equals the number of tests here. The candidate is one Cloud Run instance that
-# admits 6 requests at a time (cloud_run_env.sh). Each test sends one request at
-# a time, so at most 3 requests wait at the Cloud Run front end for a free slot:
-# Cloud Run holds them for the longer of 10 seconds and 3.5 times the average
-# instance startup time, and the tests' client allows each request 90 seconds.
-# Before adding a test, check the candidate can take one more.
+# the longest test. A new test here needs -n raised with it. The candidate is
+# one Cloud Run instance that admits 6 requests at a time (cloud_run_env.sh).
+# Each test sends one request at a time, so at most 3 requests wait at the
+# Cloud Run front end for a free slot: Cloud Run holds them for the longer of
+# 10 seconds and 3.5 times the average instance startup time, and the tests'
+# client allows each request 90 seconds. Before adding a test, check the
+# candidate can take one more.
 uv run --frozen --extra dev python -m pytest \
   -n 9 \
   --dist load \

changelog.d/release-test-shards-in-subprocesses.fixed.md

@@ -0,0 +1 @@
+Release test shards no longer split, or replace the report of, a pytest run that a test starts in its own process; that made the first release run with the split test suite fail.

docs/engineering/skills/testing.md

@@ -184,7 +184,11 @@ assigns to its shard, heaviest file first, by the measured seconds in
 `MEASURED_FILE_SECONDS` or else by test count. A file's tests stay in one
 shard, so its fixtures are set up once. The `check-release-test-shards` job
 then fails, and the staging deployment with it, unless every shard collected
-the same tests and each test ran in exactly one shard. When a test file
+the same tests and each test ran in exactly one shard. The plugin reads its
+settings once, when pytest configures it, and then removes them and its `-p`
+option from `PYTEST_ADDOPTS`, so a pytest process that a test starts collects
+its own tests without the split; a shard writes its report once, and fails if
+one is already there. When a test file
 becomes much slower or faster, measure it again with `--durations=0` and
 update `MEASURED_FILE_SECONDS`; an out-of-date value only makes the shards
 finish less evenly.

scripts/release_test_shards.py

@@ -178,12 +178,26 @@ def shard_node_ids(
 def write_report(
     settings: ShardSettings, collected: Sequence[str], selected: Sequence[str]
 ) -> None:
-    """Write what this shard collected and kept to its report directory."""
+    """Write what this shard collected and kept to its report directory.
+
+    Raise FileExistsError if the directory already has a report: a second
+    pytest run in the same shard would otherwise replace the first one's.
+    """
     settings.report_directory.mkdir(parents=True, exist_ok=True)
-    (settings.report_directory / SHARD_FILE).write_text(
-        json.dumps({"index": settings.index, "count": settings.count}) + "\n",
-        encoding="utf-8",
-    )
+    try:
+        with (settings.report_directory / SHARD_FILE).open(
+            "x", encoding="utf-8"
+        ) as shard_file:
+            shard_file.write(
+                json.dumps({"index": settings.index, "count": settings.count}) + "\n"
+            )
+    except FileExistsError as error:
+        msg = (
+            f"{settings.report_directory} already has a shard report; another "
+            "pytest run in this shard wrote it. Remove the directory to run the "
+            "shard again."
+        )
+        raise FileExistsError(msg) from error
     for file_name, node_ids in (
         (COLLECTED_FILE, collected),
         (SELECTED_FILE, selected),

tests/release_test_shard_plugin.py

@@ -7,26 +7,49 @@
 ``RELEASE_TEST_SHARD_COUNT``), reports the rest as deselected, and writes what
 it collected and kept to ``RELEASE_TEST_SHARD_REPORT_DIRECTORY`` for the
 check-release-test-shards job.
+
+It reads those settings once, when pytest configures it, and then removes them
+and its own ``-p`` option from ``PYTEST_ADDOPTS`` in the process environment.
+A test that runs pytest itself would otherwise load the plugin again in that
+inner run, split the inner run's tests and replace this shard's report.
 """
 
 from __future__ import annotations
 
 import os
+import shlex
 
 import pytest
 
-from scripts.release_test_shards import shard_node_ids, shard_settings, write_report
+from scripts.release_test_shards import (
+    REPORT_DIRECTORY_VARIABLE,
+    SHARD_COUNT_VARIABLE,
+    SHARD_INDEX_VARIABLE,
+    ShardSettings,
+    shard_node_ids,
+    shard_settings,
+    write_report,
+)
+
+PLUGIN_NAME = __name__
+SHARD_SETTINGS_KEY = pytest.StashKey[ShardSettings]()
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    """Read this shard's settings, then keep them from pytest runs tests start."""
+    try:
+        config.stash[SHARD_SETTINGS_KEY] = shard_settings(os.environ)
+    except ValueError as error:
+        raise pytest.UsageError(str(error)) from error
+    _remove_shard_settings_from_environment()
 
 
 @pytest.hookimpl(trylast=True)
 def pytest_collection_modifyitems(
     config: pytest.Config, items: list[pytest.Item]
 ) -> None:
     """Keep only this shard's tests, after every other plugin has filtered them."""
-    try:
-        settings = shard_settings(os.environ)
-    except ValueError as error:
-        raise pytest.UsageError(str(error)) from error
+    settings = config.stash[SHARD_SETTINGS_KEY]
     collected = [item.nodeid for item in items]
     selected_node_ids = set(shard_node_ids(collected, settings.index, settings.count))
     selected = [item for item in items if item.nodeid in selected_node_ids]
@@ -35,3 +58,34 @@ def pytest_collection_modifyitems(
         config.hook.pytest_deselected(items=deselected)
     items[:] = selected
     write_report(settings, collected, [item.nodeid for item in selected])
+
+
+def _remove_shard_settings_from_environment() -> None:
+    """Remove the shard settings and this plugin's option from os.environ."""
+    for variable in (
+        SHARD_INDEX_VARIABLE,
+        SHARD_COUNT_VARIABLE,
+        REPORT_DIRECTORY_VARIABLE,
+    ):
+        os.environ.pop(variable, None)
+    remaining_options = _options_without_this_plugin(
+        shlex.split(os.environ.get("PYTEST_ADDOPTS", ""))
+    )
+    if remaining_options:
+        os.environ["PYTEST_ADDOPTS"] = shlex.join(remaining_options)
+    else:
+        os.environ.pop("PYTEST_ADDOPTS", None)
+
+
+def _options_without_this_plugin(options: list[str]) -> list[str]:
+    """Return pytest ``options`` without the ones that load this plugin."""
+    remaining: list[str] = []
+    option_iterator = iter(options)
+    for option in option_iterator:
+        if option == "-p":
+            plugin_name = next(option_iterator, None)
+            if plugin_name != PLUGIN_NAME:
+                remaining += [option] if plugin_name is None else [option, plugin_name]
+        elif option != f"-p{PLUGIN_NAME}":
+            remaining.append(option)
+    return remaining

tests/unit/test_cloud_run_push_workflow.py

@@ -2,10 +2,7 @@
 
 from __future__ import annotations
 
-import os
 import re
-import subprocess
-import sys
 
 from tests.unit.cloud_run_deploy_support import (
     PRODUCTION_CLOUD_RUN_SERVICE,
@@ -29,7 +26,6 @@
     "scripts/set_cloud_run_revision.sh",
 )
 LIVE_TESTS_STEP = "Run live staging tests, one pytest worker per test"
-LIVE_TESTS_SCRIPT = REPO / ".github/scripts/run_live_staging_tests.sh"
 RELEASE_TESTS_JOB = "release-tests"
 RELEASE_COMMIT_CONDITION = (
     "(github.repository == 'PolicyEngine/policyengine-api')\n"
@@ -206,59 +202,6 @@ def _assert_live_tests_run_in_parallel_after_readiness(
     assert "github.run_attempt" in parallel_live_tests
 
 
-def _live_test_worker_count(live_test_script: str) -> int:
-    worker_counts = re.findall(r"^\s*-n (\S+) \\$", live_test_script, re.MULTILINE)
-    assert len(worker_counts) == 1, "the live test script must pass -n once"
-    assert worker_counts[0].isdigit(), "the live test worker count must be a number"
-    return int(worker_counts[0])
-
-
-def _collected_live_tests(live_test_script: str) -> list[str]:
-    test_files = re.findall(r"tests/integration/test_live_\w+\.py", live_test_script)
-    collection_env = {
-        name: value for name, value in os.environ.items() if name != "API_BASE_URL"
-    }
-    collection = subprocess.run(
-        [
-            sys.executable,
-            "-m",
-            "pytest",
-            "--collect-only",
-            "-q",
-            "-p",
-            "no:cacheprovider",
-            *test_files,
-        ],
-        cwd=REPO,
-        env=collection_env,
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    assert collection.returncode == 0, collection.stdout + collection.stderr
-    return [
-        collected_line
-        for collected_line in collection.stdout.splitlines()
-        if "::" in collected_line
-    ]
-
-
-def test_live_staging_tests_get_one_pytest_worker_each():
-    live_test_script = LIVE_TESTS_SCRIPT.read_text(encoding="utf-8")
-    worker_count = _live_test_worker_count(live_test_script)
-    collected_tests = _collected_live_tests(live_test_script)
-
-    # Each test waits on its own requests, so with one worker each the step
-    # takes about as long as the longest test. A new live test needs -n raised
-    # with it, after checking the staging candidate's capacity: one Cloud Run
-    # instance admits 6 requests at a time (cloud_run_env.sh), each test sends
-    # one at a time, and further requests wait at the Cloud Run front end.
-    assert worker_count == len(collected_tests), (
-        f"run_live_staging_tests.sh runs {len(collected_tests)} tests in "
-        f"{worker_count} pytest workers; give each test its own worker"
-    )
-
-
 def _assert_promotion_verifies_the_tested_candidate(
     workflow: str, production_check: str, cloud_run_promotion: str
 ) -> None:

tests/unit/test_release_test_shards.py

@@ -358,6 +358,141 @@ def test_the_plugin_refuses_to_run_without_shard_settings(tmp_path: Path):
     )
 
 
+INNER_PYTEST_TEST = """
+import os
+import subprocess
+import sys
+
+
+def test_a_pytest_run_this_test_starts_is_not_split():
+    inherited_shard_settings = [
+        name for name in os.environ if name.startswith("RELEASE_TEST_SHARD_")
+    ]
+    assert inherited_shard_settings == []
+    assert "release_test_shard_plugin" not in os.environ.get("PYTEST_ADDOPTS", "")
+    inner_run = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "pytest",
+            "-p",
+            "no:cacheprovider",
+            "--collect-only",
+            "-q",
+            "inner_tests",
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert inner_run.returncode == 0, inner_run.stdout + inner_run.stderr
+    assert "1 test collected" in inner_run.stdout
+"""
+
+
+def _write_test_suite_that_starts_pytest(root: Path) -> None:
+    """Write the sample suite plus a test that runs pytest on inner_tests itself.
+
+    The outer run doesn't collect inner_tests, as in release run 41063091599,
+    where a test collected the live staging tests, which make test doesn't
+    collect, in its own pytest process.
+    """
+    _write_test_suite(root)
+    (root / "sample_tests" / "test_starts_pytest.py").write_text(INNER_PYTEST_TEST)
+    (root / "inner_tests").mkdir()
+    (root / "inner_tests" / "test_inner.py").write_text("def test_inner():\n    pass\n")
+
+
+def _run_shard_as_run_release_tests_does(
+    root: Path, report_root: Path, shard_index: int, shard_count: int
+):
+    """Run one shard with the plugin loaded through PYTEST_ADDOPTS."""
+    return subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "pytest",
+            "-p",
+            "no:cacheprovider",
+            "--rootdir",
+            str(root),
+            "-q",
+            str(root / "sample_tests"),
+        ],
+        cwd=root,
+        env={
+            **os.environ,
+            "PYTHONPATH": str(REPO),
+            "PYTEST_ADDOPTS": "-p tests.release_test_shard_plugin",
+            "RELEASE_TEST_SHARD_INDEX": str(shard_index),
+            "RELEASE_TEST_SHARD_COUNT": str(shard_count),
+            "RELEASE_TEST_SHARD_REPORT_DIRECTORY": str(
+                report_root / f"release-test-shard-{shard_index}"
+            ),
+        },
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+
+def test_a_pytest_run_that_a_test_starts_runs_without_the_split(tmp_path: Path):
+    suite_root = tmp_path / "suite"
+    suite_root.mkdir()
+    _write_test_suite_that_starts_pytest(suite_root)
+    report_root = tmp_path / "reports"
+    shard_count = 3
+
+    results = [
+        _run_shard_as_run_release_tests_does(
+            suite_root, report_root, shard_index, shard_count
+        )
+        for shard_index in range(1, shard_count + 1)
+    ]
+    coverage = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "scripts.release_test_shards",
+            "check-coverage",
+            "--shard-count",
+            str(shard_count),
+            str(report_root),
+        ],
+        cwd=REPO,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    # The inner pytest run neither splits its own tests nor replaces the
+    # shard's report, so every shard passes and the reports cover the outer
+    # run's 16 tests.
+    for result in results:
+        assert result.returncode == 0, result.stdout + result.stderr
+    assert coverage.returncode == 0, coverage.stdout + coverage.stderr
+    assert "Each of the 16 collected release tests ran in exactly one of 3 shards" in (
+        coverage.stdout
+    )
+
+
+def test_a_shard_report_is_never_replaced(tmp_path: Path):
+    settings = release_test_shards.ShardSettings(
+        index=1, count=2, report_directory=tmp_path / "release-test-shard-1"
+    )
+    release_test_shards.write_report(
+        settings, ["tests/unit/test_alpha.py::test_one"], []
+    )
+
+    with pytest.raises(FileExistsError, match="already has a shard report"):
+        release_test_shards.write_report(
+            settings, ["tests/unit/test_beta.py::test_one"], []
+        )
+
+    report = release_test_shards.read_report(settings.report_directory)
+    assert report.collected == ("tests/unit/test_alpha.py::test_one",)
+
+
 # --- push.yml -------------------------------------------------------------------