add framework
This commit is contained in:
@@ -0,0 +1,89 @@
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _repo_root(cwd: Path) -> Path:
|
||||
try:
|
||||
root = subprocess.check_output(
|
||||
["git", "rev-parse", "--show-toplevel"],
|
||||
cwd=cwd,
|
||||
text=True,
|
||||
stderr=subprocess.DEVNULL,
|
||||
).strip()
|
||||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||||
return cwd
|
||||
return Path(root)
|
||||
|
||||
|
||||
def _is_git_commit(command: str) -> bool:
|
||||
return re.search(
|
||||
r"^\s*git(?:\s+(?:-[A-Za-z]\s+\S+|--[A-Za-z0-9-]+(?:=\S+)?))*\s+commit\b",
|
||||
command,
|
||||
) is not None
|
||||
|
||||
|
||||
def _deny(reason: str) -> None:
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"hookSpecificOutput": {
|
||||
"hookEventName": "PreToolUse",
|
||||
"permissionDecision": "deny",
|
||||
"permissionDecisionReason": reason,
|
||||
}
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _tail(text: str, limit: int = 1200) -> str:
|
||||
text = text.strip()
|
||||
if len(text) <= limit:
|
||||
return text
|
||||
return text[-limit:]
|
||||
|
||||
|
||||
def _build_pre_commit_commands(root: Path) -> list[list[str]]:
|
||||
return [
|
||||
[sys.executable, "-m", "unittest", "discover", "-s", "scripts", "-p", "test_*.py"],
|
||||
[sys.executable, "scripts/validate_workspace.py"],
|
||||
]
|
||||
|
||||
|
||||
def _run_checks(root: Path) -> str | None:
|
||||
for command in _build_pre_commit_commands(root):
|
||||
result = subprocess.run(command, cwd=root, capture_output=True, text=True)
|
||||
if result.returncode != 0:
|
||||
details = _tail(result.stdout + "\n" + result.stderr)
|
||||
label = " ".join(command)
|
||||
if details:
|
||||
return f"{label} failed:\n{details}"
|
||||
return f"{label} failed with exit code {result.returncode}."
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def main() -> int:
|
||||
try:
|
||||
payload = json.load(sys.stdin)
|
||||
except json.JSONDecodeError:
|
||||
return 0
|
||||
|
||||
command = payload.get("tool_input", {}).get("command", "")
|
||||
if not isinstance(command, str) or not _is_git_commit(command):
|
||||
return 0
|
||||
|
||||
cwd = Path(payload.get("cwd") or Path.cwd())
|
||||
root = _repo_root(cwd)
|
||||
failure = _run_checks(root)
|
||||
if failure:
|
||||
_deny(f"PRE-COMMIT CHECKS: {failure}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,183 @@
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
SOURCE_SUFFIXES = {".cs"}
|
||||
TEST_SUFFIXES = {".cs"}
|
||||
|
||||
|
||||
def _repo_root(cwd: Path) -> Path:
|
||||
try:
|
||||
root = subprocess.check_output(
|
||||
["git", "rev-parse", "--show-toplevel"],
|
||||
cwd=cwd,
|
||||
text=True,
|
||||
stderr=subprocess.DEVNULL,
|
||||
).strip()
|
||||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||||
return cwd
|
||||
return Path(root)
|
||||
|
||||
|
||||
def _extract_patch_paths(command: str) -> list[str]:
|
||||
prefixes = (
|
||||
"*** Add File: ",
|
||||
"*** Update File: ",
|
||||
"*** Delete File: ",
|
||||
"*** Move to: ",
|
||||
)
|
||||
paths: list[str] = []
|
||||
for raw_line in command.splitlines():
|
||||
line = raw_line.strip()
|
||||
for prefix in prefixes:
|
||||
if line.startswith(prefix):
|
||||
paths.append(line[len(prefix) :].strip())
|
||||
break
|
||||
return paths
|
||||
|
||||
|
||||
def _touched_paths(payload: dict) -> list[str]:
|
||||
tool_input = payload.get("tool_input", {})
|
||||
if not isinstance(tool_input, dict):
|
||||
return []
|
||||
|
||||
file_path = tool_input.get("file_path")
|
||||
if isinstance(file_path, str) and file_path:
|
||||
return [file_path]
|
||||
|
||||
command = tool_input.get("command")
|
||||
if isinstance(command, str):
|
||||
return _extract_patch_paths(command)
|
||||
|
||||
return []
|
||||
|
||||
|
||||
def _normalize(path_text: str) -> str:
|
||||
return path_text.replace("\\", "/").lower()
|
||||
|
||||
|
||||
def _path_segments(path_text: str) -> list[str]:
|
||||
return [segment for segment in _normalize(path_text).split("/") if segment]
|
||||
|
||||
|
||||
def _is_ignored_path(path_text: str) -> bool:
|
||||
return any(segment in {"bin", "obj"} for segment in _path_segments(path_text))
|
||||
|
||||
|
||||
def _is_test_path(path_text: str) -> bool:
|
||||
path = Path(path_text)
|
||||
if path.suffix.lower() not in TEST_SUFFIXES:
|
||||
return False
|
||||
|
||||
segments = _path_segments(path_text)
|
||||
name = segments[-1] if segments else ""
|
||||
directories = segments[:-1]
|
||||
return (
|
||||
any(
|
||||
segment in {"test", "tests"}
|
||||
or segment.endswith(".test")
|
||||
or segment.endswith(".tests")
|
||||
for segment in directories
|
||||
)
|
||||
or name.endswith("test.cs")
|
||||
or name.endswith("tests.cs")
|
||||
)
|
||||
|
||||
|
||||
def _token(text: str) -> str:
|
||||
return "".join(ch for ch in text.lower() if ch.isalnum())
|
||||
|
||||
|
||||
def _candidate_test_paths(paths: list[str], cwd: Path, root: Path) -> list[Path]:
|
||||
candidates: list[Path] = []
|
||||
for path_text in paths:
|
||||
resolved = _resolve_path(path_text, cwd)
|
||||
if _is_test_path(str(resolved)) and not _is_ignored_path(str(resolved)):
|
||||
candidates.append(resolved)
|
||||
|
||||
for candidate in root.rglob("*"):
|
||||
if (
|
||||
candidate.is_file()
|
||||
and candidate.suffix.lower() in TEST_SUFFIXES
|
||||
and _is_test_path(str(candidate))
|
||||
and not _is_ignored_path(str(candidate))
|
||||
):
|
||||
candidates.append(candidate)
|
||||
|
||||
return candidates
|
||||
|
||||
|
||||
def _has_related_test(path: Path, candidate_tests: list[Path]) -> bool:
|
||||
source_token = _token(path.stem)
|
||||
for test_path in candidate_tests:
|
||||
test_token = _token(test_path.stem)
|
||||
if source_token and source_token in test_token:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _is_exempt(path_text: str) -> bool:
|
||||
return _is_test_path(path_text)
|
||||
|
||||
|
||||
def _resolve_path(path_text: str, cwd: Path) -> Path:
|
||||
path = Path(path_text)
|
||||
if path.is_absolute():
|
||||
return path
|
||||
return (cwd / path).resolve()
|
||||
|
||||
|
||||
def _base_name(path: Path) -> str:
|
||||
return path.stem
|
||||
|
||||
|
||||
def _guarded_paths(paths: list[str], cwd: Path, root: Path) -> list[str]:
|
||||
missing_tests: list[str] = []
|
||||
candidate_tests = _candidate_test_paths(paths, cwd, root)
|
||||
for path_text in paths:
|
||||
if _is_exempt(path_text):
|
||||
continue
|
||||
|
||||
path = _resolve_path(path_text, cwd)
|
||||
if path.suffix.lower() not in SOURCE_SUFFIXES:
|
||||
continue
|
||||
if not _has_related_test(path, candidate_tests):
|
||||
missing_tests.append(_base_name(path))
|
||||
|
||||
return missing_tests
|
||||
|
||||
|
||||
def main() -> int:
|
||||
try:
|
||||
payload = json.load(sys.stdin)
|
||||
except json.JSONDecodeError:
|
||||
return 0
|
||||
|
||||
cwd = Path(payload.get("cwd") or Path.cwd())
|
||||
root = _repo_root(cwd)
|
||||
missing_tests = _guarded_paths(_touched_paths(payload), cwd, root)
|
||||
if not missing_tests:
|
||||
return 0
|
||||
|
||||
names = ", ".join(sorted(set(missing_tests)))
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"hookSpecificOutput": {
|
||||
"hookEventName": "PreToolUse",
|
||||
"permissionDecision": "deny",
|
||||
"permissionDecisionReason": (
|
||||
"TDD GUARD: missing C# test file for "
|
||||
f"{names}. Write or add the test first."
|
||||
),
|
||||
}
|
||||
}
|
||||
)
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user