c059d0b7a4
Some checks failed
ci/woodpecker/push/woodpecker Pipeline failed
- _file_to_package: infer package from file path (kotlin/java/python/src) - _top_packages: most common packages per cluster - Packages included in meta dict and JSON output - New tests: test_file_to_package, test_top_packages, test_extract_imports, test_extract_imports_empty, test_extract_imports_no_imports, test_extract_arg_info_python, test_summarize_arg_info
299 lines
10 KiB
Python
299 lines
10 KiB
Python
"""Tests for tag-preprocessor orchestrator (analyze.py)."""
|
|
|
|
from pathlib import Path, PurePath
|
|
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, _match_glob,
|
|
_file_to_package, _top_packages, _extract_imports,
|
|
)
|
|
from bex.tag_preprocessor.code import _summarize_arg_info, _classify_arg_node
|
|
|
|
|
|
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_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_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_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")
|
|
|
|
|
|
def test_file_to_package():
|
|
assert _file_to_package(
|
|
"/repo/src/main/kotlin/com/example/app/Foo.kt", ".kt"
|
|
) == "com/example/app"
|
|
assert _file_to_package(
|
|
"/repo/src/test/java/com/example/FooTest.java", ".java"
|
|
) == "com/example"
|
|
assert _file_to_package(
|
|
"/repo/src/main/python/mypackage/module.py", ".py"
|
|
) == "mypackage"
|
|
assert _file_to_package("/repo/lib/foo.py", ".py") == "lib"
|
|
print(" PASS test_file_to_package")
|
|
|
|
|
|
def test_top_packages():
|
|
fps = {
|
|
"/repo/src/main/kotlin/com/example/a/Foo.kt",
|
|
"/repo/src/main/kotlin/com/example/a/Bar.kt",
|
|
"/repo/src/main/kotlin/com/example/b/Baz.kt",
|
|
}
|
|
pkgs = _top_packages(fps, ".kt")
|
|
assert pkgs[0] == "com/example/a"
|
|
assert pkgs[1] == "com/example/b"
|
|
print(" PASS test_top_packages")
|
|
|
|
|
|
def test_extract_imports():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
d = Path(td)
|
|
(d / "a.kt").write_text(
|
|
"package com.example\nimport io.mockk.every\nimport io.mockk.verify\n\nclass Foo"
|
|
)
|
|
(d / "b.kt").write_text(
|
|
"package com.example\nimport org.junit.Test\n\nclass Bar"
|
|
)
|
|
result = _extract_imports({str(d / "a.kt"), str(d / "b.kt")})
|
|
assert "import io.mockk.every" in result
|
|
assert "import io.mockk.verify" in result
|
|
assert "import org.junit.Test" in result
|
|
assert len(result) == 3
|
|
print(" PASS test_extract_imports")
|
|
|
|
|
|
def test_extract_imports_empty():
|
|
assert _extract_imports(set()) == []
|
|
print(" PASS test_extract_imports_empty")
|
|
|
|
|
|
def test_extract_imports_no_imports():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
d = Path(td)
|
|
(d / "a.py").write_text("x = 1\ny = 2")
|
|
result = _extract_imports({str(d / "a.py")})
|
|
assert result == []
|
|
print(" PASS test_extract_imports_no_imports")
|
|
|
|
|
|
def test_extract_arg_info_python():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
d = Path(td)
|
|
(d / "a.py").write_text(
|
|
"def test_foo():\n"
|
|
" result = compute(42)\n"
|
|
" assert result == 0\n"
|
|
" run(mock, times=1)\n"
|
|
)
|
|
from bex.tag_preprocessor.code import extract_arg_info
|
|
fp = str(d / "a.py")
|
|
code = Path(fp).read_text()
|
|
info = extract_arg_info(fp, code)
|
|
# compute(42) -> 1 arg: literal
|
|
assert "compute" in info
|
|
obs = info["compute"]
|
|
assert any(n == 1 and ts == ("lit",) for n, ts in obs)
|
|
# run(mock, times=1) -> 2 args: var + kwarg
|
|
assert "run" in info
|
|
obs2 = info["run"]
|
|
assert any(n == 2 for n, ts in obs2)
|
|
assert any(ts == ("var", "kwarg") for n, ts in obs2)
|
|
print(" PASS test_extract_arg_info_python")
|
|
|
|
|
|
def test_summarize_arg_info():
|
|
info = {
|
|
"foo": [(2, ("var", "lit")), (2, ("var", "lit")), (3, ("var", "lit", "lit"))],
|
|
"bar": [(0, ()), (1, ("lambda",))],
|
|
}
|
|
summary = _summarize_arg_info(info)
|
|
assert summary["foo"]["occurrences"] == 3
|
|
assert summary["foo"]["arg_count"]["min"] == 2
|
|
assert summary["foo"]["arg_count"]["max"] == 3
|
|
assert summary["foo"]["arg_count"]["common"] == 2
|
|
assert len(summary["foo"]["patterns"]) == 2
|
|
assert summary["bar"]["occurrences"] == 2
|
|
assert summary["bar"]["arg_count"]["min"] == 0
|
|
print(" PASS test_summarize_arg_info")
|
|
|
|
|
|
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_match_glob,
|
|
test_analyze_directory_include_glob,
|
|
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_file_to_package,
|
|
test_top_packages,
|
|
test_extract_imports,
|
|
test_extract_imports_empty,
|
|
test_extract_imports_no_imports,
|
|
test_extract_arg_info_python,
|
|
test_summarize_arg_info,
|
|
]
|
|
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()
|