diff --git a/.github/mypyc_primer.toml b/.github/mypyc_primer.toml
new file mode 100644
index 0000000000000..20b887921faad
--- /dev/null
+++ b/.github/mypyc_primer.toml
@@ -0,0 +1,42 @@
+# Each entry is checked in its own Linux job. Refs are resolved once per run;
+# a target-branch comparison uses the same project revision and dependency versions.
+# setup/build/test are Bash commands, run in order from a fresh checkout.
+# The runner installs the selected mypy/mypyc after setup. Build commands must
+# preserve that installation (in particular, disable pip build isolation).
+# compiled_modules are checked before tests to prevent pure-Python false passes.
+# python_version selects the CI interpreter; local runs use the current interpreter.
+# [env] applies to all projects; [projects.env] overrides individual values.
+
+[env]
+CC = "clang"
+MYPYC_OPT_LEVEL = "0"
+MYPYC_DEBUG_LEVEL = "0"
+
+[[projects]]
+name = "black"
+repository = "https://github.com/psf/black.git"
+ref = "main"
+python_version = "3.14"
+compiled_modules = ["black"]
+setup = [
+ # Black's legacy macOS wheel tag is not supported on Apple Silicon.
+ '''
+python - <<'PY'
+import sys
+from pathlib import Path
+
+if sys.platform == "darwin":
+ path = Path("pyproject.toml")
+ path.write_text(path.read_text().replace("macos-max-compat = true", "macos-max-compat = false"))
+PY
+''',
+ 'python -m pip install "hatchling>=1.27.0" "hatch-vcs>=0.3.0" hatch-fancy-pypi-readme "hatch-mypyc>=0.16.0"',
+ 'python -m pip install --no-build-isolation ".[d,jupyter]" --group tests',
+]
+build = [
+ 'HATCH_BUILD_HOOKS_ENABLE=1 python -m pip wheel --no-build-isolation --no-deps --wheel-dir dist .',
+ 'python -m pip install --force-reinstall --no-deps dist/*.whl',
+]
+test = [
+ 'python -m pytest tests/test_black.py -k "not incompatible_with_mypyc" -n 2 -v --color=no',
+]
diff --git a/.github/scripts/mypyc_primer_comment.cjs b/.github/scripts/mypyc_primer_comment.cjs
new file mode 100644
index 0000000000000..98d1d19662d2a
--- /dev/null
+++ b/.github/scripts/mypyc_primer_comment.cjs
@@ -0,0 +1,266 @@
+// Executed exclusively from the default branch, with PR artifacts treated as data.
+const fs = require('node:fs');
+const os = require('node:os');
+const path = require('node:path');
+const { execFileSync } = require('node:child_process');
+
+const MARKER = '';
+const SHA = /^[0-9a-f]{40}$/;
+const NAME = /^[a-z0-9][a-z0-9_-]{0,63}$/;
+const MAX_JSON = 512 * 1024;
+const MAX_COMMENT = 50000;
+
+function validateManifest(manifest, run) {
+ if (manifest.version !== 1 || !Number.isSafeInteger(manifest.pr_number) ||
+ manifest.pr_number <= 0 || manifest.head_sha !== run.head_sha ||
+ !SHA.test(manifest.head_sha) || !SHA.test(manifest.base_sha) ||
+ typeof manifest.base_ref !== 'string' || !manifest.base_ref.length ||
+ manifest.base_ref.length > 1024 || /[\s\x00-\x1f\x7f]/.test(manifest.base_ref) ||
+ manifest.run_id !== String(run.id) || !Number.isSafeInteger(manifest.run_attempt) ||
+ manifest.run_attempt < 1 || manifest.run_attempt > run.run_attempt ||
+ !Array.isArray(manifest.projects) || !manifest.projects.length ||
+ manifest.projects.length > 100) {
+ throw new Error('Invalid mypyc primer manifest or workflow association');
+ }
+ const names = new Set();
+ for (const project of manifest.projects) {
+ if (typeof project.name !== 'string' || !NAME.test(project.name) ||
+ typeof project.python_version !== 'string' || !/^3\.\d+$/.test(project.python_version) ||
+ names.has(project.name)) {
+ throw new Error('Invalid or duplicate project in manifest');
+ }
+ names.add(project.name);
+ }
+}
+
+function validateResult(result, project, manifest) {
+ if (result.version !== 1 || result.name !== project.name ||
+ result.python_version !== project.python_version ||
+ result.head_sha !== manifest.head_sha || result.base_sha !== manifest.base_sha ||
+ !(result.project_sha === null || SHA.test(result.project_sha))) {
+ throw new Error(`Invalid result for ${project.name}`);
+ }
+ for (const attempt of [result.head, result.base]) {
+ if (attempt !== null && (typeof attempt !== 'object' ||
+ !['passed', 'failed', 'error'].includes(attempt.status) ||
+ !['setup', 'compiler', 'build', 'compiled-import', 'test'].includes(attempt.stage) ||
+ typeof attempt.tail !== 'string')) {
+ throw new Error(`Invalid attempt for ${project.name}`);
+ }
+ }
+ if (!result.head || (result.head.status === 'passed' &&
+ (result.base !== null || !SHA.test(result.project_sha)))) {
+ throw new Error(`Incomplete result for ${project.name}`);
+ }
+}
+
+function fencedTail(tail, limit) {
+ // ANSI/control sequences don't belong in a comment; the full Actions log is linked.
+ let text = tail.replace(/\x1b\[[0-?]*[ -/]*[@-~]/g, '')
+ .replace(/[\x00-\x08\x0b-\x1f\x7f]/g, '').split('\n').slice(-20).join('\n');
+ if (text.length > limit) text = '[excerpt truncated]\n' + text.slice(Math.max(0, text.length - limit));
+ const longest = Math.max(2, ...Array.from(text.matchAll(/`+/g), m => m[0].length));
+ const fence = '`'.repeat(longest + 1);
+ return `${fence}text\n${text}\n${fence}`;
+}
+
+function projectLogLinks(projects, jobs, runUrl, runAttempt = Infinity) {
+ const links = new Map();
+ for (const project of projects) {
+ const name = `Compile and test (${project.name}, ${project.python_version})`;
+ // Include older attempts: a failed-jobs rerun reuses other projects' results.
+ const job = jobs.filter(job => job.name === name && job.run_attempt <= runAttempt)
+ .sort((a, b) => b.run_attempt - a.run_attempt || b.id - a.id)[0];
+ if (!job) continue;
+ const attempts = {};
+ for (const [attempt, stepName] of [
+ ['head', 'Compile and test with PR mypyc'],
+ ['base', 'Compile and test with target branch mypyc'],
+ ]) {
+ const step = job.steps.find(step => step.name === stepName && step.conclusion !== 'skipped');
+ if (step) attempts[attempt] = `${runUrl}/job/${job.id}#step:${step.number}:1`;
+ }
+ links.set(project.name, attempts);
+ }
+ return links;
+}
+
+function renderComment(manifest, results, logs, runUrl, conclusion) {
+ const metadata = `${MARKER}`;
+ if (conclusion === 'success' && manifest.projects.length &&
+ manifest.projects.every(project => results.get(project.name)?.head.status === 'passed')) {
+ return `${metadata}mypyc primer: all configured projects compiled and passed their tests. โ
`;
+ }
+ const blocks = [metadata, '## mypyc primer detected failures โ'];
+ // Branch names can contain backticks; keep the entire name inside a code span.
+ const branchFence = '`'.repeat(Math.max(0,
+ ...Array.from((manifest.base_ref || '').matchAll(/`+/g), m => m[0].length)) + 1);
+ // Shared run information appears inside the first failure's details only.
+ const context = [
+ `PR commit: \`${manifest.head_sha}\``,
+ ...(manifest.base_ref ? [`Target branch: ${branchFence} ${manifest.base_ref} ${branchFence}`] : []),
+ ...(manifest.base_sha ? [`Target branch commit: \`${manifest.base_sha}\``] : []),
+ `[Workflow run](${runUrl})`,
+ ...(conclusion !== 'success' ? [`Workflow conclusion: ${conclusion}`] : []),
+ ];
+ // Leave space for all project headings and links even with a large catalog.
+ const excerpts = new Map();
+ for (const project of manifest.projects) {
+ const result = results.get(project.name);
+ if (result?.head.status === 'passed') continue;
+ const log = logs.get(project.name);
+ const prLink = log?.head ? `[PR log](${log.head})` : null;
+ const baseLink = log?.base ? `[Target branch log](${log.base})` : null;
+ const details = [`Python: ${project.python_version}`,
+ ...(result?.project_sha ? [`Project revision: \`${result.project_sha}\``] : [])];
+ let heading;
+ if (!result || result.head.status === 'error' || result.base?.status === 'error' ||
+ (result.head.status === 'failed' && (!result.base || !prLink || !baseLink))) {
+ const base = result?.base?.status;
+ const baseStatus = base === 'passed' ? 'target branch succeeded'
+ : base === 'failed' ? 'target branch failed' : 'target branch result unknown';
+ heading = `Check incomplete (${baseStatus})`;
+ details.push('Compilation and test results could not be confirmed.');
+ const links = [prLink, baseLink].filter(Boolean);
+ if (links.length) details.push(links.join(' ยท '));
+ } else if (result.base.status === 'failed') {
+ heading = 'Failed both on PR and target branch';
+ details.push(`${prLink} ยท ${baseLink}`);
+ } else {
+ heading = 'Failed on PR and succeeded on target branch';
+ const placeholder = ``;
+ excerpts.set(placeholder, result.head.tail);
+ details.push(prLink, `Last 20 lines of the PR log:\n\n${placeholder}`);
+ }
+ details.push(...context.splice(0));
+ blocks.push(`### ${project.name}: ${heading}\n\n\nDetails
\n\n${details.join('\n\n')}\n\n `);
+ }
+ if (blocks.length === 2) {
+ blocks.push(`### CI checks did not complete\n\n\nDetails
\n\nCompilation and test results could not be confirmed.\n\n${context.join('\n\n')}\n\n `);
+ }
+ let body = blocks.join('\n\n');
+ // A fence may be as long as its contents, so reserve up to three characters
+ // of Markdown for each excerpt character. Headings and log links stay intact.
+ const limit = Math.min(6000, Math.max(0,
+ Math.floor((MAX_COMMENT - body.length - 100 * excerpts.size) / (3 * excerpts.size))));
+ body = body.replace(//g,
+ placeholder => fencedTail(excerpts.get(placeholder), limit));
+ if (body.length > MAX_COMMENT) throw new Error('Result comment exceeds GitHub size limit');
+ return body;
+}
+
+async function readArtifact(github, repo, artifact, filename, directory) {
+ if (!artifact || artifact.expired || artifact.size_in_bytes > MAX_JSON) {
+ throw new Error(`Missing, expired, or oversized ${filename} artifact`);
+ }
+ const download = await github.rest.actions.downloadArtifact({
+ ...repo, artifact_id: artifact.id, archive_format: 'zip',
+ });
+ const data = Buffer.from(download.data);
+ if (data.length > MAX_JSON) throw new Error('Artifact download exceeds size limit');
+ const archive = path.join(directory, `${artifact.id}.zip`);
+ fs.writeFileSync(archive, data);
+ // Read just the expected JSON member. No archive paths are extracted to disk.
+ return JSON.parse(execFileSync('unzip', ['-p', archive, filename], {
+ encoding: 'utf8', maxBuffer: MAX_JSON, timeout: 10000,
+ }));
+}
+
+async function associatedPullRequests(github, repo, run) {
+ if (run.pull_requests.length) return run.pull_requests;
+ // Fork runs can omit this array, and unmerged fork commits may have no
+ // commit-to-PR association. Resolve the trusted run's repository and branch.
+ if (!run.head_repository) return [];
+ const prs = await github.paginate(github.rest.pulls.list, {
+ ...repo, state: 'all',
+ head: `${run.head_repository.owner.login}:${run.head_branch}`, per_page: 100,
+ });
+ return prs.filter(pr => pr.head.repo?.id === run.head_repository.id &&
+ pr.head.ref === run.head_branch);
+}
+
+async function publish({ github, context }, prNumber, body) {
+ const repo = context.repo;
+ const comments = await github.paginate(github.rest.issues.listComments, {
+ ...repo, issue_number: prNumber, per_page: 100,
+ });
+ const runMarker = body.match(//)?.[0];
+ if (!runMarker) throw new Error('Missing workflow run marker in comment');
+ if (comments.some(c => c.user.type === 'Bot' && c.body?.startsWith(MARKER) &&
+ c.body.includes(runMarker))) return;
+ const { data: posted } = await github.rest.issues.createComment({
+ ...repo, issue_number: prNumber, body,
+ });
+ for (const previous of comments) {
+ if (previous.user.type === 'Bot' && previous.user.login === posted.user.login &&
+ previous.body?.startsWith(MARKER)) {
+ await github.graphql(`mutation($id: ID!) {
+ minimizeComment(input: {subjectId: $id, classifier: OUTDATED}) {
+ minimizedComment { isMinimized }
+ }
+ }`, { id: previous.node_id });
+ }
+ }
+}
+
+async function comment({ github, context, core }) {
+ const run = context.payload.workflow_run;
+ const repo = context.repo;
+ if (run.event !== 'pull_request' || run.conclusion === 'cancelled') return;
+ const artifacts = new Map((await github.paginate(github.rest.actions.listWorkflowRunArtifacts, {
+ ...repo, run_id: run.id, per_page: 100,
+ })).filter(a => !a.expired).map(a => [a.name, a]));
+ if (!artifacts.has('mypyc-primer-manifest') && run.conclusion === 'success') {
+ // Eligibility discovery intentionally skipped this PR.
+ return;
+ }
+ const associated = await associatedPullRequests(github, repo, run);
+ const directory = fs.mkdtempSync(path.join(process.env.RUNNER_TEMP || os.tmpdir(), 'mypyc-primer-'));
+ try {
+ let manifest;
+ if (artifacts.has('mypyc-primer-manifest')) {
+ manifest = await readArtifact(github, repo, artifacts.get('mypyc-primer-manifest'), 'manifest.json', directory);
+ validateManifest(manifest, run);
+ // A failed-jobs rerun reuses the original discovery job and snapshot.
+ manifest.run_attempt = run.run_attempt;
+ } else {
+ if (associated.length !== 1) throw new Error('Cannot associate incomplete run with a PR');
+ manifest = { pr_number: associated[0].number, head_sha: run.head_sha,
+ run_id: String(run.id), run_attempt: run.run_attempt, projects: [] };
+ }
+ if (!associated.some(pr => pr.number === manifest.pr_number)) {
+ throw new Error('Manifest PR does not belong to the triggering workflow run');
+ }
+ const { data: pr } = await github.rest.pulls.get({ ...repo, pull_number: manifest.pr_number });
+ if (pr.state !== 'open' || pr.head.sha !== manifest.head_sha ||
+ (manifest.base_ref && pr.base.ref !== manifest.base_ref)) {
+ core.info('Skipping closed PR or results for an older PR head or target branch');
+ return;
+ }
+ const runUrl = `https://github.com/${repo.owner}/${repo.repo}/actions/runs/${run.id}`;
+ const jobs = manifest.projects.length ? await github.paginate(github.rest.actions.listJobsForWorkflowRun, {
+ ...repo, run_id: run.id, filter: 'all', per_page: 100,
+ }) : [];
+ const logs = projectLogLinks(manifest.projects, jobs, runUrl, run.run_attempt);
+ const results = new Map();
+ for (const project of manifest.projects) {
+ try {
+ const result = await readArtifact(github, repo, artifacts.get(`mypyc-primer-result-${project.name}`), `${project.name}.json`, directory);
+ validateResult(result, project, manifest);
+ results.set(project.name, result);
+ } catch (error) {
+ core.warning(`${project.name}: ${error.message}`);
+ }
+ }
+ const body = renderComment(manifest, results, logs, runUrl, run.conclusion);
+ // Reading a large catalog can take time; recheck for new commits or retargeting.
+ const { data: current } = await github.rest.pulls.get({ ...repo, pull_number: manifest.pr_number });
+ if (current.state !== 'open' || current.head.sha !== manifest.head_sha ||
+ (manifest.base_ref && current.base.ref !== manifest.base_ref)) return;
+ await publish({ github, context }, manifest.pr_number, body);
+ } finally {
+ fs.rmSync(directory, { recursive: true, force: true });
+ }
+}
+
+module.exports = { comment, publish, projectLogLinks, renderComment, fencedTail, validateManifest, validateResult };
diff --git a/.github/workflows/mypyc_primer.yml b/.github/workflows/mypyc_primer.yml
new file mode 100644
index 0000000000000..9e7ca8e4860cf
--- /dev/null
+++ b/.github/workflows/mypyc_primer.yml
@@ -0,0 +1,93 @@
+name: Run mypyc_primer
+
+on:
+ # Inspect the entire PR diff ourselves: GitHub path filters examine at most
+ # 300 files, which could miss a compiler change in a large PR.
+ pull_request:
+
+permissions:
+ contents: read
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.event.pull_request.number }}
+ cancel-in-progress: true
+
+jobs:
+ prepare:
+ name: Select mypyc projects
+ runs-on: ubuntu-24.04
+ timeout-minutes: 10
+ outputs:
+ eligible: ${{ steps.prepare.outputs.eligible }}
+ matrix: ${{ steps.prepare.outputs.matrix }}
+ base_sha: ${{ steps.prepare.outputs.base_sha }}
+ steps:
+ - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
+ with:
+ ref: ${{ github.sha }}
+ fetch-depth: 0
+ persist-credentials: false
+ - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6
+ with:
+ python-version: '3.14'
+ - name: Select projects and snapshot target branch
+ id: prepare
+ run: python misc/mypyc_primer.py prepare --output-dir primer
+ - name: Check PR reporter
+ if: steps.prepare.outputs.eligible == 'true'
+ run: node --check .github/scripts/mypyc_primer_comment.cjs
+ - name: Upload expected results
+ if: steps.prepare.outputs.eligible == 'true'
+ uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
+ with:
+ name: mypyc-primer-manifest
+ path: primer/manifest.json
+ if-no-files-found: error
+ overwrite: true
+
+ check:
+ name: Compile and test
+ needs: prepare
+ if: needs.prepare.outputs.eligible == 'true'
+ runs-on: ubuntu-24.04
+ timeout-minutes: 60
+ strategy:
+ fail-fast: false
+ matrix: ${{ fromJSON(needs.prepare.outputs.matrix) }}
+ env:
+ PROJECT: ${{ matrix.name }}
+ HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+ BASE_SHA: ${{ needs.prepare.outputs.base_sha }}
+ steps:
+ - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
+ with:
+ ref: ${{ github.sha }}
+ # Keep both parents so local compiler fetches resolve HEAD_SHA and BASE_SHA.
+ fetch-depth: 2
+ persist-credentials: false
+ - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6
+ with:
+ python-version: ${{ matrix.python_version }}
+ - name: Compile and test with PR mypyc
+ id: pr
+ run: |
+ python misc/mypyc_primer.py check \
+ --project "$PROJECT" --attempt head \
+ --head-sha "$HEAD_SHA" --base-sha "$BASE_SHA" \
+ --output-dir primer --work-dir "$RUNNER_TEMP"
+ - name: Compile and test with target branch mypyc
+ id: base
+ if: steps.pr.outputs.compare_base == 'true'
+ run: |
+ python misc/mypyc_primer.py check \
+ --project "$PROJECT" --attempt base \
+ --head-sha "$HEAD_SHA" --base-sha "$BASE_SHA" \
+ --output-dir primer --work-dir "$RUNNER_TEMP"
+ - name: Upload project result
+ if: ${{ always() && steps.pr.outcome != 'skipped' }}
+ uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
+ with:
+ name: mypyc-primer-result-${{ matrix.name }}
+ path: primer/${{ matrix.name }}.json
+ if-no-files-found: error
+ overwrite: true
diff --git a/.github/workflows/mypyc_primer_comment.yml b/.github/workflows/mypyc_primer_comment.yml
new file mode 100644
index 0000000000000..3634487c79834
--- /dev/null
+++ b/.github/workflows/mypyc_primer_comment.yml
@@ -0,0 +1,37 @@
+name: Comment with mypyc_primer results
+
+# Only the default-branch reporter runs with write permission. Artifacts from
+# the PR workflow are parsed as bounded JSON, never extracted or executed.
+on: # zizmor: ignore[dangerous-triggers]
+ workflow_run:
+ workflows: [Run mypyc_primer]
+ types: [completed]
+
+permissions: {}
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.event.workflow_run.id }}
+ cancel-in-progress: false
+
+jobs:
+ comment:
+ name: Comment PR with downstream compilation results
+ if: >-
+ github.event.workflow_run.event == 'pull_request' &&
+ github.event.workflow_run.conclusion != 'cancelled'
+ runs-on: ubuntu-24.04
+ timeout-minutes: 10
+ permissions:
+ actions: read
+ contents: read
+ pull-requests: write
+ steps:
+ - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
+ with:
+ ref: ${{ github.event.repository.default_branch }}
+ persist-credentials: false
+ - uses: actions/github-script@ed597411d8f924073f98dfc5c65a23a2325f34cd # v8
+ with:
+ script: |
+ const { comment } = require('./.github/scripts/mypyc_primer_comment.cjs');
+ await comment({ github, context, core });
diff --git a/MANIFEST.in b/MANIFEST.in
index b77c504fe2f9e..f6d53e662270c 100644
--- a/MANIFEST.in
+++ b/MANIFEST.in
@@ -46,6 +46,8 @@ include mypy_self_check.ini
prune misc
include misc/diff-cache.py
include misc/apply-cache-diff.py
+include misc/mypyc_primer.py
+include .github/mypyc_primer.toml
graft test-data
graft mypy/test
include conftest.py
diff --git a/misc/mypyc_primer.py b/misc/mypyc_primer.py
new file mode 100644
index 0000000000000..380269c394a8d
--- /dev/null
+++ b/misc/mypyc_primer.py
@@ -0,0 +1,653 @@
+"""Compile downstream projects with a PR's mypyc, comparing failures with its target branch.
+
+See .github/mypyc_primer.toml for the project recipes. This runner intentionally
+returns success for completed comparisons, including confirmed regressions;
+infrastructure errors return failure. A separate trusted workflow posts results.
+
+The script can also be used to validate changes locally, eg.
+ python misc/mypyc_primer.py check --project --head-sha
+ --base-sha --output-dir
+It will compile and test the project first using mypyc pinned to .
+If it fails, it will repeat the process with mypyc pinned to so
+a failure introduced by the change can be distinguished from upstream/infra failures.
+
+The second run can be skipped by passing "--attempt head" on the command line;
+in that case --base-sha can be omitted.
+Since it's meant for CI it prints a lot of logs to stdout, so for local testing it's best
+to redirect it to a file.
+"""
+
+from __future__ import annotations
+
+import argparse
+import codecs
+import json
+import os
+import re
+import selectors
+import shlex
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+from collections import deque
+from contextlib import nullcontext
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+from typing import Any, TypedDict
+
+if sys.version_info >= (3, 11):
+ import tomllib
+else:
+ import tomli as tomllib
+
+CONFIG = Path(__file__).resolve().parents[1] / ".github/mypyc_primer.toml"
+ATTEMPT_TIMEOUT = 20 * 60
+CLONE_TIMEOUT = 5 * 60
+SHA_PATTERN = re.compile(r"[0-9a-f]{40}")
+NAME_PATTERN = re.compile(r"[a-z0-9][a-z0-9_-]{0,63}")
+CI_FILES = {
+ ".github/mypyc_primer.toml",
+ ".github/workflows/mypyc_primer.yml",
+ ".github/workflows/mypyc_primer_comment.yml",
+ "misc/mypyc_primer.py",
+}
+
+
+@dataclass
+class Project:
+ name: str
+ repository: str
+ ref: str
+ python_version: str
+ compiled_modules: list[str]
+ setup: list[str]
+ build: list[str]
+ test: list[str]
+ env: dict[str, str] = field(default_factory=dict)
+
+
+@dataclass
+class Attempt:
+ status: str
+ stage: str
+ returncode: int | None = None
+ tail: str = ""
+
+
+class CommandError(Exception):
+ def __init__(self, returncode: int, message: str) -> None:
+ super().__init__(message)
+ self.returncode = returncode
+
+
+class Log:
+ """Stream output to Actions, retaining only a bounded failure excerpt."""
+
+ def __init__(self) -> None:
+ self.lines: deque[str] = deque(maxlen=20)
+
+ def write(self, text: str) -> None:
+ sys.stdout.write(text)
+ sys.stdout.flush()
+ if self.lines and not self.lines[-1].endswith("\n"):
+ text = self.lines.pop() + text
+ # Keep enough of each line to detect truncation, including its newline.
+ self.lines.extend(line[-20002:] for line in text.splitlines(keepends=True))
+
+ def tail(self) -> str:
+ tail = "".join(self.lines).rstrip("\n")
+ # Bound the JSON artifact even if a tool emits a very long final line.
+ return tail if len(tail) <= 20000 else "[excerpt truncated]\n" + tail[-20000:]
+
+
+class CommandOptions(TypedDict):
+ env: dict[str, str]
+ log: Log
+ deadline: float
+
+
+class AttemptOptions(TypedDict):
+ project_source: Path
+ project_sha: str
+ constraints: Path
+ timeout: float
+
+
+def validate_env(env: object, name: str) -> dict[str, str]:
+ if not isinstance(env, dict) or any(
+ not isinstance(key, str)
+ or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", key)
+ or not isinstance(value, str)
+ or key in {"PATH", "PYTHONPATH", "VIRTUAL_ENV", "PIP_CONSTRAINT", "MYPY_USE_MYPYC"}
+ for key, value in env.items()
+ ):
+ raise ValueError(f"{name}: invalid or reserved environment variable")
+ return env
+
+
+def load_projects(path: Path) -> list[Project]:
+ with path.open("rb") as f:
+ config = tomllib.load(f)
+ entries = config.get("projects")
+ if not isinstance(entries, list) or not 1 <= len(entries) <= 100:
+ raise ValueError("Catalog must contain 1 to 100 projects")
+ shared_env = validate_env(config.get("env", {}), "[env]")
+ projects = []
+ names: set[str] = set()
+ for entry in entries:
+ if not isinstance(entry, dict):
+ raise ValueError("Each project must be a TOML table")
+ name = entry.get("name")
+ if not isinstance(name, str) or not NAME_PATTERN.fullmatch(name):
+ raise ValueError("Project names must be short lowercase artifact-safe identifiers")
+ if name in names:
+ raise ValueError(f"Duplicate project: {name}")
+ names.add(name)
+ for key in ("repository", "ref", "python_version"):
+ value = entry.get(key)
+ if not isinstance(value, str) or not value or value.startswith("-"):
+ raise ValueError(f"{name}: invalid {key}")
+ if not entry["repository"].startswith("https://"):
+ raise ValueError(f"{name}: repository must use HTTPS")
+ if not re.fullmatch(r"3\.\d+", entry["python_version"]):
+ raise ValueError(f"{name}: python_version must be a CPython major.minor")
+ for key in ("setup", "build", "test", "compiled_modules"):
+ value = entry.get(key)
+ if not isinstance(value, list) or not value:
+ raise ValueError(f"{name}: {key} must be a nonempty list")
+ if any(not isinstance(item, str) or not item.strip() for item in value):
+ raise ValueError(f"{name}: {key} must contain nonempty strings")
+ env = validate_env(entry.get("env", {}), name)
+ entry["env"] = {**shared_env, **env}
+ projects.append(Project(**entry))
+ return projects
+
+
+def eligible_path(path: str) -> bool:
+ if path in CI_FILES or path.startswith(".github/scripts/mypyc_primer_comment"):
+ return True
+ return (
+ path.startswith("mypyc/")
+ and not path.startswith(
+ ("mypyc/doc/", "mypyc/test/", "mypyc/test-data/", "mypyc/external/googletest/")
+ )
+ and path != "mypyc/lib-rt/test_capi.cc"
+ and not path.endswith((".md", ".rst"))
+ )
+
+
+def changed_paths(source: Path, base: str, head: str) -> list[str]:
+ # --no-renames preserves both paths of a rename across the eligibility boundary.
+ data = subprocess.check_output(
+ ["git", "diff", "--name-only", "--no-renames", "-z", f"{base}...{head}"], cwd=source
+ )
+ return [os.fsdecode(path) for path in data.split(b"\0") if path]
+
+
+def write_json(path: Path, value: object) -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(json.dumps(value, indent=2) + "\n", encoding="utf8")
+
+
+def outputs(values: dict[str, str]) -> None:
+ output = os.environ.get("GITHUB_OUTPUT")
+ if output:
+ with open(output, "a", encoding="utf8") as f:
+ for key, value in values.items():
+ f.write(f"{key}={value}\n")
+
+
+def prepare(args: argparse.Namespace) -> None:
+ projects = load_projects(args.config)
+ event = json.loads(Path(os.environ["GITHUB_EVENT_PATH"]).read_text())
+ pr = event["pull_request"]
+ head = pr["head"]["sha"]
+ if not any(
+ eligible_path(path) for path in changed_paths(args.mypy_source, pr["base"]["sha"], head)
+ ):
+ outputs({"eligible": "false", "matrix": '{"include":[]}'})
+ return
+ base = pr["base"]["sha"]
+ if not SHA_PATTERN.fullmatch(base):
+ raise ValueError("Could not resolve target branch")
+ matrix = [{"name": p.name, "python_version": p.python_version} for p in projects]
+ write_json(
+ args.output_dir / "manifest.json",
+ {
+ "version": 1,
+ "pr_number": pr["number"],
+ "head_sha": head,
+ "base_sha": base,
+ "base_ref": pr["base"]["ref"],
+ "run_id": os.environ["GITHUB_RUN_ID"],
+ "run_attempt": int(os.environ["GITHUB_RUN_ATTEMPT"]),
+ "projects": matrix,
+ },
+ )
+ outputs({"eligible": "true", "matrix": json.dumps({"include": matrix}), "base_sha": base})
+
+
+def execute(
+ command: list[str],
+ *,
+ cwd: Path,
+ env: dict[str, str],
+ log: Log,
+ deadline: float,
+ capture: bool = False,
+) -> str:
+ log.write(f"\n$ {shlex.join(command)}\n")
+ captured: list[str] = []
+ with (
+ subprocess.Popen(
+ command,
+ cwd=cwd,
+ env=env,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE if capture else subprocess.STDOUT,
+ start_new_session=True,
+ ) as process,
+ selectors.DefaultSelector() as selector,
+ ):
+ for pipe in (process.stdout, process.stderr):
+ if pipe is not None:
+ selector.register(
+ pipe,
+ selectors.EVENT_READ,
+ codecs.getincrementaldecoder("utf8")(errors="replace"),
+ )
+
+ def stream(deadline: float | None) -> None:
+ while selector.get_map():
+ timeout = None if deadline is None else deadline - time.monotonic()
+ if timeout is not None and timeout <= 0:
+ raise subprocess.TimeoutExpired(command, 0)
+ for key, _ in selector.select(timeout):
+ data = os.read(key.fd, 8192)
+ text = key.data.decode(data, final=not data)
+ log.write(text)
+ if capture and key.fileobj is process.stdout:
+ captured.append(text)
+ if not data:
+ selector.unregister(key.fileobj)
+
+ try:
+ stream(deadline)
+ process.wait(timeout=max(0, deadline - time.monotonic()))
+ except subprocess.TimeoutExpired:
+ # Kill compiler/pytest children too, leaving enough time for the comparison.
+ if sys.platform != "win32":
+ os.killpg(process.pid, signal.SIGKILL)
+ else:
+ process.kill()
+ stream(None)
+ process.wait()
+ log.write("\nAttempt exceeded its time limit.\n")
+ raise CommandError(124, "Attempt exceeded its time limit") from None
+ log.write(f"\nExit code: {process.returncode}\n")
+ if process.returncode:
+ raise CommandError(process.returncode, f"Command exited with {process.returncode}")
+ return "".join(captured)
+
+
+def clone_checkout(
+ repository: str,
+ revision: str,
+ destination: Path,
+ *,
+ cwd: Path,
+ env: dict[str, str],
+ log: Log,
+ deadline: float,
+) -> None:
+ options: CommandOptions = {"env": env, "log": log, "deadline": deadline}
+ # Fetch the pinned commit only, including for local repositories where
+ # `git clone --depth=1` would otherwise ignore the depth limit.
+ execute(["git", "init", "--quiet", str(destination)], cwd=cwd, **options)
+ execute(
+ ["git", "fetch", "--quiet", "--depth=1", "--no-tags", "--", repository, revision],
+ cwd=destination,
+ **options,
+ )
+ execute(["git", "checkout", "--quiet", "--detach", "FETCH_HEAD"], cwd=destination, **options)
+
+
+def environment(project: Project, venv: Path) -> dict[str, str]:
+ env = os.environ.copy()
+ for key in (
+ "PYTHONPATH",
+ "PYTHONHOME",
+ "VIRTUAL_ENV",
+ "PIP_CONSTRAINT",
+ "PIP_BUILD_CONSTRAINT",
+ "HATCH_BUILD_HOOKS_ENABLE",
+ "MYPY_USE_MYPYC",
+ "NO_COLOR",
+ ):
+ env.pop(key, None)
+ env.update(project.env)
+ env.update(
+ PATH=str(venv / "bin") + os.pathsep + env["PATH"],
+ VIRTUAL_ENV=str(venv),
+ MYPY_USE_MYPYC="0",
+ PYTHONNOUSERSITE="1",
+ PYTHONUNBUFFERED="1",
+ PIP_DISABLE_PIP_VERSION_CHECK="1",
+ )
+ return env
+
+
+def snapshot_dependencies(data: str, path: Path, project: Project) -> None:
+ # Direct references (including the project checkout) must not leak between attempts.
+ excluded = {project.name.replace("_", "-"), "mypy", "librt"}
+ lines = []
+ for line in data.splitlines():
+ name, separator, _ = line.partition("==")
+ if separator and name.lower().replace("_", "-") not in excluded:
+ lines.append(line)
+ path.write_text("\n".join(lines) + "\n", encoding="utf8")
+
+
+def run_attempt(
+ project: Project,
+ *,
+ project_source: Path,
+ project_sha: str,
+ compiler_repository: str,
+ compiler_sha: str,
+ directory: Path,
+ log: Log,
+ constraints: Path,
+ baseline: bool,
+ timeout: float = ATTEMPT_TIMEOUT,
+) -> Attempt:
+ directory.mkdir()
+ venv = directory / "venv"
+ source = directory / "project"
+ compiler = directory / "compiler"
+ env = environment(project, venv)
+ if baseline and constraints.exists():
+ env["PIP_CONSTRAINT"] = str(constraints)
+ deadline = time.monotonic() + timeout
+ stage = "setup"
+ status = "passed"
+ returncode = None
+ log.write(
+ f"{project.name}: checking compiler {compiler_sha} ({'target branch' if baseline else 'PR'})\n"
+ )
+ log.write(
+ f"Project: {project.name} {project_sha}\nCompiler: {compiler_sha}\nPython: {sys.version}\n"
+ )
+ kwargs: CommandOptions = {"env": env, "log": log, "deadline": deadline}
+ try:
+ clone_checkout(str(project_source), project_sha, source, cwd=directory, **kwargs)
+ clone_checkout(compiler_repository, compiler_sha, compiler, cwd=directory, **kwargs)
+ execute([sys.executable, "-m", "venv", str(venv)], cwd=source, **kwargs)
+ execute(
+ ["python", "-m", "pip", "install", "--upgrade", "pip", "setuptools", "wheel"],
+ cwd=source,
+ **kwargs,
+ )
+ for command in project.setup:
+ execute(["bash", "-euo", "pipefail", "-c", command], cwd=source, **kwargs)
+ if not baseline:
+ snapshot_dependencies(
+ execute(
+ ["python", "-m", "pip", "freeze", "--all"], cwd=source, capture=True, **kwargs
+ ),
+ constraints,
+ project,
+ )
+ stage = "compiler"
+ execute(
+ ["python", "-m", "pip", "install", "-r", str(compiler / "mypy-requirements.txt")],
+ cwd=source,
+ **kwargs,
+ )
+ execute(
+ [
+ "python",
+ "-m",
+ "pip",
+ "install",
+ "--no-build-isolation",
+ "--force-reinstall",
+ "--no-deps",
+ str(compiler / "mypyc/lib-rt"),
+ str(compiler),
+ ],
+ cwd=source,
+ **kwargs,
+ )
+ verify_compiler = (
+ "import importlib.metadata, json, pathlib, sys, mypyc; "
+ "d = importlib.metadata.distribution('mypy'); "
+ "u = json.loads(d.read_text('direct_url.json')); "
+ "assert u['url'] == pathlib.Path(sys.argv[1]).as_uri(), u; "
+ "assert pathlib.Path(mypyc.__file__).is_relative_to(sys.prefix), mypyc.__file__; "
+ "print('Compiler installed from', u['url'])"
+ )
+ execute(["python", "-c", verify_compiler, str(compiler)], cwd=source, **kwargs)
+ if not baseline:
+ snapshot_dependencies(
+ execute(
+ ["python", "-m", "pip", "freeze", "--all"], cwd=source, capture=True, **kwargs
+ ),
+ constraints,
+ project,
+ )
+ stage = "build"
+ log.write(f"\n{project.name}: compiling\n")
+ for command in project.build:
+ execute(["bash", "-euo", "pipefail", "-c", command], cwd=source, **kwargs)
+ # Detect a recipe that silently replaces the selected compiler while building.
+ execute(["python", "-c", verify_compiler, str(compiler)], cwd=source, **kwargs)
+ stage = "compiled-import"
+ verify_modules = (
+ "import importlib, importlib.machinery, pathlib, sys; "
+ "modules = [importlib.import_module(n) for n in sys.argv[1:]]; "
+ "assert all(m.__file__ and any(m.__file__.endswith(s) for s in "
+ "importlib.machinery.EXTENSION_SUFFIXES) and "
+ "pathlib.Path(m.__file__).is_relative_to(sys.prefix) for m in modules), "
+ "[(m.__name__, m.__file__) for m in modules]; "
+ "print('Compiled imports:', [(m.__name__, m.__file__) for m in modules])"
+ )
+ execute(["python", "-c", verify_modules, *project.compiled_modules], cwd=source, **kwargs)
+ stage = "test"
+ log.write(f"\n{project.name}: testing\n")
+ for command in project.test:
+ execute(["bash", "-euo", "pipefail", "-c", command], cwd=source, **kwargs)
+ except (CommandError, OSError) as error:
+ status = "error" if stage == "setup" or isinstance(error, OSError) else "failed"
+ returncode = error.returncode if isinstance(error, CommandError) else None
+ log.write(f"\n{stage}: {error}\n")
+ return Attempt(status, stage, returncode, log.tail())
+
+
+def compare(
+ project: Project,
+ *,
+ output_dir: Path,
+ work_dir: Path,
+ mypy_source: Path,
+ head_sha: str | None = None,
+ base_sha: str | None = None,
+ timeout: float = ATTEMPT_TIMEOUT,
+ attempt: str = "both",
+) -> dict[str, Any]:
+ output_dir.mkdir(parents=True, exist_ok=True)
+ python_version = f"{sys.version_info.major}.{sys.version_info.minor}"
+ result: dict[str, Any] = {
+ "version": 1,
+ "name": project.name,
+ "python_version": python_version,
+ "head_sha": head_sha,
+ "base_sha": base_sha,
+ "project_sha": None,
+ "head": None,
+ "base": None,
+ }
+ result_path = output_dir / f"{project.name}.json"
+ if attempt == "base":
+ result = json.loads(result_path.read_text(encoding="utf8"))
+ if (
+ result["name"] != project.name
+ or result["python_version"] != python_version
+ or (head_sha is not None and result["head_sha"] != head_sha)
+ or not SHA_PATTERN.fullmatch(result["head_sha"])
+ or (result["base_sha"] is not None and result["base_sha"] != base_sha)
+ or result["head"]["status"] != "failed"
+ or result["base"] is not None
+ or not SHA_PATTERN.fullmatch(result["project_sha"])
+ ):
+ raise ValueError("Target-branch comparison does not match the saved PR result")
+ result["base_sha"] = base_sha
+ # Separate CI steps share the snapshot and constraints in RUNNER_TEMP.
+ # Local comparisons retain their automatic temporary-directory cleanup.
+ directory_context = (
+ tempfile.TemporaryDirectory(prefix=f"{project.name}-", dir=work_dir)
+ if attempt == "both"
+ else nullcontext(str(work_dir / f"mypyc-primer-{project.name}"))
+ )
+ with directory_context as temporary:
+ directory = Path(temporary)
+ project_source = directory / "project-source"
+ active = "base" if attempt == "base" else "head"
+ log = Log()
+ try:
+ if attempt == "head":
+ directory.mkdir()
+ if attempt != "base":
+ sha = prepare_project_source(project, directory, project_source, timeout, log)
+ if not SHA_PATTERN.fullmatch(sha):
+ raise ValueError("Could not resolve project revision")
+ result["project_sha"] = sha
+ constraints = directory / "constraints.txt"
+ common: AttemptOptions = {
+ "project_source": project_source,
+ "project_sha": result["project_sha"],
+ "constraints": constraints,
+ "timeout": timeout,
+ }
+ if attempt != "base":
+ assert head_sha is not None
+ result["head"] = asdict(
+ run_attempt(
+ project,
+ **common,
+ compiler_repository=str(mypy_source),
+ compiler_sha=head_sha,
+ directory=directory / "head",
+ log=log,
+ baseline=False,
+ )
+ )
+ if attempt != "head" and result["head"]["status"] == "failed":
+ assert base_sha is not None
+ active = "base"
+ log = Log()
+ result["base"] = asdict(
+ run_attempt(
+ project,
+ **common,
+ compiler_repository=str(mypy_source),
+ compiler_sha=base_sha,
+ directory=directory / "base",
+ log=log,
+ baseline=True,
+ )
+ )
+ except (CommandError, OSError, ValueError) as error:
+ log.write(f"\nInfrastructure error: {error}\n")
+ result[active] = asdict(Attempt("error", "setup", tail=log.tail()))
+ write_json(result_path, result)
+ outputs({"compare_base": str(result["head"]["status"] == "failed").lower()})
+ return result
+
+
+def prepare_project_source(
+ project: Project, directory: Path, project_source: Path, timeout: float, log: Log
+) -> str:
+ kwargs: CommandOptions = {
+ "env": environment(project, directory / "unused"),
+ "log": log,
+ "deadline": time.monotonic() + min(timeout, CLONE_TIMEOUT),
+ }
+ execute(
+ [
+ "git",
+ "clone",
+ "--quiet",
+ "--no-checkout",
+ "--",
+ project.repository,
+ str(project_source),
+ ],
+ cwd=directory,
+ **kwargs,
+ )
+ # Prefer the remote branch; a tag or pinned SHA is also supported.
+ try:
+ sha = execute(
+ ["git", "rev-parse", "--verify", f"refs/remotes/origin/{project.ref}^{{commit}}"],
+ cwd=project_source,
+ **kwargs,
+ capture=True,
+ ).strip()
+ except CommandError:
+ sha = execute(
+ ["git", "rev-parse", "--verify", f"{project.ref}^{{commit}}"],
+ cwd=project_source,
+ **kwargs,
+ capture=True,
+ ).strip()
+ return sha
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--config", type=Path, default=CONFIG)
+ subparsers = parser.add_subparsers(dest="command", required=True)
+ preparation = subparsers.add_parser("prepare")
+ preparation.add_argument("--mypy-source", type=Path, default=Path.cwd())
+ preparation.add_argument("--output-dir", type=Path, required=True)
+ check = subparsers.add_parser("check")
+ check.add_argument("--project", required=True)
+ check.add_argument("--mypy-source", type=Path, default=Path.cwd())
+ check.add_argument("--head-sha")
+ check.add_argument("--base-sha")
+ check.add_argument("--attempt", choices=("both", "head", "base"), default="both")
+ check.add_argument("--output-dir", type=Path, required=True)
+ check.add_argument("--work-dir", type=Path, default=Path(tempfile.gettempdir()))
+ args = parser.parse_args()
+ if args.command == "prepare":
+ prepare(args)
+ return 0
+ for revision in ("head", "base"):
+ sha = getattr(args, f"{revision}_sha")
+ if sha is None:
+ if args.attempt in ("both", revision):
+ check.error(f"--{revision}-sha is required for --attempt {args.attempt}")
+ elif not SHA_PATTERN.fullmatch(sha):
+ check.error(f"--{revision}-sha must be a full commit SHA")
+ projects = load_projects(args.config)
+ project = next(p for p in projects if p.name == args.project)
+ result = compare(
+ project,
+ output_dir=args.output_dir.resolve(),
+ work_dir=args.work_dir.resolve(),
+ mypy_source=args.mypy_source.resolve(),
+ head_sha=args.head_sha,
+ base_sha=args.base_sha,
+ attempt=args.attempt,
+ )
+ for key in ("head", "base"):
+ if run_result := result[key]:
+ print(f"{key} result: ", run_result["status"])
+ return int(any(result[key] and result[key]["status"] == "error" for key in ("head", "base")))
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/mypy/test/test_mypyc_primer.py b/mypy/test/test_mypyc_primer.py
new file mode 100644
index 0000000000000..b7c6ccb5ad661
--- /dev/null
+++ b/mypy/test/test_mypyc_primer.py
@@ -0,0 +1,497 @@
+"""Tests for downstream compiler comparisons, without downloading dependencies."""
+
+from __future__ import annotations
+
+import argparse
+import io
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from misc import mypyc_primer as primer
+
+PROJECT_CONFIG = """\
+[[projects]]
+name = "example"
+repository = "https://github.com/example/project.git"
+ref = "main"
+python_version = "3.14"
+compiled_modules = ["example"]
+setup = ["setup-project"]
+build = ["build-project"]
+test = ["test-project"]
+"""
+
+
+@unittest.skipIf(sys.platform == "win32", "mypyc primer requires POSIX subprocess handling")
+class MypycPrimerTests(unittest.TestCase):
+ def test_output_streams_before_command_exits(self) -> None:
+ with tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary)
+ release = root / "release"
+
+ class Console(io.StringIO):
+ def write(self, text: str) -> int:
+ result = super().write(text)
+ if "\nlive output" in self.getvalue():
+ release.touch()
+ return result
+
+ console = Console()
+ log = primer.Log()
+ # Even output without a newline must reach the console before exit.
+ script = (
+ "from pathlib import Path\nimport os, sys, time\n"
+ "print('live output', end='', flush=True)\n"
+ "while not Path(sys.argv[1]).exists():\n time.sleep(0.01)\n"
+ "print('stderr output', file=sys.stderr, flush=True)\n"
+ "os.write(1, b'x' * 8191 + '๐'.encode() + b'\\xff\\n')\n"
+ )
+ with patch("sys.stdout", console):
+ primer.execute(
+ [sys.executable, "-c", script, str(release)],
+ cwd=root,
+ env=os.environ.copy(),
+ log=log,
+ deadline=time.monotonic() + 5,
+ )
+ self.assertTrue(release.exists())
+ log.write("appended\n")
+ self.assertEqual(log.tail(), console.getvalue().rstrip("\n"))
+ self.assertIn("stderr output\n", console.getvalue())
+ self.assertIn("x" * 8191 + "๐๏ฟฝ\n", console.getvalue())
+
+ def test_capture_keeps_stderr_out_of_result(self) -> None:
+ with tempfile.TemporaryDirectory() as temporary:
+ console = io.StringIO()
+ # Exceed pipe capacity on both streams to catch blocked readers.
+ script = (
+ "import sys\n"
+ "sys.stderr.write('!' * 150000)\n"
+ "sys.stdout.write('?' * 150000)\n"
+ )
+ with patch("sys.stdout", console):
+ data = primer.execute(
+ [sys.executable, "-c", script],
+ cwd=Path(temporary),
+ env=os.environ.copy(),
+ log=primer.Log(),
+ deadline=time.monotonic() + 5,
+ capture=True,
+ )
+ self.assertEqual(data.count("?"), 150000)
+ self.assertEqual(len(data), 150000)
+ # Each character also occurs once in the echoed command. Pipe chunks
+ # may interleave, but neither stream should lose any output.
+ self.assertEqual(console.getvalue().count("!"), 150001)
+ self.assertEqual(console.getvalue().count("?"), 150001)
+
+ def test_attempts_can_run_in_separate_steps(self) -> None:
+ for saved_base_sha in (None, "c" * 40):
+ with self.subTest(base=saved_base_sha), tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary)
+ with (
+ patch.object(
+ primer, "prepare_project_source", return_value="a" * 40
+ ) as resolve,
+ patch.object(
+ primer,
+ "run_attempt",
+ side_effect=[
+ primer.Attempt("failed", "test"),
+ primer.Attempt("passed", "test"),
+ ],
+ ) as run,
+ patch.object(primer, "outputs") as outputs,
+ ):
+ first = primer.compare(
+ self.project(),
+ output_dir=root / "logs",
+ work_dir=root,
+ mypy_source=root,
+ head_sha="b" * 40,
+ base_sha=saved_base_sha,
+ attempt="head",
+ )
+ self.assertEqual(run.call_count, 1)
+ self.assertIsNone(first["base"])
+ self.assertEqual(first["base_sha"], saved_base_sha)
+ self.assertEqual(outputs.call_args.args[0]["compare_base"], "true")
+ constraints = run.call_args.kwargs["constraints"]
+ constraints.write_text("dependency==1.0\n")
+ # A baseline step cannot silently compare different revisions.
+ mismatches = [("d" * 40, "c" * 40)]
+ if saved_base_sha is not None:
+ mismatches.append(("b" * 40, "d" * 40))
+ for head_sha, base_sha in mismatches:
+ with (
+ self.subTest(head=head_sha, base=base_sha),
+ self.assertRaisesRegex(ValueError, "saved PR result"),
+ ):
+ primer.compare(
+ self.project(),
+ output_dir=root / "logs",
+ work_dir=root,
+ mypy_source=root,
+ head_sha=head_sha,
+ base_sha=base_sha,
+ attempt="base",
+ )
+ second = primer.compare(
+ self.project(),
+ output_dir=root / "logs",
+ work_dir=root,
+ mypy_source=root,
+ base_sha="c" * 40,
+ attempt="base",
+ )
+ self.assertEqual(resolve.call_count, 1)
+ self.assertEqual(second["head"], first["head"])
+ self.assertEqual(second["head_sha"], "b" * 40)
+ self.assertEqual(second["base_sha"], "c" * 40)
+ self.assertEqual(second["base"]["status"], "passed")
+ self.assertEqual(run.call_args.kwargs["project_sha"], "a" * 40)
+ self.assertEqual(run.call_args.kwargs["compiler_repository"], str(root))
+ self.assertEqual(run.call_args.kwargs["compiler_sha"], "c" * 40)
+ self.assertEqual(
+ run.call_args.kwargs["constraints"].read_text(), "dependency==1.0\n"
+ )
+ self.assertNotEqual(
+ run.call_args_list[0].kwargs["directory"],
+ run.call_args.kwargs["directory"],
+ )
+
+ def test_prepare_snapshot_and_docs_skip(self) -> None:
+ with tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary)
+ event = root / "event.json"
+ event.write_text(
+ json.dumps(
+ {
+ "pull_request": {
+ "number": 42,
+ "head": {"sha": "a" * 40},
+ "base": {
+ "sha": "b" * 40,
+ "ref": "release/test-target",
+ "repo": {"full_name": "example/mypy-fork"},
+ },
+ }
+ }
+ )
+ )
+ config = root / "projects.toml"
+ config.write_text(PROJECT_CONFIG)
+ args = argparse.Namespace(config=config, mypy_source=root, output_dir=root)
+ with (
+ patch.dict(
+ os.environ,
+ {
+ "GITHUB_EVENT_PATH": str(event),
+ "GITHUB_RUN_ID": "123",
+ "GITHUB_RUN_ATTEMPT": "1",
+ },
+ ),
+ patch.object(primer, "changed_paths", return_value=["mypyc/doc/conf.py"]) as paths,
+ patch.object(subprocess, "check_output") as git,
+ patch.object(primer, "outputs") as outputs,
+ ):
+ primer.prepare(args)
+ git.assert_not_called()
+ self.assertEqual(outputs.call_args.args[0]["eligible"], "false")
+ paths.return_value = ["mypyc/doc/conf.py", "mypyc/build.py"]
+ primer.prepare(args)
+ manifest = json.loads((root / "manifest.json").read_text())
+ self.assertEqual(manifest["head_sha"], "a" * 40)
+ self.assertEqual(manifest["base_sha"], "b" * 40)
+ self.assertEqual(manifest["base_ref"], "release/test-target")
+ self.assertEqual(outputs.call_args.args[0]["base_sha"], "b" * 40)
+ git.assert_not_called()
+ self.assertEqual(
+ manifest["projects"], [{"name": "example", "python_version": "3.14"}]
+ )
+ self.assertEqual(outputs.call_args.args[0]["eligible"], "true")
+
+ def project(self) -> primer.Project:
+ return primer.Project(
+ name="example",
+ repository="https://github.com/example/project.git",
+ ref="main",
+ python_version="3.14",
+ compiled_modules=["example"],
+ setup=["setup-project"],
+ build=["build-project"],
+ test=["test-project"],
+ )
+
+ def test_path_eligibility(self) -> None:
+ for path in (
+ "mypyc/irbuild/main.py",
+ "mypyc/lib-rt/CPy.h",
+ "mypyc/lib-rt/strings/librt_strings.c",
+ ".github/mypyc_primer.toml",
+ "misc/mypyc_primer.py",
+ ".github/scripts/mypyc_primer_comment.cjs",
+ ):
+ with self.subTest(path=path):
+ self.assertTrue(primer.eligible_path(path))
+ for path in (
+ "mypyc/doc/conf.py",
+ "mypyc/doc/Makefile",
+ "mypyc/README.md",
+ "mypyc/test/test_run.py",
+ "mypyc/test-data/fixtures/testutil.py",
+ "mypyc/lib-rt/test_capi.cc",
+ "mypyc/external/googletest/src/gtest.cc",
+ "mypy/checker.py",
+ "test-data/unit/check.test",
+ ):
+ with self.subTest(path=path):
+ self.assertFalse(primer.eligible_path(path))
+
+ def test_entire_diff_and_rename(self) -> None:
+ root = Path("repository")
+ # Put both sides of a rename beyond GitHub's 300-file path-filter limit.
+ expected = [f"unrelated-{index}.py" for index in range(350)]
+ expected += ["mypyc/test/renamed.py", "mypyc/renamed.py", "README.md"]
+ diff = b"\0".join(os.fsencode(path) for path in expected) + b"\0"
+ with patch.object(subprocess, "check_output", return_value=diff):
+ paths = primer.changed_paths(root, "base", "head")
+ self.assertEqual(paths, expected)
+ self.assertTrue(any(primer.eligible_path(path) for path in paths))
+
+ def test_catalog_validation(self) -> None:
+ with tempfile.TemporaryDirectory() as temporary:
+ path = Path(temporary) / "projects.toml"
+ path.write_text(PROJECT_CONFIG)
+ self.assertEqual(primer.load_projects(path), [self.project()])
+ # A duplicate entry cannot overwrite another project's artifacts.
+ path.write_text(PROJECT_CONFIG * 2)
+ with self.assertRaisesRegex(ValueError, "Duplicate"):
+ primer.load_projects(path)
+ path.write_text(PROJECT_CONFIG.replace('name = "example"', 'name = "../example"'))
+ with self.assertRaisesRegex(ValueError, "artifact-safe"):
+ primer.load_projects(path)
+ path.write_text(PROJECT_CONFIG + '\n[projects.env]\nPYTHONPATH = "src"\n')
+ with self.assertRaisesRegex(ValueError, "reserved"):
+ primer.load_projects(path)
+ for shared_env in (
+ 'env = "invalid"\n',
+ '[env]\nPYTHONPATH = "src"\n',
+ "[env]\nMYPYC_OPT_LEVEL = 0\n",
+ '[env]\n"invalid-name" = "value"\n',
+ ):
+ with self.subTest(env=shared_env):
+ path.write_text(shared_env + PROJECT_CONFIG)
+ with self.assertRaisesRegex(ValueError, "invalid or reserved"):
+ primer.load_projects(path)
+
+ def test_shared_project_environment(self) -> None:
+ with tempfile.TemporaryDirectory() as temporary:
+ path = Path(temporary) / "projects.toml"
+ path.write_text(
+ '[env]\nCC = "clang"\nMYPYC_OPT_LEVEL = "0"\n'
+ + PROJECT_CONFIG
+ + '\n[projects.env]\nMYPYC_OPT_LEVEL = "1"\nPROJECT_ONLY = "yes"\n'
+ + PROJECT_CONFIG.replace('name = "example"', 'name = "other"')
+ )
+ first, second = primer.load_projects(path)
+ self.assertEqual(
+ first.env, {"CC": "clang", "MYPYC_OPT_LEVEL": "1", "PROJECT_ONLY": "yes"}
+ )
+ self.assertEqual(second.env, {"CC": "clang", "MYPYC_OPT_LEVEL": "0"})
+ env = primer.environment(first, Path("/fresh/venv"))
+ for name, value in first.env.items():
+ self.assertEqual(env[name], value)
+
+ def test_environment_and_dependency_snapshot(self) -> None:
+ with patch.dict(
+ os.environ, {"PYTHONPATH": "/wrong", "MYPY_USE_MYPYC": "1", "NO_COLOR": "1"}
+ ):
+ env = primer.environment(self.project(), Path("/fresh/venv"))
+ self.assertNotIn("PYTHONPATH", env)
+ self.assertNotIn("NO_COLOR", env)
+ self.assertEqual(env["MYPY_USE_MYPYC"], "0")
+ self.assertTrue(env["PATH"].startswith("/fresh/venv/bin" + os.pathsep))
+ with tempfile.TemporaryDirectory() as temporary:
+ path = Path(temporary) / "constraints.txt"
+ primer.snapshot_dependencies(
+ "mypy==1.0\nlibrt==0.16\nexample==2.0\npytest==9.0\n"
+ "black @ file:///old/tree\ntyping_extensions==4.16\n",
+ path,
+ self.project(),
+ )
+ self.assertEqual(path.read_text(), "pytest==9.0\ntyping_extensions==4.16\n")
+
+ def test_command_pipefail_and_timeout(self) -> None:
+ with tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary)
+ log = primer.Log()
+ with patch("sys.stdout", io.StringIO()):
+ kwargs: primer.CommandOptions = {
+ "env": os.environ.copy(),
+ "log": log,
+ "deadline": time.monotonic() + 5,
+ }
+ with self.assertRaises(primer.CommandError) as failure:
+ primer.execute(
+ ["bash", "-euo", "pipefail", "-c", "false | cat"], cwd=root, **kwargs
+ )
+ self.assertEqual(failure.exception.returncode, 1)
+ for script in (
+ "import time; time.sleep(30)",
+ # Closing the pipes must not bypass the process deadline.
+ "import os, time; os.close(1); os.close(2); time.sleep(30)",
+ # Kill children that still hold the output pipe open.
+ "import subprocess, sys, time; "
+ "subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)']); "
+ "time.sleep(30)",
+ # A continuously readable pipe must still respect the deadline.
+ "import os\nwhile True: os.write(1, b'x' * 8192)",
+ ):
+ kwargs["deadline"] = time.monotonic() + 0.1
+ with (
+ self.subTest(script=script),
+ self.assertRaises(primer.CommandError) as failure,
+ ):
+ primer.execute([sys.executable, "-c", script], cwd=root, **kwargs)
+ self.assertEqual(failure.exception.returncode, 124)
+ self.assertIn("time limit", log.tail())
+
+ def test_infrastructure_failure_keeps_excerpt(self) -> None:
+ with tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary)
+
+ def fail(
+ project: primer.Project,
+ directory: Path,
+ source: Path,
+ timeout: float,
+ log: primer.Log,
+ ) -> str:
+ log.write("Source checkout failed\n")
+ raise primer.CommandError(1, "Cannot fetch project")
+
+ with (
+ patch.object(primer, "prepare_project_source", side_effect=fail),
+ patch("sys.stdout", io.StringIO()) as console,
+ ):
+ result = primer.compare(
+ self.project(),
+ output_dir=root / "results",
+ work_dir=root,
+ mypy_source=root,
+ head_sha="b" * 40,
+ base_sha="c" * 40,
+ )
+ self.assertEqual(result["head"]["status"], "error")
+ self.assertEqual(result["head"]["tail"], console.getvalue().rstrip("\n"))
+ self.assertEqual(list((root / "results").iterdir()), [root / "results/example.json"])
+
+ def test_failures_run_base_in_fresh_directory(self) -> None:
+ for pr_status, base_status in (
+ ("passed", None),
+ ("failed", "passed"),
+ ("failed", "failed"),
+ ("failed", "error"),
+ ("error", None),
+ ):
+ with self.subTest(pr=pr_status, base=base_status):
+ with tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary)
+ attempts = [primer.Attempt(pr_status, "build")]
+ if base_status:
+ attempts.append(primer.Attempt(base_status, "test"))
+ with (
+ patch.object(primer, "execute", return_value="a" * 40),
+ patch.object(primer, "run_attempt", side_effect=attempts) as run,
+ patch.object(primer, "outputs"),
+ ):
+ result = primer.compare(
+ self.project(),
+ output_dir=root / "logs",
+ work_dir=root,
+ mypy_source=root,
+ head_sha="b" * 40,
+ base_sha="c" * 40,
+ )
+ self.assertEqual(run.call_count, 2 if base_status else 1)
+ self.assertEqual(result["head"]["status"], pr_status)
+ if base_status:
+ self.assertNotEqual(
+ run.call_args_list[0].kwargs["directory"],
+ run.call_args_list[1].kwargs["directory"],
+ )
+ self.assertEqual(
+ run.call_args_list[0].kwargs["project_sha"],
+ run.call_args_list[1].kwargs["project_sha"],
+ )
+ self.assertEqual(run.call_args_list[1].kwargs["compiler_sha"], "c" * 40)
+ self.assertEqual(
+ run.call_args_list[1].kwargs["compiler_repository"], str(root)
+ )
+ self.assertTrue(run.call_args_list[1].kwargs["baseline"])
+ saved = json.loads((root / "logs/example.json").read_text())
+ self.assertEqual(saved, result)
+
+ def test_attempt_failure_stages(self) -> None:
+ for stage in ("setup", "compiler", "build", "compiled-import", "test"):
+ with self.subTest(stage=stage), tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary)
+
+ def execute(command: list[str], **kwargs: object) -> str:
+ match = {
+ "setup": command[-1] == "setup-project",
+ "compiler": "--force-reinstall" in command,
+ "build": command[-1] == "build-project",
+ "compiled-import": command[:2] == ["python", "-c"]
+ and "Compiled imports:" in command[2],
+ "test": command[-1] == "test-project",
+ }
+ if match[stage]:
+ raise primer.CommandError(124, "Timed out")
+ return ""
+
+ with patch.object(primer, "execute", side_effect=execute):
+ result = primer.run_attempt(
+ self.project(),
+ project_source=root,
+ project_sha="a" * 40,
+ compiler_repository=str(root),
+ compiler_sha="b" * 40,
+ directory=root / "attempt",
+ log=primer.Log(),
+ constraints=root / "constraints.txt",
+ baseline=False,
+ )
+ self.assertEqual(result.stage, stage)
+ self.assertEqual(result.status, "error" if stage == "setup" else "failed")
+ self.assertEqual(result.returncode, 124)
+
+ def test_log_keeps_last_20_lines(self) -> None:
+ log = primer.Log()
+ console = io.StringIO()
+ data = "".join(f"line {index}\n" for index in range(100)) + "last line"
+ with patch("sys.stdout", console):
+ for offset in range(0, len(data), 7):
+ log.write(data[offset : offset + 7])
+ self.assertEqual(
+ log.tail().splitlines(), [f"line {i}" for i in range(81, 100)] + ["last line"]
+ )
+ self.assertEqual(console.getvalue(), data)
+
+ with patch("sys.stdout", console):
+ for _ in range(100):
+ log.write("x" * 10000)
+ self.assertEqual(log.tail(), "[excerpt truncated]\n" + "x" * 20000)
+ # Once a long line falls out of the excerpt, truncation is unnecessary.
+ log.write("\n" + data)
+ self.assertEqual(
+ log.tail().splitlines(), [f"line {i}" for i in range(81, 100)] + ["last line"]
+ )