chore: merge dev into eigen execution prep
This commit is contained in:
KOKO\Mimi
+45
-1385
File diff suppressed because it is too large
Load Diff
@@ -1,158 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Codex PreToolUse 정책: 위험 명령과 테스트 없는 구현 파일 수정을 차단한다."""
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
SCRIPTS_DIR = Path(__file__).resolve().parents[1]
|
||||
if str(SCRIPTS_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(SCRIPTS_DIR))
|
||||
|
||||
from msvc_harness.config import ConfigError, load_config
|
||||
from msvc_harness.tdd_policy import CPP_SUFFIXES, evaluate_paths
|
||||
from hooks.validation_scope import EDIT_TOOL_NAMES, record_edit_scope
|
||||
|
||||
|
||||
DANGEROUS_PATTERNS = (
|
||||
re.compile(r"\bgit\s+reset\s+--hard\b", re.IGNORECASE),
|
||||
re.compile(r"\bgit\s+push\b[^\n]*--force(?:-with-lease)?\b", re.IGNORECASE),
|
||||
re.compile(r"\brm\s+-rf\b", re.IGNORECASE),
|
||||
re.compile(
|
||||
r"\bRemove-Item\b(?=[^\n]*-Recurse\b)(?=[^\n]*-Force\b)",
|
||||
re.IGNORECASE,
|
||||
),
|
||||
re.compile(
|
||||
r"\b(?:rmdir|rd)\b(?=[^\n]*/s\b)(?=[^\n]*/q\b)",
|
||||
re.IGNORECASE,
|
||||
),
|
||||
re.compile(r"\bDROP\s+TABLE\b", re.IGNORECASE),
|
||||
)
|
||||
EXECUTOR_OWNED_GIT = re.compile(
|
||||
r"\bgit(?:\.exe)?\b[^\r\n;&|]*\b(?:commit|push|switch|checkout|branch|tag|"
|
||||
r"merge|rebase|reset|cherry-pick|revert|stash|clean|update-ref|symbolic-ref|"
|
||||
r"worktree|fetch|pull)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
PATCH_PATH = re.compile(
|
||||
r"^\*\*\* (?:(?:Add|Update|Delete) File:|Move to:) (?P<path>.+?)\s*$",
|
||||
re.MULTILINE,
|
||||
)
|
||||
|
||||
|
||||
class HookInputError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def _tool_input(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
value = payload.get("tool_input")
|
||||
if not isinstance(value, dict):
|
||||
raise HookInputError("hook tool_input must be an object")
|
||||
return value
|
||||
|
||||
|
||||
def _candidate_paths(payload: dict[str, Any], root: Path) -> list[Path]:
|
||||
tool_input = _tool_input(payload)
|
||||
raw_paths: list[str] = []
|
||||
for key in ("path", "file_path"):
|
||||
value = tool_input.get(key)
|
||||
if isinstance(value, str) and value.strip():
|
||||
raw_paths.append(value.strip())
|
||||
|
||||
edits = tool_input.get("edits")
|
||||
if isinstance(edits, list):
|
||||
for edit in edits:
|
||||
if not isinstance(edit, dict):
|
||||
continue
|
||||
value = edit.get("path")
|
||||
if isinstance(value, str) and value.strip():
|
||||
raw_paths.append(value.strip())
|
||||
|
||||
for key in ("command", "patch", "input"):
|
||||
value = tool_input.get(key)
|
||||
if isinstance(value, str):
|
||||
raw_paths.extend(match.group("path") for match in PATCH_PATH.finditer(value))
|
||||
|
||||
paths: list[Path] = []
|
||||
for raw_path in raw_paths:
|
||||
path = Path(raw_path)
|
||||
paths.append(path.resolve() if path.is_absolute() else (root / path).resolve())
|
||||
return list(dict.fromkeys(paths))
|
||||
|
||||
|
||||
def evaluate(payload: dict[str, Any], root: Path) -> str | None:
|
||||
"""Return a blocking reason, or None when the tool call is allowed."""
|
||||
tool_name = payload.get("tool_name")
|
||||
if not isinstance(tool_name, str) or not tool_name:
|
||||
raise HookInputError("hook tool_name must be a non-empty string")
|
||||
tool_input = _tool_input(payload)
|
||||
|
||||
if tool_name in {"Bash", "shell_command", "PowerShell"}:
|
||||
command = tool_input.get("command")
|
||||
if not isinstance(command, str):
|
||||
raise HookInputError("shell tool_input.command must be a string")
|
||||
if EXECUTOR_OWNED_GIT.search(command):
|
||||
return (
|
||||
"Harness-owned Git operation detected. The implementation session "
|
||||
"must leave branch, commit, push, and history operations to the Executor."
|
||||
)
|
||||
if any(
|
||||
pattern.search(command) for pattern in DANGEROUS_PATTERNS
|
||||
):
|
||||
return "위험한 명령어가 감지되어 실행을 차단했습니다."
|
||||
|
||||
if tool_name not in EDIT_TOOL_NAMES:
|
||||
return None
|
||||
|
||||
paths = _candidate_paths(payload, root)
|
||||
if not paths:
|
||||
raise HookInputError(
|
||||
f"{tool_name} input did not contain a path that can be checked"
|
||||
)
|
||||
try:
|
||||
config = load_config(root)
|
||||
except ConfigError as exc:
|
||||
if any(path.suffix.lower() in CPP_SUFFIXES for path in paths):
|
||||
return f"TDD GUARD: .harness/config.json must be repaired: {exc}"
|
||||
return None
|
||||
return evaluate_paths(paths, root, config.tdd)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
try:
|
||||
payload = json.load(sys.stdin)
|
||||
if not isinstance(payload, dict):
|
||||
raise HookInputError("hook input must be a JSON object")
|
||||
cwd = payload.get("cwd")
|
||||
if cwd is not None and not isinstance(cwd, str):
|
||||
raise HookInputError("hook cwd must be a string")
|
||||
|
||||
root = Path(cwd or Path.cwd()).resolve()
|
||||
reason = evaluate(payload, root)
|
||||
except (
|
||||
json.JSONDecodeError,
|
||||
HookInputError,
|
||||
ConfigError,
|
||||
KeyError,
|
||||
ValueError,
|
||||
OSError,
|
||||
) as exc:
|
||||
print(f"TDD GUARD: hook input rejected: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
if reason:
|
||||
print(reason, file=sys.stderr)
|
||||
return 2
|
||||
if payload["tool_name"] in EDIT_TOOL_NAMES:
|
||||
try:
|
||||
record_edit_scope(payload, root, _candidate_paths(payload, root))
|
||||
except (ValueError, OSError) as exc:
|
||||
print(f"TDD GUARD: edit scope could not be recorded: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,168 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Codex Stop hook: validate discovered C/C++ projects with MSVC."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
SCRIPTS = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(SCRIPTS))
|
||||
|
||||
from msvc_harness.adapters.base import AdapterError
|
||||
from msvc_harness.adapters.cmake import CMakeAdapter
|
||||
from msvc_harness.adapters.msbuild import MSBuildAdapter
|
||||
from msvc_harness.config import ConfigError, load_config
|
||||
from msvc_harness.discovery import DiscoveryError, discover_project
|
||||
from msvc_harness.models import ProjectKind
|
||||
from msvc_harness.process import ValidationFailure, execute_plan
|
||||
from msvc_harness.toolchain import ToolchainError, discover_toolchain
|
||||
from hooks.validation_scope import is_document_only_turn
|
||||
|
||||
|
||||
REENTRY_ENV = "CODEX_STOP_VALIDATION_ACTIVE"
|
||||
TOTAL_TIMEOUT_SECONDS = 7200
|
||||
|
||||
|
||||
def _remaining(deadline, clock, stage):
|
||||
remaining = deadline - clock()
|
||||
if remaining <= 0:
|
||||
raise ValidationFailure(f"{stage} timed out before it could start")
|
||||
return remaining
|
||||
|
||||
|
||||
def _project_root(
|
||||
cwd: Path,
|
||||
*,
|
||||
deadline: float,
|
||||
run=subprocess.run,
|
||||
clock=time.monotonic,
|
||||
) -> Path:
|
||||
argv = ["git", "rev-parse", "--show-toplevel"]
|
||||
resolved_cwd = cwd.resolve()
|
||||
try:
|
||||
result = run(
|
||||
argv,
|
||||
cwd=resolved_cwd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
shell=False,
|
||||
timeout=_remaining(deadline, clock, "repository discovery"),
|
||||
)
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
raise ValidationFailure(
|
||||
f"repository discovery timed out; argv={argv!r}; "
|
||||
f"cwd={str(resolved_cwd)!r}"
|
||||
) from exc
|
||||
if result.returncode == 0 and result.stdout.strip():
|
||||
return Path(result.stdout.strip()).resolve()
|
||||
return resolved_cwd
|
||||
|
||||
|
||||
def run_validations(
|
||||
root: Path,
|
||||
*,
|
||||
deadline: float | None = None,
|
||||
clock=time.monotonic,
|
||||
) -> tuple[bool, str]:
|
||||
"""Build and test a discovered C/C++ project, when one exists."""
|
||||
if deadline is None:
|
||||
deadline = clock() + TOTAL_TIMEOUT_SECONDS
|
||||
try:
|
||||
config = load_config(root)
|
||||
discovery = discover_project(root, config)
|
||||
if discovery.selection is None:
|
||||
return True, ""
|
||||
selection = discovery.selection
|
||||
tools = discover_toolchain(
|
||||
selection.kind,
|
||||
deadline=deadline,
|
||||
clock=clock,
|
||||
)
|
||||
adapter = (
|
||||
CMakeAdapter()
|
||||
if selection.kind is ProjectKind.CMAKE
|
||||
else MSBuildAdapter()
|
||||
)
|
||||
plan = adapter.create_plan(root, selection, config, tools)
|
||||
child_env = os.environ.copy()
|
||||
child_env[REENTRY_ENV] = "1"
|
||||
execute_plan(
|
||||
plan,
|
||||
root,
|
||||
env=child_env,
|
||||
deadline=deadline,
|
||||
clock=clock,
|
||||
)
|
||||
return True, ""
|
||||
except (
|
||||
ConfigError,
|
||||
DiscoveryError,
|
||||
ToolchainError,
|
||||
AdapterError,
|
||||
ValidationFailure,
|
||||
OSError,
|
||||
) as exc:
|
||||
return False, str(exc)
|
||||
|
||||
|
||||
def _emit_stop_response(message: str) -> None:
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"decision": "block",
|
||||
"reason": message,
|
||||
},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def main(*, run=subprocess.run, clock=time.monotonic) -> int:
|
||||
if os.environ.get(REENTRY_ENV) == "1":
|
||||
return 0
|
||||
|
||||
try:
|
||||
raw_payload = sys.stdin.read()
|
||||
payload = json.loads(raw_payload) if raw_payload.strip() else {}
|
||||
if not isinstance(payload, dict):
|
||||
raise TypeError("hook input must be a JSON object")
|
||||
cwd = payload.get("cwd")
|
||||
if cwd is not None and not isinstance(cwd, str):
|
||||
raise TypeError("hook cwd must be a string")
|
||||
stop_hook_active = payload.get("stop_hook_active", False)
|
||||
if not isinstance(stop_hook_active, bool):
|
||||
raise TypeError("stop_hook_active must be a boolean")
|
||||
except (json.JSONDecodeError, TypeError) as exc:
|
||||
_emit_stop_response(f"validation hook input rejected: {exc}")
|
||||
return 0
|
||||
|
||||
if stop_hook_active:
|
||||
return 0
|
||||
|
||||
deadline = clock() + TOTAL_TIMEOUT_SECONDS
|
||||
try:
|
||||
root = _project_root(
|
||||
Path(cwd or Path.cwd()),
|
||||
deadline=deadline,
|
||||
run=run,
|
||||
clock=clock,
|
||||
)
|
||||
if is_document_only_turn(payload, root):
|
||||
return 0
|
||||
ok, message = run_validations(root, deadline=deadline, clock=clock)
|
||||
except Exception as exc:
|
||||
_emit_stop_response(f"validation hook failed: {exc}")
|
||||
return 0
|
||||
|
||||
if ok:
|
||||
return 0
|
||||
_emit_stop_response(message)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,109 +0,0 @@
|
||||
"""Track whether a Codex turn edited Markdown files only."""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable
|
||||
|
||||
|
||||
EDIT_TOOL_NAMES = frozenset({"apply_patch", "Edit", "MultiEdit", "Write"})
|
||||
STATE_VERSION = 1
|
||||
|
||||
|
||||
def _identifier(payload: dict[str, Any], name: str) -> str | None:
|
||||
value = payload.get(name)
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def _digest(value: str) -> str:
|
||||
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def _state_path(root: Path, session_id: str) -> Path:
|
||||
return root / ".harness" / "hook-state" / f"{_digest(session_id)}.json"
|
||||
|
||||
|
||||
def _is_markdown_in_root(path: Path, root: Path) -> bool:
|
||||
resolved_root = root.resolve()
|
||||
resolved_path = path.resolve()
|
||||
try:
|
||||
resolved_path.relative_to(resolved_root)
|
||||
except ValueError:
|
||||
return False
|
||||
return resolved_path.suffix.lower() == ".md"
|
||||
|
||||
|
||||
def record_edit_scope(
|
||||
payload: dict[str, Any],
|
||||
root: Path,
|
||||
paths: Iterable[Path],
|
||||
) -> bool:
|
||||
"""Record an allowed edit; return False when turn identity is unavailable."""
|
||||
session_id = _identifier(payload, "session_id")
|
||||
turn_id = _identifier(payload, "turn_id")
|
||||
if session_id is None or turn_id is None:
|
||||
return False
|
||||
|
||||
edit_paths = list(paths)
|
||||
if not edit_paths:
|
||||
raise ValueError("edit scope requires at least one path")
|
||||
|
||||
turn_digest = _digest(turn_id)
|
||||
state_path = _state_path(root, session_id)
|
||||
document_only = all(
|
||||
_is_markdown_in_root(path, root) for path in edit_paths
|
||||
)
|
||||
|
||||
if state_path.is_file():
|
||||
try:
|
||||
previous = json.loads(state_path.read_text(encoding="utf-8"))
|
||||
valid_previous = (
|
||||
isinstance(previous, dict)
|
||||
and previous.get("version") == STATE_VERSION
|
||||
and isinstance(previous.get("turn"), str)
|
||||
and isinstance(previous.get("document_only"), bool)
|
||||
)
|
||||
if not valid_previous:
|
||||
document_only = False
|
||||
elif previous["turn"] == turn_digest:
|
||||
document_only = previous["document_only"] and document_only
|
||||
except (json.JSONDecodeError, OSError):
|
||||
document_only = False
|
||||
|
||||
state_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary_path = state_path.with_suffix(".tmp")
|
||||
temporary_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"version": STATE_VERSION,
|
||||
"turn": turn_digest,
|
||||
"document_only": document_only,
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
temporary_path.replace(state_path)
|
||||
return True
|
||||
|
||||
|
||||
def is_document_only_turn(payload: dict[str, Any], root: Path) -> bool:
|
||||
"""Return True only for a valid state proving Markdown-only edits."""
|
||||
session_id = _identifier(payload, "session_id")
|
||||
turn_id = _identifier(payload, "turn_id")
|
||||
if session_id is None or turn_id is None:
|
||||
return False
|
||||
|
||||
state_path = _state_path(root, session_id)
|
||||
try:
|
||||
state = json.loads(state_path.read_text(encoding="utf-8"))
|
||||
except (FileNotFoundError, json.JSONDecodeError, OSError):
|
||||
return False
|
||||
|
||||
return (
|
||||
isinstance(state, dict)
|
||||
and state.get("version") == STATE_VERSION
|
||||
and state.get("turn") == _digest(turn_id)
|
||||
and state.get("document_only") is True
|
||||
)
|
||||
@@ -1,16 +1,9 @@
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from string import Formatter
|
||||
from typing import Any
|
||||
|
||||
from .models import CMakeConfig, HarnessConfig, MSBuildConfig, TDDConfig
|
||||
|
||||
DEFAULT_TEST_PATTERNS = (
|
||||
"{stem}_test.cpp",
|
||||
"{stem}_tests.cpp",
|
||||
"test_{stem}.cpp",
|
||||
"{stem}.test.cpp",
|
||||
)
|
||||
from .models import CMakeConfig, HarnessConfig, MSBuildConfig
|
||||
|
||||
|
||||
class ConfigError(ValueError):
|
||||
@@ -60,38 +53,6 @@ def _string_array(value: Any, label: str) -> tuple[str, ...]:
|
||||
)
|
||||
|
||||
|
||||
def _validate_test_pattern(pattern: str) -> None:
|
||||
candidate = Path(pattern)
|
||||
if (
|
||||
candidate.is_absolute()
|
||||
or bool(candidate.drive)
|
||||
or "/" in pattern
|
||||
or "\\" in pattern
|
||||
or candidate.name != pattern
|
||||
):
|
||||
raise ConfigError(
|
||||
"every tdd.testPatterns entry must be a safe basename pattern"
|
||||
)
|
||||
try:
|
||||
fields = [
|
||||
(field_name, format_spec, conversion)
|
||||
for _, field_name, format_spec, conversion in Formatter().parse(pattern)
|
||||
if field_name is not None
|
||||
]
|
||||
except ValueError as exc:
|
||||
raise ConfigError(f"invalid tdd.testPatterns entry: {exc}") from exc
|
||||
if (
|
||||
not fields
|
||||
or any(
|
||||
field_name != "stem" or format_spec or conversion
|
||||
for field_name, format_spec, conversion in fields
|
||||
)
|
||||
):
|
||||
raise ConfigError(
|
||||
"every tdd.testPatterns entry may contain only the {stem} field"
|
||||
)
|
||||
|
||||
|
||||
def load_config(root: Path) -> HarnessConfig:
|
||||
root = root.resolve()
|
||||
config_path = root / ".harness" / "config.json"
|
||||
@@ -166,21 +127,12 @@ def load_config(root: Path) -> HarnessConfig:
|
||||
else None
|
||||
)
|
||||
|
||||
tdd_data = _mapping(data.get("tdd", {}), "tdd")
|
||||
_reject_unknown(tdd_data, {"testRoots", "testPatterns", "exclude"}, "tdd")
|
||||
raw_test_roots = _string_array(tdd_data["testRoots"], "tdd.testRoots") if "testRoots" in tdd_data else ("tests", "test")
|
||||
test_roots = tuple(_repo_path(root, item, "tdd.testRoots") for item in raw_test_roots)
|
||||
test_patterns = _string_array(tdd_data["testPatterns"], "tdd.testPatterns") if "testPatterns" in tdd_data else DEFAULT_TEST_PATTERNS
|
||||
for pattern in test_patterns:
|
||||
_validate_test_pattern(pattern)
|
||||
exclude = _string_array(tdd_data["exclude"], "tdd.exclude") if "exclude" in tdd_data else ()
|
||||
for item in exclude:
|
||||
_repo_path(root, item, "tdd.exclude")
|
||||
if "tdd" in data:
|
||||
print("Deprecated config field 'tdd' is ignored; it may be removed.", file=sys.stderr)
|
||||
|
||||
return HarnessConfig(
|
||||
version=1,
|
||||
project_type=project_type,
|
||||
cmake=CMakeConfig(source_dir, binary_dir, configure_preset, build_preset, test_preset),
|
||||
msbuild=MSBuildConfig(solution, configuration, platform, test_command),
|
||||
tdd=TDDConfig(test_roots, test_patterns, exclude),
|
||||
)
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
from pathlib import Path
|
||||
|
||||
from .models import DiscoveryResult, HarnessConfig, ProjectKind, ProjectSelection
|
||||
from .tdd_policy import find_cpp_files
|
||||
|
||||
|
||||
class DiscoveryError(RuntimeError):
|
||||
@@ -68,10 +67,4 @@ def discover_project(root: Path, config: HarnessConfig) -> DiscoveryResult:
|
||||
ProjectSelection(ProjectKind.MSBUILD, projects[0])
|
||||
)
|
||||
|
||||
cpp_files = find_cpp_files(root)
|
||||
if cpp_files:
|
||||
raise DiscoveryError(
|
||||
"C/C++ files exist but no CMakeLists.txt, preset, .sln, or .vcxproj "
|
||||
"was found; configure projectType and project path"
|
||||
)
|
||||
return DiscoveryResult(None, ())
|
||||
return DiscoveryResult(None)
|
||||
|
||||
@@ -30,20 +30,12 @@ class MSBuildConfig:
|
||||
test_command: tuple[str, ...] | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TDDConfig:
|
||||
test_roots: tuple[Path, ...]
|
||||
test_patterns: tuple[str, ...]
|
||||
exclude: tuple[str, ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HarnessConfig:
|
||||
version: int
|
||||
project_type: str
|
||||
cmake: CMakeConfig
|
||||
msbuild: MSBuildConfig
|
||||
tdd: TDDConfig
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -55,7 +47,6 @@ class ProjectSelection:
|
||||
@dataclass(frozen=True)
|
||||
class DiscoveryResult:
|
||||
selection: ProjectSelection | None
|
||||
cpp_files: tuple[Path, ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
|
||||
@@ -1,88 +0,0 @@
|
||||
from fnmatch import fnmatch
|
||||
from pathlib import Path
|
||||
|
||||
CPP_SUFFIXES = frozenset({".c", ".cc", ".cpp", ".cxx", ".h", ".hpp", ".hxx"})
|
||||
DEFAULT_EXCLUDES = (
|
||||
".harness/build/**",
|
||||
"build/**",
|
||||
"out/**",
|
||||
"cmake-build-*/**",
|
||||
"third_party/**",
|
||||
"external/**",
|
||||
"vendor/**",
|
||||
"generated/**",
|
||||
)
|
||||
|
||||
|
||||
def is_excluded_path(path: Path, root: Path, extra: tuple[str, ...] = ()) -> bool:
|
||||
try:
|
||||
relative = path.resolve().relative_to(root.resolve()).as_posix()
|
||||
except ValueError:
|
||||
return False
|
||||
return any(fnmatch(relative, pattern) for pattern in DEFAULT_EXCLUDES + extra)
|
||||
|
||||
|
||||
def find_cpp_files(
|
||||
root: Path,
|
||||
extra_excludes: tuple[str, ...] = (),
|
||||
) -> tuple[Path, ...]:
|
||||
found = (
|
||||
path
|
||||
for path in root.resolve().rglob("*")
|
||||
if path.is_file()
|
||||
and path.suffix.lower() in CPP_SUFFIXES
|
||||
and not is_excluded_path(path, root, extra_excludes)
|
||||
)
|
||||
return tuple(sorted(found))
|
||||
|
||||
|
||||
def is_test_file(path: Path, root: Path, config) -> bool:
|
||||
resolved = path.resolve()
|
||||
if any(root == resolved or root in resolved.parents for root in config.test_roots):
|
||||
return True
|
||||
relative = resolved.relative_to(root.resolve())
|
||||
if {"test", "tests"} & {part.lower() for part in relative.parts}:
|
||||
return True
|
||||
stem = resolved.stem.lower()
|
||||
return stem.startswith("test_") or stem.endswith(("_test", "_tests", ".test"))
|
||||
|
||||
|
||||
def matching_test_exists(path: Path, config) -> bool:
|
||||
names = tuple(pattern.format(stem=path.stem) for pattern in config.test_patterns)
|
||||
roots = (
|
||||
*config.test_roots,
|
||||
path.parent / "tests",
|
||||
path.parent / "test",
|
||||
)
|
||||
for root in roots:
|
||||
if not root.is_dir():
|
||||
continue
|
||||
for name in names:
|
||||
if any(candidate.is_file() for candidate in root.rglob(name)):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def evaluate_paths(paths, root: Path, config) -> str | None:
|
||||
root = root.resolve()
|
||||
for raw in paths:
|
||||
path = raw.resolve()
|
||||
try:
|
||||
path.relative_to(root)
|
||||
except ValueError:
|
||||
return f"TDD GUARD: '{path}' is outside the repository"
|
||||
if path.suffix.lower() not in CPP_SUFFIXES:
|
||||
continue
|
||||
if (
|
||||
path.name.lower() == "main.cpp"
|
||||
or is_excluded_path(path, root, config.exclude)
|
||||
or is_test_file(path, root, config)
|
||||
):
|
||||
continue
|
||||
if not matching_test_exists(path, config):
|
||||
expected = config.test_patterns[0].format(stem=path.stem)
|
||||
return (
|
||||
f"TDD GUARD: '{path.name}' requires an existing test such as "
|
||||
f"'{expected}'. Add the test in a configured test root first."
|
||||
)
|
||||
return None
|
||||
Reference in New Issue
Block a user