ci: label pull requests by size (#3560)
This commit is contained in:
Carlos Florêncio
@@ -0,0 +1,407 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Contract tests for the pull request size label workflow."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
WORKFLOW = REPO_ROOT / ".github" / "workflows" / "pr-size-label.yml"
|
||||
LINT_WORKFLOW = REPO_ROOT / ".github" / "workflows" / "lint-action-pinning.yml"
|
||||
|
||||
|
||||
def extract_function(source: str, name: str) -> str:
|
||||
"""Return one complete JavaScript function from the trusted workflow."""
|
||||
|
||||
marker = f"function {name}("
|
||||
start = source.find(marker)
|
||||
if start < 0:
|
||||
raise AssertionError(f"{name} is missing")
|
||||
|
||||
body_start = source.find("{", start)
|
||||
if body_start < 0:
|
||||
raise AssertionError(f"{name} has no body")
|
||||
|
||||
depth = 0
|
||||
quote: str | None = None
|
||||
escaped = False
|
||||
for index in range(body_start, len(source)):
|
||||
character = source[index]
|
||||
if quote is not None:
|
||||
if escaped:
|
||||
escaped = False
|
||||
elif character == "\\":
|
||||
escaped = True
|
||||
elif character == quote:
|
||||
quote = None
|
||||
continue
|
||||
|
||||
if character in {"'", '"', "`"}:
|
||||
quote = character
|
||||
elif character == "{":
|
||||
depth += 1
|
||||
elif character == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return source[start : index + 1]
|
||||
|
||||
raise AssertionError(f"{name} has an incomplete body")
|
||||
|
||||
|
||||
def extract_call_argument(source: str, call: str) -> str:
|
||||
"""Return one balanced array/object expression passed to a call."""
|
||||
|
||||
call_start = source.find(call)
|
||||
if call_start < 0:
|
||||
raise AssertionError(f"{call} is missing")
|
||||
|
||||
expression_start = call_start + len(call)
|
||||
while expression_start < len(source) and source[expression_start].isspace():
|
||||
expression_start += 1
|
||||
|
||||
opening = source[expression_start]
|
||||
closing = {"[": "]", "{": "}"}.get(opening)
|
||||
if closing is None:
|
||||
raise AssertionError(f"{call} has no array/object argument")
|
||||
|
||||
depth = 0
|
||||
quote: str | None = None
|
||||
escaped = False
|
||||
for index in range(expression_start, len(source)):
|
||||
character = source[index]
|
||||
if quote is not None:
|
||||
if escaped:
|
||||
escaped = False
|
||||
elif character == "\\":
|
||||
escaped = True
|
||||
elif character == quote:
|
||||
quote = None
|
||||
continue
|
||||
|
||||
if character in {"'", '"', "`"}:
|
||||
quote = character
|
||||
elif character == opening:
|
||||
depth += 1
|
||||
elif character == closing:
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return source[expression_start : index + 1]
|
||||
|
||||
raise AssertionError(f"{call} has an incomplete argument")
|
||||
|
||||
|
||||
def run_javascript(function_source: str, function_name: str, values: list[object]) -> object:
|
||||
"""Execute a pure workflow helper without loading the GitHub runtime."""
|
||||
|
||||
script = (
|
||||
f"{function_source}\n"
|
||||
f"process.stdout.write(JSON.stringify({function_name}(...JSON.parse(process.argv[1]))));"
|
||||
)
|
||||
result = subprocess.run(
|
||||
["node", "-e", script, json.dumps(values)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise AssertionError(result.stderr or result.stdout)
|
||||
return json.loads(result.stdout)
|
||||
|
||||
|
||||
def run_javascript_for_each(
|
||||
function_source: str, function_name: str, values: list[object]
|
||||
) -> object:
|
||||
"""Execute a pure workflow helper once for every value."""
|
||||
|
||||
script = (
|
||||
f"{function_source}\n"
|
||||
"const values = JSON.parse(process.argv[1]);"
|
||||
f"process.stdout.write(JSON.stringify(values.map(value => {function_name}(value))));"
|
||||
)
|
||||
result = subprocess.run(
|
||||
["node", "-e", script, json.dumps(values)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise AssertionError(result.stderr or result.stdout)
|
||||
return json.loads(result.stdout)
|
||||
|
||||
|
||||
class PullRequestSizeLabelWorkflowContractTest(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.assertTrue(WORKFLOW.is_file(), "Pull request size workflow is missing")
|
||||
self.workflow = WORKFLOW.read_text(encoding="utf-8")
|
||||
|
||||
# @covers AC-CI-PR-SIZE-001.1, AC-CI-PR-SIZE-001.9
|
||||
def test_recalculates_on_current_pull_request_events_and_serializes_by_number(self) -> None:
|
||||
trigger = self.workflow.partition("on:\n")[2].partition("\nconcurrency:")[0]
|
||||
self.assertEqual(
|
||||
trigger,
|
||||
" pull_request_target: # zizmor: ignore[dangerous-triggers] "
|
||||
"base-controlled metadata-only workflow\n"
|
||||
" types: [opened, reopened, synchronize]\n",
|
||||
)
|
||||
self.assertRegex(
|
||||
self.workflow,
|
||||
r"group: pr-size-label-\$\{\{ github\.event\.pull_request\.number \}\}",
|
||||
)
|
||||
self.assertIn("cancel-in-progress: false", self.workflow)
|
||||
|
||||
# @covers AC-CI-PR-SIZE-001.2
|
||||
def test_classifier_assigns_boundaries_and_confirmed_examples(self) -> None:
|
||||
classifier = extract_function(self.workflow, "classifySize")
|
||||
expected_boundaries = {
|
||||
0: "small",
|
||||
10: "small",
|
||||
11: "medium",
|
||||
50: "medium",
|
||||
51: "big",
|
||||
}
|
||||
for count, expected in expected_boundaries.items():
|
||||
with self.subTest(count=count):
|
||||
self.assertEqual(run_javascript(classifier, "classifySize", [count]), expected)
|
||||
|
||||
examples = {
|
||||
"small": (2, 7),
|
||||
"medium": (15, 17, 41),
|
||||
"big": (80, 96),
|
||||
}
|
||||
for expected, counts in examples.items():
|
||||
for count in counts:
|
||||
with self.subTest(expected=expected, count=count):
|
||||
self.assertEqual(run_javascript(classifier, "classifySize", [count]), expected)
|
||||
|
||||
# @covers AC-CI-PR-SIZE-001.3, AC-CI-PR-SIZE-001.4
|
||||
def test_file_classifier_counts_application_and_translation_paths_only(self) -> None:
|
||||
classifier = extract_function(self.workflow, "isCountedFile")
|
||||
counted = [
|
||||
"apps/backend/internal/workflow/workflow.go",
|
||||
"apps/backend/internal/workflow/workflow_test.go",
|
||||
"apps/backend/internal/common/scripts/scripts.go",
|
||||
"apps/web/src/components/task/task.tsx",
|
||||
"apps/web/src/vite-preload-recovery.ts",
|
||||
"apps/web/src/vite-preload-recovery.test.ts",
|
||||
"apps/web/vitest-environment.test.tsx",
|
||||
"apps/web/vitest-monaco-editor.test.ts",
|
||||
"apps/web/src/locales/en/common.json",
|
||||
"apps/web/src/locales/zh-cn/common.json",
|
||||
"apps/desktop/src-tauri/src/main.rs",
|
||||
]
|
||||
excluded = [
|
||||
"docs/specs/ci/requirements/pull-request-size-labels.md",
|
||||
".github/workflows/pr-size-label.yml",
|
||||
"scripts/pr-size-label.py",
|
||||
"apps/web/src/assets/icon.svg",
|
||||
"apps/web/src/assets/compiled.ts",
|
||||
"apps/web/lib/assets/icons.ts",
|
||||
"apps/web/public/style.css",
|
||||
"apps/backend/internal/notifications/providers/assets/icon.ts",
|
||||
"apps/desktop/src-tauri/icons/generated.ts",
|
||||
"apps/web/src/locales/en/common.yaml",
|
||||
"apps/web/node_modules/example/index.ts",
|
||||
"apps/backend/internal/webapp/embedded/generated/model.go",
|
||||
"apps/web/generated/model.ts",
|
||||
"apps/backend/scripts/check.go",
|
||||
"apps/backend/internal/agentctl/server/api/scripts/inspector.js",
|
||||
"apps/web/e2e/scripts/check.ts",
|
||||
"apps/web/scripts/check.ts",
|
||||
"apps/backend/cmd/sqlguard/check.go",
|
||||
"apps/backend/internal/db/sqlguard/check.go",
|
||||
"apps/web/eslint-rules/no-literal.ts",
|
||||
"apps/commitlint.config.mjs",
|
||||
"apps/prettier.config.mjs",
|
||||
"apps/web/eslint.config.mjs",
|
||||
"apps/web/e2e/playwright.config.ts",
|
||||
"apps/web/postcss.config.mjs",
|
||||
"apps/web/tailwind.config.ts",
|
||||
]
|
||||
|
||||
self.assertEqual(
|
||||
run_javascript_for_each(classifier, "isCountedFile", counted),
|
||||
[True] * len(counted),
|
||||
)
|
||||
self.assertEqual(
|
||||
run_javascript_for_each(classifier, "isCountedFile", excluded),
|
||||
[False] * len(excluded),
|
||||
)
|
||||
|
||||
# @covers AC-CI-PR-SIZE-001.3, AC-CI-PR-SIZE-001.4
|
||||
def test_file_filter_is_explicit_and_normalizes_filenames(self) -> None:
|
||||
self.assertIn("typeof filename !== 'string'", self.workflow)
|
||||
self.assertIn("filename.startsWith('apps/')", self.workflow)
|
||||
self.assertIn("filename.split('/')", self.workflow)
|
||||
self.assertIn("pathSegments.includes('node_modules')", self.workflow)
|
||||
for directory in (
|
||||
"apps/backend/scripts/",
|
||||
"apps/backend/internal/agentctl/server/api/scripts/",
|
||||
"apps/backend/internal/webapp/embedded/generated/",
|
||||
"apps/web/e2e/scripts/",
|
||||
"apps/web/generated/",
|
||||
"apps/web/scripts/",
|
||||
):
|
||||
self.assertIn(directory, self.workflow)
|
||||
for directory in (
|
||||
"apps/backend/internal/notifications/providers/assets/",
|
||||
"apps/desktop/src-tauri/icons/",
|
||||
"apps/web/lib/assets/",
|
||||
"apps/web/public/",
|
||||
"apps/web/src/assets/",
|
||||
):
|
||||
self.assertIn(directory, self.workflow)
|
||||
for extension in (
|
||||
"c",
|
||||
"cc",
|
||||
"cpp",
|
||||
"css",
|
||||
"go",
|
||||
"graphql",
|
||||
"gql",
|
||||
"h",
|
||||
"html",
|
||||
"js",
|
||||
"jsx",
|
||||
"mjs",
|
||||
"mts",
|
||||
"proto",
|
||||
"rs",
|
||||
"scss",
|
||||
"sql",
|
||||
"ts",
|
||||
"tsx",
|
||||
):
|
||||
self.assertRegex(self.workflow, rf"['\"]{extension}['\"]")
|
||||
self.assertIn(r"apps\/web\/src\/locales\/[^/]+\/[^/]+\.json", self.workflow)
|
||||
|
||||
# @covers AC-CI-PR-SIZE-001.5, AC-CI-PR-SIZE-001.6
|
||||
def test_label_definitions_and_convergence_preserve_unrelated_labels(self) -> None:
|
||||
self.assertIn("const sizeLabelDefinitions =", self.workflow)
|
||||
for name, color in (
|
||||
("small", "0E8A16"),
|
||||
("medium", "FBCA04"),
|
||||
("big", "D93F0B"),
|
||||
):
|
||||
self.assertRegex(self.workflow, rf"{name}: \{{")
|
||||
self.assertIn(f"color: '{color}'", self.workflow)
|
||||
self.assertIn("github.rest.issues.getLabel", self.workflow)
|
||||
self.assertIn("github.rest.issues.createLabel", self.workflow)
|
||||
self.assertIn("github.rest.issues.listLabelsOnIssue", self.workflow)
|
||||
self.assertIn("github.rest.issues.addLabels", self.workflow)
|
||||
self.assertIn("github.rest.issues.removeLabel", self.workflow)
|
||||
|
||||
stale_labels = extract_function(self.workflow, "staleSizeLabels")
|
||||
self.assertEqual(
|
||||
run_javascript(stale_labels, "staleSizeLabels", [["bug", "medium", "big"], "small"]),
|
||||
["medium", "big"],
|
||||
)
|
||||
self.assertEqual(
|
||||
run_javascript(stale_labels, "staleSizeLabels", [["bug", "small"], "small"]),
|
||||
[],
|
||||
)
|
||||
|
||||
add_index = self.workflow.index("github.rest.issues.addLabels")
|
||||
remove_index = self.workflow.index("github.rest.issues.removeLabel")
|
||||
self.assertLess(add_index, remove_index)
|
||||
|
||||
already_exists = extract_function(self.workflow, "isAlreadyExistingLabelError")
|
||||
self.assertTrue(
|
||||
run_javascript(
|
||||
already_exists,
|
||||
"isAlreadyExistingLabelError",
|
||||
[
|
||||
{
|
||||
"status": 422,
|
||||
"response": {
|
||||
"data": {"errors": [{"code": "already_exists"}]}
|
||||
},
|
||||
}
|
||||
],
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
run_javascript(
|
||||
already_exists,
|
||||
"isAlreadyExistingLabelError",
|
||||
[{"status": 422, "response": {"data": {"errors": [{"code": "invalid"}]}}}],
|
||||
)
|
||||
)
|
||||
create_index = self.workflow.index("github.rest.issues.createLabel")
|
||||
retry_read_index = self.workflow.index(
|
||||
"await github.rest.issues.getLabel", create_index
|
||||
)
|
||||
self.assertLess(create_index, retry_read_index)
|
||||
|
||||
# @covers AC-CI-PR-SIZE-001.5
|
||||
def test_summary_row_contains_two_string_cells(self) -> None:
|
||||
summary_row = extract_function(self.workflow, "summaryTableRow")
|
||||
self.assertEqual(
|
||||
run_javascript(summary_row, "summaryTableRow", [4, "small"]),
|
||||
["4", "small"],
|
||||
)
|
||||
table_expression = extract_call_argument(self.workflow, "core.summary.addTable(")
|
||||
script = (
|
||||
f"{summary_row}\n"
|
||||
"const countedFileCount = 4;\n"
|
||||
"const targetLabel = 'small';\n"
|
||||
f"const table = {table_expression};\n"
|
||||
"process.stdout.write(JSON.stringify(table));"
|
||||
)
|
||||
result = subprocess.run(
|
||||
["node", "-e", script],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise AssertionError(result.stderr or result.stdout)
|
||||
table = json.loads(result.stdout)
|
||||
self.assertEqual(table[1], ["4", "small"])
|
||||
self.assertTrue(all(isinstance(cell, str) for cell in table[1]))
|
||||
|
||||
# @covers AC-CI-PR-SIZE-001.7
|
||||
def test_incomplete_file_results_fail_before_label_mutations(self) -> None:
|
||||
self.assertIn("github.paginate(", self.workflow)
|
||||
self.assertIn("github.rest.pulls.listFiles", self.workflow)
|
||||
self.assertIn("per_page: 100", self.workflow)
|
||||
self.assertIn("pull_number: context.payload.pull_request.number", self.workflow)
|
||||
limit_index = self.workflow.index("changedFiles.length >= 3000")
|
||||
self.assertIn("throw new Error", self.workflow[limit_index : limit_index + 240])
|
||||
for mutation in (
|
||||
"github.rest.issues.getLabel",
|
||||
"github.rest.issues.createLabel",
|
||||
"github.rest.issues.addLabels",
|
||||
"github.rest.issues.removeLabel",
|
||||
):
|
||||
self.assertLess(limit_index, self.workflow.index(mutation))
|
||||
|
||||
# @covers AC-CI-PR-SIZE-001.8
|
||||
def test_workflow_uses_trusted_least_privilege_and_pinned_execution(self) -> None:
|
||||
permissions = self.workflow.partition("permissions:\n")[2].partition("\njobs:")[0]
|
||||
self.assertEqual("pull-requests: write", permissions.strip())
|
||||
self.assertNotIn("actions/checkout", self.workflow)
|
||||
self.assertNotIn("\n - run:", self.workflow)
|
||||
self.assertNotIn("child_process", self.workflow)
|
||||
self.assertNotIn("eval(", self.workflow)
|
||||
self.assertRegex(
|
||||
self.workflow,
|
||||
r"uses: actions/github-script@[0-9a-f]{40} # v9\.[0-9]+\.[0-9]+",
|
||||
)
|
||||
|
||||
def test_contract_is_registered_in_the_required_lint_workflow(self) -> None:
|
||||
lint_workflow = LINT_WORKFLOW.read_text(encoding="utf-8")
|
||||
self.assertRegex(
|
||||
lint_workflow,
|
||||
r"(?m)^ - name: Test pull request size label workflow contract$\n"
|
||||
r"^ run: python3 .github/scripts/pr-size-label-workflow-contract_test.py$",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user