yorickr/plezy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

build: make toolchain inputs reproducible

Browse Source
This commit is contained in:
edde746
2026-07-24 03:40:06 +02:00
parent 1d1c301f61
commit 658da37b48
19 changed files with 1849 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/check_apple_spm_locks.py
Executable
+161
View File
@@ -0,0 +1,161 @@
#!/usr/bin/env python3
"""Validate duplicate Apple SwiftPM locks against resolved Flutter plugins."""
from __future__ import annotations
import argparse
import json
import re
from pathlib import Path
from urllib.parse import unquote, urlparse
LOCK_PAIRS = {
"iOS": (
Path("ios/Runner.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved"),
Path("ios/Runner.xcworkspace/xcshareddata/swiftpm/Package.resolved"),
),
"macOS": (
Path("macos/Runner.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved"),
Path("macos/Runner.xcworkspace/xcshareddata/swiftpm/Package.resolved"),
),
}
SENTRY_IDENTITY = "sentry-cocoa"
EXACT_REQUIREMENT = re.compile(
r"\.package\s*\(\s*url\s*:\s*[\"'][^\"']*sentry-cocoa(?:\.git)?[\"']\s*,\s*exact\s*:\s*[\"']([^\"']+)[\"']",
re.DOTALL,
)
def _load_pin_map(path: Path, errors: list[str]) -> dict[str, dict]:
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
errors.append(f"{path}: cannot read version-2 SwiftPM lock: {error}")
return {}
if not isinstance(payload, dict) or payload.get("version") != 2 or not isinstance(payload.get("pins"), list):
errors.append(f"{path}: expected a version-2 SwiftPM lock with a pins array")
return {}
pins: dict[str, dict] = {}
for pin in payload["pins"]:
identity = pin.get("identity") if isinstance(pin, dict) else None
if not isinstance(identity, str):
errors.append(f"{path}: pin without a string identity")
continue
canonical = {
"kind": pin.get("kind"),
"location": pin.get("location"),
"state": pin.get("state"),
}
if identity in pins:
errors.append(f"{path}: duplicate pin identity {identity}")
pins[identity] = canonical
return pins
def _resolved_sentry_root(root: Path, errors: list[str]) -> Path | None:
config_path = root / ".dart_tool/package_config.json"
try:
config = json.loads(config_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
errors.append(f"{config_path}: unavailable ({error}); run flutter pub get first")
return None
packages = config.get("packages") if isinstance(config, dict) else None
if not isinstance(packages, list):
errors.append(f"{config_path}: missing packages list; run flutter pub get first")
return None
package = next(
(item for item in packages if isinstance(item, dict) and item.get("name") == "sentry_flutter"),
None,
)
if package is None or not isinstance(package.get("rootUri"), str):
errors.append(f"{config_path}: sentry_flutter is not resolved; run flutter pub get first")
return None
uri = package["rootUri"]
parsed = urlparse(uri)
if parsed.scheme == "file":
return Path(unquote(parsed.path))
if parsed.scheme:
errors.append(f"{config_path}: unsupported sentry_flutter root URI {uri!r}")
return None
return (config_path.parent / unquote(uri)).resolve()
def _manifest_requirement(path: Path, errors: list[str]) -> str | None:
try:
source = path.read_text(encoding="utf-8")
except OSError as error:
errors.append(f"{path}: unavailable ({error}); run flutter pub get first")
return None
versions = EXACT_REQUIREMENT.findall(source)
if len(versions) != 1:
errors.append(f"{path}: expected exactly one exact sentry-cocoa requirement, found {len(versions)}")
return None
return versions[0]
def validate(root: Path) -> list[str]:
root = root.resolve()
errors: list[str] = []
lock_maps: dict[Path, dict[str, dict]] = {}
for platform, relative_paths in LOCK_PAIRS.items():
first_path, second_path = (root / path for path in relative_paths)
first = _load_pin_map(first_path, errors)
second = _load_pin_map(second_path, errors)
lock_maps[first_path] = first
lock_maps[second_path] = second
if first != second:
identities = sorted(set(first) | set(second))
differing = [identity for identity in identities if first.get(identity) != second.get(identity)]
errors.append(
f"{platform} SwiftPM locks differ between {first_path} and {second_path}: "
+ ", ".join(differing)
)
sentry_root = _resolved_sentry_root(root, errors)
required_versions: list[str] = []
if sentry_root is not None:
for platform in ("ios", "macos"):
version = _manifest_requirement(sentry_root / platform / "sentry_flutter/Package.swift", errors)
if version is not None:
required_versions.append(version)
if len(set(required_versions)) > 1:
errors.append("iOS and macOS sentry_flutter manifests require different sentry-cocoa versions")
required_version = required_versions[0] if required_versions and len(set(required_versions)) == 1 else None
sentry_pins: list[tuple[Path, dict]] = []
for path, pins in lock_maps.items():
pin = pins.get(SENTRY_IDENTITY)
if pin is None:
errors.append(f"{path}: missing {SENTRY_IDENTITY} pin")
continue
sentry_pins.append((path, pin))
state = pin.get("state")
version = state.get("version") if isinstance(state, dict) else None
if required_version is not None and version != required_version:
errors.append(f"{path}: {SENTRY_IDENTITY} is {version!r}, manifest requires exactly {required_version}")
if sentry_pins:
canonical_path, canonical_pin = sentry_pins[0]
for path, pin in sentry_pins[1:]:
if pin != canonical_pin:
errors.append(f"{path}: {SENTRY_IDENTITY} state differs from {canonical_path}")
return errors
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[1])
args = parser.parse_args(argv)
errors = validate(args.root)
if errors:
for error in errors:
print(f"error: {error}")
return 1
print("Apple SwiftPM locks match each other and the resolved Sentry manifests.")
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/check_codegen.py
Executable
+144
View File
@@ -0,0 +1,144 @@
#!/usr/bin/env python3
"""Check generated outputs in an isolated worktree without mutating the caller."""
from __future__ import annotations
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
EXPLICIT_GENERATED = {"server/relay_protocol_gen.go"}
DEPENDENCY_STATE = ("package_config.json", "package_graph.json")
def _run(root: Path, *args: str) -> subprocess.CompletedProcess[bytes]:
return subprocess.run(args, cwd=root, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
def _is_generated(relative: str) -> bool:
return relative in EXPLICIT_GENERATED or (
relative.startswith("lib/") and (relative.endswith(".g.dart") or relative.endswith(".freezed.dart"))
)
def _generated_state(root: Path) -> dict[str, bytes]:
state = {}
lib = root / "lib"
if lib.exists():
for path in lib.rglob("*.dart"):
relative = path.relative_to(root).as_posix()
if _is_generated(relative) and path.is_file():
state[relative] = path.read_bytes()
for relative in EXPLICIT_GENERATED:
path = root / relative
if path.is_file():
state[relative] = path.read_bytes()
return state
def _nul_paths(result: bytes) -> list[str]:
return [os.fsdecode(value) for value in result.split(b"\0") if value]
def _copy_overlay_path(source_root: Path, target_root: Path, relative: str) -> None:
source = source_root / relative
target = target_root / relative
if not source.exists() and not source.is_symlink():
if target.is_dir() and not target.is_symlink():
shutil.rmtree(target)
else:
target.unlink(missing_ok=True)
return
target.parent.mkdir(parents=True, exist_ok=True)
if target.is_dir() and not target.is_symlink():
shutil.rmtree(target)
else:
target.unlink(missing_ok=True)
if source.is_symlink():
target.symlink_to(os.readlink(source))
else:
shutil.copy2(source, target)
def _copy_dependency_state(source_root: Path, target_root: Path) -> None:
source = source_root / ".dart_tool"
target = target_root / ".dart_tool"
for relative in DEPENDENCY_STATE:
source_path = source / relative
if source_path.is_file():
target.mkdir(parents=True, exist_ok=True)
shutil.copy2(source_path, target / relative)
def _create_isolated_checkout(root: Path, destination: Path) -> None:
_run(root, "git", "worktree", "add", "--detach", "--quiet", str(destination), "HEAD")
changed = _nul_paths(_run(root, "git", "diff", "--name-only", "-z", "HEAD", "--").stdout)
untracked = _nul_paths(_run(root, "git", "ls-files", "--others", "--exclude-standard", "-z").stdout)
for relative in sorted(set(changed + untracked)):
_copy_overlay_path(root, destination, relative)
_copy_dependency_state(root, destination)
def _dirty_generated_paths(root: Path) -> set[str]:
changed = _nul_paths(_run(root, "git", "diff", "--name-only", "-z", "--").stdout)
untracked = _nul_paths(
_run(
root,
"git",
"ls-files",
"--others",
"--exclude-standard",
"-z",
"--",
"lib",
"server/relay_protocol_gen.go",
).stdout
)
return {relative for relative in changed + untracked if _is_generated(relative)}
def main(argv: list[str] | None = None) -> int:
arguments = list(sys.argv[1:] if argv is None else argv)
root = Path.cwd().resolve()
caller_state = _generated_state(root)
temporary = Path(tempfile.mkdtemp(prefix="plezy-codegen-check-"))
checkout = temporary / "checkout"
registered = True
try:
_create_isolated_checkout(root, checkout)
result = subprocess.run(["bash", "scripts/codegen.sh", *arguments], cwd=checkout, check=False)
if result.returncode != 0:
return result.returncode
expected_state = _generated_state(checkout)
stale = {
relative
for relative in caller_state.keys() | expected_state.keys()
if caller_state.get(relative) != expected_state.get(relative)
}
stale.update(_dirty_generated_paths(root))
if stale:
print("Generated files are out of date:", file=sys.stderr)
for relative in sorted(stale):
print(f" {relative}", file=sys.stderr)
print("Run 'scripts/codegen.sh' and commit the result.", file=sys.stderr)
return 1
return 0
finally:
if registered:
subprocess.run(
["git", "worktree", "remove", "--force", str(checkout)],
cwd=root,
check=False,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
shutil.rmtree(temporary, ignore_errors=True)
if __name__ == "__main__":
raise SystemExit(main())
scripts/check_workflow_action_pins.py
Executable
+373
View File
@@ -0,0 +1,373 @@
#!/usr/bin/env python3
"""Require immutable commit pins for remote GitHub Actions dependencies."""
from __future__ import annotations
import re
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
WORKFLOWS = ROOT / ".github" / "workflows"
MAPPING_RE = re.compile(
r"""^\s*(?:-\s*)?(?P<key>uses|'(?:''|[^'])*'|"(?:\\.|[^"\\])*")\s*:\s*(?P<value>.*?)\s*$"""
)
EXPLICIT_KEY_RE = re.compile(
r"""^\s*(?:-\s*)?\?\s*(?P<key>uses|'(?:''|[^'])*'|"(?:\\.|[^"\\])*")\s*$"""
)
EXPLICIT_VALUE_RE = re.compile(r"^\s*:\s*(?P<value>.*?)\s*$")
REMOTE_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+(?:/[A-Za-z0-9_./-]+)?@[0-9a-fA-F]{40}$")
BLOCK_SCALAR_RE = re.compile(r":\s*[|>](?:[1-9][+-]?|[+-][1-9]?)?\s*(?:#.*)?$")
BLOCK_SCALAR_VALUE_RE = re.compile(r"^[|>](?:[1-9][+-]?|[+-][1-9]?)?$")
YAML_DOUBLE_ESCAPES = {
"0": "\0",
"a": "\a",
"b": "\b",
"t": "\t",
"\t": "\t",
"n": "\n",
"v": "\v",
"f": "\f",
"r": "\r",
"e": "\x1b",
" ": " ",
'"': '"',
"/": "/",
"\\": "\\",
"N": "\u0085",
"_": "\u00a0",
"L": "\u2028",
"P": "\u2029",
}
def iter_workflow_files(directory: Path = WORKFLOWS):
yield from sorted((*directory.glob("*.yml"), *directory.glob("*.yaml")))
def _strip_yaml_comment(value: str) -> str:
quote = None
escaped = False
for index, char in enumerate(value):
if escaped:
escaped = False
continue
if char == "\\" and quote == '"':
escaped = True
continue
if char in ("'", '"'):
if quote is None:
quote = char
elif quote == char:
quote = None
continue
if char == "#" and quote is None and (index == 0 or value[index - 1].isspace()):
return value[:index].rstrip()
return value.rstrip()
def _decode_quoted_yaml_string(value: str) -> str | None:
if len(value) < 2 or value[0] != value[-1] or value[0] not in ("'", '"'):
return None
if value[0] == "'":
return value[1:-1].replace("''", "'")
decoded = []
index = 1
end = len(value) - 1
while index < end:
char = value[index]
if char != "\\":
decoded.append(char)
index += 1
continue
index += 1
if index >= end:
return None
escape = value[index]
if escape in YAML_DOUBLE_ESCAPES:
decoded.append(YAML_DOUBLE_ESCAPES[escape])
index += 1
continue
width = {"x": 2, "u": 4, "U": 8}.get(escape)
if width is None or index + width >= end:
return None
digits = value[index + 1 : index + 1 + width]
if not re.fullmatch(rf"[0-9a-fA-F]{{{width}}}", digits):
return None
try:
decoded.append(chr(int(digits, 16)))
except ValueError:
return None
index += width + 1
return "".join(decoded)
def _unquote(value: str) -> str:
decoded = _decode_quoted_yaml_string(value)
return value if decoded is None else decoded
def _flow_value(line: str, start: int, mapping_depth: int) -> str:
index = start
quote = None
escaped = False
depth = mapping_depth
while index < len(line):
char = line[index]
if escaped:
escaped = False
elif char == "\\" and quote == '"':
escaped = True
elif quote is not None:
if char == quote:
quote = None
elif char in ("'", '"'):
quote = char
elif char in ("{", "["):
depth += 1
elif char in ("}", "]"):
if depth == mapping_depth:
break
depth -= 1
elif char == "," and depth == mapping_depth:
break
index += 1
return _unquote(line[start:index].strip())
def _has_unsupported_block_mapping_key(line: str) -> bool:
candidate = line.lstrip()
if candidate.startswith("-") and not candidate.startswith("---"):
candidate = candidate[1:].lstrip()
if not candidate:
return False
if candidate[0] in "!&*":
return True
if candidate[0] not in ("'", '"'):
return False
quote = candidate[0]
escaped = False
index = 1
while index < len(candidate):
char = candidate[index]
if quote == "'" and char == "'" and index + 1 < len(candidate) and candidate[index + 1] == "'":
index += 2
continue
if escaped:
escaped = False
elif quote == '"' and char == "\\":
escaped = True
elif char == quote:
return False
index += 1
return True
def _flow_uses_references(line: str, initial_depth: int) -> tuple[list[str], int]:
references = []
depth = initial_depth
index = 0
while index < len(line):
char = line[index]
if char in ("'", '"'):
quote = char
escaped = False
end = index + 1
while end < len(line):
quoted_char = line[end]
if escaped:
escaped = False
elif quoted_char == "\\" and quote == '"':
escaped = True
elif quoted_char == quote:
break
end += 1
if end >= len(line):
if depth > 0:
references.append("<unsupported multiline flow scalar>")
return references, depth
key = _decode_quoted_yaml_string(line[index : end + 1])
after_key = end + 1
while after_key < len(line) and line[after_key].isspace():
after_key += 1
if depth > 0 and after_key < len(line) and line[after_key] == ":":
if key == "uses":
references.append(_flow_value(line, after_key + 1, depth))
elif key is None:
references.append("<unsupported quoted flow mapping key>")
index = end + 1
continue
if line.startswith("${{", index):
expression_end = line.find("}}", index + 3)
if expression_end < 0:
references.append("<unterminated GitHub expression>")
return references, depth
index = expression_end + 2
continue
if char in ("{", "["):
depth += 1
index += 1
continue
if char in ("}", "]"):
depth = max(0, depth - 1)
index += 1
continue
if depth > 0 and char == "?":
references.append("<unsupported explicit flow mapping>")
index += 1
continue
if depth > 0 and char in "!&*":
references.append("<unsupported tagged, anchored, or aliased flow mapping>")
index += 1
continue
if depth > 0 and (char.isalpha() or char == "_"):
end = index + 1
while end < len(line) and (line[end].isalnum() or line[end] in "_-"):
end += 1
after_key = end
while after_key < len(line) and line[after_key].isspace():
after_key += 1
if line[index:end] == "uses" and after_key < len(line) and line[after_key] == ":":
references.append(_flow_value(line, after_key + 1, depth))
index = end
continue
index += 1
return references, depth
def iter_uses_references(path: Path):
block_parent_indent = None
block_content_indent = None
block_uses_line = None
block_uses_content: list[str] = []
explicit_uses_line = None
flow_start_line = None
flow_depth = 0
for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
stripped = raw_line.lstrip()
indent = len(raw_line) - len(stripped)
if block_parent_indent is not None:
if not stripped:
if block_uses_line is not None:
block_uses_content.append("")
continue
if indent <= block_parent_indent:
if block_uses_line is not None:
yield block_uses_line, "\n".join(block_uses_content).strip()
block_parent_indent = None
block_content_indent = None
block_uses_line = None
block_uses_content = []
elif block_content_indent is None:
block_content_indent = indent
if block_uses_line is not None:
block_uses_content.append(raw_line[block_content_indent:])
continue
elif indent >= block_content_indent:
if block_uses_line is not None:
block_uses_content.append(raw_line[block_content_indent:])
continue
else:
if block_uses_line is not None:
yield block_uses_line, "\n".join(block_uses_content).strip()
block_parent_indent = None
block_content_indent = None
block_uses_line = None
block_uses_content = []
if stripped.startswith("#") or not stripped:
continue
active_line = _strip_yaml_comment(raw_line)
if explicit_uses_line is not None:
explicit_value = EXPLICIT_VALUE_RE.match(active_line)
if explicit_value is None:
yield explicit_uses_line, "<missing explicit mapping value>"
else:
value = explicit_value.group("value").strip()
if BLOCK_SCALAR_VALUE_RE.fullmatch(value):
block_parent_indent = indent
block_content_indent = None
block_uses_line = explicit_uses_line
block_uses_content = []
else:
yield explicit_uses_line, _unquote(value)
explicit_uses_line = None
continue
explicit_uses_line = None
explicit_key = EXPLICIT_KEY_RE.match(active_line)
if explicit_key:
if _unquote(explicit_key.group("key")) == "uses":
explicit_uses_line = line_number
continue
if re.match(r"^\s*(?:-\s*)?\?", active_line):
yield line_number, "<unsupported explicit mapping key>"
continue
if _has_unsupported_block_mapping_key(active_line):
yield line_number, "<unsupported multiline, tagged, anchored, or aliased mapping key>"
continue
match = MAPPING_RE.match(active_line) if flow_depth == 0 else None
if match:
key = _unquote(match.group("key"))
value = match.group("value").strip()
if BLOCK_SCALAR_VALUE_RE.fullmatch(value):
block_parent_indent = indent
block_content_indent = None
if key == "uses":
block_uses_line = line_number
block_uses_content = []
continue
if key == "uses":
yield line_number, _unquote(value)
if BLOCK_SCALAR_RE.search(raw_line):
block_parent_indent = indent
block_content_indent = None
continue
previous_flow_depth = flow_depth
flow_references, flow_depth = _flow_uses_references(active_line, flow_depth)
for reference in flow_references:
yield line_number, reference
if previous_flow_depth == 0 and flow_depth > 0:
flow_start_line = line_number
elif flow_depth == 0:
flow_start_line = None
if explicit_uses_line is not None:
yield explicit_uses_line, "<missing explicit mapping value>"
if flow_depth > 0:
yield flow_start_line or 1, "<unterminated flow collection>"
if block_uses_line is not None:
yield block_uses_line, "\n".join(block_uses_content).strip()
def validate_reference(reference: str) -> str | None:
if reference.startswith("./"):
return None
if REMOTE_RE.fullmatch(reference):
return None
return "remote actions must use a full 40-character commit SHA"
def main(argv: list[str] | None = None) -> int:
args = list(sys.argv[1:] if argv is None else argv)
paths = [Path(value) for value in args] if args else list(iter_workflow_files())
violations = []
for path in paths:
for line_number, reference in iter_uses_references(path):
reason = validate_reference(reference)
if reason:
try:
display_path = path.resolve().relative_to(ROOT)
except ValueError:
display_path = path
violations.append(f"{display_path}:{line_number}: {reference!r}: {reason}")
if violations:
print("Mutable or malformed GitHub Actions references:", file=sys.stderr)
for violation in violations:
print(f" {violation}", file=sys.stderr)
return 1
print(f"Workflow action pins verified ({len(paths)} files).")
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/ci_checks.sh
+7 -1
View File
@@ -83,8 +83,14 @@ fi
# 4. Workflow and script regression guards
section "workflow and script guards"
if python3 scripts/check_build_workflow.py &&
python3 scripts/check_apple_spm_locks.py &&
python3 scripts/test_check_apple_spm_locks.py &&
python3 scripts/check_workflow_security.py &&
python3 scripts/test_check_workflow_security.py &&
python3 scripts/check_workflow_action_pins.py &&
python3 scripts/test_check_workflow_action_pins.py &&
python3 scripts/test_check_codegen.py &&
python3 scripts/test_format_native.py &&
python3 scripts/check_update_packages_workflow.py &&
python3 scripts/test_pubspec_version.py &&
python3 scripts/test_clean_translations.py &&
@@ -111,7 +117,7 @@ out="$(mktemp)"
if scripts/format_native.sh --check >"$out" 2>&1; then
ok "native files correctly formatted"
else
fail "native formatting issues"
fail "native formatting check failed"
sed 's/^/ /' "$out"
FAILED=1
fi
scripts/codegen.sh
+1 -19
View File
@@ -2,29 +2,11 @@
set -euo pipefail
cd "$(dirname "${BASH_SOURCE[0]}")/.."
check=false
if [[ "${1:-}" == "--check" ]]; then
check=true
shift
exec python3 scripts/check_codegen.py "$@"
fi
python3 scripts/generate_relay_protocol.py
dart run slang
dart run build_runner build --delete-conflicting-outputs "$@"
if $check; then
generated_changes="$({
git diff --name-only -- lib server/relay_protocol_gen.go
git ls-files --others --exclude-standard -- \
':(glob)lib/**/*.g.dart' \
':(glob)lib/**/*.freezed.dart' \
server/relay_protocol_gen.go
} | grep -E '(\.(g|freezed)\.dart|relay_protocol_gen\.go)$' || true)"
if [[ -n "$generated_changes" ]]; then
echo "Generated files are out of date:" >&2
printf ' %s\n' "$generated_changes" >&2
echo "Run 'scripts/codegen.sh' and commit the result." >&2
exit 1
fi
fi
scripts/format_native.sh
+121 -11
View File
@@ -20,9 +20,22 @@ case "${1:---check}" in
;;
esac
KTLINT_VERSION="${KTLINT_VERSION:-1.5.0}"
KTLINT_VERSION="1.5.0"
KTLINT_SHA256="a16be01dcc480aab2f55f444b620142152f66e31564b3b9376506d624c28a2ad"
KTLINT_URL="https://github.com/ktlint/ktlint/releases/download/$KTLINT_VERSION/ktlint"
KTLINT_BIN="$ROOT/.dart_tool/native-format/ktlint-$KTLINT_VERSION"
KTLINT_TMP=""
JAVA_BIN_DIR=""
cleanup() {
if [ -n "$KTLINT_TMP" ]; then
rm -f -- "$KTLINT_TMP"
fi
}
trap cleanup EXIT
trap 'exit 129' HUP
trap 'exit 130' INT
trap 'exit 143' TERM
has_command() {
command -v "$1" >/dev/null 2>&1
}
@@ -51,22 +64,118 @@ run_swift_format() {
fi
}
java_diagnostic() {
echo "A working JDK 17+ is required for Kotlin formatting. Set JAVA_HOME to a JDK 17+ installation or put java on PATH." >&2
}
resolve_java() {
local candidate output version major
if [ -n "${JAVA_HOME:-}" ]; then
candidate="$JAVA_HOME/bin/java"
else
candidate="$(command -v java 2>/dev/null || true)"
fi
if [ -z "$candidate" ] || [ ! -x "$candidate" ]; then
java_diagnostic
return 127
fi
if [[ "$candidate" != /* ]]; then
candidate="$PWD/$candidate"
fi
if ! output="$("$candidate" -version 2>&1)"; then
java_diagnostic
return 127
fi
if [[ "$output" =~ version[[:space:]]+\"([^\"]+)\" ]]; then
version="${BASH_REMATCH[1]}"
else
java_diagnostic
return 127
fi
if [[ "$version" =~ ^1\.([0-9]+)([._-]|$) ]]; then
major="${BASH_REMATCH[1]}"
elif [[ "$version" =~ ^([0-9]+)([._-]|$) ]]; then
major="${BASH_REMATCH[1]}"
else
java_diagnostic
return 127
fi
if ((major < 17)); then
java_diagnostic
return 127
fi
JAVA_BIN_DIR="${candidate%/*}"
}
sha256_file() {
local output
if has_command shasum; then
output="$(shasum -a 256 "$1")" || {
echo "Failed to calculate the ktlint SHA-256 digest with shasum." >&2
return 2
}
elif has_command sha256sum; then
output="$(sha256sum "$1")" || {
echo "Failed to calculate the ktlint SHA-256 digest with sha256sum." >&2
return 2
}
else
echo "A SHA-256 tool is required to verify ktlint (shasum or sha256sum)." >&2
return 127
fi
printf '%s\n' "${output%%[[:space:]]*}"
}
verify_ktlint() {
local digest
digest="$(sha256_file "$1")" || return $?
[[ "$digest" =~ ^[0-9a-f]{64}$ ]] && [ "$digest" = "$KTLINT_SHA256" ]
}
ensure_ktlint() {
if [ -x "$KTLINT_BIN" ]; then
return 0
local status
if ! has_command shasum && ! has_command sha256sum; then
echo "A SHA-256 tool is required to verify ktlint (shasum or sha256sum)." >&2
return 127
fi
if [ -f "$KTLINT_BIN" ]; then
if verify_ktlint "$KTLINT_BIN"; then
chmod +x "$KTLINT_BIN"
return 0
else
status=$?
if [ "$status" -ne 1 ]; then
return "$status"
fi
fi
fi
if ! has_command curl; then
echo "curl not found. Install curl to download ktlint." >&2
return 127
fi
if ! has_command java; then
echo "java not found. Install JDK 17+ to run ktlint." >&2
return 127
fi
mkdir -p "$(dirname "$KTLINT_BIN")"
curl -fsSL "https://github.com/pinterest/ktlint/releases/download/$KTLINT_VERSION/ktlint" -o "$KTLINT_BIN"
chmod +x "$KTLINT_BIN"
KTLINT_TMP="$(mktemp "$(dirname "$KTLINT_BIN")/.ktlint-$KTLINT_VERSION.XXXXXX")"
if ! curl -fsSL "$KTLINT_URL" -o "$KTLINT_TMP"; then
echo "Failed to download ktlint $KTLINT_VERSION." >&2
return 1
fi
if verify_ktlint "$KTLINT_TMP"; then
:
else
status=$?
if [ "$status" -eq 1 ]; then
echo "Downloaded ktlint $KTLINT_VERSION failed SHA-256 verification." >&2
fi
return "$status"
fi
chmod +x "$KTLINT_TMP"
mv -f "$KTLINT_TMP" "$KTLINT_BIN"
KTLINT_TMP=""
}
run_ktlint() {
PATH="$JAVA_BIN_DIR:$PATH" "$KTLINT_BIN" "$@"
}
append_native_files() {
@@ -106,11 +215,12 @@ append_native_files \
FAILED=0
if [ "${#ktlint_files[@]}" -gt 0 ]; then
resolve_java
ensure_ktlint
if [ "$MODE" = "fix" ]; then
"$KTLINT_BIN" -F "${ktlint_files[@]}"
run_ktlint -F "${ktlint_files[@]}"
else
"$KTLINT_BIN" "${ktlint_files[@]}" || FAILED=1
run_ktlint "${ktlint_files[@]}" || FAILED=1
fi
else
echo "No Kotlin files found."
scripts/test_check_apple_spm_locks.py
Executable
+121
View File
@@ -0,0 +1,121 @@
#!/usr/bin/env python3
import importlib.util
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
SCRIPT = Path(__file__).with_name("check_apple_spm_locks.py")
SPEC = importlib.util.spec_from_file_location("check_apple_spm_locks", SCRIPT)
CHECKER = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
SPEC.loader.exec_module(CHECKER)
GOOD_VERSION = "8.58.3"
GOOD_REVISION = "dad229c665bfd043c5d80ac7aa77717cbd19a1c3"
class AppleSpmLockCheckerTest(unittest.TestCase):
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
self.root = Path(self.temporary.name)
self._write_package_config()
self._write_manifests(GOOD_VERSION)
self._write_all_locks(GOOD_VERSION, GOOD_REVISION)
def tearDown(self) -> None:
self.temporary.cleanup()
def _write_package_config(self) -> None:
path = self.root / ".dart_tool/package_config.json"
path.parent.mkdir(parents=True)
path.write_text(
json.dumps(
{
"configVersion": 2,
"packages": [{"name": "sentry_flutter", "rootUri": "../packages/sentry_flutter"}],
}
),
encoding="utf-8",
)
def _write_manifests(self, version: str) -> None:
source = f'.package(url: "https://github.com/getsentry/sentry-cocoa", exact: "{version}")\n'
for platform in ("ios", "macos"):
path = self.root / f"packages/sentry_flutter/{platform}/sentry_flutter/Package.swift"
path.parent.mkdir(parents=True)
path.write_text(source, encoding="utf-8")
def _lock_payload(self, version: str, revision: str) -> dict:
return {
"originHash": "fixture",
"pins": [
{
"identity": "sentry-cocoa",
"kind": "remoteSourceControl",
"location": "https://github.com/getsentry/sentry-cocoa",
"state": {"revision": revision, "version": version},
}
],
"version": 2,
}
def _write_lock(self, relative: Path, version: str, revision: str) -> None:
path = self.root / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(self._lock_payload(version, revision)), encoding="utf-8")
def _write_all_locks(self, version: str, revision: str) -> None:
for paths in CHECKER.LOCK_PAIRS.values():
for path in paths:
self._write_lock(path, version, revision)
def test_consistent_graph_passes_library_and_cli(self) -> None:
self.assertEqual([], CHECKER.validate(self.root))
completed = subprocess.run(
[sys.executable, str(SCRIPT), "--root", str(self.root)],
check=False,
capture_output=True,
text=True,
)
self.assertEqual(0, completed.returncode, completed.stdout + completed.stderr)
self.assertIn("Apple SwiftPM locks match", completed.stdout)
def test_reports_project_workspace_version_mismatch(self) -> None:
workspace = CHECKER.LOCK_PAIRS["iOS"][1]
self._write_lock(workspace, "8.58.0", "old")
errors = CHECKER.validate(self.root)
self.assertTrue(any(str(workspace) in error and "sentry-cocoa" in error for error in errors))
self.assertTrue(any("iOS SwiftPM locks differ" in error for error in errors))
def test_reports_all_locks_stale_against_manifests(self) -> None:
self._write_all_locks("8.58.0", "old")
errors = CHECKER.validate(self.root)
self.assertEqual(4, sum("manifest requires exactly 8.58.3" in error for error in errors))
def test_reports_same_version_with_different_revision(self) -> None:
workspace = CHECKER.LOCK_PAIRS["macOS"][1]
self._write_lock(workspace, GOOD_VERSION, "different")
errors = CHECKER.validate(self.root)
self.assertTrue(any("macOS SwiftPM locks differ" in error and "sentry-cocoa" in error for error in errors))
self.assertTrue(any("state differs" in error for error in errors))
def test_reports_missing_resolved_package_inputs(self) -> None:
(self.root / ".dart_tool/package_config.json").unlink()
errors = CHECKER.validate(self.root)
self.assertTrue(any("flutter pub get first" in error for error in errors))
if __name__ == "__main__":
unittest.main()
scripts/test_check_codegen.py
+212
View File
@@ -0,0 +1,212 @@
import hashlib
import os
import shutil
import stat
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
GENERATED_PATHS = (
"lib/i18n/strings.g.dart",
"lib/models/model.freezed.dart",
"lib/models/model.g.dart",
"lib/watch_together/services/relay_protocol.g.dart",
"server/relay_protocol_gen.go",
)
def executable(path: Path, contents: str) -> None:
path.write_text(contents, encoding="utf-8")
path.chmod(path.stat().st_mode | stat.S_IXUSR)
class CodegenCheckTest(unittest.TestCase):
def setUp(self) -> None:
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name)
self.codegen_temp = Path(tempfile.mkdtemp(prefix="plezy-codegen-test-temp-"))
subprocess.run(["git", "init", "-q", str(self.root)], check=True)
subprocess.run(["git", "config", "user.email", "fixture@example.invalid"], cwd=self.root, check=True)
subprocess.run(["git", "config", "user.name", "Fixture"], cwd=self.root, check=True)
(self.root / "scripts").mkdir()
shutil.copy2(SCRIPT_DIR / "codegen.sh", self.root / "scripts" / "codegen.sh")
shutil.copy2(SCRIPT_DIR / "check_codegen.py", self.root / "scripts" / "check_codegen.py")
(self.root / "scripts" / "generate_relay_protocol.py").write_text("fixture\n", encoding="utf-8")
(self.root / "source.txt").write_text("version one\n", encoding="utf-8")
for relative in GENERATED_PATHS:
path = self.root / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("version one\n", encoding="utf-8")
self.bin = self.root / "fake-bin"
self.bin.mkdir()
executable(
self.bin / "python3",
"""#!/usr/bin/env bash
if [[ "$1" == *check_codegen.py ]]; then exec "$REAL_PYTHON" "$@"; fi
mkdir -p lib/watch_together/services server
cp source.txt lib/watch_together/services/relay_protocol.g.dart
cp source.txt server/relay_protocol_gen.go
""",
)
executable(
self.bin / "dart",
"""#!/usr/bin/env bash
if [ "${FAIL_DART:-0}" -ne 0 ]; then exit "$FAIL_DART"; fi
case "$*" in
"run slang")
mkdir -p lib/i18n
cp source.txt lib/i18n/strings.g.dart
;;
*"build_runner"*)
mkdir -p lib/models
cp source.txt lib/models/model.g.dart
cp source.txt lib/models/model.freezed.dart
printf '%s\n' "$*" > build-runner-args.txt
;;
esac
""",
)
subprocess.run(["git", "add", "."], cwd=self.root, check=True)
subprocess.run(["git", "commit", "-qm", "fixture"], cwd=self.root, check=True)
self.env = os.environ | {
"PATH": f"{self.bin}:{os.environ['PATH']}",
"REAL_PYTHON": sys.executable,
"TMPDIR": str(self.codegen_temp),
}
def tearDown(self) -> None:
self.temp.cleanup()
shutil.rmtree(self.codegen_temp, ignore_errors=True)
def generated_state(self) -> dict[str, str]:
state = {}
for relative in GENERATED_PATHS:
path = self.root / relative
if path.is_file():
state[relative] = hashlib.sha256(path.read_bytes()).hexdigest()
for path in (self.root / "lib").rglob("*.dart"):
relative = path.relative_to(self.root).as_posix()
if relative.endswith(".g.dart") or relative.endswith(".freezed.dart"):
state.setdefault(relative, hashlib.sha256(path.read_bytes()).hexdigest())
return state
def git_status(self) -> bytes:
return subprocess.run(
["git", "status", "--porcelain=v1", "-z"],
cwd=self.root,
check=True,
capture_output=True,
).stdout
def run_codegen(self, *arguments: str, env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["bash", "scripts/codegen.sh", *arguments],
cwd=self.root,
env=self.env if env is None else env,
check=False,
capture_output=True,
text=True,
)
def assert_isolation_cleaned_up(self) -> None:
worktrees = subprocess.run(
["git", "worktree", "list", "--porcelain"],
cwd=self.root,
check=True,
capture_output=True,
text=True,
).stdout
self.assertEqual(worktrees.count("worktree "), 1)
self.assertEqual(list(self.codegen_temp.glob("plezy-codegen-check-*")), [])
def test_stale_check_reports_sorted_paths_without_changing_caller(self) -> None:
(self.root / "source.txt").write_text("version two\n", encoding="utf-8")
before = self.generated_state()
status_before = self.git_status()
result = self.run_codegen("--check")
self.assertEqual(result.returncode, 1)
self.assertEqual(
result.stderr.splitlines(),
[
"Generated files are out of date:",
*(f" {relative}" for relative in GENERATED_PATHS),
"Run 'scripts/codegen.sh' and commit the result.",
],
)
self.assertEqual(self.generated_state(), before)
self.assertEqual(self.git_status(), status_before)
self.assert_isolation_cleaned_up()
def test_generator_failure_propagates_without_partial_writes(self) -> None:
(self.root / "source.txt").write_text("version two\n", encoding="utf-8")
before = self.generated_state()
status_before = self.git_status()
result = self.run_codegen("--check", env=self.env | {"FAIL_DART": "7"})
self.assertEqual(result.returncode, 7)
self.assertEqual(self.generated_state(), before)
self.assertEqual(self.git_status(), status_before)
self.assert_isolation_cleaned_up()
def test_index_boundary_requires_matching_staged_outputs(self) -> None:
(self.root / "source.txt").write_text("version two\n", encoding="utf-8")
self.assertEqual(self.run_codegen().returncode, 0)
unstaged_state = self.generated_state()
self.assertEqual(self.run_codegen("--check").returncode, 1)
self.assertEqual(self.generated_state(), unstaged_state)
subprocess.run(["git", "add", "source.txt", "lib", "server/relay_protocol_gen.go"], cwd=self.root, check=True)
self.assertEqual(self.run_codegen("--check").returncode, 0)
incorrect = self.root / "lib" / "models" / "model.g.dart"
incorrect.write_text("incorrect staged output\n", encoding="utf-8")
subprocess.run(["git", "add", str(incorrect.relative_to(self.root))], cwd=self.root, check=True)
status_before = self.git_status()
result = self.run_codegen("--check")
self.assertEqual(result.returncode, 1)
self.assertIn("lib/models/model.g.dart", result.stderr)
self.assertEqual(incorrect.read_text(encoding="utf-8"), "incorrect staged output\n")
self.assertEqual(self.git_status(), status_before)
def test_deleted_and_untracked_outputs_are_reported_without_repair(self) -> None:
deleted = self.root / "lib" / "models" / "model.g.dart"
deleted.unlink()
untracked = self.root / "lib" / "models" / "extra.g.dart"
untracked.write_text("caller sentinel\n", encoding="utf-8")
status_before = self.git_status()
result = self.run_codegen("--check")
self.assertEqual(result.returncode, 1)
self.assertIn("lib/models/extra.g.dart", result.stderr)
self.assertIn("lib/models/model.g.dart", result.stderr)
self.assertFalse(deleted.exists())
self.assertEqual(untracked.read_text(encoding="utf-8"), "caller sentinel\n")
self.assertEqual(self.git_status(), status_before)
def test_write_mode_updates_outputs_and_forwards_build_runner_arguments(self) -> None:
(self.root / "source.txt").write_text("version two\n", encoding="utf-8")
result = self.run_codegen("--build-filter=lib/models/**")
self.assertEqual(result.returncode, 0)
for relative in GENERATED_PATHS:
self.assertEqual((self.root / relative).read_text(encoding="utf-8"), "version two\n")
self.assertEqual(
(self.root / "build-runner-args.txt").read_text(encoding="utf-8"),
"run build_runner build --delete-conflicting-outputs --build-filter=lib/models/**\n",
)
if __name__ == "__main__":
unittest.main()
scripts/test_check_workflow_action_pins.py
+289
View File
@@ -0,0 +1,289 @@
import contextlib
import io
import sys
import tempfile
import unittest
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPT_DIR))
from check_workflow_action_pins import iter_uses_references, main, validate_reference
SHA = "0123456789abcdef0123456789abcdef01234567"
class WorkflowActionPinsTest(unittest.TestCase):
def test_accepts_local_and_full_sha_references(self) -> None:
self.assertIsNone(validate_reference("./.github/actions/local"))
self.assertIsNone(validate_reference(f"actions/checkout@{SHA}"))
self.assertIsNone(validate_reference(f"owner/repository/sub/action@{SHA}"))
def test_rejects_mutable_dynamic_and_malformed_references(self) -> None:
references = [
"actions/checkout@v4",
"owner/action@latest",
"owner/action@main",
"owner/action@0123456",
"owner/action@${{ inputs.ref }}",
"docker://alpine:3",
"not-a-reference",
]
for reference in references:
with self.subTest(reference=reference):
self.assertIsNotNone(validate_reference(reference))
def test_parses_mapping_locations_without_comments_or_block_text(self) -> None:
fixture = f'''\
name: policy
jobs:
reusable:
uses: "owner/workflows/.github/workflows/check.yml@{SHA}" # reviewed
steps:
runs-on: ubuntu-latest
steps:
# - uses: actions/checkout@v4
- uses: 'actions/checkout@{SHA}' # v4
- run: |
echo "uses: owner/action@main"
'''
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "workflow.yaml"
path.write_text(fixture, encoding="utf-8")
references = list(iter_uses_references(path))
self.assertEqual(
references,
[
(4, f"owner/workflows/.github/workflows/check.yml@{SHA}"),
(9, f"actions/checkout@{SHA}"),
],
)
def test_rejects_mutable_flow_style_reference(self) -> None:
fixture = f"""\
jobs: {{ pinned: {{ uses: actions/checkout@{SHA} }}, mutable: {{ uses: owner/action@main }} }}
"""
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "workflow.yaml"
path.write_text(fixture, encoding="utf-8")
references = list(iter_uses_references(path))
stderr = io.StringIO()
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(stderr):
status = main([str(path)])
self.assertEqual(
references,
[
(1, f"actions/checkout@{SHA}"),
(1, "owner/action@main"),
],
)
self.assertEqual(status, 1)
self.assertIn("owner/action@main", stderr.getvalue())
def test_rejects_quoted_and_escaped_block_style_keys(self) -> None:
fixture = f"""\
jobs:
test:
steps:
- "uses": owner/action@main
- "us\\x65s": owner/other@latest
- "us\\u0065s": actions/checkout@{SHA}
- 'uses': actions/checkout@{SHA}
"""
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "workflow.yaml"
path.write_text(fixture, encoding="utf-8")
references = list(iter_uses_references(path))
stderr = io.StringIO()
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(stderr):
status = main([str(path)])
self.assertEqual(
references,
[
(4, "owner/action@main"),
(5, "owner/other@latest"),
(6, f"actions/checkout@{SHA}"),
(7, f"actions/checkout@{SHA}"),
],
)
self.assertEqual(status, 1)
self.assertIn("owner/action@main", stderr.getvalue())
self.assertIn("owner/other@latest", stderr.getvalue())
def test_parses_folded_and_literal_uses_scalars(self) -> None:
fixture = f"""\
jobs:
test:
steps:
- uses: >-
owner/folded@main
- uses: |
owner/literal@latest
- uses: >-
actions/checkout@{SHA}
- uses: |-
actions/checkout@{SHA}
"""
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "workflow.yaml"
path.write_text(fixture, encoding="utf-8")
references = list(iter_uses_references(path))
stderr = io.StringIO()
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(stderr):
status = main([str(path)])
self.assertEqual(
references,
[
(4, "owner/folded@main"),
(6, "owner/literal@latest"),
(8, f"actions/checkout@{SHA}"),
(10, f"actions/checkout@{SHA}"),
],
)
self.assertEqual(status, 1)
self.assertIn("owner/folded@main", stderr.getvalue())
self.assertIn("owner/literal@latest", stderr.getvalue())
def test_parses_explicit_mapping_uses_keys(self) -> None:
fixture = f"""\
jobs:
test:
steps:
- ? uses
: owner/explicit@main
- ? "us\\x65s"
: actions/checkout@{SHA}
"""
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "workflow.yaml"
path.write_text(fixture, encoding="utf-8")
references = list(iter_uses_references(path))
stderr = io.StringIO()
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(stderr):
status = main([str(path)])
self.assertEqual(
references,
[
(4, "owner/explicit@main"),
(6, f"actions/checkout@{SHA}"),
],
)
self.assertEqual(status, 1)
self.assertIn("owner/explicit@main", stderr.getvalue())
def test_parses_flow_sequence_mapping_pairs(self) -> None:
fixture = (
f"steps: [uses: owner/sequence@main, uses: actions/checkout@{SHA}]\n"
)
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "workflow.yaml"
path.write_text(fixture, encoding="utf-8")
references = list(iter_uses_references(path))
stderr = io.StringIO()
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(stderr):
status = main([str(path)])
self.assertEqual(
references,
[
(1, "owner/sequence@main"),
(1, f"actions/checkout@{SHA}"),
],
)
self.assertEqual(status, 1)
self.assertIn("owner/sequence@main", stderr.getvalue())
def test_fails_closed_on_multiline_aliased_and_tagged_keys(self) -> None:
fixture = """\
uses_key: &uses-key uses
jobs:
test:
steps:
- "u\\
ses": owner/multiline@main
- *uses-key: owner/alias@main
- !!str uses: owner/tagged@main
"""
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "workflow.yaml"
path.write_text(fixture, encoding="utf-8")
references = list(iter_uses_references(path))
stderr = io.StringIO()
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(stderr):
status = main([str(path)])
self.assertEqual(
references,
[
(5, "<unsupported multiline, tagged, anchored, or aliased mapping key>"),
(7, "<unsupported multiline, tagged, anchored, or aliased mapping key>"),
(8, "<unsupported multiline, tagged, anchored, or aliased mapping key>"),
],
)
self.assertEqual(status, 1)
self.assertIn("workflow.yaml:5", stderr.getvalue())
self.assertIn("workflow.yaml:7", stderr.getvalue())
self.assertIn("workflow.yaml:8", stderr.getvalue())
def test_does_not_treat_github_expressions_as_yaml_flow_mappings(self) -> None:
fixture = f"""\
if: ${{{{ always() && !contains(needs.*.result, 'failure') }}}}
steps:
- uses: actions/checkout@{SHA}
"""
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "workflow.yaml"
path.write_text(fixture, encoding="utf-8")
references = list(iter_uses_references(path))
self.assertEqual(references, [(3, f"actions/checkout@{SHA}")])
def test_block_scalar_ends_at_inferred_content_indentation(self) -> None:
fixture = '''\
jobs:
steps:
runs-on: ubuntu-latest
steps:
- name: |
Multiline step name
uses: actions/checkout@v4
'''
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "workflow.yaml"
path.write_text(fixture, encoding="utf-8")
references = list(iter_uses_references(path))
self.assertEqual(references, [(7, "actions/checkout@v4")])
def test_cli_reports_all_violations_with_file_and_line(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
first = root / "first.yml"
second = root / "second.yaml"
first.write_text("steps:\n - uses: actions/checkout@v4\n", encoding="utf-8")
second.write_text("jobs:\n call:\n uses: owner/workflow@main\n", encoding="utf-8")
stderr = io.StringIO()
with contextlib.redirect_stderr(stderr):
status = main([str(first), str(second)])
self.assertEqual(status, 1)
output = stderr.getvalue()
self.assertIn("first.yml:2", output)
self.assertIn("second.yaml:3", output)
self.assertIn("actions/checkout@v4", output)
self.assertIn("owner/workflow@main", output)
if __name__ == "__main__":
unittest.main()
scripts/test_format_native.py
+251
View File
@@ -0,0 +1,251 @@
import os
import shutil
import stat
import subprocess
import tempfile
import unittest
from pathlib import Path
SCRIPT = Path(__file__).resolve().parent / "format_native.sh"
DIGEST = "a16be01dcc480aab2f55f444b620142152f66e31564b3b9376506d624c28a2ad"
JAVA_DIAGNOSTIC = (
"A working JDK 17+ is required for Kotlin formatting. "
"Set JAVA_HOME to a JDK 17+ installation or put java on PATH."
)
def executable(path: Path, contents: str) -> Path:
path.write_text(contents, encoding="utf-8")
path.chmod(path.stat().st_mode | stat.S_IXUSR)
return path
class NativeFormatterTest(unittest.TestCase):
def setUp(self) -> None:
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name)
(self.root / "scripts").mkdir()
shutil.copy2(SCRIPT, self.root / "scripts" / "format_native.sh")
(self.root / "android").mkdir()
(self.root / "android" / "fixture.kt").write_text("class Fixture\n", encoding="utf-8")
self.bin = self.root / "bin"
self.bin.mkdir()
executable(self.bin / "git", "#!/bin/bash\nprintf 'android/fixture.kt\\0'\n")
executable(
self.bin / "shasum",
f"""#!/bin/bash
for file do :; done
case "${{HASH_MODE:-valid}}" in
valid) digest={DIGEST} ;;
invalid) digest={'0' * 64} ;;
cache-invalid)
case "$file" in
*/ktlint-1.5.0) digest={'0' * 64} ;;
*) digest={DIGEST} ;;
esac
;;
esac
printf '%s %s\\n' "$digest" "$file"
printf 'shasum\\n' >> "$HASH_MARKER"
""",
)
self.download = executable(
self.root / "download-ktlint",
"""#!/bin/bash
set -e
java -version >/dev/null 2>&1
printf '%s\n' "$*" >> "$KTLINT_ARGS_MARKER"
printf 'launched\n' >> "$KTLINT_MARKER"
""",
)
executable(
self.bin / "curl",
"""#!/bin/bash
printf 'downloaded\n' >> "$CURL_MARKER"
out=''
while [ $# -gt 0 ]; do
if [ "$1" = '-o' ]; then out=$2; shift 2; else shift; fi
done
cp "$FAKE_DOWNLOAD" "$out"
""",
)
self.env = os.environ.copy()
self.env.pop("JAVA_HOME", None)
self.env.update(
{
"PATH": f"{self.bin}:{self.env['PATH']}",
"FAKE_DOWNLOAD": str(self.download),
"CURL_MARKER": str(self.root / "curl.marker"),
"HASH_MARKER": str(self.root / "hash.marker"),
"JAVA_MARKER": str(self.root / "java.marker"),
"KTLINT_ARGS_MARKER": str(self.root / "ktlint-args.marker"),
"KTLINT_MARKER": str(self.root / "ktlint.marker"),
}
)
def tearDown(self) -> None:
self.temp.cleanup()
@property
def cache(self) -> Path:
return self.root / ".dart_tool" / "native-format" / "ktlint-1.5.0"
def java(self, directory: Path, version: str | None, *, identity: str = "path", status: int = 0) -> Path:
directory.mkdir(parents=True, exist_ok=True)
if version is None:
output = "echo 'unrecognized java output' >&2"
else:
output = f"echo 'openjdk version \"{version}\"' >&2"
return executable(
directory / "java",
f"""#!/bin/bash
printf '{identity} %s\\n' "$*" >> "$JAVA_MARKER"
{output}
exit {status}
""",
)
def install_cache(self, contents: bytes | None = None) -> None:
self.cache.parent.mkdir(parents=True, exist_ok=True)
if contents is None:
shutil.copy2(self.download, self.cache)
else:
self.cache.write_bytes(contents)
self.cache.chmod(self.cache.stat().st_mode | stat.S_IXUSR)
def reset_markers(self) -> None:
for marker in self.root.glob("*.marker"):
marker.unlink()
def run_script(
self, *arguments: str, env: dict[str, str] | None = None
) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["/bin/bash", "scripts/format_native.sh", *arguments],
cwd=self.root,
env=self.env if env is None else env,
check=False,
capture_output=True,
text=True,
)
def assert_java_rejection(self, result: subprocess.CompletedProcess[str]) -> None:
self.assertNotEqual(result.returncode, 0)
self.assertEqual(result.stderr.splitlines().count(JAVA_DIAGNOSTIC), 1)
self.assertNotIn("integer expression expected", result.stderr)
self.assertFalse((self.root / "curl.marker").exists())
self.assertFalse((self.root / "ktlint.marker").exists())
def test_unusable_java_rejects_cold_and_warm_cache_before_execution(self) -> None:
self.java(self.bin, "17.0.12", status=1)
self.assert_java_rejection(self.run_script("--check"))
self.install_cache()
self.reset_markers()
self.assert_java_rejection(self.run_script("--check"))
def test_java_home_runtime_and_ktlint_arguments_are_used_by_check_and_fix(self) -> None:
self.java(self.bin, "17.0.12", identity="path-shim", status=1)
home = self.root / "jdk home"
self.java(home / "bin", "17.0.12", identity="java-home")
env = self.env | {"JAVA_HOME": str(home)}
expected_arguments = {
"--check": "android/fixture.kt",
"--fix": "-F android/fixture.kt",
}
for mode, expected in expected_arguments.items():
with self.subTest(mode=mode):
self.reset_markers()
result = self.run_script(mode, env=env)
self.assertEqual(result.returncode, 0, result.stderr)
java_calls = (self.root / "java.marker").read_text(encoding="utf-8").splitlines()
self.assertEqual(java_calls, ["java-home -version", "java-home -version"])
self.assertEqual(
(self.root / "ktlint-args.marker").read_text(encoding="utf-8").strip(),
expected,
)
def test_java_version_support_floor_fails_closed(self) -> None:
for version in ("17.0.12", "21.0.2"):
with self.subTest(accepted=version):
self.java(self.bin, version)
self.reset_markers()
result = self.run_script("--check")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertTrue((self.root / "ktlint.marker").exists())
for version, status in (("1.8.0_402", 0), ("16.0.2", 0), (None, 0), ("17.0.12", 1)):
with self.subTest(rejected=version, status=status):
self.java(self.bin, version, status=status)
self.reset_markers()
self.assert_java_rejection(self.run_script("--check"))
def test_cache_verification_replacement_and_mismatch_cleanup(self) -> None:
self.java(self.bin, "17.0.12")
self.install_cache()
valid = self.run_script("--check")
self.assertEqual(valid.returncode, 0, valid.stderr)
self.assertFalse((self.root / "curl.marker").exists())
self.assertEqual((self.root / "ktlint.marker").read_text(encoding="utf-8").splitlines(), ["launched"])
self.install_cache(b"invalid cache sentinel\n")
self.reset_markers()
replaced = self.run_script("--check", env=self.env | {"HASH_MODE": "cache-invalid"})
self.assertEqual(replaced.returncode, 0, replaced.stderr)
self.assertEqual(self.cache.read_bytes(), self.download.read_bytes())
self.assertTrue((self.root / "curl.marker").exists())
self.assertTrue((self.root / "ktlint.marker").exists())
sentinel = b"preserve invalid cache\n"
self.install_cache(sentinel)
self.reset_markers()
mismatch = self.run_script("--check", env=self.env | {"HASH_MODE": "invalid"})
self.assertNotEqual(mismatch.returncode, 0)
self.assertIn("failed SHA-256 verification", mismatch.stderr)
self.assertEqual(self.cache.read_bytes(), sentinel)
self.assertFalse((self.root / "ktlint.marker").exists())
self.assertEqual(list(self.cache.parent.glob(".ktlint-1.5.0.*")), [])
def test_sha256sum_fallback_verifies_and_executes(self) -> None:
self.java(self.bin, "17.0.12")
minimal = self.root / "minimal-bin"
minimal.mkdir()
for name in ("git", "java", "curl"):
shutil.copy2(self.bin / name, minimal / name)
executable(
minimal / "sha256sum",
f"#!/bin/bash\nfor file do :; done\nprintf '{DIGEST} %s\\n' \"$file\"\nprintf 'sha256sum\\n' >> \"$HASH_MARKER\"\n",
)
for name, source in {
"chmod": "/bin/chmod",
"cp": "/bin/cp",
"dirname": "/usr/bin/dirname",
"mkdir": "/bin/mkdir",
"mktemp": "/usr/bin/mktemp",
"mv": "/bin/mv",
"rm": "/bin/rm",
}.items():
(minimal / name).symlink_to(source)
result = self.run_script("--check", env=self.env | {"PATH": str(minimal)})
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual((self.root / "hash.marker").read_text(encoding="utf-8").splitlines(), ["sha256sum"])
self.assertTrue((self.root / "ktlint.marker").exists())
def test_repository_without_kotlin_does_not_require_java_or_ktlint(self) -> None:
executable(self.bin / "git", "#!/bin/bash\nexit 0\n")
self.java(self.bin, "17.0.12", status=1)
result = self.run_script("--check")
self.assertEqual(result.returncode, 0, result.stderr)
for marker in ("curl.marker", "hash.marker", "java.marker", "ktlint.marker"):
self.assertFalse((self.root / marker).exists())
self.assertFalse(self.cache.parent.exists())
if __name__ == "__main__":
unittest.main()