diff --git a/test/source_corpus.py b/test/source_corpus.py index 85592a95279..47c9313faa5 100644 --- a/test/source_corpus.py +++ b/test/source_corpus.py @@ -33,6 +33,7 @@ import ast import functools +import unicodedata from collections.abc import Iterator, Sequence from pathlib import Path @@ -78,6 +79,30 @@ def unreadable_files() -> tuple[Path, ...]: return _read_tree()[1] +def _nfkc(text: str) -> str: + """NFKC-normalise, matching how CPython folds identifiers at parse time.""" + return unicodedata.normalize("NFKC", text) + + +@functools.lru_cache(maxsize=1) +def _normalized_texts() -> tuple[str, ...]: + """NFKC-normalised copy of every file's text, in ``source_texts`` order. + + A gate matches on a bare identifier, and CPython NFKC-folds identifiers at + parse time -- so a call written with a Unicode compatibility homoglyph of a + guarded name (``delete_items_b\uff41tch``) is that ASCII name in the AST but + NOT in the raw bytes. Filtering on raw text would skip the file and let the + offender through green. Normalising the haystack (here) and the needle (in + ``candidate_sources``) the same way closes that hole while keeping the + narrowing: NFKC is a fixpoint on ASCII, so every raw ASCII match is + preserved and only homoglyph spellings are newly caught. Computed once over + the whole tree (~0.3s) and cached, like the read itself. ``source_texts`` + still returns the RAW text, which gates that scan comments or string + literals (a ``# render-ok`` marker, an import alias) depend on. + """ + return tuple(_nfkc(text) for _path, text in source_texts()) + + def candidate_sources( require_all: Sequence[str] = (), require_any: Sequence[str] = (), @@ -87,11 +112,18 @@ def candidate_sources( An empty ``require_any`` imposes no alternation, so passing neither argument returns the whole corpus. """ + # Match on the NFKC-normalised text with NFKC-normalised needles, so a call + # whose identifier is a Unicode compatibility homoglyph of a literal (which + # CPython folds to that literal at parse time, making it a real AST match) is + # not skipped by a raw-byte pre-filter. The yielded ``text`` stays RAW. + all_n = tuple(_nfkc(lit) for lit in require_all) + any_n = tuple(_nfkc(lit) for lit in require_any) + texts = source_texts() + norm = _normalized_texts() return tuple( (path, text) - for path, text in source_texts() - if all(lit in text for lit in require_all) - and (not require_any or any(lit in text for lit in require_any)) + for (path, text), ntext in zip(texts, norm) + if all(lit in ntext for lit in all_n) and (not any_n or any(lit in ntext for lit in any_n)) ) diff --git a/test/test_knowledge_delete_off_loop.py b/test/test_knowledge_delete_off_loop.py index 699f200e760..20630343d9b 100644 --- a/test/test_knowledge_delete_off_loop.py +++ b/test/test_knowledge_delete_off_loop.py @@ -21,11 +21,11 @@ import ast import asyncio import json -import pathlib import threading from unittest.mock import MagicMock import pytest +from source_corpus import parsed_candidates, src_root from kiro_crew.knowledge.folder_watcher import FolderWatcher from kiro_crew.knowledge.store import KnowledgeStore @@ -35,7 +35,7 @@ # repo's other on-loop guards use. _NESTED_SCOPES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda) -_SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "kiro_crew" +_SRC = src_root() def _called_names(body: list[ast.stmt]) -> set[tuple[str, int]]: @@ -101,11 +101,11 @@ def _on_loop_call_sites(name: str) -> list[str]: reaches it (see ``_sync_helpers_reaching``). """ found: list[str] = [] - for path in sorted(_SRC.rglob("*.py")): - try: - tree = ast.parse(path.read_text(errors="replace")) - except SyntaxError: # pragma: no cover - syntax is enforced elsewhere - continue + # Only files whose TEXT holds ``name`` can call it (directly, or through a + # same-module sync helper that does), so the shared corpus parses just those + # instead of re-walking all ~1250 modules for this gate. ``src_root()`` is the + # same tree the old ``_SRC`` named, so the relative paths below are unchanged. + for path, _text, tree in parsed_candidates(require_all=(name,)): indirect = _sync_helpers_reaching(tree, name) for fn in (n for n in ast.walk(tree) if isinstance(n, ast.AsyncFunctionDef)): for called, lineno in sorted(_called_names(fn.body), key=lambda c: c[1]): diff --git a/test/test_slack_render_pipeline.py b/test/test_slack_render_pipeline.py index 6464acf08aa..f412e082778 100644 --- a/test/test_slack_render_pipeline.py +++ b/test/test_slack_render_pipeline.py @@ -53,6 +53,7 @@ import tokenize import pytest +from source_corpus import parsed_candidates from kiro_crew.slack.format import ( CONTINUATION, @@ -174,10 +175,15 @@ def find_violations(source: str, path: str = "") -> list[tuple[str, int, def collect_repo_violations() -> list[tuple[str, int, str]]: """Scan every ``kiro_crew/**/*.py`` except the owning module.""" - root = _src_root() - base = root.parent + base = _src_root().parent out: list[tuple[str, int, str]] = [] - for py in sorted(root.rglob("*.py")): + # A direct call to to_slack_mrkdwn -- bare-imported or reached as + # ``.to_slack_mrkdwn`` -- can only exist in a file whose TEXT holds the + # literal ``to_slack_mrkdwn`` (the import that binds the alias, or the attribute + # call itself), so the shared corpus parses just those files rather than + # re-walking the whole package for this one gate. Narrowing on the literal + # drops non-matches only (proved lossless in the PR body). + for py, text, _tree in parsed_candidates(require_all=(_BANNED_FUNC,)): try: rel = str(py.relative_to(base)) except ValueError: # pragma: no cover - defensive @@ -185,11 +191,7 @@ def collect_repo_violations() -> list[tuple[str, int, str]]: if rel.replace("\\", "/").endswith(_OWNER_MODULE): continue try: - src = py.read_text(encoding="utf-8") - except (OSError, UnicodeDecodeError): # pragma: no cover - defensive - continue - try: - out.extend(find_violations(src, rel)) + out.extend(find_violations(text, rel)) except SyntaxError: # pragma: no cover - defensive continue return out