diff --git a/.github/workflows/experimental.yml b/.github/workflows/experimental.yml index 6249ad3be..85fcc3281 100644 --- a/.github/workflows/experimental.yml +++ b/.github/workflows/experimental.yml @@ -25,26 +25,88 @@ on: permissions: contents: read -# #1903, the owner's option 1: a run is superseded by the next one on its ref, -# whether a pull request's or a push to main's. Nothing waits on these legs, so -# main's newest commit is the one worth testing, and a merge wave no longer -# queues a two-hour GraalPy leg per push. unit-tests.yml still gives each push -# a run of its own, since its `unittest-ordering` leg runs only on main. Any -# other event -- a push to another branch, should `on:` ever allow one, or a -# trigger added later -- keeps a group of its own, by run id, and is never -# cancelled. -concurrency: - group: experimental-tests-${{ (github.event_name == 'pull_request' || (github.event_name == 'push' && github.ref == 'refs/heads/main')) && github.ref || github.run_id }} - cancel-in-progress: ${{ github.event_name == 'pull_request' || (github.event_name == 'push' && github.ref == 'refs/heads/main') }} +# #1903 has a newer commit supersede a run, a pull request's or a push to +# main's: nothing waits on these legs, so the newest commit is the one worth +# testing. #1938, the owner's choice: a superseded run is skipped or stopped, +# never cancelled, since GitHub counts a cancelled check against its commit and +# marks it red. So there is no `concurrency:` block at all: without +# `cancel-in-progress` a group still holds one pending run and cancels it once +# a newer one arrives (docs.github.com, "Control the concurrency of workflows +# and jobs"), and `queue: max` would hold a new commit's run behind the +# superseded one until that one stops. Instead `current` skips every other job +# of a run whose commit is superseded before it starts, and `Run unit tests` +# stops one superseded while its tests run; util/superseded.py holds the rule +# for both. A run whose commit stays current is unaffected. jobs: + # #1938: the gate. util/superseded.py's `check` asks the API whether this + # run's commit is still the head of main or of its open pull request, and + # writes `current`; any other event, or an API it cannot read, is current. + # Every other job needs this one and skips where `current` is `false`, so a + # superseded run's checks read `skipped`, which marks no commit red. Every + # step is `continue-on-error`: without `latest`, setup-python or requests the + # check fails or answers current, so a step's failure leaves `current` empty + # or `true`, which runs the rest; each job's `if:` opens with `!cancelled()`, + # so a failure of the job itself, which replaces the implicit `success()`, + # runs the rest too. + current: + name: Check the commit is still current + runs-on: ubuntu-latest + timeout-minutes: 5 + permissions: + contents: read + pull-requests: read + outputs: + current: ${{ steps.check.outputs.current }} + + steps: + - uses: actions/checkout@v7 + continue-on-error: true + with: + sparse-checkout: | + .github/python-versions.toml + util/python_versions.py + util/superseded.py + sparse-checkout-cone-mode: false + persist-credentials: false + + # #1660: `latest` from .github/python-versions.toml, read with the + # runner's own python3, for the interpreter util/superseded.py fetches + # with requests on. + - name: Read the Python versions + id: versions + continue-on-error: true + run: python3 util/python_versions.py --github-output + + - uses: actions/setup-python@v7 + id: python + continue-on-error: true + with: + python-version: ${{ steps.versions.outputs.latest }} + + - name: Install requests + id: requests + continue-on-error: true + run: python -m pip install requests + + - name: Ask whether this run's commit is still current + id: check + continue-on-error: true + env: + GH_TOKEN: ${{ github.token }} + run: python util/superseded.py check + # unit-tests.yml's job of the same name, reading the one output this file needs. python-versions: name: Read the Python versions + needs: current + if: ${{ !cancelled() && needs.current.outputs.current != 'false' }} runs-on: ubuntu-latest timeout-minutes: 5 outputs: experimental: ${{ steps.read.outputs.experimental }} + # #1938: the watcher's own interpreter, apart from the leg's. + latest: ${{ steps.read.outputs.latest }} steps: - uses: actions/checkout@v7 @@ -65,6 +127,8 @@ jobs: # pull_request run. changes: name: Classify changed files + needs: current + if: ${{ !cancelled() && needs.current.outputs.current != 'false' }} runs-on: ubuntu-latest timeout-minutes: 5 outputs: @@ -127,11 +191,12 @@ jobs: # # A job past its `timeout-minutes` is cancelled (same page, # `jobs..timeout-minutes`), and the page does not say whether the - # job-level guard covers that. So setup-python, the install and the tests - # carry caps of their own, 5, 10 and 120 minutes, under the job's 140: a step - # past its cap fails rather than cancels, and `continue-on-error` then applies - # as to any failure (actions/runner src/Runner.Worker/StepsRunner.cs:322-353, - # at 5f0b3ddcc). That leaves 5 minutes for the rest, which took at most 0.32, + # job-level guard covers that. So setup-python, the install, the watcher's + # setup-python and requests (#1938) and the tests carry caps of their own, + # 5, 10, 1, 1 and 120 minutes, under the job's 140: a step past its cap + # fails rather than cancels, and `continue-on-error` then applies as to any + # failure (actions/runner src/Runner.Worker/StepsRunner.cs:322-353, at + # 5f0b3ddcc). That leaves 3 minutes for the rest, which took at most 0.32, # setup-python included, on every uncovered leg of runs # 38569321543-38569522442. # @@ -151,11 +216,20 @@ jobs: # expected, blocks nothing, and leaves the caps as they are. experimental: name: Python ${{ matrix.python-version }} (experimental) - if: ${{ needs.changes.outputs.code != 'false' }} - needs: [changes, python-versions] + # #1938: `!cancelled()` so a failed `current` skips nothing, and then the + # two results stand in for the implicit `success()` it replaces: a failed + # `changes` or `python-versions` still skips the legs. + if: >- + ${{ !cancelled() && needs.changes.result == 'success' && needs.python-versions.result == 'success' + && needs.current.outputs.current != 'false' && needs.changes.outputs.code != 'false' }} + needs: [current, changes, python-versions] runs-on: ${{ matrix.os || 'ubuntu-latest' }} timeout-minutes: 140 continue-on-error: true + # #1938: `Run unit tests` reads the pull request, as `current` does. + permissions: + contents: read + pull-requests: read # #1672: every annotation a failing step writes is a warning. env: ANNOTATION_LEVEL: warning @@ -202,6 +276,35 @@ jobs: nproc python -c "import os; print('cpu_count', os.cpu_count())" + # #1938: the interpreter `Run unit tests` runs util/superseded.py on, a + # CPython of its own: `latest`, never the leg's GraalPy or free-threaded + # build, which would also have to take requests. `update-environment: + # false` leaves PATH, `pythonLocation` and `LD_LIBRARY_PATH` as the leg's + # setup-python set them, so the tests see the leg's interpreter alone; + # this one is run by its `python-path`, and finds its own libpython by + # the rpath actions/python-versions builds with + # (builders/ubuntu-python-builder.psm1), since its soname is not the + # leg's. No `cache:` and `--no-cache-dir`, so the leg's pip cache is + # untouched. Each step has a cap of 1 minute (see the job's comment), + # and a failure leaves the tests to run unwatched. + - uses: actions/setup-python@v7 + id: watcher-python + if: ${{ steps.install.outcome == 'success' }} + continue-on-error: true + timeout-minutes: 1 + with: + python-version: ${{ needs.python-versions.outputs.latest }} + update-environment: false + + - name: Install the watcher's requests + id: watcher + if: ${{ steps.install.outcome == 'success' && steps.watcher-python.outcome == 'success' }} + continue-on-error: true + timeout-minutes: 1 + env: + WATCHER_PYTHON: ${{ steps.watcher-python.outputs.python-path }} + run: '"$WATCHER_PYTHON" -m pip install --no-cache-dir requests' + # #1740: GraalPy's leg loses its runner 18-30 minutes into its tests ("The # runner has received a shutdown signal"), under a 120-minute cap. So every # 30 s this samples `free -m`, `df -h` of `/` and the workspace's volume, @@ -313,6 +416,26 @@ jobs: # crashed, by xdist's `[gwN] node down:` lines (under `-q` it prints no # `replacing crashed worker`, xdist/dsession.py:78-80 at v3.8.0), for # `Containment summary`. + # + # #1938: util/superseded.py's `watch` runs the whole command -- the scope's + # sudo where there is one, then prlimit, timeout and pytest -- in a + # session of its own, outside the scope, and asks `current`'s question + # every 60 s. Once the commit is superseded it sends SIGTERM, SIGKILL 30 s + # on, to that session and, with `--scope`, to every process of + # pcapkit-tests.scope through `systemctl kill`: the scope's processes run + # as the runner's user under a root-owned sudo, which a signal to the + # session cannot pass. Ubuntu's own /etc/sudoers sets `use_pty` (sudo + # 1.9.9-1ubuntu2.7 and 1.9.15p5-3ubuntu5.24.04.4), and sudoers(5) says it + # has no effect where sudo has no terminal, as in a step; the scope is + # reached by name either way. Once the command exits it writes a + # `Superseded` notice and a job-summary line and exits 0, so the step + # succeeds and frees the runner; where it outlives SIGKILL by 15 s, a + # warning and exit 70, since the tests may run on until `timeout` ends + # them. Otherwise it exits with the command's own status, so a failure on + # a current commit reads as it would unwatched, and an API it cannot read + # counts as current. It runs on `watcher-python`, with requests; where + # that interpreter or its requests did not install, the tests run + # unwatched, as they would without #1938. - name: Run unit tests id: tests if: ${{ steps.install.outcome == 'success' }} @@ -321,6 +444,9 @@ jobs: env: GRAAL_PYTHON_VM_ARGS: ${{ startsWith(matrix.python-version, 'graalpy') && '--vm.Xmx2g --vm.XX:+ExitOnOutOfMemoryError' || '' }} WORKERS: ${{ startsWith(matrix.python-version, 'graalpy') && '3' || 'auto' }} + GH_TOKEN: ${{ github.token }} + WATCHER: ${{ steps.watcher.outcome }} + WATCHER_PYTHON: ${{ steps.watcher-python.outputs.python-path }} run: | tail -n +1 -F "$RUNNER_TEMP/resource-monitor.log" 2>/dev/null & monitor=$! @@ -350,6 +476,18 @@ jobs: # A log line, not an annotation: on a lost runner this log is all that is left. echo "No 12G scope: the probe failed (${probe}), so the tests run uncapped." fi + if [ "$WATCHER" = success ] && [ -x "$WATCHER_PYTHON" ]; then + watched= + if [ "$scope" = 12G ]; then + # `scoped` is a function, which the watcher runs through a shell. + export -f scoped + set -- bash -c '"$@"' bash "$@" + watched=pcapkit-tests.scope + fi + set -- "$WATCHER_PYTHON" util/superseded.py watch --interval 60 --grace 30 --scope "$watched" -- "$@" + else + echo "No watcher: its interpreter or requests did not install (${WATCHER:-skipped}), so the tests run unwatched." + fi oom_kills() { awk '$1 == "oom_kill" { print $2 }' /proc/vmstat; } before=$(oom_kills) started=$SECONDS diff --git a/docs/source/contributing/workflows.rst b/docs/source/contributing/workflows.rst index c350b8334..cef833516 100644 --- a/docs/source/contributing/workflows.rst +++ b/docs/source/contributing/workflows.rst @@ -758,13 +758,50 @@ Tests run completes without it, and with it ``Required checks passed``, ``unittest-ordering`` and the workflows that wait on Unit Tests completing (``coverage-comment.yml``, ``deploy-pages.yml``, ``cron-vendor.yml`` and ``cron-conda.yml``). It runs on Unit Tests' ``push`` and ``pull_request`` events, -skips a docs-only pull request by the same ``changes`` script, and cancels a -superseded run, a pull request's or, unlike Unit Tests, a push to ``main``'s +skips a docs-only pull request by the same ``changes`` script, and lets a newer +commit supersede a run, a pull request's or, unlike Unit Tests, a push to ``main``'s (`#1903 `__): since nothing waits on the legs, ``main``'s newest commit is the one worth testing, whereas Unit -Tests keeps a run per push for ``unittest-ordering``, which runs only there. Any -other event keeps a concurrency group of its own, by run id. The legs never -block, by the KTLO mechanics above: every step that can fail has +Tests keeps a run per push for ``unittest-ordering``, which runs only there. + +A superseded run is skipped or stopped, never cancelled (`#1938 +`__): GitHub counts a +``cancelled`` check against its commit, so a cancelled run marks the commit red. +The workflow has no ``concurrency`` group either, since a group cancels its pending +run once a newer one queues. Its first job, ``current`` ("Check the commit is still +current"), sets up ``latest`` and installs ``requests``, which the script fetches with, +then runs :file:`util/superseded.py`'s ``check``. A push's commit is current +while it is the head of ``main``; a pull request's while the pull request is open and +its head is still the event's ``pull_request.head.sha``; any other event's always. +Every other job needs ``current`` and skips where its output is ``false``, so a run +superseded before it starts ends with ``skipped`` checks and a ``Superseded`` notice. +Each of those ``if:`` conditions opens with ``!cancelled()``, which drops the implicit +``success()``, so a failed ``current`` step, which leaves the output empty, or a failed +``current`` job runs the rest; the legs still need ``changes`` and ``python-versions`` +to succeed. ``Run unit tests`` runs its whole command -- the scope's ``sudo`` where +there is one, then ``prlimit``, ``timeout`` and pytest -- under the script's ``watch``, +outside the scope, which asks again every 60 seconds. The watcher runs on a CPython of +its own, ``latest`` from a second setup-python step, ``watcher-python``, with +``update-environment: false``, so the leg's ``PATH``, ``LD_LIBRARY_PATH`` and pip +cache are as its own setup-python left them, and ``requests`` goes into that +interpreter, never the leg's GraalPy or free-threaded build. The two steps are capped +at 1 minute each, and where either fails the tests run unwatched. Once the commit is superseded, +or the pull request merged or closed, it sends SIGTERM, then SIGKILL 30 seconds on: +to ``pcapkit-tests.scope`` by name (``sudo -n systemctl kill``) where the tests run +in it, since they run as the runner's user under a root-owned ``sudo`` and in +``timeout``'s own process group, and to the command's session otherwise. Ubuntu's +``/etc/sudoers`` sets ``use_pty``, which sudoers(5) says has no effect where ``sudo`` +has no terminal, as in a step; the scope is reached by name either way. Once the +command exits, the watcher writes a ``Superseded`` notice and a job-summary line and +exits 0, so the leg ends green and frees its runner. Every wait is bounded: where the +command outlives SIGKILL by 15 seconds, it writes a ``Superseded run not stopped`` +warning and exits 70, not 0, since the tests may run on, holding the step's output, +until ``timeout`` or the step's cap ends them, and the failed step is then a warning +like any other on these legs. Otherwise the step ends with the tests' own status. An +API the script cannot read counts as current, so it never stops the tests, and it +asks with ``If-None-Match``, as a ``304`` answer costs no rate limit. + +The legs never block, by the KTLO mechanics above: every step that can fail has ``continue-on-error: true``, their annotations are warnings, and ``Experimental summary``, the job's last step, reports each failed step. A failure no step contains, such as a lost runner, fails the job, and the job's own @@ -773,8 +810,10 @@ run from failing when a job fails" (`Workflow syntax `__), though the leg's check run still reads ``failure``. A job past its ``timeout-minutes`` is cancelled, and that page does not say whether the job-level -guard covers a cancellation. So setup-python, the install and the tests have caps -of their own, of 5, 10 and 120 minutes, under the job's 140: a step at its cap +guard covers a cancellation. So setup-python, the install, the watcher's setup-python +and ``requests`` (`#1938 `__, +below) and the tests have caps of their own, of 5, 10, 1, 1 and 120 minutes, under +the job's 140, which leaves 3 for the rest: a step at its cap fails rather than cancels, and the guard holds (actions/runner ``src/Runner.Worker/StepsRunner.cs:322-353``). The runner's ``has timed out`` error stays, changing no conclusion, as ``Process completed with exit code N.`` does. diff --git a/tests/project/test_experimental_concurrency_1903.py b/tests/project/test_experimental_concurrency_1903.py index fd16c5c3e..3935e0639 100644 --- a/tests/project/test_experimental_concurrency_1903.py +++ b/tests/project/test_experimental_concurrency_1903.py @@ -1,12 +1,14 @@ # -*- coding: utf-8 -*- """#1903: a newer run on ``main`` supersedes Experimental Tests' older one. -Every push to ``main`` used to get a concurrency group of its own, by run id, so -a merge wave queued a GraalPy leg of up to two hours per commit and held runners -that the open pull requests' CI then waited behind. The owner chose option 1: in -experimental.yml only, a push to ``main`` joins one group per ref and cancels the -run in progress, as a pull request's run already did. Nothing waits on these legs -(#1732), so the newest commit's run is enough. +A group per push queues a GraalPy leg of up to two hours per commit in a merge +wave, holding runners the open pull requests' CI then waits behind. The owner +chose option 1: in experimental.yml only, a push to ``main`` is superseded by +the next one, as a pull request's run is. Nothing waits on these legs (#1732), +so the newest commit's run is enough. #1938 keeps that rule and drops the +cancelling: experimental.yml has no group (:class:`TestExperimentalHasNoGroup`), +and its gate and watcher skip or stop a superseded run instead, which +:mod:`tests.project.test_superseded_runs_1938` pins. unit-tests.yml keeps a group per push, because its ``unittest-ordering`` leg runs only on ``main`` and a cancelled run there would leave a commit with no ordering @@ -32,7 +34,6 @@ WORKFLOWS = _tiers.ROOT / '.github' / 'workflows' MAIN = 'refs/heads/main' -PULL = 'refs/pull/1903/merge' #: One ``${{ }}`` interpolation in a ``group:`` value, its body captured. _INTERPOLATION = re.compile(r'\$\{\{(.*?)\}\}', re.S) @@ -73,45 +74,17 @@ def cancels(self, run: 'dict[str, object]') -> 'bool': return evaluate_condition(value, run) -class TestExperimentalSupersedesOnMain(unittest.TestCase): - """experimental.yml: a pull request's and ``main``'s runs are superseded per ref.""" +class TestExperimentalHasNoGroup(unittest.TestCase): + """#1938: experimental.yml has no group, so nothing of it is ever cancelled. - def setUp(self) -> None: - self.own = Concurrency('experimental.yml') + A newer commit still supersedes a run, by #1903's rule, through a gate and a + watcher: :mod:`tests.project.test_superseded_runs_1938`. + + """ - def test_the_cases_below_cover_every_trigger(self) -> None: - # Pushes to `main` and pull requests are all experimental.yml runs on; - # the other events below are the guard for a trigger added later. + def test_no_concurrency_block(self) -> None: with open(WORKFLOWS / 'experimental.yml', encoding='utf-8') as file: - spec = yaml.safe_load(file) - on = spec.get('on', spec.get(True)) - self.assertEqual(set(on), {'push', 'pull_request'}) - self.assertEqual(on['push'], {'branches': ['main']}) - - def test_pushes_to_main_share_one_cancelling_group(self) -> None: - """#1903: the next push's run cancels the one in progress.""" - first, second = context('push', MAIN, '1'), context('push', MAIN, '2') - self.assertEqual(self.own.group(first), f'experimental-tests-{MAIN}') - self.assertEqual(self.own.group(first), self.own.group(second)) - self.assertTrue(self.own.cancels(second)) - - def test_a_pull_request_still_shares_one_per_ref(self) -> None: - first, second = context('pull_request', PULL, '1'), context('pull_request', PULL, '2') - self.assertEqual(self.own.group(first), f'experimental-tests-{PULL}') - self.assertEqual(self.own.group(first), self.own.group(second)) - self.assertTrue(self.own.cancels(second)) - other = context('pull_request', 'refs/pull/1904/merge', '3') - self.assertNotEqual(self.own.group(other), self.own.group(first)) - self.assertNotEqual(self.own.group(first), self.own.group(context('push', MAIN, '1'))) - - def test_any_other_run_gets_a_group_of_its_own_and_is_never_cancelled(self) -> None: - """Including a dispatch or schedule on ``main``, which must not join the pushes' group.""" - for event_name, ref in (('push', 'refs/heads/feature'), ('schedule', MAIN), - ('workflow_dispatch', MAIN), ('workflow_run', MAIN)): - with self.subTest(event_name=event_name, ref=ref): - run = context(event_name, ref, '38497299656') - self.assertEqual(self.own.group(run), 'experimental-tests-38497299656') - self.assertFalse(self.own.cancels(run)) + self.assertNotIn('concurrency', yaml.safe_load(file)) class TestUnitTestsKeepsARunPerCommit(unittest.TestCase): diff --git a/tests/project/test_ktlo_legs_stay_green_1672.py b/tests/project/test_ktlo_legs_stay_green_1672.py index 60677fda9..815fc0ead 100644 --- a/tests/project/test_ktlo_legs_stay_green_1672.py +++ b/tests/project/test_ktlo_legs_stay_green_1672.py @@ -119,17 +119,19 @@ KTLO_REASON = 'KTLO legs never block (#1633), so this job still succeeds (#1672)' EXPERIMENTAL_REASON = 'Experimental legs never block (#1660), so this job still succeeds' #: ``experimental``'s steps with a ``timeout-minutes`` of their own (#1660), by ``id``: -#: setup-python, the install and the tests. #1732: the tests' is GraalPy's unit tier -#: less the round-trip sweeps, 88-113 minutes (estimate) from the 26% its leg reached -#: in 25 (job 115915976024). -EXPERIMENTAL_CAPS = {'python': 5, 'install': 10, 'tests': 120} +#: setup-python, the install, the watcher's setup-python and requests (#1938) and the +#: tests. #1732: the tests' is GraalPy's unit tier less the round-trip sweeps, 88-113 +#: minutes (estimate) from the 26% its leg reached in 25 (job 115915976024). +EXPERIMENTAL_CAPS = {'python': 5, 'install': 10, 'watcher-python': 1, 'watcher': 1, 'tests': 120} #: ``experimental``'s own ``timeout-minutes`` (#1732), and ``test``'s, main's 45 again. EXPERIMENTAL_JOB_CAP = 140 JOB_CAP = 45 -#: The job's minutes those caps leave for the rest of an experimental leg. Everything -#: but the install and the tests took at most 0.32 minutes, setup-python included, on -#: every uncovered leg of runs 38569321543, 38569429406 and 38569522442. -UNCAPPED_MINUTES = 5 +#: The job's minutes those caps leave for the rest of an experimental leg: the steps +#: without a cap of their own -- checkout, the parallelism report, the resource +#: monitor, the uploads and the summaries. Everything but the install and the tests +#: took at most 0.32 minutes, setup-python included, on every uncovered leg of runs +#: 38569321543, 38569429406 and 38569522442. +UNCAPPED_MINUTES = 3 #: Those steps' cap on every other leg: the largest a step's ``timeout-minutes`` may be #: (docs.github.com, workflow-syntax, ``jobs..steps[*].timeout-minutes``). DOCUMENTED_MAXIMUM = 360 @@ -692,7 +694,8 @@ def test_test_keeps_mains_caps(self) -> None: def test_the_workflows_page_states_the_caps(self) -> None: page = prose(ROOT / 'docs' / 'source' / 'contributing' / 'workflows.rst') - self.assertEqual(re.findall(r'caps of their own, of (\d+), (\d+) and (\d+) minutes', page), + stated = re.findall(r'caps of their own, of ((?:\d+, )+\d+) and (\d+) minutes', page) + self.assertEqual([tuple(head.split(', ')) + (last,) for head, last in stated], [tuple(str(cap) for cap in EXPERIMENTAL_CAPS.values())]) self.assertEqual(re.findall(r"minutes, under the job's (\d+)", page), [str(EXPERIMENTAL_JOB_CAP)]) diff --git a/tests/project/test_resource_monitor_1740.py b/tests/project/test_resource_monitor_1740.py index da26a76d2..2828e1be7 100644 --- a/tests/project/test_resource_monitor_1740.py +++ b/tests/project/test_resource_monitor_1740.py @@ -50,7 +50,8 @@ PROBES = ('free -m', 'df -h / "$GITHUB_WORKSPACE"', 'cat /proc/loadavg', 'ps -eo pid,rss,vsz,etime,args --sort=-rss | head -6') #: The experimental job's caps and its check name, which the monitor leaves alone. -CAPS = {'python': 5, 'install': 10, 'tests': 120} +#: #1938 adds the watcher's interpreter and its requests, capped as the rest are. +CAPS = {'python': 5, 'install': 10, 'watcher-python': 1, 'watcher': 1, 'tests': 120} CHECK_NAME = 'Python ${{ matrix.python-version }} (experimental)' #: A sample's first line, ``[monitor] sample ``. HEADER = re.compile(r'^\[monitor\] \d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ sample (\d+)$') @@ -550,7 +551,7 @@ def test_no_reader_takes_the_artifact_for_a_junit_one(self) -> None: self.assertTrue(ci_timing.JUNIT_ARTIFACT.startswith('junit-')) self.assertEqual([path.name for path in sorted((ROOT / 'util').glob('*.py')) if 'resource-monitor' in path.read_text(encoding='utf-8')], []) - self.assertEqual(list(workflow(EXPERIMENTAL[0])['jobs']), ['python-versions', 'changes', 'experimental']) + self.assertEqual(list(workflow(EXPERIMENTAL[0])['jobs']), ['current', 'python-versions', 'changes', 'experimental']) class TestTheDocsSayIt(unittest.TestCase): diff --git a/tests/project/test_superseded_runs_1938.py b/tests/project/test_superseded_runs_1938.py new file mode 100644 index 000000000..77c65a12c --- /dev/null +++ b/tests/project/test_superseded_runs_1938.py @@ -0,0 +1,809 @@ +# -*- coding: utf-8 -*- +"""#1938: Experimental Tests skips or stops a superseded run instead of cancelling it. + +GitHub counts a ``cancelled`` check against its commit in the status rollup, so +#1903's ``cancel-in-progress`` marked every superseded commit red, and a merged +pull request's last run, which nothing supersedes, kept its runner to the end. +The owner chose both remedies together: a gate job that skips the rest of a run +whose commit is no longer current, and a watcher that stops the tests once it +stops being current, each ending green. + +* :class:`TestNothingIsCancelled` and :class:`TestTheGate` read experimental.yml. +* :class:`TestCheck` drives util/superseded.py's rule through a fake transport. +* :class:`TestGateCommand` and :class:`TestWatch` run its two actions. +* :class:`TestTheTestsStep` runs ``Run unit tests`` itself, against a local + HTTP server standing in for the API. + +""" + +from __future__ import annotations + +import http.server +import io +import json +import os +import pathlib +import shutil +import signal +import subprocess # nosec B404 +import sys +import tempfile +import threading +import time +import unittest +import urllib.error +from typing import TYPE_CHECKING + +from tests.project.test_python_matrix_tiers_1633 import job, workflow, workflow_commands +from tests.project.test_python_versions_1660 import ROOT, prose +from tests.project.test_release_gates import STATUS, calls_status_check, evaluate_condition +from tests.project.test_resource_monitor_1740 import DEFAULT, TESTS, sh, step +from util import superseded + +if TYPE_CHECKING: + from typing import Any, Callable, Dict, List, Optional, Tuple + +NAME = 'experimental.yml' +GATE = 'current' +#: The commit a run tests, and the one that supersedes it. +OURS, NEWER = 'a' * 40, 'b' * 40 +REPO = 'JarryShaw/PyPCAPKit' +#: What every job but the gate skips on. +SKIPS_ON = "needs.current.outputs.current != 'false'" +POSIX = os.name == 'posix' + + +class FakeApi: + """A transport answering from ``routes``, recording each request. + + A route's value is a JSON-able body, answered 200 with an ``ETag`` of its + own and 304 to a request that sends that ``ETag`` back; an ``int`` status; + or an exception to raise. + + """ + + def __init__(self, routes: 'Dict[str, Any]') -> 'None': + self.routes = routes + self.requests = [] # type: List[Tuple[str, Dict[str, str]]] + + def __call__(self, url: 'str', headers: 'Dict[str, str]') -> 'Tuple[int, Dict[str, str], bytes]': + self.requests.append((url, dict(headers))) + path = url[len(superseded.API) + 1:] + answer = self.routes[path] + if isinstance(answer, BaseException): + raise answer + if isinstance(answer, int): + return answer, {}, b'{"message": "Server Error"}' + etag = f'W/"{path}"' + if headers.get('If-None-Match') == etag: + return 304, {'etag': etag}, b'' + return 200, {'etag': etag}, json.dumps(answer).encode() + + +class FakeResponse: + """What :class:`FakeSession` answers: the parts of a :class:`requests.Response` the transport reads.""" + + def __init__(self, status: 'int', headers: 'Dict[str, str]', body: 'Any') -> 'None': + self.status_code = status + self.headers = headers + self.content = b'' if body is None else json.dumps(body).encode() + + +class FakeSession: + """A :class:`requests.Session` answering from ``answers`` in turn, recording each GET. + + An answer is ``(status, headers, body)`` or an exception to raise. + + """ + + def __init__(self, answers: 'List[Any]') -> 'None': + self.answers = list(answers) + self.calls = [] # type: List[Tuple[str, Dict[str, Any]]] + + def get(self, url: 'str', **kwargs: 'Any') -> 'FakeResponse': + self.calls.append((url, kwargs)) + answer = self.answers.pop(0) + if isinstance(answer, BaseException): + raise answer + return FakeResponse(*answer) + + +def push(sha: 'str' = OURS) -> 'superseded.Context': + return superseded.Context('push', 'refs/heads/main', sha, REPO) + + +def pull(sha: 'str' = OURS, number: 'int' = 1938) -> 'superseded.Context': + return superseded.Context('pull_request', f'refs/pull/{number}/merge', sha, REPO, number) + + +def main_at(sha: 'str') -> 'Dict[str, Any]': + return {f'repos/{REPO}/commits/main': {'sha': sha}} + + +def pull_is(state: 'str', head: 'str' = OURS, merged: 'bool' = False, number: 'int' = 1938) -> 'Dict[str, Any]': + return {f'repos/{REPO}/pulls/{number}': {'state': state, 'merged': merged, 'head': {'sha': head}}} + + +class TestNothingIsCancelled(unittest.TestCase): + """No ``cancel-in-progress``, and no ``concurrency:`` group to cancel a pending run either.""" + + def test_no_cancel_in_progress_is_true(self) -> None: + text = (ROOT / '.github' / 'workflows' / NAME).read_text(encoding='utf-8') + code = [line for line in text.splitlines() if not line.lstrip().startswith('#')] + self.assertFalse([line for line in code if 'cancel-in-progress' in line]) + + def test_no_concurrency_group_at_all(self) -> None: + """A group without ``cancel-in-progress`` still cancels its pending run when a newer one queues.""" + spec = workflow(NAME) + self.assertNotIn('concurrency', spec) + self.assertEqual([job_id for job_id, body in spec['jobs'].items() if 'concurrency' in body], []) + + +class TestTheGate(unittest.TestCase): + """``current`` runs first, always succeeds, and every other job skips on its output.""" + + def setUp(self) -> None: + self.jobs = workflow(NAME)['jobs'] + + def test_the_gate_job(self) -> None: + gate = self.jobs[GATE] + self.assertNotIn('needs', gate) + self.assertNotIn('if', gate) + self.assertEqual(gate['outputs'], {'current': '${{ steps.check.outputs.current }}'}) + # Read-only: the default token, and nothing it can write with. + self.assertEqual(set(gate['permissions'].values()), {'read'}) + self.assertIs(gate.get('continue-on-error'), None) + self.assertTrue(all(entry.get('continue-on-error') is True for entry in gate['steps']), gate['steps']) + check = next(entry for entry in gate['steps'] if entry.get('id') == 'check') + self.assertEqual(check['run'], 'python util/superseded.py check') + self.assertEqual(check['env'], {'GH_TOKEN': '${{ github.token }}'}) + checkout = gate['steps'][0] + self.assertTrue(checkout['uses'].startswith('actions/checkout@')) + for path in ('util/superseded.py', 'util/python_versions.py', '.github/python-versions.toml'): + with self.subTest(path=path): + self.assertIn(path, checkout['with']['sparse-checkout']) + self.assertIs(checkout['with']['persist-credentials'], False) + + def test_the_gate_checks_on_latest_with_requests(self) -> None: + """The gate reads ``latest`` as every job reads it, sets it up and installs requests before the check.""" + ids = [entry.get('id') for entry in self.jobs[GATE]['steps']] + self.assertEqual(ids, [None, 'versions', 'python', 'requests', 'check']) + steps = {entry.get('id'): entry for entry in self.jobs[GATE]['steps']} + self.assertEqual(steps['versions']['run'], 'python3 util/python_versions.py --github-output') + setup = steps['python'] + self.assertEqual(setup['uses'], job(NAME, 'experimental')['steps'][1]['uses']) + self.assertEqual(setup['with'], {'python-version': '${{ steps.versions.outputs.latest }}'}) + self.assertEqual(steps['requests']['run'], 'python -m pip install requests') + + def test_the_watcher_runs_on_a_cpython_of_its_own_with_requests(self) -> None: + """The watcher runs on ``latest``, outside the leg's environment, and requests goes into it alone.""" + legs = job(NAME, 'experimental') + self.assertEqual(self.jobs['python-versions']['outputs']['latest'], '${{ steps.read.outputs.latest }}') + steps = {entry.get('id'): entry for entry in legs['steps']} + leg, own = steps['python'], steps['watcher-python'] + self.assertEqual(own['uses'], leg['uses']) + self.assertEqual(own['with'], {'python-version': '${{ needs.python-versions.outputs.latest }}', + 'update-environment': False}) + self.assertIs(own.get('continue-on-error'), True) + # The leg's own setup-python comes first: its interpreter, its environment, its cache. + self.assertEqual(leg['with'], {'python-version': '${{ matrix.python-version }}', 'cache': 'pip'}) + self.assertLess(legs['steps'].index(leg), legs['steps'].index(own)) + install = steps['watcher'] + self.assertEqual(install['env'], {'WATCHER_PYTHON': '${{ steps.watcher-python.outputs.python-path }}'}) + self.assertEqual(install['run'], '"$WATCHER_PYTHON" -m pip install --no-cache-dir requests') + self.assertIs(install.get('continue-on-error'), True) + self.assertIn("steps.watcher-python.outcome == 'success'", install['if']) + tests = step(TESTS) + self.assertEqual(tests['env']['WATCHER'], '${{ steps.watcher.outcome }}') + self.assertEqual(tests['env']['WATCHER_PYTHON'], '${{ steps.watcher-python.outputs.python-path }}') + self.assertIn('set -- "$WATCHER_PYTHON" util/superseded.py watch ', tests['run']) + self.assertNotIn('/usr/bin/python3', tests['run']) + # Nothing the leg's interpreter runs installs requests. + for entry in legs['steps']: + if entry.get('id') != 'watcher': + with self.subTest(step=entry.get('name') or entry.get('uses')): + self.assertNotIn('install requests', str(entry.get('run', ''))) + + def test_every_other_job_needs_it_and_skips_on_its_output(self) -> None: + others = [job_id for job_id in self.jobs if job_id != GATE] + self.assertEqual(sorted(others), ['changes', 'experimental', 'python-versions']) + for job_id in others: + body = self.jobs[job_id] + needs = body['needs'] if isinstance(body['needs'], list) else [body['needs']] + with self.subTest(job=job_id): + self.assertIn(GATE, needs) + self.assertIn(SKIPS_ON, body['if']) + # No implicit `success()`, so a failed gate job skips nothing. + self.assertTrue(body['if'].startswith('${{ !cancelled() && '), body['if']) + + def run_if(self, job_id: 'str', current: 'str', code: 'str' = 'true', *, gate: 'str' = 'success', + versions: 'str' = 'success', changes: 'str' = 'success', cancelled: 'bool' = False) -> 'bool': + """Whether ``job_id`` runs, GitHub's implicit ``success()`` applied where its ``if:`` calls no status check.""" + body = self.jobs[job_id] + needs = body['needs'] if isinstance(body['needs'], list) else [body['needs']] + results = {GATE: gate, 'changes': changes, 'python-versions': versions} + context = {'needs': {GATE: {'outputs': {'current': current}, 'result': gate}, + 'changes': {'outputs': {'code': code}, 'result': changes}, + 'python-versions': {'outputs': {}, 'result': versions}}, + STATUS: {'cancelled': cancelled, 'always': True, + 'success': not cancelled and all(results[name] == 'success' for name in needs), + 'failure': any(results[name] == 'failure' for name in needs)}} + runs = evaluate_condition(body['if'], context) + if not calls_status_check(body['if']): + runs = runs and bool(context[STATUS]['success']) # type: ignore[index] + return runs + + def test_a_superseded_commit_skips_every_job_and_anything_else_runs_them(self) -> None: + for job_id in ('python-versions', 'changes', 'experimental'): + with self.subTest(job=job_id): + self.assertFalse(self.run_if(job_id, 'false')) + self.assertTrue(self.run_if(job_id, 'true')) + # A gate step that failed writes nothing; the run goes on as it would without the gate. + self.assertTrue(self.run_if(job_id, '')) + # So does a gate job that failed, or timed out, as a whole. + self.assertTrue(self.run_if(job_id, '', gate='failure')) + self.assertTrue(self.run_if(job_id, '', gate='cancelled')) + # A cancelled run stays cancelled. + self.assertFalse(self.run_if(job_id, 'true', cancelled=True)) + # A docs-only pull request still skips the legs. + self.assertFalse(self.run_if('experimental', 'true', code='false')) + # And the legs still need the jobs they read: `!cancelled()` drops only the implicit `success()`. + for job_id in ('changes', 'python-versions'): + for result in ('failure', 'skipped', 'cancelled'): + with self.subTest(needs=job_id, result=result): + self.assertFalse(self.run_if('experimental', 'true', **{ + 'versions' if job_id == 'python-versions' else 'changes': result})) + # Superseded: `changes` and `python-versions` skip, and the legs with them. + self.assertFalse(self.run_if('experimental', 'false', versions='skipped', changes='skipped')) + + def test_the_legs_and_the_tests_step_read_with_the_token(self) -> None: + self.assertEqual(set(job(NAME, 'experimental')['permissions'].values()), {'read'}) + self.assertEqual(step(TESTS)['env']['GH_TOKEN'], '${{ github.token }}') + + +class TestCheck(unittest.TestCase): + """The rule: the head of ``main``, or of an open pull request; an API failure is current.""" + + def verdict(self, context: 'superseded.Context', routes: 'Dict[str, Any]') -> 'superseded.Verdict': + self.api = FakeApi(routes) + return superseded.check(context, superseded.Api('token', self.api)) + + def test_a_push_to_main(self) -> None: + self.assertEqual(self.verdict(push(), main_at(OURS)).state, superseded.CURRENT) + stale = self.verdict(push(), main_at(NEWER)) + self.assertEqual(stale.state, superseded.SUPERSEDED) + self.assertEqual(stale.reason, f'main is at {NEWER[:9]}, no longer at {OURS[:9]}') + url, headers = self.api.requests[0] + self.assertEqual(url, f'https://api.github.com/repos/{REPO}/commits/main') + self.assertEqual(headers['Authorization'], 'Bearer token') + + def test_a_pull_request(self) -> None: + self.assertEqual(self.verdict(pull(), pull_is('open')).state, superseded.CURRENT) + moved = self.verdict(pull(), pull_is('open', head=NEWER)) + self.assertEqual((moved.state, moved.reason), + (superseded.SUPERSEDED, f'pull request #1938 is at {NEWER[:9]}, no longer at {OURS[:9]}')) + merged = self.verdict(pull(), pull_is('closed', merged=True)) + self.assertEqual((merged.state, merged.reason), (superseded.SUPERSEDED, 'pull request #1938 is merged')) + closed = self.verdict(pull(), pull_is('closed')) + self.assertEqual((closed.state, closed.reason), (superseded.SUPERSEDED, 'pull request #1938 is closed')) + + def test_an_api_failure_is_current(self) -> None: + for answer in (500, 404, 403, urllib.error.URLError('timed out'), OSError('reset'), ValueError('not JSON')): + for context, path in ((push(), f'repos/{REPO}/commits/main'), (pull(), f'repos/{REPO}/pulls/1938')): + with self.subTest(answer=answer, event=context.event): + verdict = self.verdict(context, {path: answer}) + self.assertEqual(verdict.state, superseded.CURRENT) + self.assertTrue(verdict.error) + self.assertIn('so the commit counts as current', verdict.reason) + # A body that is not the expected shape is a failure too. + self.assertEqual(self.verdict(push(), {f'repos/{REPO}/commits/main': {}}).state, superseded.CURRENT) + + def test_any_other_event_is_current_and_asks_nothing(self) -> None: + for context in (superseded.Context('push', 'refs/heads/feature', OURS, REPO), + superseded.Context('workflow_dispatch', 'refs/heads/main', OURS, REPO), + superseded.Context('', '', '', ''), + superseded.Context('pull_request', 'refs/pull/1/merge', '', REPO, None)): + with self.subTest(event=context.event, ref=context.ref): + self.assertEqual(self.verdict(context, {}).state, superseded.CURRENT) + self.assertEqual(self.api.requests, []) + + def test_the_default_transport_is_a_requests_session(self) -> None: + """One :class:`requests.Session`, a bounded timeout, the status returned, a 304 answered from the cache.""" + session = FakeSession([(200, {'ETag': 'W/"1"'}, {'sha': NEWER}), (304, {'ETag': 'W/"1"'}, None)]) + api = superseded.Api('token') + api.session = session # type: ignore[assignment] + self.assertEqual(superseded.check(push(), api).state, superseded.SUPERSEDED) + self.assertEqual(superseded.check(push(), api).state, superseded.SUPERSEDED) + (url, first), (_, second) = session.calls + self.assertEqual(url, f'https://api.github.com/repos/{REPO}/commits/main') + self.assertEqual(first['timeout'], superseded.TIMEOUT) + self.assertEqual(first['headers']['Authorization'], 'Bearer token') + self.assertNotIn('If-None-Match', first['headers']) + self.assertEqual(second['headers']['If-None-Match'], 'W/"1"') + # The session is opened once and kept. + self.assertIs(api.session, session) + + def test_a_requests_error_is_current(self) -> None: + import requests # pylint: disable=import-outside-toplevel + for error in (requests.ConnectionError('reset'), requests.Timeout('slow'), requests.TooManyRedirects('loop')): + with self.subTest(error=type(error).__name__): + api = superseded.Api('token') + api.session = FakeSession([error]) # type: ignore[assignment] + verdict = superseded.check(push(), api) + self.assertEqual((verdict.state, verdict.error), (superseded.CURRENT, True)) + + def test_without_requests_everything_is_current(self) -> None: + """A missing requests is one more failed request, not a crash.""" + original = superseded.requests + superseded.requests = None # type: ignore[assignment] + self.addCleanup(setattr, superseded, 'requests', original) + verdict = superseded.check(push(), superseded.Api('token')) + self.assertEqual((verdict.state, verdict.error), (superseded.CURRENT, True)) + self.assertIn('requests is not installed', verdict.reason) + files, out = Files(self), io.StringIO() + self.assertEqual(superseded.main(['check'], env=files.env(), out=out), 0) + self.assertEqual(files.read(files.output), 'current=true\n') + + def test_the_script_runs_where_requests_cannot_import(self) -> None: + """In an interpreter whose ``import requests`` fails, the gate answers current and exits 0.""" + with tempfile.TemporaryDirectory() as tmp: + pathlib.Path(tmp, 'requests.py').write_text('raise ImportError("no requests here")\n', encoding='utf-8') + output = pathlib.Path(tmp, 'output') + env = {key: value for key, value in os.environ.items() if not key.startswith('GITHUB_')} + env.update(PYTHONPATH=tmp, GITHUB_EVENT_NAME='push', GITHUB_REF='refs/heads/main', GITHUB_SHA=OURS, + GITHUB_REPOSITORY=REPO, GITHUB_OUTPUT=str(output), GITHUB_API_URL='http://127.0.0.1:9') + result = subprocess.run([sys.executable, str(ROOT / 'util' / 'superseded.py'), 'check'], # nosec B603 + env=env, capture_output=True, text=True, check=False, timeout=60) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn('requests is not installed', result.stdout) + self.assertEqual(output.read_text(encoding='utf-8'), 'current=true\n') + + def test_a_repeated_request_is_conditional(self) -> None: + """``If-None-Match`` with the last ``ETag``; a 304 answers from the last body.""" + api = FakeApi(main_at(NEWER)) + client = superseded.Api('token', api) + first, second = superseded.check(push(), client), superseded.check(push(), client) + self.assertEqual((first.state, second.state), (superseded.SUPERSEDED, superseded.SUPERSEDED)) + self.assertEqual(second.reason, first.reason) + self.assertNotIn('If-None-Match', api.requests[0][1]) + self.assertEqual(api.requests[1][1]['If-None-Match'], f'W/"repos/{REPO}/commits/main"') + # A 304 with nothing cached is no answer, so it is current. + self.assertEqual(superseded.check(push(), superseded.Api('token', lambda url, headers: (304, {}, b''))).state, + superseded.CURRENT) + + def test_the_context_comes_from_the_job_environment(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + event = pathlib.Path(tmp, 'event.json') + event.write_text(json.dumps({'pull_request': {'number': 1938, 'head': {'sha': OURS}}}), encoding='utf-8') + env = {'GITHUB_EVENT_NAME': 'pull_request', 'GITHUB_REF': 'refs/pull/1938/merge', + 'GITHUB_SHA': NEWER, 'GITHUB_REPOSITORY': REPO, 'GITHUB_EVENT_PATH': str(event)} + context = superseded.Context.from_env(env) + # The pull request's head, not GITHUB_SHA, which is the merge commit. + self.assertEqual((context.sha, context.number, context.repo), (OURS, 1938, REPO)) + event.write_text('not JSON', encoding='utf-8') + broken = superseded.Context.from_env(env) + self.assertEqual((broken.sha, broken.number), ('', None)) + self.assertEqual(self.verdict(broken, {}).state, superseded.CURRENT) + context = superseded.Context.from_env({'GITHUB_EVENT_NAME': 'push', 'GITHUB_REF': 'refs/heads/main', + 'GITHUB_SHA': OURS, 'GITHUB_REPOSITORY': REPO}) + self.assertEqual((context.event, context.sha, context.watched), ('push', OURS, True)) + + +class Files: + """``$GITHUB_OUTPUT`` and ``$GITHUB_STEP_SUMMARY`` in a temporary directory.""" + + def __init__(self, case: 'unittest.TestCase') -> 'None': + tmp = tempfile.TemporaryDirectory(prefix='pcapkit-1938-') # pylint: disable=consider-using-with + case.addCleanup(tmp.cleanup) + self.root = pathlib.Path(tmp.name) + self.output = self.root / 'output' + self.summary = self.root / 'summary.md' + + def env(self, **extra: 'str') -> 'Dict[str, str]': + return dict({'GITHUB_EVENT_NAME': 'push', 'GITHUB_REF': 'refs/heads/main', 'GITHUB_SHA': OURS, + 'GITHUB_REPOSITORY': REPO, 'GH_TOKEN': 'token', 'GITHUB_OUTPUT': str(self.output), + 'GITHUB_STEP_SUMMARY': str(self.summary)}, **extra) + + def read(self, path: 'pathlib.Path') -> 'str': + return path.read_text(encoding='utf-8') if path.exists() else '' + + +class TestGateCommand(unittest.TestCase): + """``check``: ``current=`` in ``$GITHUB_OUTPUT``, a notice and a summary line where superseded, exit 0.""" + + def run_gate(self, routes: 'Dict[str, Any]') -> 'Tuple[int, str, Files]': + files, out = Files(self), io.StringIO() + code = superseded.main(['check'], env=files.env(), transport=FakeApi(routes), out=out) + return code, out.getvalue(), files + + def test_superseded(self) -> None: + code, out, files = self.run_gate(main_at(NEWER)) + self.assertEqual(code, 0) + self.assertEqual(files.read(files.output), 'current=false\n') + self.assertEqual(workflow_commands(out), [ + f'::notice title=Superseded::main is at {NEWER[:9]}, no longer at {OURS[:9]}. ' + 'The rest of this run is skipped, so none of its jobs fails or is cancelled.']) + self.assertEqual(files.read(files.summary), + f'- Superseded: main is at {NEWER[:9]}, no longer at {OURS[:9]}. ' + 'The rest of this run is skipped, so none of its jobs fails or is cancelled.\n') + + def test_current_and_on_an_api_failure(self) -> None: + for routes in (main_at(OURS), {f'repos/{REPO}/commits/main': 502}): + with self.subTest(routes=routes): + code, out, files = self.run_gate(routes) + self.assertEqual(code, 0) + self.assertEqual(files.read(files.output), 'current=true\n') + self.assertEqual(workflow_commands(out), []) + self.assertEqual(files.read(files.summary), '') + + +#: A stand-in for pytest that runs until it is stopped. +SLEEPER = [sys.executable, '-c', 'import time; time.sleep(60)'] + + +@unittest.skipUnless(POSIX, 'the watcher signals a process group, as on the Ubuntu runners') +class TestWatch(unittest.TestCase): + """``watch``: the command's own status while current; stopped, a notice and 0 once superseded.""" + + def watch(self, command: 'List[str]', routes: 'Dict[str, Any]', grace: 'str' = '5', *, + options: 'Tuple[str, ...]' = (), run: 'Optional[Callable[[Any], int]]' = None, + kill: 'Optional[Callable[[int, int], None]]' = None, **env: 'str') -> 'Tuple[int, str, Files, float]': + files, out = Files(self), io.StringIO() + started = time.monotonic() + extra = {} # type: Dict[str, Any] + if run is not None: + extra['run'] = run + if kill is not None: + extra['kill'] = kill + code = superseded.main(['watch', '--interval', '0.2', '--grace', grace, *options, '--', *command], + env=files.env(**env), transport=FakeApi(routes), out=out, **extra) + return code, out.getvalue(), files, time.monotonic() - started + + def pid_command(self, body: 'str') -> 'Tuple[List[str], pathlib.Path]': + """A command that writes its pid to a file, then runs ``body``; the pid is killed at cleanup.""" + tmp = tempfile.TemporaryDirectory(prefix='pcapkit-1938-') # pylint: disable=consider-using-with + self.addCleanup(tmp.cleanup) + pidfile = pathlib.Path(tmp.name, 'pid') + + def cleanup() -> 'None': + if pidfile.exists(): + try: + os.kill(int(pidfile.read_text(encoding='utf-8')), signal.SIGKILL) + except (OSError, ValueError): + pass + self.addCleanup(cleanup) + return [sys.executable, '-c', f'import os, signal, time; open({str(pidfile)!r}, "w").write(str(os.getpid())); ' + + body], pidfile + + def test_the_scoped_tests_are_stopped_through_the_scope(self) -> None: + """Under sudo a signal to the session is refused, so the scope's processes are killed by name.""" + command, pidfile = self.pid_command('time.sleep(60)') + calls = [] # type: List[List[str]] + + def run(argv: 'Any') -> 'int': + # systemd, as it kills every process of the scope. + calls.append(list(argv)) + signum = getattr(signal, argv[4].split('=', 1)[1]) + try: + os.kill(int(pidfile.read_text(encoding='utf-8')), signum) + except (OSError, ValueError): + pass + return 0 + + refused = [] # type: List[Tuple[int, int]] + code, out, _, elapsed = self.watch(command, main_at(NEWER), options=('--scope', 'pcapkit-tests.scope'), + run=run, kill=lambda pid, signum: refused.append((pid, signum))) + self.assertEqual(code, 0) + self.assertLess(elapsed, 30) + self.assertIn('::notice title=Superseded::main is at', out) + self.assertEqual(calls[0], ['sudo', '-n', 'systemctl', 'kill', '--signal=SIGTERM', 'pcapkit-tests.scope']) + # And a sweep, for whatever outlived the scope's main process. + self.assertEqual(calls[-1], ['sudo', '-n', 'systemctl', 'kill', '--signal=SIGKILL', 'pcapkit-tests.scope']) + # SIGTERM to the scope alone, so the tests keep their grace; the session only gets the sweep. + self.assertEqual([signum for _, signum in refused], [signal.SIGKILL]) + + def test_without_a_scope_nothing_runs_through_sudo(self) -> None: + calls = [] # type: List[Any] + code, _, _, _ = self.watch(SLEEPER, main_at(NEWER), run=lambda argv: calls.append(argv) or 0) + self.assertEqual((code, calls), (0, [])) + + def test_a_command_that_cannot_be_stopped_ends_the_watch_with_a_warning(self) -> None: + """Every wait is bounded; a stop that fails warns and exits STOP_FAILED, not 0.""" + command, pidfile = self.pid_command('time.sleep(60)') + refused = [] # type: List[Tuple[int, int]] + calls = [] # type: List[Any] + code, out, files, elapsed = self.watch( + command, main_at(NEWER), grace='0.5', options=('--kill-wait', '0.5', '--scope', 'pcapkit-tests.scope'), + run=lambda argv: calls.append(list(argv)) or 1, kill=lambda pid, signum: refused.append((pid, signum))) + self.assertEqual(code, superseded.STOP_FAILED) + self.assertNotIn(code, (0, 1, 2, 3, 4, 5, 124, 125, 126, 127, 137)) + self.assertLess(elapsed, 20) + self.assertEqual([signum for _, signum in refused], [signal.SIGKILL]) + self.assertEqual([argv[4] for argv in calls], ['--signal=SIGTERM', '--signal=SIGKILL']) + commands = workflow_commands(out) + self.assertEqual(len(commands), 1, out) + self.assertTrue(commands[0].startswith( + f'::warning title=Superseded run not stopped::main is at {NEWER[:9]}, no longer at {OURS[:9]}, but SIGTERM ' + 'and SIGKILL to the scope pcapkit-tests.scope and the session of pid '), commands[0]) + self.assertTrue(commands[0].endswith(' did not end it within 1 s. The tests may run on until their timeout.'), + commands[0]) + self.assertTrue(files.read(files.summary).startswith('- Superseded run not stopped: main is at')) + # It is still running: the watcher gave up on it rather than waiting. + os.kill(int(pidfile.read_text(encoding='utf-8')), 0) + + def test_a_failing_systemctl_is_bounded(self) -> None: + """The default runner: a command that hangs past its timeout, or cannot start, is a failure.""" + self.assertEqual(superseded.run_command(['/nonexistent/systemctl']), -1) + original = superseded.TIMEOUT + superseded.TIMEOUT = 0.5 + self.addCleanup(setattr, superseded, 'TIMEOUT', original) + started = time.monotonic() + self.assertEqual(superseded.run_command([sys.executable, '-c', 'import time; time.sleep(30)']), -1) + self.assertLess(time.monotonic() - started, 10) + + def test_a_superseded_commit_stops_the_tests_and_exits_0_with_the_notice(self) -> None: + code, out, files, elapsed = self.watch(SLEEPER, main_at(NEWER)) + self.assertEqual(code, 0) + self.assertLess(elapsed, 30) + commands = workflow_commands(out) + self.assertEqual(len(commands), 1, out) + self.assertTrue(commands[0].startswith( + f'::notice title=Superseded::main is at {NEWER[:9]}, no longer at {OURS[:9]}. The tests were stopped after '), + commands[0]) + self.assertTrue(commands[0].endswith(' s, and this step succeeds.'), commands[0]) + self.assertRegex(files.read(files.summary), r'^- Superseded: main is at b{9}, no longer at a{9}\. ' + r'The tests were stopped after \d+ s, and this step succeeds\.\n$') + + def test_a_closed_pull_request_stops_the_tests(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + event = pathlib.Path(tmp, 'event.json') + event.write_text(json.dumps({'pull_request': {'number': 1938, 'head': {'sha': OURS}}}), encoding='utf-8') + code, out, _, _ = self.watch(SLEEPER, pull_is('closed', merged=True), GITHUB_EVENT_NAME='pull_request', + GITHUB_EVENT_PATH=str(event)) + self.assertEqual(code, 0) + self.assertIn('::notice title=Superseded::pull request #1938 is merged. The tests were stopped', out) + + def test_a_command_that_ignores_sigterm_is_killed_after_the_grace(self) -> None: + stubborn = [sys.executable, '-c', 'import signal, time; signal.signal(signal.SIGTERM, signal.SIG_IGN); ' + 'print("ready", flush=True); time.sleep(60)'] + code, out, _, elapsed = self.watch(stubborn, main_at(NEWER), grace='1') + self.assertEqual(code, 0) + self.assertLess(elapsed, 30) + self.assertIn('::notice title=Superseded::', out) + + def test_the_whole_session_is_stopped(self) -> None: + """A grandchild that outlives its parent is stopped too, so nothing holds the step's pipe.""" + with tempfile.TemporaryDirectory() as tmp: + pidfile = pathlib.Path(tmp, 'pid') + spawner = [sys.executable, '-c', + 'import subprocess, sys, time; ' + 'child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]); ' + f'open({str(pidfile)!r}, "w").write(str(child.pid)); time.sleep(60)'] + code, _, _, _ = self.watch(spawner, main_at(NEWER)) + self.assertEqual(code, 0) + grandchild = int(pidfile.read_text(encoding='utf-8')) + for _ in range(50): + try: + os.kill(grandchild, 0) + except ProcessLookupError: + break + time.sleep(0.1) + else: + os.kill(grandchild, signal.SIGKILL) + self.fail('the grandchild outlived the watcher') + + def test_a_failure_on_a_current_commit_keeps_its_status(self) -> None: + """The KTLO and experimental semantics: the tests' status, unchanged, current or not askable.""" + failing = [sys.executable, '-c', 'import sys, time; time.sleep(0.7); sys.exit(3)'] + for routes in (main_at(OURS), {f'repos/{REPO}/commits/main': 500}, + {f'repos/{REPO}/commits/main': urllib.error.URLError('down')}): + with self.subTest(routes=routes): + code, out, files, _ = self.watch(failing, routes) + self.assertEqual(code, 3) + self.assertEqual(workflow_commands(out), []) + self.assertEqual(files.read(files.summary), '') + code, _, _, _ = self.watch([sys.executable, '-c', 'pass'], main_at(OURS)) + self.assertEqual(code, 0) + + def test_an_api_failure_is_logged_once_and_the_tests_run_on(self) -> None: + failing = [sys.executable, '-c', 'import sys, time; time.sleep(1); sys.exit(1)'] + code, out, _, _ = self.watch(failing, {f'repos/{REPO}/commits/main': 503}) + self.assertEqual(code, 1) + self.assertEqual(out.count('util/superseded.py: could not ask the API (ApiError: HTTP 503'), 1, out) + + def test_a_signal_ends_as_a_shell_reports_it(self) -> None: + code, _, _, _ = self.watch([sys.executable, '-c', 'import os, signal; os.kill(os.getpid(), signal.SIGKILL)'], + main_at(OURS)) + self.assertEqual(code, 128 + signal.SIGKILL) + + def test_an_unwatched_event_never_asks(self) -> None: + api = FakeApi({}) + files = Files(self) + code = superseded.main(['watch', '--interval', '0.1', '--', sys.executable, '-c', 'import time; time.sleep(0.5)'], + env=files.env(GITHUB_EVENT_NAME='workflow_dispatch'), transport=api, out=io.StringIO()) + self.assertEqual((code, api.requests), (0, [])) + + +class Server(http.server.ThreadingHTTPServer): + """The API, answering ``/repos//commits/main`` with ``head``.""" + + head = NEWER + + +class Handler(http.server.BaseHTTPRequestHandler): + + def do_GET(self) -> 'None': # pylint: disable=invalid-name + if self.path != f'/repos/{REPO}/commits/main': + self.send_error(404) + return + body = json.dumps({'sha': self.server.head}).encode() # type: ignore[attr-defined] + self.send_response(200) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: 'str', *args: 'Any') -> 'None': # pylint: disable=redefined-builtin + pass + + +@unittest.skipUnless(sys.platform.startswith('linux') and all(map(shutil.which, ('bash', 'timeout', 'prlimit'))), + 'the tests step runs coreutils timeout and util-linux prlimit, as on the Ubuntu runners') +class TestTheTestsStep(unittest.TestCase): + """``Run unit tests`` itself, with ``python`` a stub that runs until stopped and the API a local server.""" + + def setUp(self) -> None: + tmp = tempfile.TemporaryDirectory(prefix='pcapkit-1938-') # pylint: disable=consider-using-with + self.addCleanup(tmp.cleanup) + self.tmp = pathlib.Path(tmp.name) + self.temp = self.tmp / 'temp' + self.temp.mkdir() + bindir = self.tmp / 'bin' + bindir.mkdir() + (self.temp / 'resource-monitor.log').write_text('', encoding='utf-8') + # No real sudo: the probe fails, so the tests run uncapped, the same argv as in the scope. + for name, body in (('sudo', 'echo "sudo: a password is required" >&2\nexit 1\n'), + ('python', 'echo collected\nexec sleep 60\n')): + (bindir / name).write_text('#!/bin/sh\n' + body, encoding='utf-8') + (bindir / name).chmod(0o755) + self.server = Server(('127.0.0.1', 0), Handler) + thread = threading.Thread(target=self.server.serve_forever, daemon=True) + thread.start() + self.addCleanup(self.server.server_close) + self.addCleanup(self.server.shutdown) + self.env = {key: value for key, value in os.environ.items() if not key.startswith('GITHUB_')} + self.env.update(RUNNER_TEMP=str(self.temp), GITHUB_WORKSPACE=str(ROOT), WORKERS='3', GRAAL_PYTHON_VM_ARGS='', + GITHUB_STEP_SUMMARY=str(self.tmp / 'summary.md'), GH_TOKEN='token', + GITHUB_API_URL=f'http://127.0.0.1:{self.server.server_address[1]}', + GITHUB_EVENT_NAME='push', GITHUB_REF='refs/heads/main', GITHUB_SHA=OURS, + GITHUB_REPOSITORY=REPO, PATH=str(bindir) + os.pathsep + os.environ.get('PATH', ''), + # `watcher-python` and its requests: this interpreter, which has requests. + WATCHER='success', WATCHER_PYTHON=sys.executable) + + def test_without_the_watchers_interpreter_the_tests_run_unwatched(self) -> None: + """A failed `watcher-python` or requests install runs the tests unwatched, with their own status.""" + bindir = pathlib.Path(self.env['PATH'].split(os.pathsep)[0]) + (bindir / 'python').write_text('#!/bin/sh\nsleep 1.5\nexit 3\n', encoding='utf-8') + for outcome, path in (('failure', sys.executable), ('skipped', ''), ('success', '/nonexistent/python')): + with self.subTest(outcome=outcome, path=path): + env = dict(self.env, WATCHER=outcome, WATCHER_PYTHON=path) + # Superseded, so a watcher would have stopped the tests and exited 0. + result = sh(self.script(), env, DEFAULT) + self.assertEqual(result.returncode, 3, result.stderr) + self.assertIn('No watcher: its interpreter or requests did not install', result.stdout) + self.assertEqual(workflow_commands(result.stdout), []) + + def script(self) -> 'str': + script = step(TESTS)['run'] + quick = script.replace('watch --interval 60 --grace 30 ', 'watch --interval 0.5 --grace 2 --kill-wait 2 ') + self.assertNotEqual(quick, script) + return quick + + def status(self) -> 'Dict[str, str]': + text = (self.temp / 'tests-status').read_text(encoding='utf-8') + return dict(line.split('=', 1) for line in text.splitlines()) + + def test_a_superseded_run_stops_the_tests_and_the_step_succeeds(self) -> None: + result = sh(self.script(), self.env, DEFAULT) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn('collected', result.stdout) + commands = workflow_commands(result.stdout) + self.assertEqual(len(commands), 1, result.stdout) + self.assertTrue(commands[0].startswith( + f'::notice title=Superseded::main is at {NEWER[:9]}, no longer at {OURS[:9]}. '), commands[0]) + self.assertIn('- Superseded: main is at', (self.tmp / 'summary.md').read_text(encoding='utf-8')) + self.assertEqual((self.status()['exit'], self.status()['scope']), ('0', 'none')) + + def test_a_scoped_run_is_stopped_through_its_scope(self) -> None: + """In the scope, ``timeout`` and pytest are out of the session's reach; ``systemctl kill`` is not. + + ``sudo`` is a stub that runs the scope's command and plays systemd for + ``systemctl kill``, signalling the pid the tests wrote. The tests leave + the session as the scoped ones are out of its reach, so only the scope's + kill can end them: without it, they hold the step's output for 60 s. + + """ + bindir = pathlib.Path(self.env['PATH'].split(os.pathsep)[0]) + pidfile, calls = self.tmp / 'pid', self.tmp / 'systemctl' + (bindir / 'sudo').write_text( + '#!/bin/sh\n' + f'if [ "$2" = systemctl ]; then echo "$*" >> {calls}; ' + f'[ -s {pidfile} ] && kill -s "${{4#--signal=SIG}}" "$(cat {pidfile})"; exit 0; fi\n' + 'while [ "$#" -gt 0 ] && [ "$1" != -- ]; do shift; done\n' + '[ "$#" -gt 0 ] || exit 0\nshift\nexec "$@"\n', encoding='utf-8') + (bindir / 'python').write_text(f'#!/bin/sh\necho $$ > {pidfile}\necho collected\nexec setsid sleep 60\n', + encoding='utf-8') + started = time.monotonic() + result = sh(self.script(), self.env, DEFAULT) + self.assertLess(time.monotonic() - started, 30) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(self.status()['scope'], '12G') + commands = workflow_commands(result.stdout) + self.assertEqual(len(commands), 1, result.stdout) + self.assertTrue(commands[0].startswith('::notice title=Superseded::main is at '), commands[0]) + self.assertEqual(calls.read_text(encoding='utf-8').splitlines()[0], + '-n systemctl kill --signal=SIGTERM pcapkit-tests.scope') + + def test_a_current_run_keeps_the_tests_status(self) -> None: + self.server.head = OURS + bindir = pathlib.Path(self.env['PATH'].split(os.pathsep)[0]) + (bindir / 'python').write_text('#!/bin/sh\nsleep 1.5\nexit 1\n', encoding='utf-8') + result = sh(self.script(), self.env, DEFAULT) + self.assertEqual(result.returncode, 1, result.stderr) + self.assertEqual(workflow_commands(result.stdout), []) + self.assertFalse((self.tmp / 'summary.md').exists()) + + def test_the_watcher_runs_the_scope_and_names_it(self) -> None: + """Every cap stays on pytest, and the watcher runs outside the scope, which it names to ``--scope``.""" + script = step(TESTS)['run'] + capped = 'set -- prlimit --fsize=2147483648 timeout --signal=INT --kill-after=3m 110m python -m pytest "$@"\n' + scoped = 'set -- scoped pcapkit-tests.scope "$@"\n' + named = 'watched=pcapkit-tests.scope\n' + watched = ('set -- "$WATCHER_PYTHON" util/superseded.py watch --interval 60 --grace 30 --scope "$watched" ' + '-- "$@"\n') + self.assertLess(script.index(capped), script.index(scoped)) + self.assertLess(script.index(scoped), script.index(named)) + self.assertLess(script.index(named), script.index(watched)) + self.assertLess(script.index(watched), script.index('"$@" 2>&1 | tee "$RUNNER_TEMP/pytest.log"')) + # Named only where the probe passed, and the one name the tests' scope is started under. + self.assertIn('if [ "$scope" = 12G ]; then\n', script[script.index('scope=none'):script.index(named)]) + self.assertEqual(script.count('pcapkit-tests.scope'), 2) + self.assertIn('if [ "$WATCHER" = success ] && [ -x "$WATCHER_PYTHON" ]; then\n', script) + + +class TestTheUpdaterReadsTheWatchersLatest(unittest.TestCase): + """util/update_python_versions.py reads where ``latest`` is set up, the watcher's included.""" + + @staticmethod + def workflows(runs_on: 'str', leg: 'bool' = True) -> 'Dict[str, Any]': + """One job setting up ``latest`` on ``runs_on``, after its leg's own version where ``leg``.""" + steps = [{'id': 'v', 'run': 'python3 util/python_versions.py --github-output'}, + {'uses': 'actions/setup-python@v7', 'with': {'python-version': '${{ steps.v.outputs.latest }}'}}] + if leg: + steps.insert(1, {'uses': 'actions/setup-python@v7', 'with': {'python-version': '${{ matrix.python-version }}'}}) + return {'w.yml': {'jobs': {'j': {'runs-on': runs_on, 'steps': steps}}}} + + def test_a_matrix_jobs_latest_is_on_its_fallback_label(self) -> None: + from tests.project.test_python_versions_update_1660 import updater # pylint: disable=import-outside-toplevel + module = updater() + labels = module.workflow_labels(self.workflows("${{ matrix.os || 'ubuntu-24.04' }}")) + self.assertEqual(labels['latest'], ['ubuntu-24.04']) + with self.assertRaisesRegex(ValueError, 'installs latest on .*which this cannot read'): + module.workflow_labels(self.workflows('${{ matrix.runner }}', leg=False)) + # The real workflows, experimental's watcher among them, are readable too. + self.assertIn('ubuntu-latest', module.workflow_labels(module.load_workflows(ROOT))['latest']) + + +class TestTheDocsSayIt(unittest.TestCase): + + def test_the_workflows_page_describes_it(self) -> None: + page = prose(ROOT / 'docs' / 'source' / 'contributing' / 'workflows.rst') + for phrase in ('`#1938 `__', ':file:`util/superseded.py`', + '``Superseded``', '``skipped``', '``current``'): + with self.subTest(phrase=phrase): + self.assertIn(phrase, page) + self.assertNotIn('cancels a superseded run', page) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/project/test_sweeps_on_latest_only_1732.py b/tests/project/test_sweeps_on_latest_only_1732.py index 443765c29..9153a1ea2 100644 --- a/tests/project/test_sweeps_on_latest_only_1732.py +++ b/tests/project/test_sweeps_on_latest_only_1732.py @@ -59,7 +59,8 @@ CONSUMERS = ('coverage-comment.yml', 'deploy-pages.yml', 'cron-vendor.yml', 'cron-conda.yml') #: Each job's ``timeout-minutes``, and its capped steps', by ``id``: ``test``'s are #: main's, ``experimental``'s are sized for GraalPy (88-113 minutes, estimate). -CAPS = {JOB: (45, {}), EXPERIMENTAL: (140, {'python': 5, 'install': 10, 'tests': 120})} +#: #1938 adds the watcher's interpreter and its requests, capped too. +CAPS = {JOB: (45, {}), EXPERIMENTAL: (140, {'python': 5, 'install': 10, 'watcher-python': 1, 'watcher': 1, 'tests': 120})} #: The jobs whose result reaches a gate: ``required-checks`` and what it needs, and #: ``unittest-ordering``, which runs on main's pushes and gates publishing. GATING = ('required-checks', 'unittest-ordering') @@ -97,7 +98,9 @@ def stringify(value: 'object') -> 'str': def step_env(leg: 'dict[str, Any]', where: 'tuple[str, str]' = JOB) -> 'dict[str, str]': """The step's ``env:`` on ``leg``, ``needs`` read from the tier file.""" - context = {'matrix': leg, 'needs': {'python-versions': {'outputs': live_outputs()}}} + # #1938: experimental's tests step reads the token and the watcher's interpreter, unwatched here. + context = {'matrix': leg, 'needs': {'python-versions': {'outputs': live_outputs()}}, 'github': {'token': ''}, + 'steps': {'watcher': {'outcome': 'skipped'}, 'watcher-python': {'outputs': {'python-path': ''}}}} return {key: re.sub(r'\$\{\{.*?\}\}', lambda match: stringify(evaluate_expression(match.group(0), context)), str(value)) for key, value in (step(where).get('env') or {}).items()} @@ -294,18 +297,18 @@ def test_the_workflow_holds_every_experimental_leg_and_unit_tests_none(self) -> self.assertFalse([leg for leg in legs('unit-tests.yml', job_id) if is_experimental(leg)]) self.assertNotIn('experimental', str(spec.get('name'))) - def test_the_same_events_and_its_own_concurrency(self) -> None: - """unit-tests.yml's push and pull_request, no ``workflow_call``, read-only, a group of its own. + def test_the_same_events_and_no_concurrency_group(self) -> None: + """unit-tests.yml's push and pull_request, no ``workflow_call``, read-only, sharing no group. - Which runs it supersedes is #1903's, evaluated in - :mod:`tests.project.test_experimental_concurrency_1903`. + #1938 drops its group, so a superseded run is skipped or stopped, not + cancelled: :mod:`tests.project.test_superseded_runs_1938`. """ events = triggers(self.own) self.assertEqual(events, {key: value for key, value in triggers(self.unit).items() if key != 'workflow_call'}) self.assertEqual(set(events), {'push', 'pull_request'}) self.assertEqual(self.own['permissions'], {'contents': 'read'}) - self.assertNotEqual(self.own['concurrency']['group'], self.unit['concurrency']['group']) + self.assertNotIn('concurrency', self.own) def test_it_skips_a_docs_only_diff_as_test_does(self) -> None: """unit-tests.yml's ``changes`` script, byte for byte, gating the job as it gates ``test``.""" @@ -313,8 +316,13 @@ def classify(spec: 'dict[str, Any]') -> 'str': return next(entry['run'] for entry in spec['jobs']['changes']['steps'] if entry.get('id') == 'classify') self.assertEqual(classify(self.own), classify(self.unit)) spec = self.own['jobs']['experimental'] - self.assertEqual(spec['needs'], ['changes', 'python-versions']) - self.assertEqual(spec['if'], "${{ needs.changes.outputs.code != 'false' }}") + # #1938 adds `current`, the gate every job needs, and `!cancelled()`, so the two results stand in + # for the implicit `success()`. + self.assertEqual(spec['needs'], ['current', 'changes', 'python-versions']) + self.assertEqual(spec['if'], + "${{ !cancelled() && needs.changes.result == 'success' && " + "needs.python-versions.result == 'success' && needs.current.outputs.current != 'false' && " + "needs.changes.outputs.code != 'false' }}") self.assertIn("needs.changes.outputs.code != 'false'", self.unit['jobs']['test']['if']) def test_nothing_waits_on_it(self) -> None: diff --git a/util/superseded.py b/util/superseded.py new file mode 100644 index 000000000..4973d7fe4 --- /dev/null +++ b/util/superseded.py @@ -0,0 +1,433 @@ +# -*- coding: utf-8 -*- +"""Whether an Experimental Tests run's commit is still the one worth testing (#1938). + +GitHub counts a ``cancelled`` check against its commit in the status rollup, +so cancelling a superseded run (#1903) marks that commit with a red X, and a +merged pull request's last run, which no newer run supersedes, keeps its runner +for the whole GraalPy leg. So experimental.yml cancels nothing and asks this +script instead, which ends a superseded run green. + +A run's commit is **current** when + +* for a ``push`` to ``main``: the run's ``GITHUB_SHA`` is still the head of + ``main`` (``GET /repos/{repo}/commits/main``); +* for a ``pull_request``: the pull request is still open, and its head is still + the event's ``pull_request.head.sha`` (``GET /repos/{repo}/pulls/{number}``); +* for any other event: always. + +Anything the script cannot read -- an HTTP error, a timeout, a missing token, +an event file without a pull request -- counts as current, so a failing API +never stops a test run that would otherwise have finished. + +``check`` is the gate job: it writes ``current=true`` or ``current=false`` to +``$GITHUB_OUTPUT``, and where the commit is superseded a ``Superseded`` notice +and a job-summary line, so every later job can skip on ``if:``. It exits 0. + +``watch -- COMMAND...`` runs ``COMMAND`` in a session of its own and asks the +same question every ``--interval`` seconds while it runs. Once the commit is +superseded, it sends SIGTERM, then SIGKILL after ``--grace`` seconds, to the +session or, with ``--scope UNIT``, to every process of that systemd scope +(``sudo -n systemctl kill``): ``COMMAND`` then starts a root-owned ``sudo``, and +the tests run in the scope as the runner's user, out of reach of a signal to +the session's group (:func:`stop`). Where ``COMMAND`` exits, it writes the notice and the summary +line and exits 0. Otherwise it exits with ``COMMAND``'s own status, ``128 + N`` +for a signal ``N`` as a shell reports it, so a test failure on a current commit +reads as it did without the watcher. SIGINT, SIGTERM and SIGHUP sent to the +watcher are passed on to ``COMMAND``. + +Every wait is bounded. Where ``COMMAND`` outlives SIGKILL by ``--kill-wait`` +seconds, the watcher writes a ``Superseded run not stopped`` warning and exits +:data:`STOP_FAILED`, not 0: the tests may still be running, and holding the +step's output open, until their own ``timeout`` or the step's cap ends them, so +the step reports a failure -- a warning on these non-blocking legs -- rather +than claiming it stopped them. + +The API is read with conditional requests: a ``304 Not Modified`` answer to an +``If-None-Match`` request does not count against the primary rate limit +(docs.github.com, "Best practices for using the REST API"), and ``GITHUB_TOKEN`` +has a limit per repository that every concurrent leg's polling shares. + +It fetches with :mod:`requests`, as util/ci_timing.py does. Without it the +script still runs, and every request fails, so the commit counts as current: +the gate writes ``current=true``, and the watcher runs the tests to the end. + +""" + +from __future__ import annotations + +import argparse +import json +import os +import signal +import subprocess # nosec B404 +import sys +import time +from typing import TYPE_CHECKING + +try: + import requests +except ImportError: # pragma: no cover -- every test interpreter has it + requests = None # type: ignore[assignment] + +if TYPE_CHECKING: + from types import FrameType + from typing import IO, Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple + + #: One GET: the URL and request headers in, the status, headers and body out. + Transport = Callable[[str, Dict[str, str]], Tuple[int, Dict[str, str], bytes]] + +#: The REST API's root, where ``GITHUB_API_URL`` does not name one. +API = 'https://api.github.com' +#: The two answers. +CURRENT, SUPERSEDED = 'current', 'superseded' +#: Seconds between two checks while ``watch`` runs its command. +INTERVAL = 60.0 +#: Seconds between SIGTERM and SIGKILL once the commit is superseded. +GRACE = 30.0 +#: Seconds to wait for the command to exit after SIGKILL. +KILL_WAIT = 15.0 +#: Seconds one API request, or one ``systemctl kill``, may take before it counts as failed. +TIMEOUT = 15.0 +#: ``watch``'s status where the command outlives SIGKILL: sysexits.h's ``EX_SOFTWARE``, +#: which neither pytest (0-5) nor ``timeout`` (124-137) exits with. +STOP_FAILED = 70 +#: The signals ``watch`` passes on to its command. +FORWARDED = (signal.SIGINT, signal.SIGTERM, signal.SIGHUP) + + +class ApiError(Exception): + """The API gave no usable answer.""" + + +class Context: + """What a run tests: its event, ref, commit, repository and pull request. + + Args: + event: ``GITHUB_EVENT_NAME``. + ref: ``GITHUB_REF``. + sha: The commit under test -- ``GITHUB_SHA`` for a push, the + pull request's head for a ``pull_request`` run, whose + ``GITHUB_SHA`` is a merge commit. + repo: ``owner/name``. + number: The pull request's number, for a ``pull_request`` run. + + """ + + def __init__(self, event: 'str', ref: 'str', sha: 'str', repo: 'str', number: 'Optional[int]' = None) -> 'None': + self.event = event + self.ref = ref + self.sha = sha + self.repo = repo + self.number = number + + @classmethod + def from_env(cls, env: 'Mapping[str, str]') -> 'Context': + """The run's context, from the variables every step of a job has.""" + event = env.get('GITHUB_EVENT_NAME', '') + sha = env.get('GITHUB_SHA', '') + number = None # type: Optional[int] + if event == 'pull_request': + sha, number = '', None + try: + with open(env.get('GITHUB_EVENT_PATH', ''), encoding='utf-8') as file: + pull = json.load(file).get('pull_request') or {} + sha = str((pull.get('head') or {}).get('sha') or '') + number = int(pull['number']) if pull.get('number') else None + except (OSError, ValueError, TypeError, AttributeError): + pass + return cls(event, env.get('GITHUB_REF', ''), sha, env.get('GITHUB_REPOSITORY', ''), number) + + @property + def watched(self) -> 'bool': + """Whether a newer commit can supersede this run at all.""" + return (self.event == 'push' and self.ref == 'refs/heads/main') or self.event == 'pull_request' + + +class Verdict: + """:data:`CURRENT` or :data:`SUPERSEDED`, and why. + + ``error`` is set where the API could not answer, and the commit counts as + current for that reason alone. + + """ + + def __init__(self, state: 'str', reason: 'str', error: 'bool' = False) -> 'None': + self.state = state + self.reason = reason + self.error = error + + @property + def current(self) -> 'bool': + return self.state == CURRENT + + def __repr__(self) -> 'str': + return f'Verdict({self.state!r}, {self.reason!r}, error={self.error!r})' + + +class Api: + """GET requests to the REST API, each repeated with its last ``ETag``. + + Args: + token: Sent as a bearer token where not empty. + transport: Replaces the network, for tests. + base: The API's root. + + """ + + def __init__(self, token: 'str', transport: 'Optional[Transport]' = None, base: 'str' = API) -> 'None': + self.token = token + self.transport = transport or self._get + self.base = base.rstrip('/') + self._cache = {} # type: Dict[str, Tuple[str, Any]] + #: The connection :meth:`_get` reuses from one check to the next, opened on its first call. + self.session = None # type: Optional[requests.Session] + + def _get(self, url: 'str', headers: 'Dict[str, str]') -> 'Tuple[int, Dict[str, str], bytes]': + """The default transport: one GET, its status returned rather than raised, as a 304 is an answer.""" + if requests is None: + raise ApiError('requests is not installed') + if self.session is None: + self.session = requests.Session() + response = self.session.get(url, timeout=TIMEOUT, headers=headers) + return response.status_code, {k.lower(): v for k, v in response.headers.items()}, response.content + + def get(self, path: 'str') -> 'Any': + """``path``'s JSON, or the last copy where the API answers ``304 Not Modified``.""" + url = f'{self.base}/{path}' + headers = {'Accept': 'application/vnd.github+json', 'X-GitHub-Api-Version': '2022-11-28', + 'User-Agent': 'PyPCAPKit-superseded'} + if self.token: + headers['Authorization'] = f'Bearer {self.token}' + cached = self._cache.get(url) + if cached is not None: + headers['If-None-Match'] = cached[0] + status, response_headers, body = self.transport(url, headers) + if status == 304 and cached is not None: + return cached[1] + if status != 200: + raise ApiError(f'HTTP {status} from GET /{path}') + data = json.loads(body.decode('utf-8')) + etag = next((value for key, value in response_headers.items() if key.lower() == 'etag'), '') + if etag: + self._cache[url] = (etag, data) + return data + + +def check(context: 'Context', api: 'Api') -> 'Verdict': + """Whether ``context``'s commit is still current; an API failure is current.""" + short = context.sha[:9] + try: + if context.event == 'push' and context.ref == 'refs/heads/main': + head = str(api.get(f'repos/{context.repo}/commits/main')['sha']) + if head != context.sha: + return Verdict(SUPERSEDED, f'main is at {head[:9]}, no longer at {short}') + return Verdict(CURRENT, f'{short} is the head of main') + if context.event == 'pull_request': + if not (context.number and context.sha): + return Verdict(CURRENT, 'the event file names no pull request head, so nothing can supersede it') + pull = api.get(f'repos/{context.repo}/pulls/{context.number}') + if pull['state'] != 'open': + state = 'merged' if pull.get('merged') else str(pull['state']) + return Verdict(SUPERSEDED, f'pull request #{context.number} is {state}') + head = str(pull['head']['sha']) + if head != context.sha: + return Verdict(SUPERSEDED, f'pull request #{context.number} is at {head[:9]}, no longer at {short}') + return Verdict(CURRENT, f'{short} is the head of pull request #{context.number}') + return Verdict(CURRENT, f"a {context.event or 'unnamed'} event's run is never superseded") + except Exception as error: # pylint: disable=broad-except + # Whatever the API did, the tests go on: stopping them is the one thing a failure must not do. + return Verdict(CURRENT, f'could not ask the API ({type(error).__name__}: {error}), ' + 'so the commit counts as current', error=True) + + +def escape(message: 'str') -> 'str': + """``message`` as a workflow command's message must hold it.""" + return message.replace('%', '%25').replace('\r', '%0D').replace('\n', '%0A') + + +def report(verdict: 'Verdict', outcome: 'str', env: 'Mapping[str, str]', out: 'IO[str]') -> 'None': + """The ``Superseded`` notice, and its job-summary line where the job has a summary file.""" + print(f'::notice title=Superseded::{escape(verdict.reason)}. {escape(outcome)}', file=out, flush=True) + summary = env.get('GITHUB_STEP_SUMMARY') + if summary: + try: + with open(summary, 'a', encoding='utf-8') as file: + print(f'- Superseded: {verdict.reason}. {outcome}', file=file) + except OSError as error: + print(f'Could not write the job summary: {error}', file=out, flush=True) + + +def gate(context: 'Context', api: 'Api', env: 'Mapping[str, str]', out: 'IO[str]') -> 'int': + """The gate job: ``current=true`` or ``current=false``, in ``$GITHUB_OUTPUT``.""" + verdict = check(context, api) + print(f'{verdict.state}: {verdict.reason}', file=out, flush=True) + if not verdict.current: + report(verdict, 'The rest of this run is skipped, so none of its jobs fails or is cancelled.', env, out) + output = env.get('GITHUB_OUTPUT') + if output: + with open(output, 'a', encoding='utf-8') as file: + print(f'current={str(verdict.current).lower()}', file=file) + return 0 + + +def exit_status(returncode: 'int') -> 'int': + """``returncode`` as a shell reports it: ``128 + N`` for a process ended by signal ``N``.""" + return 128 - returncode if returncode < 0 else returncode + + +def signal_session(pid: 'int', signum: 'int') -> 'None': + """``signum`` to every process in ``pid``'s process group that will take it. + + A refusal is not an error here: under ``sudo`` the group's leader is root, + and the scope (:func:`stop`) is what reaches the tests there. + + """ + try: + os.killpg(pid, signum) + except OSError: + try: + os.kill(pid, signum) + except OSError: + pass + + +def run_command(argv: 'Sequence[str]') -> 'int': + """The default command runner: ``argv``'s exit status, ``-1`` where it could not run or finish in time.""" + try: + return subprocess.run(list(argv), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, # nosec B603 + stderr=subprocess.DEVNULL, timeout=TIMEOUT, check=False).returncode + except (OSError, subprocess.SubprocessError): + return -1 + + +def scope_kill(scope: 'str', signum: 'int') -> 'List[str]': + """The command that sends ``signum`` to every process in the systemd unit ``scope``.""" + return ['sudo', '-n', 'systemctl', 'kill', f'--signal={signal.Signals(signum).name}', scope] + + +def ended(child: 'subprocess.Popen[bytes]', timeout: 'float') -> 'bool': + """Whether ``child`` has exited within ``timeout`` seconds.""" + try: + child.wait(timeout=timeout) + except subprocess.TimeoutExpired: + return False + return True + + +def stop(child: 'subprocess.Popen[bytes]', grace: 'float', *, scope: 'str' = '', kill_wait: 'float' = KILL_WAIT, + run: 'Callable[[Sequence[str]], int]' = run_command, + kill: 'Callable[[int, int], None]' = signal_session) -> 'bool': + """SIGTERM, then SIGKILL ``grace`` seconds on; whether ``child`` exited, each wait bounded. + + Where the tests run in ``scope``, a systemd unit, each signal goes to every + process of it, through ``sudo``: they run as the runner's user under a + root-owned ``sudo``, and ``timeout`` puts them in a process group of its + own, so no signal to ``child``'s session reaches them. SIGTERM goes to the + scope alone there, since the session's shell would end at once and the + tests would lose their grace; SIGKILL goes to the session too. Where there + is no scope, ``child`` is ``timeout`` itself, its session's leader, and its + session is the tests. + + """ + for signum, wait in ((signal.SIGTERM, grace), (signal.SIGKILL, kill_wait)): + if scope: + run(scope_kill(scope, signum)) + if not scope or signum == signal.SIGKILL: + kill(child.pid, signum) + if ended(child, wait): + # A process that outlived the leader would hold the step's pipe open. + if scope: + run(scope_kill(scope, signal.SIGKILL)) + kill(child.pid, signal.SIGKILL) + return True + return False + + +def watch(command: 'Sequence[str]', context: 'Context', api: 'Api', env: 'Mapping[str, str]', out: 'IO[str]', *, + interval: 'float' = INTERVAL, grace: 'float' = GRACE, scope: 'str' = '', kill_wait: 'float' = KILL_WAIT, + run: 'Callable[[Sequence[str]], int]' = run_command, kill: 'Callable[[int, int], None]' = signal_session, + clock: 'Callable[[], float]' = time.monotonic) -> 'int': + """Run ``command``, stopping it once ``context``'s commit is superseded. + + :data:`STOP_FAILED` where it cannot be stopped: see the module's docstring. + + """ + started = clock() + # A session of its own, so a signal to the group reaches every process the command started. + child = subprocess.Popen(list(command), start_new_session=True) # nosec B603 pylint: disable=consider-using-with + + def forward(signum: 'int', frame: 'Optional[FrameType]') -> 'None': # pylint: disable=unused-argument + try: + os.kill(child.pid, signum) + except OSError: + pass + + previous = {signum: signal.signal(signum, forward) for signum in FORWARDED} + try: + last_error = '' + while True: + try: + return exit_status(child.wait(timeout=interval if context.watched else None)) + except subprocess.TimeoutExpired: + pass + verdict = check(context, api) + if verdict.error and verdict.reason != last_error: + print(f'util/superseded.py: {verdict.reason}', file=out, flush=True) + last_error = verdict.reason if verdict.error else '' + if verdict.current: + continue + if stop(child, grace, scope=scope, kill_wait=kill_wait, run=run, kill=kill): + elapsed = int(clock() - started) + report(verdict, f'The tests were stopped after {elapsed} s, and this step succeeds.', env, out) + return 0 + where = f'the scope {scope} and ' if scope else '' + reason = (f'{verdict.reason}, but SIGTERM and SIGKILL to {where}the session of pid {child.pid} ' + f'did not end it within {int(grace + kill_wait)} s. The tests may run on until their timeout.') + print(f'::warning title=Superseded run not stopped::{escape(reason)}', file=out, flush=True) + summary = env.get('GITHUB_STEP_SUMMARY') + if summary: + try: + with open(summary, 'a', encoding='utf-8') as file: + print(f'- Superseded run not stopped: {reason}', file=file) + except OSError: + pass + return STOP_FAILED + finally: + for signum, handler in previous.items(): + signal.signal(signum, handler) + + +def main(argv: 'Optional[List[str]]' = None, *, env: 'Optional[Mapping[str, str]]' = None, + transport: 'Optional[Transport]' = None, out: 'Optional[IO[str]]' = None, + run: 'Callable[[Sequence[str]], int]' = run_command, + kill: 'Callable[[int, int], None]' = signal_session) -> 'int': + """The command line; ``transport``, ``run`` and ``kill`` replace the network, commands and signals in tests.""" + parser = argparse.ArgumentParser(description=__doc__.split('\n', 1)[0]) + commands = parser.add_subparsers(dest='action') + commands.add_parser('check', help="the gate: write current=true or current=false to $GITHUB_OUTPUT") + watcher = commands.add_parser('watch', help='run a command, stopping it once the commit is superseded') + watcher.add_argument('--interval', type=float, default=INTERVAL, help='seconds between checks') + watcher.add_argument('--grace', type=float, default=GRACE, help='seconds from SIGTERM to SIGKILL') + watcher.add_argument('--kill-wait', type=float, default=KILL_WAIT, help='seconds to wait after SIGKILL') + watcher.add_argument('--scope', default='', help='the systemd scope the command runs its tests in, if any') + watcher.add_argument('command', nargs=argparse.REMAINDER, help='the command, after --') + args = parser.parse_args(argv) + + env = os.environ if env is None else env + out = sys.stdout if out is None else out + context = Context.from_env(env) + api = Api(env.get('GH_TOKEN') or env.get('GITHUB_TOKEN') or '', transport, env.get('GITHUB_API_URL') or API) + if args.action == 'check': + return gate(context, api, env, out) + if args.action == 'watch': + command = args.command[1:] if args.command[:1] == ['--'] else args.command + if not command: + parser.error('watch needs a command after --') + return watch(command, context, api, env, out, interval=args.interval, grace=args.grace, scope=args.scope, + kill_wait=args.kill_wait, run=run, kill=kill) + parser.error('name an action: check or watch') + return 2 # unreachable: parser.error exits + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/util/update_python_versions.py b/util/update_python_versions.py index ca8065439..7212aafad 100644 --- a/util/update_python_versions.py +++ b/util/update_python_versions.py @@ -553,7 +553,8 @@ def workflow_labels(workflows: 'dict[str, Any]') -> 'dict[str, list[str]]': ``supported`` ``default``, plus each label of a literal ``os`` axis (the conda jobs). ``latest`` - The ``runs-on`` of every job whose setup-python reads ``latest``. + The ``runs-on`` of every job whose setup-python reads ``latest``, or the + label its ``matrix.os || '...'`` falls back to. """ default, axis, latest = set(), set(), set() # type: set[str], set[str], set[str] @@ -567,9 +568,15 @@ def workflow_labels(workflows: 'dict[str, Any]') -> 'dict[str, list[str]]': if output is None and 'matrix.python-version' not in text: continue # a pin the tier file exempts if output == 'latest': - if '${{' in runs_on: + # #1938: experimental's legs set up `latest` beside their own version, on the + # label `matrix.os || '...'` falls back to, as `default` reads it. + match = DEFAULT_LABEL.fullmatch(runs_on) + if match: + latest.add(match.group(1)) + elif '${{' in runs_on: raise ValueError(f'{name} {job_id} installs latest on {runs_on!r}, which this cannot read') - latest.add(runs_on) + else: + latest.add(runs_on) continue match = DEFAULT_LABEL.fullmatch(runs_on) if match: