493 lines
18 KiB
Python
493 lines
18 KiB
Python
#!/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.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)
|
|
self.assertIn("isMissingLabelDefinitionError", self.workflow)
|
|
self.assertNotIn(
|
|
"for (const [name, definition] of Object.entries(sizeLabelDefinitions))",
|
|
self.workflow,
|
|
)
|
|
self.assertEqual(self.workflow.count("github.rest.issues.listLabelsOnIssue"), 1)
|
|
self.assertEqual(self.workflow.count("github.rest.issues.getLabel"), 0)
|
|
|
|
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")
|
|
first_add_index = self.workflow.index("github.rest.issues.addLabels")
|
|
retry_add_index = self.workflow.index(
|
|
"github.rest.issues.addLabels", first_add_index + 1
|
|
)
|
|
list_index = self.workflow.index("github.rest.issues.listLabelsOnIssue")
|
|
self.assertLess(list_index, first_add_index)
|
|
self.assertLess(first_add_index, create_index)
|
|
self.assertLess(create_index, retry_add_index)
|
|
|
|
missing_definition = extract_function(
|
|
self.workflow, "isMissingLabelDefinitionError"
|
|
)
|
|
self.assertTrue(
|
|
run_javascript(
|
|
missing_definition,
|
|
"isMissingLabelDefinitionError",
|
|
[
|
|
{
|
|
"status": 422,
|
|
"response": {
|
|
"data": {
|
|
"errors": [
|
|
{
|
|
"resource": "Label",
|
|
"code": "invalid",
|
|
"field": "name",
|
|
"value": "small",
|
|
}
|
|
]
|
|
}
|
|
},
|
|
},
|
|
"small",
|
|
],
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
run_javascript(
|
|
missing_definition,
|
|
"isMissingLabelDefinitionError",
|
|
[
|
|
{
|
|
"status": 422,
|
|
"response": {
|
|
"data": {
|
|
"errors": [
|
|
{
|
|
"resource": "Label",
|
|
"code": "invalid",
|
|
"field": "name",
|
|
"value": "medium",
|
|
}
|
|
]
|
|
}
|
|
},
|
|
},
|
|
"small",
|
|
],
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
run_javascript(
|
|
missing_definition,
|
|
"isMissingLabelDefinitionError",
|
|
[
|
|
{
|
|
"status": 422,
|
|
"response": {
|
|
"data": {
|
|
"errors": [
|
|
{
|
|
"resource": "Issue",
|
|
"code": "invalid",
|
|
"field": "name",
|
|
"value": "small",
|
|
}
|
|
]
|
|
}
|
|
},
|
|
},
|
|
"small",
|
|
],
|
|
)
|
|
)
|
|
|
|
# @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.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()
|