feat(ci)(hard-gate): lint-required-context-exists-in-bp (Tier 2g)
Some checks failed
Block internal-flavored paths / Block forbidden paths (pull_request) Successful in 9s
Lint curl status-code capture / Scan workflows for curl status-capture pollution (pull_request) Successful in 10s
CI / Detect changes (pull_request) Successful in 20s
E2E API Smoke Test / detect-changes (pull_request) Successful in 22s
E2E Staging Canvas (Playwright) / detect-changes (pull_request) Successful in 24s
Handlers Postgres Integration / detect-changes (pull_request) Successful in 24s
Secret scan / Scan diff for credential-shaped strings (pull_request) Successful in 15s
Runtime PR-Built Compatibility / detect-changes (pull_request) Successful in 29s
qa-review / approved (pull_request) Failing after 15s
sop-checklist / all-items-acked (pull_request) [soft-fail tier:low] acked: 0/7 — missing: comprehensive-testing, local-postgres-e2e, staging-smoke, +4 — body-unfilled: 7
security-review / approved (pull_request) Failing after 15s
sop-checklist-gate / gate (pull_request) Successful in 13s
gate-check-v3 / gate-check (pull_request) Successful in 21s
sop-tier-check / tier-check (pull_request) Successful in 11s
CI / Shellcheck (E2E scripts) (pull_request) Successful in 5s
CI / Platform (Go) (pull_request) Successful in 13s
CI / Python Lint & Test (pull_request) Successful in 4s
lint-required-no-paths / lint-required-no-paths (pull_request) Successful in 1m13s
Handlers Postgres Integration / Handlers Postgres Integration (pull_request) Successful in 6s
E2E API Smoke Test / E2E API Smoke Test (pull_request) Successful in 13s
Runtime PR-Built Compatibility / PR-built wheel + import smoke (pull_request) Successful in 6s
lint-continue-on-error-tracking / lint-continue-on-error-tracking (pull_request) Failing after 1m29s
CI / Canvas (Next.js) (pull_request) Successful in 27s
Lint workflow YAML (Gitea-1.22.6-hostile shapes) / Lint workflow YAML for Gitea-1.22.6-hostile shapes (pull_request) Failing after 1m27s
E2E Staging Canvas (Playwright) / Canvas tabs E2E (pull_request) Successful in 30s
lint-required-context-exists-in-bp / lint-required-context-exists-in-bp (pull_request) Successful in 1m45s
CI / Canvas Deploy Reminder (pull_request) Has been skipped
CI / all-required (pull_request) Successful in 3s

PR-time diff-based lint: when a PR adds a NEW commit-status emission,
the workflow file must carry one of three directives adjacent to the
new job:
  - `# bp-required: yes`           AND the context is in BP
  - `# bp-required: pending #NNN`  acknowledged asymmetry + tracker
  - `# bp-exempt: <reason>`        informational job, not a gate

Default (no directive on a new emitter) = FAIL with 3-option hint.

The class this prevents
-----------------------
PR#656 added `CI / all-required (pull_request)` as a sentinel context
that workflows emit, but BP did NOT list it. When platform-build
failed, all-required failed, but BP let the PR merge anyway → mc#664.

Cousin to Tier 2f
-----------------
Tier 2g blocks at PR-time (diff-based); Tier 2f files a drift issue
at scheduled-time. They share enumeration helpers (workflow_contexts,
event-map) but the semantics differ — Tier 2g is PR-time block,
Tier 2f is scheduled audit + issue. Co-design documented in #350.

Why the directive lives in the YAML, not the PR body
----------------------------------------------------
PR-body claim evaporates on merge; the directive must persist with
the emitter so Tier 2f's daily audit reads the same contract.

Implementation
--------------
- `.gitea/scripts/lint_required_context_exists_in_bp.py` — git diff
  base..head, enumerate emitted contexts on each side via PyYAML AST
  (mirror Tier 2f), `new = head - base`. For each new context resolve
  back to (file, job-key), scan ±3 lines above the job-key line for a
  directive comment. Validate against BP context list when directive
  is `bp-required: yes`. Graceful-degrade 403/404 per Tier 2a.
- `.gitea/workflows/lint-required-context-exists-in-bp.yml` —
  pull_request with paths-filter on .gitea/workflows/**. Phase 3
  (continue-on-error: true).
- `tests/test_lint_required_context_exists_in_bp.py` — 11 unit tests:
  no new emissions skip, bp-required:yes+in-BP pass, bp-required:yes
  not-in-BP fail, bp-required:pending pass, bp-exempt pass, no-directive
  fail, new-job-in-existing-workflow flagged, job-rename flagged,
  comment-only edit no-flag, 403 graceful, PR-body directive
  insufficient.

Refs: #350
This commit is contained in:
core-devops 2026-05-11 23:19:31 -07:00 committed by Molecule AI Core-DevOps
parent 84ec7fe728
commit 27bf3680fa
3 changed files with 1074 additions and 0 deletions

View File

@ -0,0 +1,526 @@
#!/usr/bin/env python3
"""lint_required_context_exists_in_bp — Tier 2g per internal#350.
Rule
----
When a PR adds a NEW commit-status emission (a context that didn't
exist on the base side), the workflow file must carry one of three
directive comments adjacent to the new job:
(a) `# bp-required: yes`
The new context MUST already be in
`branch_protections/<branch>.status_check_contexts`. Verified
via Gitea API at PR time.
(b) `# bp-required: pending #NNN`
Acknowledged asymmetry; references an OPEN tracking issue that
will follow up with the BP PATCH.
(c) `# bp-exempt: <free-text reason>`
Informational job, not intended to be a required gate.
No directive on a new emitter FAIL with a 3-option fix-hint.
The class this prevents
-----------------------
PR#656 added `CI / all-required (pull_request)` as a sentinel context
that workflows emit, but BP did NOT list it. When `platform-build`
failed, `all-required` failed, but BP let the PR merge anyway
cascade to mc#664. With this lint, PR#656 would have been blocked
until either the BP PATCH ran alongside OR the author added a
`bp-required: pending` directive.
Why directives MUST live in the workflow YAML
---------------------------------------------
The directive comment lives with the emitter so a scheduled
audit (Tier 2f, daily) can read the same source. PR-body-only
directives invisibly evaporate on merge the asymmetry would
return to undetected. PR-body claims are advisory; workflow-file
comments are the contract.
How "new emission" is detected
------------------------------
Diff base..head over `.gitea/workflows/*.yml`. For each YAML file
that's added or modified:
- Parse both base-side and head-side via PyYAML AST.
- Enumerate emitted contexts on each side using the same rules as
Tier 2f (workflow.name + job.name|key + event-mapping).
- `new_contexts = head_contexts - base_contexts`.
If `new_contexts` is empty after de-dup, no rule applies pass.
Per `feedback_behavior_based_ast_gates`: comment scanning uses raw
text in a small window around the job-key line, NOT regex over the
full file. This avoids matching `bp-required:` mentioned in a
comment unrelated to the new job.
Exit codes
----------
0 no new emissions, all new emissions have valid directives,
or BP read errored (graceful-degrade per Tier 2a contract).
1 at least one new emission lacks a directive, or has
`bp-required: yes` but the context is missing from BP.
2 env contract violation or YAML parse error.
Env
---
BASE_SHA PR base SHA
HEAD_SHA PR head SHA
GITEA_TOKEN DRIFT_BOT_TOKEN (repo-admin for BP read)
GITEA_HOST e.g. git.moleculesai.app
REPO owner/name
BRANCH defaults to `main`
WORKFLOWS_DIR defaults to `.gitea/workflows`
Memory cross-links
------------------
- internal#350 (the RFC that specs this lint)
- PR#656 (the empirical case that prompted Tier 2g)
- mc#664 (the surfaced cascade)
- feedback_phantom_required_check_after_gitea_migration (Tier 2f cousin)
- feedback_behavior_based_ast_gates
"""
from __future__ import annotations
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from typing import Any
try:
import yaml
except ImportError:
sys.stderr.write(
"::error::PyYAML is required. Install with: pip install PyYAML\n"
)
sys.exit(2)
# Directive comment patterns. We match `# bp-required:` OR `# bp-exempt:`,
# both with optional surrounding whitespace and case-sensitive on the
# `bp-` prefix (convention).
BP_REQUIRED_YES_RE = re.compile(
r"#\s*bp-required:\s*yes\b", re.IGNORECASE
)
BP_REQUIRED_PENDING_RE = re.compile(
r"#\s*bp-required:\s*pending\s*#(?P<num>\d+)\b", re.IGNORECASE
)
BP_EXEMPT_RE = re.compile(
r"#\s*bp-exempt:\s*\S", re.IGNORECASE
)
# Gitea event-mapping (same as Tier 2f).
_EVENT_MAP = {
"pull_request": "pull_request",
"pull_request_target": "pull_request",
"push": "push",
}
# ---------------------------------------------------------------------------
# Env
# ---------------------------------------------------------------------------
def _env(key: str, default: str | None = None) -> str:
v = os.environ.get(key, default)
return v if v is not None else ""
def _require_env(key: str) -> str:
v = os.environ.get(key)
if not v:
sys.stderr.write(f"::error::missing required env var: {key}\n")
sys.exit(2)
return v
# ---------------------------------------------------------------------------
# API helper (same contract as Tier 2f).
# ---------------------------------------------------------------------------
def api(
method: str,
path: str,
*,
body: dict | None = None,
query: dict[str, str] | None = None,
) -> tuple[str, Any]:
host = _env("GITEA_HOST")
token = _env("GITEA_TOKEN")
url = f"https://{host}/api/v1{path}"
if query:
url = f"{url}?{urllib.parse.urlencode(query)}"
data = None
headers = {
"Authorization": f"token {token}",
"Accept": "application/json",
}
if body is not None:
data = json.dumps(body).encode("utf-8")
headers["Content-Type"] = "application/json"
req = urllib.request.Request(url, method=method, data=data, headers=headers)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
raw = resp.read()
if not raw:
return ("ok", None)
return ("ok", json.loads(raw))
except urllib.error.HTTPError as e:
if e.code == 404:
return ("not_found", None)
if e.code in (401, 403):
return ("forbidden", None)
return ("error", None)
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError):
return ("error", None)
# ---------------------------------------------------------------------------
# git helpers
# ---------------------------------------------------------------------------
def git_show(sha: str, path: str) -> str | None:
r = subprocess.run(
["git", "show", f"{sha}:{path}"], capture_output=True, text=True
)
if r.returncode != 0:
return None
return r.stdout
def git_diff_paths(base: str, head: str) -> list[str]:
r = subprocess.run(
["git", "diff", "--name-only", f"{base}..{head}"],
capture_output=True,
text=True,
)
if r.returncode != 0:
return []
return [p for p in r.stdout.splitlines() if p.strip()]
# ---------------------------------------------------------------------------
# Workflow context enumeration (mirror Tier 2f).
# ---------------------------------------------------------------------------
def _get_on(d: Any) -> Any:
if not isinstance(d, dict):
return None
if "on" in d:
return d["on"]
if True in d:
return d[True]
return None
def _on_events(doc: Any) -> set[str]:
on = _get_on(doc)
raw: set[str] = set()
if on is None:
return raw
if isinstance(on, str):
raw.add(on)
elif isinstance(on, list):
for e in on:
if isinstance(e, str):
raw.add(e)
elif isinstance(on, dict):
for k in on:
if isinstance(k, str):
raw.add(k)
return {_EVENT_MAP[e] for e in raw if e in _EVENT_MAP}
def _job_display(jbody: dict, jkey: str) -> str:
n = jbody.get("name") if isinstance(jbody, dict) else None
if isinstance(n, str) and n:
return n
return jkey
def workflow_contexts(doc: Any) -> set[str]:
if not isinstance(doc, dict):
return set()
wf_name = doc.get("name")
if not isinstance(wf_name, str) or not wf_name:
return set()
events = _on_events(doc)
if not events:
return set()
jobs = doc.get("jobs")
if not isinstance(jobs, dict):
return set()
out: set[str] = set()
for jkey, jbody in jobs.items():
if jkey == "__lines__":
continue
if not isinstance(jbody, dict):
continue
disp = _job_display(jbody, jkey)
for ev in events:
out.add(f"{wf_name} / {disp} ({ev})")
return out
# ---------------------------------------------------------------------------
# Find the source line of a job-key in a workflow YAML's raw text.
# Used to scan for nearby directive comments.
# ---------------------------------------------------------------------------
def _find_job_key_line(raw_lines: list[str], jkey: str) -> int | None:
"""Return 1-based line of `<jkey>:` under jobs:."""
in_jobs = False
jobs_indent = -1
for i, line in enumerate(raw_lines, start=1):
stripped = line.lstrip()
if stripped.startswith("jobs:"):
in_jobs = True
jobs_indent = len(line) - len(stripped)
continue
if in_jobs:
# Job key is the next indent level under `jobs:`.
indent = len(line) - len(stripped)
if stripped and indent <= jobs_indent:
# Left the jobs: block
in_jobs = False
continue
if re.match(rf"^\s*{re.escape(jkey)}\s*:", line):
return i
return None
_DIRECTIVE_WINDOW = 3 # lines above the job-key line (inclusive)
def find_directive_for_job(
raw_text: str, jkey: str
) -> tuple[str, str | None] | None:
"""Return (kind, value) tuple for the first directive in a small
window above the job-key line.
kind {"required-yes", "required-pending", "exempt"}.
value is the pending-issue number for required-pending, else None.
Returns None if no directive found.
We scan ABOVE the line only (the convention is the directive
precedes the job matches how `# mc#NNN` comments are placed
above `continue-on-error: true`). We don't scan inside the job
body because steps can produce false positives.
"""
lines = raw_text.splitlines()
line_no = _find_job_key_line(lines, jkey)
if line_no is None:
return None
lo = max(1, line_no - _DIRECTIVE_WINDOW)
for i in range(lo, line_no):
line = lines[i - 1]
m = BP_REQUIRED_PENDING_RE.search(line)
if m:
return ("required-pending", m.group("num"))
if BP_REQUIRED_YES_RE.search(line):
return ("required-yes", None)
if BP_EXEMPT_RE.search(line):
return ("exempt", None)
return None
# ---------------------------------------------------------------------------
# Map a context back to its emitting (workflow_path, job_key) pair so
# we know WHERE to look for the directive comment.
# ---------------------------------------------------------------------------
def _resolve_emitter(
ctx: str, head_workflows: dict[str, tuple[str, Any]]
) -> tuple[str, str] | None:
"""Return (file_path, job_key) emitting ctx, or None."""
m = re.match(r"^(?P<wf>.+?) / (?P<job>.+) \((?P<event>[^)]+)\)$", ctx)
if not m:
return None
target_wf = m.group("wf")
target_job_disp = m.group("job")
for path, (_raw, doc) in head_workflows.items():
if not isinstance(doc, dict):
continue
if doc.get("name") != target_wf:
continue
jobs = doc.get("jobs") or {}
if not isinstance(jobs, dict):
continue
for jkey, jbody in jobs.items():
if jkey == "__lines__":
continue
if not isinstance(jbody, dict):
continue
disp = _job_display(jbody, jkey)
if disp == target_job_disp:
return (path, jkey)
return None
# ---------------------------------------------------------------------------
# Driver
# ---------------------------------------------------------------------------
def run() -> int:
base_sha = _require_env("BASE_SHA")
head_sha = _require_env("HEAD_SHA")
_require_env("GITEA_TOKEN")
_require_env("GITEA_HOST")
repo = _require_env("REPO")
branch = _env("BRANCH", "main")
wf_dir = _env("WORKFLOWS_DIR", ".gitea/workflows")
# Step 1 — find workflow files changed in the PR.
changed = git_diff_paths(base_sha, head_sha)
changed_workflows = [
p
for p in changed
if p.startswith(wf_dir + "/")
and (p.endswith(".yml") or p.endswith(".yaml"))
]
if not changed_workflows:
print(
"::notice::no workflow file changes in this PR; "
"lint-required-context-exists-in-bp skipped."
)
return 0
# Step 2 — load base+head + compute new contexts.
head_workflows: dict[str, tuple[str, Any]] = {}
new_contexts: set[str] = set()
for path in changed_workflows:
base_raw = git_show(base_sha, path)
head_raw = git_show(head_sha, path)
if head_raw is None:
# File deleted on head — no new emission contribution.
continue
try:
head_doc = yaml.safe_load(head_raw)
except yaml.YAMLError as e:
sys.stderr.write(
f"::error file={path}::YAML parse error on head: {e}\n"
)
return 2
head_workflows[path] = (head_raw, head_doc)
head_ctx = workflow_contexts(head_doc)
base_ctx: set[str] = set()
if base_raw is not None:
try:
base_doc = yaml.safe_load(base_raw)
except yaml.YAMLError:
base_doc = None
if base_doc is not None:
base_ctx = workflow_contexts(base_doc)
new_contexts |= (head_ctx - base_ctx)
if not new_contexts:
print(
"::notice::no new context emissions detected in this PR; "
"lint-required-context-exists-in-bp skipped."
)
return 0
# Step 3 — fetch BP context list.
status, bp = api("GET", f"/repos/{repo}/branch_protections/{branch}")
bp_contexts: set[str] = set()
if status == "forbidden":
sys.stderr.write(
f"::error::GET branch_protections/{branch} returned HTTP 403 — "
f"DRIFT_BOT_TOKEN lacks repo-admin scope. Cannot verify "
f"bp-required directives; skipping lint with exit 0 per "
f"Tier 2a contract. Fix the token, not the lint.\n"
)
return 0
elif status == "not_found":
# Branch has no protection — nothing to verify against; the
# bp-required: yes directive can't be satisfied. Treat as
# graceful-skip rather than red-X.
print(
f"::notice::branch '{branch}' has no protection; cannot verify "
f"bp-required directives. Skipping (exit 0)."
)
return 0
elif status == "ok" and isinstance(bp, dict):
bp_contexts = set(bp.get("status_check_contexts") or [])
else:
sys.stderr.write(
f"::error::branch_protections/{branch} response unexpected; "
f"status={status}. Treating as transient; exit 0.\n"
)
return 0
# Step 4 — validate each new emission's directive.
violations: list[str] = []
for ctx in sorted(new_contexts):
emitter = _resolve_emitter(ctx, head_workflows)
if emitter is None:
# Shouldn't happen — we just derived ctx from head_workflows.
# Belt-and-suspenders fallback.
violations.append(
f"::error::new emission '{ctx}' (could not resolve emitter "
f"file/job — bug in lint?)"
)
continue
file_path, jkey = emitter
raw_text, _ = head_workflows[file_path]
directive = find_directive_for_job(raw_text, jkey)
if directive is None:
violations.append(
f"::error file={file_path}::lint-required-context-exists-in-bp "
f"(Tier 2g): NEW emission `{ctx}` (job '{jkey}') has no "
f"directive comment. Add ONE of these comments on the line "
f"directly above `{jkey}:` (within {_DIRECTIVE_WINDOW} lines):\n"
f" - `# bp-required: yes` — and ensure the context is "
f"already in branch_protections/{branch}.status_check_contexts.\n"
f" - `# bp-required: pending #NNN` — acknowledged asymmetry, "
f"references the tracking issue for the BP PATCH.\n"
f" - `# bp-exempt: <reason>` — informational job, not a gate.\n"
f"Memory: internal#350 (PR#656 + mc#664 empirical case)."
)
continue
kind, value = directive
if kind == "exempt":
print(f"::notice::{ctx}: bp-exempt directive present, OK.")
continue
if kind == "required-pending":
print(
f"::notice::{ctx}: bp-required: pending #{value}"
f"acknowledged asymmetry, OK."
)
continue
if kind == "required-yes":
if ctx in bp_contexts:
print(
f"::notice::{ctx}: bp-required: yes, and context is in "
f"BP, OK."
)
else:
violations.append(
f"::error file={file_path}::lint-required-context-exists-in-bp "
f"(Tier 2g): job '{jkey}' has `bp-required: yes` "
f"directive but its emitted context `{ctx}` is NOT in "
f"`branch_protections/{branch}.status_check_contexts`. "
f"FIX: either (a) add `{ctx}` to BP (Owners-tier PATCH), "
f"or (b) downgrade the directive to "
f"`# bp-required: pending #NNN` referencing the tracker "
f"for the pending BP PATCH."
)
if violations:
print(
f"::error::lint-required-context-exists-in-bp: "
f"{len(violations)} violation(s) across "
f"{len(changed_workflows)} changed workflow file(s)."
)
for v in violations:
print(v)
return 1
print(
f"::notice::lint-required-context-exists-in-bp: "
f"{len(new_contexts)} new emission(s) all directive-validated."
)
return 0
if __name__ == "__main__":
sys.exit(run())

View File

@ -0,0 +1,118 @@
name: lint-required-context-exists-in-bp
# Tier 2g hard-gate lint (per internal#350) — diff-based PR-time
# check. When a PR adds a NEW commit-status emission (workflow YAML
# `name:` + job `name:`-or-key + on:-event), the workflow file must
# carry one of three directives adjacent to the new job:
#
# - `# bp-required: yes` — and BP must list the context
# - `# bp-required: pending #NNN` — acknowledged asymmetry + tracker
# - `# bp-exempt: <reason>` — informational job, not a gate
#
# Default (no directive on a new emitter) = FAIL.
#
# Why this exists
# ---------------
# PR#656 added `CI / all-required (pull_request)` as a sentinel
# context that workflows emit, but BP did NOT list it. When
# platform-build failed, all-required failed, but BP let the PR
# merge anyway → cascade to mc#664. With this lint, PR#656 would
# have been blocked until either the BP PATCH ran alongside OR
# the author added a `bp-required: pending` directive.
#
# Tier 2g vs Tier 2f
# ------------------
# Tier 2g runs at PR-time (diff-based) and BLOCKS the merge.
# Tier 2f runs daily (scheduled) and FILES a drift issue. They
# share the workflow-context enumeration helpers
# (`_event_map`, `workflow_contexts`, `_job_display`) but the
# semantics are intentionally distinct so they're separate scripts.
# Co-design is documented in internal#350.
#
# Directive comment lives in the workflow file (NOT PR body)
# ----------------------------------------------------------
# A PR-body claim of "BP exempt" evaporates on merge — the
# asymmetry returns to undetected state and Tier 2f's daily
# scheduled audit can't see it. The directive must live with the
# emitter so both PR-time (Tier 2g) and post-merge (Tier 2f)
# readers consume the same source.
#
# Phase contract (RFC internal#219 §1 ladder)
# -------------------------------------------
# Lands at `continue-on-error: true` (Phase 3 — surface the
# pattern without blocking PRs while the directive convention
# beds in). After 7 days of clean runs on `main` with no false
# positives, follow-up flips to `false`. Tracking: internal#350.
#
# Cross-links
# -----------
# - internal#350 (the RFC that specs this lint)
# - PR#656 (the empirical case)
# - mc#664 (the surfaced cascade)
# - feedback_phantom_required_check_after_gitea_migration (Tier 2f cousin)
# - feedback_behavior_based_ast_gates
#
# Auth: DRIFT_BOT_TOKEN (repo-admin for branch_protections read).
on:
pull_request:
types: [opened, synchronize, reopened]
paths:
- '.gitea/workflows/**'
- '.gitea/scripts/lint_required_context_exists_in_bp.py'
- '.gitea/workflows/lint-required-context-exists-in-bp.yml'
- 'tests/test_lint_required_context_exists_in_bp.py'
env:
GITHUB_SERVER_URL: https://git.moleculesai.app
permissions:
contents: read
concurrency:
group: lint-required-context-exists-in-bp-${{ github.event.pull_request.number || github.ref }}
cancel-in-progress: true
jobs:
# bp-exempt: this lint is a PR-time advisory and is not intended to
# be a required gate on main. The directive eat-our-own-dogfood
# confirms the convention works on the lint that defines it.
lint:
name: lint-required-context-exists-in-bp
runs-on: ubuntu-latest
timeout-minutes: 5
# Phase 3 (RFC #219 §1): surface the pattern without blocking PRs
# while the directive convention beds in. Follow-up flip to false
# after 7 clean days on main. internal#350.
continue-on-error: true
steps:
- name: Check out PR head with full history (need base SHA blobs)
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
with:
# `git show <base-sha>:<path>` needs the base SHA's blobs.
# Same rationale as PR#673 and check-migration-collisions.yml.
fetch-depth: 0
- uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: '3.12'
- name: Install PyYAML
run: python -m pip install --quiet 'PyYAML==6.0.2'
- name: Ensure base ref is reachable locally
# Cheap insurance against runner-version drift.
run: |
git fetch origin "${{ github.event.pull_request.base.ref }}" || true
- name: Run lint-required-context-exists-in-bp
env:
# DRIFT_BOT_TOKEN — repo-admin (needed for branch_protections).
GITEA_TOKEN: ${{ secrets.DRIFT_BOT_TOKEN }}
GITEA_HOST: git.moleculesai.app
REPO: ${{ github.repository }}
BRANCH: main
BASE_SHA: ${{ github.event.pull_request.base.sha }}
HEAD_SHA: ${{ github.event.pull_request.head.sha }}
WORKFLOWS_DIR: .gitea/workflows
run: python3 .gitea/scripts/lint_required_context_exists_in_bp.py
- name: Run lint-required-context-exists-in-bp unit tests
run: |
python -m pip install --quiet pytest
python3 -m pytest tests/test_lint_required_context_exists_in_bp.py -v

View File

@ -0,0 +1,430 @@
"""Tests for `.gitea/scripts/lint_required_context_exists_in_bp.py` — Tier 2g lint.
Structural enforcement of internal#350 Tier 2g: when a PR adds a NEW
commit-status emission (a workflow's `name:` + a new job-key/name pair
that didn't exist on the base side), the PR must EITHER:
(a) Include a `# bp-required: yes` directive comment on the workflow
AND the new context must already be in
`branch_protections/<branch>.status_check_contexts`, OR
(b) Include a `# bp-required: pending #NNN` directive (acknowledged
asymmetry with a tracking issue), OR
(c) Include a `# bp-exempt: <reason>` directive (informational job,
not intended to be a required gate).
Default (no directive on a new emitter) = FAIL.
The class this prevents
-----------------------
PR#656 added `CI / all-required (pull_request)` as a sentinel context
that workflows emit, but BP did NOT list it so when `platform-build`
failed, `all-required` failed, but BP let the PR merge anyway. Cascade
to mc#664. With Tier 2g, PR#656 would have been blocked until either
the BP PATCH ran alongside OR the author marked the emission with a
`bp-required: pending #NNN` directive.
Test classes (per `feedback_branch_count_before_approving`):
- test_no_new_emissions_skips diff doesn't add any
new emitter; pass.
- test_new_emission_with_bp_required_yes_in_bp directive set AND
BP lists the context; pass.
- test_new_emission_with_bp_required_yes_not_in_bp directive set
BUT BP doesn't list; fail.
- test_new_emission_with_bp_required_pending `# bp-required:
pending #800` directive references an open tracker; pass.
- test_new_emission_with_bp_exempt `# bp-exempt:
informational` directive; pass.
- test_new_emission_no_directive_fails no directive on a
new emission; fail with the 3-option fix-hint.
- test_modified_workflow_with_new_job_is_new pre-existing
workflow gains a new job with a new name counted as new
emission. Apply rule.
- test_modified_workflow_job_renamed_is_new same workflow,
same job-key, but job `name:` changed counted as new emission
(the OLD context name disappears; the NEW one needs validation).
- test_unrelated_workflow_edit_is_not_new edit a comment in
an existing emitter; no new context introduced; pass.
- test_api_403_skips_gracefully BP read 403; exit 0
with stderr ::error::.
- test_directive_must_be_in_workflow_yml directive in PR
body alone is NOT sufficient; the comment must live in the
workflow file so future scheduled Tier 2f runs can see it.
Run:
python3 -m pytest tests/test_lint_required_context_exists_in_bp.py -v
"""
from __future__ import annotations
import importlib.util
import os
import subprocess
import sys
from pathlib import Path
from unittest import mock
import pytest
SCRIPT_PATH = (
Path(__file__).resolve().parent.parent
/ ".gitea"
/ "scripts"
/ "lint_required_context_exists_in_bp.py"
)
def _import_lint():
spec = importlib.util.spec_from_file_location(
f"lint_required_ctx_in_bp_{os.getpid()}", SCRIPT_PATH
)
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)
return m
# Sample workflows used across multiple tests.
WF_CI_BASE = """name: CI
on:
pull_request:
branches: [main]
jobs:
all-required:
runs-on: x
steps:
- run: echo hi
"""
# CI with a new job added.
WF_CI_NEW_JOB = """name: CI
on:
pull_request:
branches: [main]
jobs:
all-required:
runs-on: x
steps:
- run: echo hi
brand-new:
runs-on: x
steps:
- run: echo new
"""
WF_CI_NEW_JOB_BP_YES = """name: CI
on:
pull_request:
branches: [main]
jobs:
all-required:
runs-on: x
steps:
- run: echo hi
# bp-required: yes
brand-new:
runs-on: x
steps:
- run: echo new
"""
WF_CI_NEW_JOB_BP_PENDING = """name: CI
on:
pull_request:
branches: [main]
jobs:
all-required:
runs-on: x
steps:
- run: echo hi
# bp-required: pending #800
brand-new:
runs-on: x
steps:
- run: echo new
"""
WF_CI_NEW_JOB_BP_EXEMPT = """name: CI
on:
pull_request:
branches: [main]
jobs:
all-required:
runs-on: x
steps:
- run: echo hi
# bp-exempt: informational sticker, not a gate
brand-new:
runs-on: x
steps:
- run: echo new
"""
# Same WF, job rename only (CI/all-required → CI/sentinel).
WF_CI_JOB_RENAMED = """name: CI
on:
pull_request:
branches: [main]
jobs:
all-required:
runs-on: x
name: sentinel
steps:
- run: echo hi
"""
# Comment-only edit — should NOT count as new emission.
WF_CI_COMMENT_ONLY = """# a fresh comment line
name: CI
on:
pull_request:
branches: [main]
jobs:
all-required:
runs-on: x
steps:
- run: echo hi
"""
def _stub_git_and_api(
monkeypatch,
lint_mod,
base_files: dict[str, str | None],
head_files: dict[str, str | None],
bp_response,
):
"""Stub `subprocess.run` for git, and `lint_mod.api` for HTTP."""
def fake_run(cmd, *args, **kwargs):
if not isinstance(cmd, list):
raise AssertionError(f"unexpected cmd: {cmd!r}")
if cmd[:2] == ["git", "show"] and ":" in cmd[2]:
sha, path = cmd[2].split(":", 1)
side = base_files if "base" in sha else head_files
content = side.get(path)
if content is None:
return subprocess.CompletedProcess(cmd, 128, "", "fatal: path not in tree")
return subprocess.CompletedProcess(cmd, 0, content, "")
if cmd[:2] == ["git", "diff"]:
# Names of files that changed (any side has differing contents
# from the other, or only appears on one side).
all_paths = set(base_files) | set(head_files)
changed = sorted(p for p in all_paths if base_files.get(p) != head_files.get(p))
return subprocess.CompletedProcess(cmd, 0, "\n".join(changed) + "\n", "")
raise AssertionError(f"unexpected cmd: {cmd!r}")
monkeypatch.setattr(subprocess, "run", fake_run)
def fake_api(method, path, *, body=None, query=None):
if "branch_protections" in path:
return bp_response
return ("ok", {})
monkeypatch.setattr(lint_mod, "api", fake_api)
@pytest.fixture()
def env(monkeypatch):
monkeypatch.setenv("BASE_SHA", "base-x")
monkeypatch.setenv("HEAD_SHA", "head-x")
monkeypatch.setenv("GITEA_TOKEN", "stub")
monkeypatch.setenv("GITEA_HOST", "git.example.test")
monkeypatch.setenv("REPO", "owner/molecule-core")
monkeypatch.setenv("BRANCH", "main")
monkeypatch.setenv("WORKFLOWS_DIR", ".gitea/workflows")
return monkeypatch
# ---------------------------------------------------------------------------
# No new emissions — pass.
# ---------------------------------------------------------------------------
def test_no_new_emissions_skips(env, monkeypatch, capsys):
m = _import_lint()
_stub_git_and_api(
monkeypatch,
m,
base_files={".gitea/workflows/ci.yml": WF_CI_BASE},
head_files={".gitea/workflows/ci.yml": WF_CI_BASE},
bp_response=("ok", {"status_check_contexts": []}),
)
rc = m.run()
assert rc == 0
# ---------------------------------------------------------------------------
# New emission + bp-required: yes + in BP → pass.
# ---------------------------------------------------------------------------
def test_new_emission_with_bp_required_yes_in_bp(env, monkeypatch, capsys):
m = _import_lint()
_stub_git_and_api(
monkeypatch,
m,
base_files={".gitea/workflows/ci.yml": WF_CI_BASE},
head_files={".gitea/workflows/ci.yml": WF_CI_NEW_JOB_BP_YES},
bp_response=(
"ok",
{"status_check_contexts": ["CI / brand-new (pull_request)"]},
),
)
rc = m.run()
assert rc == 0
# ---------------------------------------------------------------------------
# bp-required: yes but NOT in BP → fail.
# ---------------------------------------------------------------------------
def test_new_emission_with_bp_required_yes_not_in_bp(env, monkeypatch, capsys):
m = _import_lint()
_stub_git_and_api(
monkeypatch,
m,
base_files={".gitea/workflows/ci.yml": WF_CI_BASE},
head_files={".gitea/workflows/ci.yml": WF_CI_NEW_JOB_BP_YES},
bp_response=("ok", {"status_check_contexts": []}),
)
rc = m.run()
assert rc == 1
out = capsys.readouterr().out
assert "brand-new" in out
# ---------------------------------------------------------------------------
# bp-required: pending #NNN → pass.
# ---------------------------------------------------------------------------
def test_new_emission_with_bp_required_pending(env, monkeypatch, capsys):
m = _import_lint()
_stub_git_and_api(
monkeypatch,
m,
base_files={".gitea/workflows/ci.yml": WF_CI_BASE},
head_files={".gitea/workflows/ci.yml": WF_CI_NEW_JOB_BP_PENDING},
bp_response=("ok", {"status_check_contexts": []}),
)
rc = m.run()
assert rc == 0
# ---------------------------------------------------------------------------
# bp-exempt → pass.
# ---------------------------------------------------------------------------
def test_new_emission_with_bp_exempt(env, monkeypatch, capsys):
m = _import_lint()
_stub_git_and_api(
monkeypatch,
m,
base_files={".gitea/workflows/ci.yml": WF_CI_BASE},
head_files={".gitea/workflows/ci.yml": WF_CI_NEW_JOB_BP_EXEMPT},
bp_response=("ok", {"status_check_contexts": []}),
)
rc = m.run()
assert rc == 0
# ---------------------------------------------------------------------------
# New emission, no directive → fail with 3-option fix hint.
# ---------------------------------------------------------------------------
def test_new_emission_no_directive_fails(env, monkeypatch, capsys):
m = _import_lint()
_stub_git_and_api(
monkeypatch,
m,
base_files={".gitea/workflows/ci.yml": WF_CI_BASE},
head_files={".gitea/workflows/ci.yml": WF_CI_NEW_JOB},
bp_response=("ok", {"status_check_contexts": []}),
)
rc = m.run()
assert rc == 1
out = capsys.readouterr().out
assert "brand-new" in out
assert "bp-required" in out
assert "bp-exempt" in out
# ---------------------------------------------------------------------------
# Pre-existing workflow gains a new job → counted as new emission.
# ---------------------------------------------------------------------------
def test_modified_workflow_with_new_job_is_new(env, monkeypatch, capsys):
m = _import_lint()
_stub_git_and_api(
monkeypatch,
m,
base_files={".gitea/workflows/ci.yml": WF_CI_BASE},
head_files={".gitea/workflows/ci.yml": WF_CI_NEW_JOB},
bp_response=("ok", {"status_check_contexts": []}),
)
rc = m.run()
# No directive → fail
assert rc == 1
# ---------------------------------------------------------------------------
# Same workflow, same job-key, but job `name:` changed → new context.
# ---------------------------------------------------------------------------
def test_modified_workflow_job_renamed_is_new(env, monkeypatch, capsys):
m = _import_lint()
_stub_git_and_api(
monkeypatch,
m,
base_files={".gitea/workflows/ci.yml": WF_CI_BASE},
head_files={".gitea/workflows/ci.yml": WF_CI_JOB_RENAMED},
bp_response=("ok", {"status_check_contexts": []}),
)
rc = m.run()
assert rc == 1
out = capsys.readouterr().out
assert "sentinel" in out
# ---------------------------------------------------------------------------
# Comment-only edit → no new emission.
# ---------------------------------------------------------------------------
def test_unrelated_workflow_edit_is_not_new(env, monkeypatch, capsys):
m = _import_lint()
_stub_git_and_api(
monkeypatch,
m,
base_files={".gitea/workflows/ci.yml": WF_CI_BASE},
head_files={".gitea/workflows/ci.yml": WF_CI_COMMENT_ONLY},
bp_response=("ok", {"status_check_contexts": []}),
)
rc = m.run()
assert rc == 0
# ---------------------------------------------------------------------------
# BP API 403 → exit 0 with ::error::.
# ---------------------------------------------------------------------------
def test_api_403_skips_gracefully(env, monkeypatch, capsys):
m = _import_lint()
_stub_git_and_api(
monkeypatch,
m,
base_files={".gitea/workflows/ci.yml": WF_CI_BASE},
head_files={".gitea/workflows/ci.yml": WF_CI_NEW_JOB},
bp_response=("forbidden", None),
)
rc = m.run()
assert rc == 0
err = capsys.readouterr().err
assert "403" in err or "scope" in err.lower() or "token" in err.lower()
# ---------------------------------------------------------------------------
# Directive must be in the workflow YML, not PR body.
# ---------------------------------------------------------------------------
def test_directive_must_be_in_workflow_yml(env, monkeypatch, capsys):
monkeypatch = env
monkeypatch.setenv("PR_BODY", "bp-required: yes — see comment above")
m = _import_lint()
_stub_git_and_api(
monkeypatch,
m,
base_files={".gitea/workflows/ci.yml": WF_CI_BASE},
head_files={".gitea/workflows/ci.yml": WF_CI_NEW_JOB},
bp_response=("ok", {"status_check_contexts": []}),
)
rc = m.run()
# Even though PR body claims, the workflow itself lacks the directive.
assert rc == 1