#!/usr/bin/env python3 """Inventory and ratchet common low-value or flaky test patterns.""" import argparse import re import sys from collections import Counter from collections import namedtuple from pathlib import Path from typing import Dict, List REPO = Path(__file__).resolve().parents[2] FIRST_PARTY_ROOTS = ( "internal", "cmd", "desktop", "dev", "web", "e2e", "kits", "miniapp", "scripts", ) EXCLUDED_PARTS = { ".git", "node_modules", "dist", "coverage", ".data", } PATTERNS = { "fixed_waits": re.compile(r"\b(time\.Sleep|waitForTimeout)\s*\("), "overbroad_mocks": re.compile(r"\b(mock\.Anything|gomock\.Any\s*\(|AnyTimes\s*\(|Maybe\s*\()"), "snapshot_assertions": re.compile(r"\b(toMatchSnapshot|toMatchInlineSnapshot)\s*\("), "focused_tests": re.compile(r"\.(only)\s*\("), "skipped_tests": re.compile(r"\.(skip)\s*\("), } SOURCE_READ_PATTERN = re.compile( r"\b(?:os\.)?Read(?:Dir|File)\s*\(|\breadFile(?:Sync)?\s*\(|\bfs\.readFile(?:Sync)?\s*\(" ) SOURCE_ASSERTION_PATTERN = re.compile( r"\b(?:assert|require)\.(?:match|doesNotMatch|Contains|NotContains|Regexp|NotRegexp)\s*\(" r"|\bexpect\s*\([\s\S]{0,240}?\)\.(?:toContain|toMatch)\s*\(" r"|\bstrings\.(?:Contains|HasPrefix|HasSuffix)\s*\(" ) SOURCE_FILE_LITERAL_PATTERN = re.compile( r"""["`][^"`\n]*\.(?:go|ts|tsx|js|mjs|cjs)["`]""" ) GO_SUBTEST_CALLBACK_PATTERN = re.compile( r"\b[A-Za-z_]\w*\.Run\s*\([\s\S]{0,300}?\bfunc\s*\([^)]*\)\s*\{" ) TIDB_FIXTURE_CALL_PATTERN = re.compile( r"\b(?:" r"start\w*(?:Rollback|Isolated|Committed)\w*DB" r"|Start(?:Initialized|Isolated)TiDB\w*" r"|[A-Za-z_]\w*\.RollbackDB" r")\s*\(" ) GO_COMMENT_OR_LITERAL_PATTERN = re.compile( r"//[^\n]*|/\*[\s\S]*?\*/|`[\s\S]*?`|\"(?:\\[\s\S]|[^\"\\])*\"|'(?:\\[\s\S]|[^'\\])*'" ) Hit = namedtuple("Hit", "file line code") FileHit = namedtuple("FileHit", "file lines") def rel(path: Path) -> str: return path.relative_to(REPO).as_posix() def is_excluded(path: Path) -> bool: return bool(EXCLUDED_PARTS.intersection(path.parts)) def is_test_file(path: Path) -> bool: name = path.name return name.endswith("_test.go") or ".test." in name or ".spec." in name def test_files() -> List[Path]: files: List[Path] = [] for root_name in FIRST_PARTY_ROOTS: root = REPO / root_name if not root.exists(): continue for path in root.rglob("*"): if not path.is_file() or is_excluded(path): continue if is_test_file(path): files.append(path) return sorted(files) def scan() -> Dict[str, List[Hit]]: hits = {name: [] for name in PATTERNS} for path in test_files(): for lineno, raw in enumerate(path.read_text(errors="ignore").splitlines(), 1): for name, pattern in PATTERNS.items(): if pattern.search(raw): hits[name].append(Hit(rel(path), lineno, raw.strip())) return hits def scan_large_test_files(line_threshold: int) -> List[FileHit]: hits: List[FileHit] = [] for path in test_files(): line_count = len(path.read_text(errors="ignore").splitlines()) if line_count > line_threshold: hits.append(FileHit(rel(path), line_count)) return sorted(hits, key=lambda hit: (-hit.lines, hit.file)) def is_source_assertion_test(source: str) -> bool: return bool( SOURCE_READ_PATTERN.search(source) and SOURCE_ASSERTION_PATTERN.search(source) and SOURCE_FILE_LITERAL_PATTERN.search(source) ) def scan_source_assertion_test_files() -> List[FileHit]: hits: List[FileHit] = [] for path in test_files(): source = path.read_text(errors="ignore") if is_source_assertion_test(source): hits.append(FileHit(rel(path), len(source.splitlines()))) return sorted(hits, key=lambda hit: (-hit.lines, hit.file)) def strip_go_comments_and_literals(source: str) -> str: """Blank comments and literals while preserving offsets and newlines.""" return GO_COMMENT_OR_LITERAL_PATTERN.sub( lambda match: "".join("\n" if char == "\n" else " " for char in match.group()), source, ) def matching_go_brace(source: str, open_brace: int) -> int: depth = 0 for index in range(open_brace, len(source)): if source[index] == "{": depth += 1 elif source[index] == "}": depth -= 1 if depth == 0: return index return len(source) def tidb_fixture_calls_in_subtests(source: str) -> List[int]: stripped = strip_go_comments_and_literals(source) call_offsets = set() for callback in GO_SUBTEST_CALLBACK_PATTERN.finditer(stripped): open_brace = callback.end() - 1 close_brace = matching_go_brace(stripped, open_brace) for call in TIDB_FIXTURE_CALL_PATTERN.finditer(stripped, open_brace, close_brace): call_offsets.add(call.start()) return sorted(call_offsets) def scan_tidb_fixtures_in_subtests() -> List[Hit]: hits: List[Hit] = [] for path in test_files(): if path.suffix != ".go": continue source = path.read_text(errors="ignore") lines = source.splitlines() for offset in tidb_fixture_calls_in_subtests(source): line = source.count("\n", 0, offset) + 1 hits.append(Hit(rel(path), line, lines[line - 1].strip())) return hits def print_inventory(name: str, hits: List[Hit], *, max_examples: int = 8) -> None: print(f"{name}: {len(hits)}") if not hits: return by_file = Counter(hit.file for hit in hits) for file, count in by_file.most_common(max_examples): print(f" {count:4d} {file}") def print_file_inventory(name: str, hits: List[FileHit], *, max_examples: int = 8) -> None: print(f"{name}: {len(hits)}") for hit in hits[:max_examples]: print(f" {hit.lines:5d} {hit.file}") def fail_if_over(name: str, actual: int, maximum: int, failures: List[str]) -> None: print(f"{name}: {actual} (max {maximum})") if actual > maximum: failures.append(f"{name} grew above baseline: {actual} > {maximum}") def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--max-fixed-waits", type=int, required=True) parser.add_argument("--max-overbroad-mocks", type=int, required=True) parser.add_argument("--max-snapshot-assertions", type=int, required=True) parser.add_argument("--max-focused-tests", type=int, required=True) parser.add_argument("--max-skipped-tests", type=int, required=True) parser.add_argument("--large-test-file-line-threshold", type=int, default=800) parser.add_argument("--max-large-test-files", type=int, required=True) parser.add_argument("--max-source-assertion-test-files", type=int, required=True) parser.add_argument("--max-tidb-fixtures-in-subtests", type=int, required=True) args = parser.parse_args() hits = scan() large_test_files = scan_large_test_files(args.large_test_file_line_threshold) source_assertion_test_files = scan_source_assertion_test_files() tidb_fixtures_in_subtests = scan_tidb_fixtures_in_subtests() print("== Test quality debt inventory ==") for name in ( "fixed_waits", "overbroad_mocks", "snapshot_assertions", "focused_tests", "skipped_tests", ): print_inventory(name, hits[name]) print_file_inventory(f"large_test_files_over_{args.large_test_file_line_threshold}_lines", large_test_files) print_file_inventory("source_assertion_test_files", source_assertion_test_files) print_inventory("tidb_fixtures_in_subtests", tidb_fixtures_in_subtests) failures: List[str] = [] print("\n== Ratchet baselines ==") fail_if_over("fixed_waits", len(hits["fixed_waits"]), args.max_fixed_waits, failures) fail_if_over("overbroad_mocks", len(hits["overbroad_mocks"]), args.max_overbroad_mocks, failures) fail_if_over("snapshot_assertions", len(hits["snapshot_assertions"]), args.max_snapshot_assertions, failures) fail_if_over("focused_tests", len(hits["focused_tests"]), args.max_focused_tests, failures) fail_if_over("skipped_tests", len(hits["skipped_tests"]), args.max_skipped_tests, failures) fail_if_over("large_test_files", len(large_test_files), args.max_large_test_files, failures) fail_if_over( "source_assertion_test_files", len(source_assertion_test_files), args.max_source_assertion_test_files, failures, ) fail_if_over( "tidb_fixtures_in_subtests", len(tidb_fixtures_in_subtests), args.max_tidb_fixtures_in_subtests, failures, ) if failures: for failure in failures: print(f"FAIL: {failure}", file=sys.stderr) print(f"\nTest quality guardrail failed with {len(failures)} issue(s).", file=sys.stderr) return 1 print("\nTest quality guardrail passed.") return 0 if __name__ == "__main__": raise SystemExit(main())