"""Tests for tag-preprocessor orchestrator (analyze.py).""" from pathlib import Path import tempfile import sys sys.path.insert(0, str(Path(__file__).parent.parent)) from bex.tag_preprocessor.analyze import ( scan_directory, frequency_filter, infer, analyze_directory, _filter_glob, _group_by_package, _count_optionals, _should_try_idregex, ) def test_scan_directory_empty(): with tempfile.TemporaryDirectory() as td: result = scan_directory(td) assert result == {} print(" PASS test_scan_directory_empty") def test_scan_directory_groups_by_extension(): with tempfile.TemporaryDirectory() as td: d = Path(td) (d / "a.py").write_text("x = 1") (d / "b.py").write_text("y = 2") (d / "c.js").write_text("let x = 1;") (d / "d.rs").write_text("fn main() {}") result = scan_directory(td) assert ".py" in result assert ".js" in result assert ".rs" in result assert len(result[".py"]) == 2 assert len(result[".js"]) == 1 assert len(result[".rs"]) == 1 print(" PASS test_scan_directory_groups_by_extension") def test_scan_directory_skips_unsupported(): with tempfile.TemporaryDirectory() as td: d = Path(td) (d / "f.py").write_text("x = 1") (d / "f.txt").write_text("hello") (d / "f.md").write_text("# doc") result = scan_directory(td) assert ".py" in result assert ".txt" not in result assert ".md" not in result print(" PASS test_scan_directory_skips_unsupported") def test_scan_directory_nested(): with tempfile.TemporaryDirectory() as td: d = Path(td) (d / "sub").mkdir() (d / "sub" / "a.py").write_text("x = 1") (d / "b.py").write_text("y = 2") result = scan_directory(td) assert len(result[".py"]) == 2 print(" PASS test_scan_directory_nested") def test_scan_directory_skips_build_dirs(): with tempfile.TemporaryDirectory() as td: d = Path(td) (d / "src" / "main").mkdir(parents=True) (d / "build" / "reports").mkdir(parents=True) (d / "node_modules" / "pkg").mkdir(parents=True) (d / "src" / "main" / "app.py").write_text("x = 1") (d / "build" / "reports" / "report.js").write_text("let x = 1;") (d / "node_modules" / "pkg" / "index.js").write_text("let y = 2;") result = scan_directory(td) assert ".py" in result assert ".js" not in result assert len(result[".py"]) == 1 print(" PASS test_scan_directory_skips_build_dirs") def test_frequency_filter_nothing_to_filter(): sequences = [ [("function", "foo", 1), ("keyword.return", "return", 2)], [("function", "bar", 1), ("keyword.return", "return", 2)], ] filtered = frequency_filter(sequences, min_coverage=0.5) assert len(filtered) == 2 assert filtered == sequences print(" PASS test_frequency_filter_nothing_to_filter") def test_frequency_filter_removes_infrequent_symbol(): sequences = [ [("keyword.return", "return", 1)], [("keyword.return", "return", 1)], [("function", "rare_fn", 1)], ] filtered = frequency_filter(sequences, min_coverage=0.67) assert len(filtered) == 3 assert len(filtered[0]) == 1 # "return" kept assert len(filtered[1]) == 1 # "return" kept assert len(filtered[2]) == 0 # "rare_fn" removed print(" PASS test_frequency_filter_removes_infrequent_symbol") def test_frequency_filter_edge_empty_sequences(): assert frequency_filter([], min_coverage=0.5) == [] assert frequency_filter([[], []], min_coverage=0.5) == [[], []] print(" PASS test_frequency_filter_edge_empty_sequences") def test_infer_returns_ensemble_dict(): with tempfile.TemporaryDirectory() as td: d = Path(td) (d / "a.py").write_text("def foo():\n return 1") (d / "b.py").write_text("def bar():\n return 2") result = infer([str(d / "a.py"), str(d / "b.py")], ".py", min_coverage=0.5) assert isinstance(result, dict) assert "best" in result assert "all" in result assert "why" in result print(" PASS test_infer_returns_ensemble_dict") 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_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(): with tempfile.TemporaryDirectory() as td: d = Path(td) (d / "common.py").write_text( "def setup():\n pass\ndef teardown():\n pass" ) (d / "rare.py").write_text( "def setup():\n pass\ndef one_off():\n raise Exception('boom')" ) result = infer( [str(d / "common.py"), str(d / "rare.py")], ".py", min_coverage=0.6 ) assert result["best"] is not None assert result["best"]["grammar"] is not None print(" PASS test_infer_low_coverage_filters_noise") from bex.grammar import ( Symbol, Concat, Alt, Optional, Plus, Star, Empty, ) def test_count_optionals_flat_chain(): g = Concat([Optional(Symbol('a')), Optional(Symbol('b')), Optional(Symbol('c')), Optional(Symbol('d')), Optional(Symbol('e'))]) n_opt, n_total = _count_optionals(g) assert n_opt == 5 assert n_total == 5 def test_count_optionals_no_optionals(): g = Concat([Symbol('a'), Symbol('b'), Symbol('c')]) n_opt, n_total = _count_optionals(g) assert n_opt == 0 assert n_total == 3 def test_count_optionals_mixed(): # return . error? . (request|response)? . data? -> 3 optional of 4 parts g = Concat([ Symbol('return'), Optional(Symbol('error')), Optional(Alt([Symbol('request'), Symbol('response')])), Optional(Symbol('data')), ]) n_opt, n_total = _count_optionals(g) assert n_opt == 3 assert n_total == 4 def test_count_optionals_repetition_not_optional(): # a . b+ . c? -> only c? is optional g = Concat([Symbol('a'), Plus(Symbol('b')), Optional(Symbol('c'))]) n_opt, n_total = _count_optionals(g) assert n_opt == 1 assert n_total == 3 def test_should_try_idregex_small_many_optionals(): g = Concat([Optional(Symbol('a')), Optional(Symbol('b')), Optional(Symbol('c')), Optional(Symbol('d')), Optional(Symbol('e'))]) assert _should_try_idregex(g, 5) is True def test_should_try_idregex_large_group(): g = Concat([Optional(Symbol('a')), Optional(Symbol('b')), Optional(Symbol('c')), Optional(Symbol('d')), Optional(Symbol('e'))]) assert _should_try_idregex(g, 15) is False def test_should_try_idregex_few_optionals(): g = Concat([Symbol('a'), Symbol('b'), Symbol('c'), Symbol('d'), Symbol('e')]) assert _should_try_idregex(g, 5) is False def test_should_short_concat(): g = Concat([Optional(Symbol('a')), Symbol('b')]) assert _should_try_idregex(g, 5) is False # too few parts def run_all(): tests = [ test_scan_directory_empty, test_scan_directory_groups_by_extension, test_scan_directory_skips_unsupported, test_scan_directory_nested, test_scan_directory_skips_build_dirs, 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, test_infer_returns_ensemble_dict, test_infer_low_coverage_filters_noise, test_count_optionals_flat_chain, test_count_optionals_no_optionals, test_count_optionals_mixed, test_count_optionals_repetition_not_optional, test_should_try_idregex_small_many_optionals, test_should_try_idregex_large_group, test_should_try_idregex_few_optionals, test_should_short_concat, ] passed = 0 failed = 0 for t in tests: try: t() passed += 1 except Exception as e: print(f" FAIL {t.__name__}: {e}") failed += 1 print(f"\n{passed} passed, {failed} failed") if __name__ == "__main__": run_all()