tobi/grammar-inference-engine
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat: add --slice package (per-directory grouping), --include/--exclude, --verbose, drop root group inference, run full ensemble on all groups

Browse source
This commit is contained in:
tobjend 2026-07-04 02:43:03 +02:00
parent 4a506cd39b
commit e3ad256321
4 changed files with 361 additions and 223 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

282
bex/tag_preprocessor/analyze.py
View file

@ -3,7 +3,7 @@
Usage:
python -m bex.tag_preprocessor.analyze <directory> [options]
Runs the full Phase 1.0 pipeline over a directory of source files.
Runs the full pipeline over a directory of source files.
"""
import argparse
@ -11,12 +11,13 @@ import json
import os
import re
import sys
import time
from pathlib import Path
from collections import Counter
import pathspec
from .code import preprocess_by_method, _extract_call_tokens, extract_arg_info, _summarize_arg_info
from .code import preprocess_by_method, extract_arg_info, _summarize_arg_info
from bex.ensemble import infer_ensemble
SUPPORTED_EXTENSIONS = {
@ -157,6 +158,15 @@ def scan_directory(dir_path, gitignore_spec=None):
DEFAULT_COVERAGE = 0.8
VERBOSE = False
_vstart = 0.0
def _vprint(*args, **kwargs):
if VERBOSE:
elapsed = time.time() - _vstart
print(f"[{elapsed:6.1f}s]", *args, file=sys.stderr, **kwargs)
def frequency_filter(sequences, min_coverage=0.2):
"""Remove symbols appearing in fewer than min_coverage fraction of files.
@ -193,78 +203,8 @@ def frequency_filter(sequences, min_coverage=0.2):
return filtered
def cluster_methods(sequences, min_cluster_size=3, ngram_size=3, max_clusters=20):
"""Group method sequences by shared n-gram call patterns.
Extracts call tokens from each sequence, builds an n-gram index,
and assigns methods to all matching clusters (multi-assignment).
Methods that match NO n-gram pattern go to an '(other)' cluster.
Args:
sequences: list of (capture, text, line) lists.
min_cluster_size: minimum methods to form a cluster.
ngram_size: length of n-grams to match (default 3).
max_clusters: max clusters before dumping rest to (other).
Returns:
list of (label, [sequences]) tuples.
"""
if not sequences:
return []
call_seqs = [_extract_call_tokens(seq) for seq in sequences]
ngram_to_methods = {}
for idx, cseq in enumerate(call_seqs):
for i in range(len(cseq) - ngram_size + 1):
ngram = tuple(cseq[i:i + ngram_size])
if len(ngram) == ngram_size:
ngram_to_methods.setdefault(ngram, set()).add(idx)
clusters = []
assigned = set()
for ngram, indices in sorted(ngram_to_methods.items(), key=lambda x: -len(x[1])):
if len(indices) >= min_cluster_size:
if len(clusters) >= max_clusters:
break
label = " → ".join(ngram)
cluster_seqs = [sequences[i] for i in indices]
clusters.append((label, cluster_seqs))
assigned.update(indices)
remaining = [i for i in range(len(sequences)) if i not in assigned]
if remaining:
clusters.append(("(other)", [sequences[i] for i in remaining]))
return clusters
def cluster_methods_adaptive(sequences, min_cluster_size=3, ngram_size=3, other_threshold=0.6):
"""Adaptive clustering: shrink ngram until (other) <= threshold or ngram=1.
When the (other) cluster swallows > other_threshold of methods,
retry with ngram-1. Keeps the smallest ngram that gives acceptable coverage.
"""
for n in range(ngram_size, 0, -1):
clusters = cluster_methods(sequences, min_cluster_size=min_cluster_size, ngram_size=n)
other_count = 0
total = 0
for label, seqs in clusters:
total += len(seqs)
if label == "(other)":
other_count = len(seqs)
if total == 0:
return clusters
if other_count / total <= other_threshold:
break
return clusters
def analyze_clusters(file_paths, extension, project_root="", min_coverage=DEFAULT_COVERAGE, prefer=None, kmax=2, N=3):
"""Run full pipeline with clustering: preprocess → cluster → per-cluster infer.
Each cluster gets its own ensemble inference with BEX core/outlier
detection at the given min_coverage threshold.
"""Run full pipeline: preprocess → frequency filter → ensemble infer.
Returns:
list of (label, ensemble_result_dict, method_count, meta) tuples.
@ -284,21 +224,79 @@ def analyze_clusters(file_paths, extension, project_root="", min_coverage=DEFAUL
return []
sequences = frequency_filter(sequences, min_coverage=0.2)
clusters = cluster_methods_adaptive(sequences)
cluster_fps = set(seq_files)
imports = _extract_imports(cluster_fps)
arg_patterns = _build_arg_patterns(cluster_fps)
packages = _top_packages(cluster_fps, project_root)
symbol_seqs = [[text for _, text, _ in seq] for seq in sequences]
result = infer_ensemble(symbol_seqs, kmax=kmax, N=N, prefer=prefer, min_coverage=min_coverage)
meta = {"files": cluster_fps, "imports": imports, "arg_patterns": arg_patterns, "packages": packages}
return [("(all methods)", result, len(sequences), meta)]
def analyze_by_package(file_paths, extension, project_root="", min_coverage=DEFAULT_COVERAGE, prefer=None, kmax=2, N=3, min_pkg_size=3):
"""Preprocess and group by package directory, infer per group.
Groups methods by their file's relative directory path, merging
small packages (< min_pkg_size methods) upward to their parent.
Returns:
list of (package_label, ensemble_result_dict, method_count, meta).
"""
sequences = []
seq_files = []
t0 = time.time()
for fp in file_paths:
with open(fp) as f:
code = f.read()
for method_seq in preprocess_by_method(fp, code):
if method_seq:
sequences.append(method_seq)
seq_files.append(fp)
if not sequences:
return []
_vprint(f"Preprocess: {len(sequences)} methods from {len(file_paths)} {extension} files ({time.time()-t0:.1f}s)")
seq_packages = [_file_to_package(fp, project_root) for fp in seq_files]
groups, ungrouped = _group_by_package(
list(enumerate(seq_packages)), min_size=min_pkg_size
)
_vprint(f"Groups: {len(groups)} named, {len(ungrouped)} ungrouped methods")
for label, idxs in groups:
_vprint(f" ├ {label} ({len(idxs)} methods)")
if ungrouped:
_vprint(f" └ (other) ({len(ungrouped)} methods)")
results = []
for label, cluster_seqs in clusters:
cluster_fps = set()
for seq in cluster_seqs:
idx = next(i for i, s in enumerate(sequences) if s is seq)
cluster_fps.add(seq_files[idx])
imports = _extract_imports(cluster_fps)
arg_patterns = _build_arg_patterns(cluster_fps)
symbol_seqs = [[text for _, text, _ in seq] for seq in cluster_seqs]
result = infer_ensemble(symbol_seqs, kmax=kmax, N=N, prefer=prefer, min_coverage=min_coverage)
packages = _top_packages(cluster_fps, project_root)
meta = {"files": cluster_fps, "imports": imports, "arg_patterns": arg_patterns, "packages": packages}
results.append((label, result, len(cluster_seqs), meta))
for label, indices in groups:
t1 = time.time()
group_seqs = [sequences[i] for i in indices]
group_files = set(seq_files[i] for i in indices)
group_seqs = frequency_filter(group_seqs, min_coverage=0.2)
imports = _extract_imports(group_files)
arg_patterns = _build_arg_patterns(group_files)
packages = _top_packages(group_files, project_root)
group_prefer = prefer
_vprint(f"Infer {label} ({len(group_seqs)} methods, full ensemble) ... ", end="")
symbol_seqs = [[text for _, text, _ in seq] for seq in group_seqs]
result = infer_ensemble(symbol_seqs, kmax=kmax, N=N, prefer=group_prefer, min_coverage=min_coverage)
_vprint(f"done ({time.time()-t1:.1f}s)")
meta = {"files": group_files, "imports": imports, "arg_patterns": arg_patterns, "packages": packages}
results.append((label, result, len(group_seqs), meta))
if ungrouped:
ungrouped_files = set(seq_files[i] for i in ungrouped)
ungrouped_seqs = [sequences[i] for i in ungrouped]
results.append(("(other)", None, len(ungrouped_seqs), {"files": ungrouped_files, "imports": [], "arg_patterns": {}, "packages": []}))
return results
@ -332,23 +330,84 @@ def infer(file_paths, extension, min_coverage=DEFAULT_COVERAGE, prefer=None, kma
return infer_ensemble(symbol_seqs, kmax=kmax, N=N, prefer=prefer, min_coverage=min_coverage)
def _merge_up(pkg):
"""Go one directory level up from a package path."""
parts = pkg.replace(os.sep, "/").rstrip("/").split("/")
if len(parts) <= 1:
return ""
return "/".join(parts[:-1])
def _group_by_package(indices_and_packages, min_size=3):
"""Group method indices by package, merging small packages upward.
Small packages that would merge into root are discarded (too small
to form a meaningful group and not worth root-level inference).
Args:
indices_and_packages: list of (index, package_path) tuples.
min_size: minimum methods to keep a standalone group.
Returns:
(groups, ungrouped) where groups is [(label, [indices])] sorted by label,
and ungrouped is a list of indices that couldn't form a named group.
"""
pkg_to_indices = {}
for idx, pkg in indices_and_packages:
pkg_to_indices.setdefault(pkg, []).append(idx)
ungrouped = []
while True:
to_merge = {p for p, idxs in pkg_to_indices.items()
if p != "" and len(idxs) < min_size}
if not to_merge:
break
new_groups = {}
for pkg, indices in pkg_to_indices.items():
if pkg in to_merge:
parent = _merge_up(pkg)
if parent == "":
ungrouped.extend(indices)
else:
new_groups.setdefault(parent, []).extend(indices)
else:
new_groups.setdefault(pkg, []).extend(indices)
pkg_to_indices = new_groups
return sorted(pkg_to_indices.items(), key=lambda x: x[0]), ungrouped
def _filter_glob(files, include=None, exclude=None):
"""Filter file list by include/exclude glob patterns."""
if include:
spec = pathspec.PathSpec.from_lines("gitwildmatch", [include])
files = [f for f in files if spec.match_file(f)]
if exclude:
spec = pathspec.PathSpec.from_lines("gitwildmatch", [exclude])
files = [f for f in files if not spec.match_file(f)]
return files
def analyze_directory(
dir_path,
min_coverage=DEFAULT_COVERAGE,
prefer=None,
kmax=2,
slice="flat",
include=None,
cluster=True,
exclude=None,
):
"""Scan a directory and run cluster analysis for each language found.
"""Scan a directory and run analysis for each language found.
Args:
dir_path: directory to scan.
min_coverage: minimum file fraction for symbol to be kept.
min_coverage: BEX core coverage threshold for outlier removal.
prefer: algorithm preference.
kmax: max k for k-ORE algorithms.
include: optional glob pattern to filter files (e.g. '**/src/main/**').
cluster: if True, run method-level clustering (default).
slice: grouping strategy — "flat" (one per language) or "package" (per directory).
include: optional glob — only process files matching this pattern.
exclude: optional glob — skip files matching this pattern.
Returns:
dict mapping extension → list of (label, result_dict, count, meta) tuples.
@ -358,12 +417,11 @@ def analyze_directory(
for ext, files in groups.items():
if len(files) < 1:
continue
if include:
files = [f for f in files if _match_glob(f, include)]
files = _filter_glob(files, include=include, exclude=exclude)
if not files:
continue
if cluster:
results[ext] = analyze_clusters(
if slice == "package":
results[ext] = analyze_by_package(
files, ext,
project_root=dir_path,
min_coverage=min_coverage,
@ -371,9 +429,13 @@ def analyze_directory(
kmax=kmax,
)
else:
r = infer(files, ext, min_coverage=min_coverage, prefer=prefer, kmax=kmax)
imports = _extract_imports(files)
results[ext] = [("(all methods)", r, 0, {"files": set(files), "imports": imports, "arg_patterns": {}, "packages": _top_packages(files, dir_path)})]
results[ext] = analyze_clusters(
files, ext,
project_root=dir_path,
min_coverage=min_coverage,
prefer=prefer,
kmax=kmax,
)
return results
@ -419,17 +481,17 @@ def _parse_args(argv=None):
"--min-coverage", type=float, default=DEFAULT_COVERAGE,
help="BEX core coverage threshold — outlier methods are removed until this fraction remains (default: 0.8)",
)
parser.add_argument(
"--slice", choices=["flat", "package"], default="flat",
help="Grouping strategy: flat (one per language) or package (per directory) (default: flat)",
)
parser.add_argument(
"--include",
help="Glob pattern to filter files (e.g. '**/src/main/**')",
help="Glob pattern to filter files (e.g. '**/src/test/**')",
)
parser.add_argument(
"--min-cluster-size", type=int, default=3,
help="Minimum methods to form a cluster (default: 3)",
)
parser.add_argument(
"--ngram-size", type=int, default=3,
help="N-gram length for clustering (default: 3)",
"--exclude",
help="Glob pattern to skip files (e.g. '**/build/**')",
)
parser.add_argument(
"--format", choices=["text", "json"], default="text",
@ -439,17 +501,27 @@ def _parse_args(argv=None):
"--json", action="store_true", dest="json_flag",
help="Shortcut for --format json",
)
parser.add_argument(
"--verbose", action="store_true",
help="Print progress to stderr",
)
return parser.parse_args(argv)
def main():
args = _parse_args()
global VERBOSE, _vstart
VERBOSE = args.verbose
_vstart = time.time()
_vprint(f"Scanning {args.directory} ...")
results = analyze_directory(
args.directory,
min_coverage=args.min_coverage,
prefer=args.prefer,
kmax=args.kmax,
slice=args.slice,
include=args.include,
exclude=args.exclude,
)
if args.json_flag or args.format == "json":

75
docs/adr/0012-remove-ngram-clustering.md Normal file
View file

@ -0,0 +1,75 @@
# 12. Remove n-gram clustering from pipeline
**Date:** 2026-07-04
**Status:** Accepted
## Context
The pipeline grouped method sequences by shared n-gram call patterns before
inference. The idea: methods in the same category (test, config, helper) would
cluster together, and each cluster would get a cleaner, more specific grammar.
## What Was Removed
- `cluster_methods()` — built n-gram→method index from call tokens, assigned
methods to shared-pattern clusters, dumped remainder to `(other)`.
- `cluster_methods_adaptive()` — retried clustering at ngram=2 then ngram=1
when `(other)` exceeded 60% of methods.
- `_extract_call_tokens()` import from `code.py` (still in `code.py` but no
longer called by the pipeline).
- `--min-cluster-size` and `--ngram-size` CLI flags.
- `cluster` parameter in `analyze_directory()`.
- Per-cluster loop in `analyze_clusters()` — metadata extraction (files,
imports, arg patterns) now runs once across all methods.
## Why Removed
1. **No benefit for diverse codebases.** On Kotlin SpringRAG (1581 methods),
21 named clusters formed (3-13 methods each) but 1254 landed in `(other)`.
The named clusters were too small to produce better grammars than running
inference once on the filtered set.
2. **Added complexity for zero signal gain.** The n-gram labels (e.g.
`locator → click → waitForTimeout`) restated what CRX already outputs as
`(locator+click+waitForTimeout)+`.per-cluster infer_ensemble call was
redundant with the single-pass result.
3. **Slower.** 22 extra BEX calls (one per named cluster) for grammars
that would appear in the single-pass result anyway.
## Pipeline After Removal
```
preprocess_by_method → frequency_filter(0.2) → infer_ensemble(0.8)
```
Single pass. Metadata extracted once.
## How to Reintroduce
The removed code is preserved in the archive branch:
```
git archive/unreverted-25898c2
```
Files:
- `bex/tag_preprocessor/analyze.py` contains `cluster_methods()`,
`cluster_methods_adaptive()`, and the per-cluster loop body.
To restore:
1. Cherry-pick or copy the two function definitions.
2. Re-add `_extract_call_tokens` to the import from `code.py`.
3. Re-add the `--min-cluster-size` and `--ngram-size` CLI flags.
4. Change `analyze_clusters()` back to: cluster → per-cluster filter → per-cluster infer.
5. Restore the cluster parameter in `analyze_directory()`.
## Performance Impact
| Metric | Before (with clustering) | After (single pass) |
|--------|--------------------------|---------------------|
| Pipeline time (460 .kt files) | ~42s | ~3s |
| Inference time per cluster | ~22s on (other) | ~0.1s total |
| Named clusters | 21 tiny + (other) | 1 group |
| Grammar quality | same `assertEquals+` | same `assertEquals+` |

172
docs/pipeline-overview.txt
View file

@ -1,101 +1,85 @@
```
┌───────────────────────────────┐
│ Source Code Directory │
│ (.py .js .ts .kt .rb .go │
│ .rs .java .c .cpp .h) │
└──────────┬────────────────────┘
│ scan_directory()
▼
┌───────────────────────────────┐
│ Files grouped by extension │
│ .kt → [a.kt, b.kt, ...] │
│ .py → [x.py, y.py, ...] │
└──────────┬────────────────────┘
│ for each extension
▼
┌───────────────────────────────────────────┐
│ preprocess_by_method(file_path, code) │
│ │
│ tree-sitter parser ◄── _load_grammar() │
│ + │
│ highlights.scm query ◄── _load_query() │
│ │ │
│ ▼ │
│ BEHAVIORAL_PREFIXES filter │
│ (definition./reference./keyword./ │
│ function/attribute/constructor/ │
│ label/type.definition/module) │
│ │ │
│ ▼ │
│ _find_method_bodies() │
│ grouped by body boundaries │
│ │ │
│ ▼ │
│ [(cap, text, line), ...] per method │
└──────────┬────────────────────────────────┘
│ list of sequences
▼
┌───────────────────────────────────────────┐
│ frequency_filter(sequences, 0.2) │
│ removes symbols in <20% of files │
└──────────┬────────────────────────────────┘
│ filtered sequences
▼
┌───────────────────────────────────────────┐
│ cluster_methods_adaptive() │
│ │
│ Try ngram=3 │
│ ├── _extract_call_tokens() → call seqs │
│ ├── build ngram→indices index │
│ ├── top 20 by freq → clusters │
│ └── if (other) > 60% → ngram=2 → ... │
│ │
│ Multi-assignment: no used set │
│ Methods can be in N clusters │
└──────────┬────────────────────────────────┘
│ clusters: [(label, [seqs]), ...]
▼
┌────────────────────────────────────────────────────────┐
│ For each cluster: │
│ │
│ ┌─────────────────────┐ ┌────────────────────────┐ │
│ │ _extract_imports() │ │ _build_arg_patterns() │ │
│ │ scan 200 lines │ │ extract_arg_info() │ │
│ │ for import/from/ │ │ + _classify_arg_node │ │
│ │ require/#include/ │ │ + _find_arglist_node │ │
│ │ use/include │ │ + _iterate_arg_nodes │ │
│ └─────────┬──────────┘ │ + _summarize_arg_info│ │
│ │ └───────────┬────────────┘ │
│ ▼ ▼ │
│ ┌──────────────────────────────────────────────────┐ │
│ │ infer_ensemble(symbol_seqs, min_coverage=0.8) │ │
┌───────────────────────────────┐
│ Source Code Directory │
│ (.py .js .ts .kt .rb .go │
│ .rs .java .c .cpp .h) │
└──────────┬────────────────────┘
│ scan_directory()
▼
┌───────────────────────────────┐
│ Files grouped by extension │
│ .kt → [a.kt, b.kt, ...] │
│ .py → [x.py, y.py, ...] │
└──────────┬────────────────────┘
│ for each extension
▼
┌───────────────────────────────────────────┐
│ preprocess_by_method(file_path, code) │
│ │
│ tree-sitter parser ◄── _load_grammar() │
│ + │
│ highlights.scm query ◄── _load_query() │
│ │ │
│ ▼ │
│ BEHAVIORAL_PREFIXES filter │
│ (definition./reference./keyword./ │
│ function/attribute/constructor/ │
│ label/type.definition/module) │
│ │ │
│ ▼ │
│ _find_method_bodies() │
│ grouped by body boundaries │
│ │ │
│ ▼ │
│ [(cap, text, line), ...] per method │
└──────────┬────────────────────────────────┘
│ list of sequences
▼
┌───────────────────────────────────────────┐
│ frequency_filter(sequences, 0.2) │
│ removes symbols in <20% of methods │
└──────────┬────────────────────────────────┘
│ filtered sequences
▼
┌────────────────────────────────────────────────────────┐
│ │
│ ┌─────────────────────┐ ┌────────────────────────┐ │
│ │ _extract_imports() │ │ _build_arg_patterns() │ │
│ │ scan 200 lines │ │ extract_arg_info() │ │
│ │ for import/from/ │ │ + _classify_arg_node │ │
│ │ require/#include/ │ │ + _find_arglist_node │ │
│ │ use/include │ │ + _iterate_arg_nodes │ │
│ └─────────┬──────────┘ │ + _summarize_arg_info│ │
│ │ └───────────┬────────────┘ │
│ ▼ ▼ │
│ ┌──────────────────────────────────────────────────┐ │
│ │ infer_ensemble(symbol_seqs, min_coverage=0.8) │ │
│ │ ├── CRX (fast, unordered) │ │
│ │ ├── iDRegEx (ordered regex) │ │
│ │ └── kOREInference (noisy, probabilistic) │ │
│ │ └── pick best by MDL score │ │
│ │ └── core/outlier split via _find_core(0.8) │ │
│ └──────────────────────┬───────────────────────────┘ │
│ ▼ │
│ (label, result, count, meta) │
│ meta = {files, imports, arg_patterns, │
│ packages: _file_to_package(relpath)} │
└─────────────────────────┬──────────────────────────────┘
▼
┌─────────────────────────────────────────────────────┐
│ Output │
│ │
│ --format text (default) --format json │
│ ┌──────────────────┐ ┌──────────────────┐ │
│ │ .kt: │ │ [{ │ │
│ │ ├─ assertEquals │ │ "language": │ │
│ │ │ (327 meth.) │ │ ".kt", │ │
│ │ │ Grammar: │ │ "conventions": │ │
│ │ │ assertEquals+│ │ [{...}, ...] │ │
│ │ │ Imports: ... │ │ }] │ │
│ │ │ Args(assertEquals): │ │ │
│ │ │ n=2 [lit,var]│ │ → inject into │ │
│ │ ├─ (other) │ │ LLM prompt │ │
│ │ │ (1254 meth.) │ └──────────────────┘ │
│ └──────────────────┘ │
└─────────────────────────────────────────────────────┘
│ └──────────────────────┬───────────────────────────┘ │
│ ▼ │
│ (result, meta) │
│ meta = {files, imports, arg_patterns, │
│ packages: _file_to_package(relpath)} │
└─────────────────────────┬──────────────────────────────┘
▼
┌─────────────────────────────────────────────────────┐
│ Output │
│ │
│ --format text (default) --format json │
│ ┌──────────────────┐ ┌──────────────────┐ │
│ │ .kt: │ │ [{ │ │
│ │ Grammar: │ │ "language": │ │
│ │ assertEquals+ │ │ ".kt", │ │
│ │ Imports: ... │ │ "conventions": │ │
│ │ Args(assertEquals): │ [{...}, ...] │ │
│ │ n=2 [lit,var] │ │ }] │ │
│ └──────────────────┘ │ │ │
│ │ → inject into │ │
│ │ LLM prompt │ │
│ └──────────────────┘ │
└─────────────────────────────────────────────────────┘
```

55
tests/test_analyze.py
View file

@ -7,7 +7,8 @@ import sys
sys.path.insert(0, str(Path(__file__).parent.parent))
from bex.tag_preprocessor.analyze import (
scan_directory, frequency_filter, infer, analyze_directory, _match_glob,
scan_directory, frequency_filter, infer, analyze_directory, _filter_glob,
_group_by_package,
)
@ -119,29 +120,35 @@ def test_infer_returns_ensemble_dict():
print(" PASS test_infer_returns_ensemble_dict")
def test_match_glob():
assert _match_glob("/repo/src/main/app.kt", "**/src/main/**")
assert _match_glob("/repo/src/main/org/app.kt", "**/src/main/**")
assert _match_glob("/repo/src/main/deep/nested/app.kt", "**/src/main/**")
assert not _match_glob("/repo/src/test/app.kt", "**/src/main/**")
assert not _match_glob("/repo/build/app.kt", "**/src/main/**")
print(" PASS test_match_glob")
def test_filter_glob():
files = [
"/repo/src/main/app.kt",
"/repo/src/main/org/app.kt",
"/repo/src/test/app.kt",
"/repo/build/app.kt",
]
inc = _filter_glob(files, include="**/src/main/**")
assert inc == ["/repo/src/main/app.kt", "/repo/src/main/org/app.kt"]
exc = _filter_glob(files, exclude="**/build/**")
assert exc == ["/repo/src/main/app.kt", "/repo/src/main/org/app.kt", "/repo/src/test/app.kt"]
both = _filter_glob(files, include="**/src/**", exclude="**/test/**")
assert both == ["/repo/src/main/app.kt", "/repo/src/main/org/app.kt"]
print(" PASS test_filter_glob")
def test_analyze_directory_include_glob():
with tempfile.TemporaryDirectory() as td:
d = Path(td)
(d / "src" / "main").mkdir(parents=True)
(d / "src" / "test").mkdir(parents=True)
(d / "src" / "main" / "prod.py").write_text("def setup():\n pass\ndef run():\n return x")
(d / "src" / "test" / "test_prod.py").write_text("def test_run():\n assert run() == x")
results = analyze_directory(td, include="**/src/main/**")
assert ".py" in results
assert len(results[".py"]) >= 1
for label, r, count, meta in results[".py"]:
if r and r.get("best"):
assert r["best"]["grammar"] is not None
print(" PASS test_analyze_directory_include_glob")
def test_group_by_package():
data = [(0, "src/main"), (1, "src/main"), (2, "src/main"),
(3, "src/test"), (4, "src/test"),
(5, "docs")]
groups, ungrouped = _group_by_package(data, min_size=3)
labels = [l for l, _ in groups]
assert "src/main" in labels
assert "src" not in labels
assert "" not in labels # no root group — small packages discarded
assert len(groups) == 1
assert len(ungrouped) == 3 # src/test (2) + docs (1) → discarded
print(" PASS test_group_by_package")
def test_infer_low_coverage_filters_noise():
@ -168,8 +175,8 @@ def run_all():
test_scan_directory_skips_unsupported,
test_scan_directory_nested,
test_scan_directory_skips_build_dirs,
test_match_glob,
test_analyze_directory_include_glob,
test_filter_glob,
test_group_by_package,
test_frequency_filter_nothing_to_filter,
test_frequency_filter_removes_infrequent_symbol,
test_frequency_filter_edge_empty_sequences,