1940 lines
77 KiB
Python
1940 lines
77 KiB
Python
"""
|
|
Test Selector - Precision test selector based on coverage data (line, function, file granularity)
|
|
|
|
Workflow:
|
|
1. Build 'test case -> covered lines' mapping from coverage SQLite data
|
|
2. Parse code changes (supports GitHub PR or local file hash comparison)
|
|
3. Select affected test cases (by line, function, file granularity)
|
|
"""
|
|
|
|
import argparse
|
|
import ast
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import ssl
|
|
import subprocess
|
|
import tempfile
|
|
import textwrap
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
import regex as re
|
|
|
|
# ==================== Configuration ====================
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
|
|
# Repository name: used for filtering and path normalization
|
|
REPO_NAME = "sglang"
|
|
|
|
# Product code path prefix in coverage data / diff paths.
|
|
# sglang: coverage path is /__w/sglang/sglang/python/sglang/xxx.py, diff path is python/sglang/xxx.py
|
|
# -> PRODUCT_PREFIX = "python/sglang/"
|
|
# After stripping this prefix, both sides produce the same relative path (e.g. srt/models/qwen3_vl.py)
|
|
PRODUCT_PREFIX = "python/sglang/"
|
|
|
|
|
|
# Directory prefix of test case folders under coverage data dir.
|
|
TEST_CASE_DIR_PREFIX = "____w__sglang__sglang__test__"
|
|
|
|
# Coverage density threshold: proportion of changed lines covered
|
|
# Range: 0.0 ~ 1.0, higher value = stricter filtering
|
|
# Example: 0.05 means at least 5% of changed lines must be covered
|
|
# Recommendation: start at 0.05, increase to 0.10/0.15/0.20 if too many results
|
|
COVERAGE_DENSITY_THRESHOLD = 0.0
|
|
|
|
# Minimum affected lines threshold
|
|
MIN_AFFECTED_LINES = 1
|
|
|
|
|
|
def _get_test_files_from_pr_diff(diff_file: str) -> list[str]:
|
|
"""
|
|
Extract new/modified test files from PR diff.
|
|
Test files must be in tests/ directory and start with test_
|
|
|
|
Args:
|
|
diff_file: Path to the PR diff file
|
|
|
|
Returns:
|
|
List of test case names that correspond to new/modified test files
|
|
"""
|
|
test_files_found = []
|
|
|
|
try:
|
|
with open(diff_file, encoding="utf-8-sig") as f:
|
|
diff_content = f.read()
|
|
except Exception as e:
|
|
print(f" Warning: Failed to read diff file for test file detection: {e}")
|
|
return test_files_found
|
|
|
|
# Pattern to match test file paths: test/registered/ directory
|
|
# In diff output:
|
|
# - +++ b/test/registered/unit/xxx/test_xxx.py (new/modified test file)
|
|
# - rename to test/registered/unit/xxx/test_xxx.py (renamed test file)
|
|
# Test files must be in test/ directory and start with test_
|
|
test_file_pattern = re.compile(
|
|
r"^(?:\+\+\+ [ab]/|rename to )((?:test/registered(?:/.+)?/test_\w+\.py|test/(?:unit|e2e|integration)(?:/.+)?/test_\w+\.py))",
|
|
re.MULTILINE,
|
|
)
|
|
|
|
changed_test_files = set()
|
|
for match in test_file_pattern.finditer(diff_content):
|
|
test_file_path = match.group(1)
|
|
changed_test_files.add(test_file_path)
|
|
|
|
if not changed_test_files:
|
|
return test_files_found
|
|
|
|
print(
|
|
f" Found {len(changed_test_files)} changed test file(s): {changed_test_files}"
|
|
)
|
|
|
|
# Add all changed test files directly to recommended list (no matching with test_case_map)
|
|
for changed_file in changed_test_files:
|
|
if changed_file not in test_files_found:
|
|
test_files_found.append(changed_file)
|
|
|
|
return test_files_found
|
|
|
|
|
|
def _get_deleted_test_files_from_pr(diff_file: str, test_case_map: dict) -> list[str]:
|
|
"""
|
|
Extract deleted test files from PR diff.
|
|
Test files are in test/registered/ directory with test_*.py pattern.
|
|
|
|
Args:
|
|
diff_file: Path to the PR diff file
|
|
test_case_map: Mapping of test case names to their coverage info
|
|
|
|
Returns:
|
|
List of test case names that correspond to deleted test files
|
|
"""
|
|
deleted_test_files = []
|
|
|
|
try:
|
|
with open(diff_file, encoding="utf-8-sig") as f:
|
|
diff_content = f.read()
|
|
except Exception as e:
|
|
print(f" Warning: Failed to read diff file for deleted test detection: {e}")
|
|
return deleted_test_files
|
|
|
|
# Pattern to match deleted test files (test/registered/ directory)
|
|
# Match --- a/test/... followed by +++ /dev/null (deleted file marker)
|
|
deleted_pattern = re.compile(
|
|
r"^--- a/(test/registered(?:/.+)?/test_\w+\.py|test/(?:unit|e2e|integration)(?:/.+)?/test_\w+\.py)\s*\n\s*\+\+\+ [ab]?/dev/null",
|
|
re.MULTILINE,
|
|
)
|
|
|
|
for match in deleted_pattern.finditer(diff_content):
|
|
test_file_path = match.group(1)
|
|
deleted_test_files.append(test_file_path)
|
|
|
|
if deleted_test_files:
|
|
print(
|
|
f" Found {len(deleted_test_files)} deleted test file(s): {deleted_test_files}"
|
|
)
|
|
|
|
return deleted_test_files
|
|
|
|
|
|
class CoverageSelector:
|
|
"""Coverage-based test selector"""
|
|
|
|
def __init__(
|
|
self, coverage_data_dir: str | None = None, source_dir: str | None = None
|
|
):
|
|
"""
|
|
Args:
|
|
coverage_data_dir: Coverage data directory (only needed for building map)
|
|
source_dir: Source code directory (only needed for function-level matching)
|
|
"""
|
|
self.coverage_data_dir = Path(coverage_data_dir) if coverage_data_dir else None
|
|
self.source_dir = Path(source_dir) if source_dir else None
|
|
self.test_case_map = {} # test_case_name -> {files: {filepath: {lines}}}
|
|
self._noise_lines_cache = {} # filepath -> set of noise lines (import + def)
|
|
|
|
def scan_test_cases(self) -> list[str]:
|
|
"""
|
|
Scan all test case directories.
|
|
"""
|
|
test_cases = []
|
|
if not self.coverage_data_dir or not self.coverage_data_dir.exists():
|
|
print(
|
|
f" Warning: Coverage data directory not found: {self.coverage_data_dir}"
|
|
)
|
|
return test_cases
|
|
for item in self.coverage_data_dir.iterdir():
|
|
if not item.is_dir():
|
|
continue
|
|
name = item.name
|
|
# sglang naming: ____w__sglang__sglang__test__... (GitHub Actions encoded path)
|
|
is_sglang_layout = TEST_CASE_DIR_PREFIX and name.startswith(
|
|
TEST_CASE_DIR_PREFIX
|
|
)
|
|
if not is_sglang_layout:
|
|
continue
|
|
# coverage.* files directly under test case dir (sglang)
|
|
has_cov_files = any(item.glob("coverage.*"))
|
|
if has_cov_files:
|
|
test_cases.append(name)
|
|
return sorted(test_cases)
|
|
|
|
@staticmethod
|
|
def normalize_test_name(test_name: str) -> str:
|
|
"""
|
|
Convert test case directory name to standard script name format.
|
|
sglang (GitHub Actions encoded dir name: /__w/sglang/sglang/test/... -> ____w__sglang__sglang__test__...):
|
|
- ____w__sglang__sglang__test__registered__npu__xxx__test_foo.py
|
|
-> test/registered/npu/xxx/test_foo.py (file-level)
|
|
- ...--test_foo -> test/registered/npu/xxx/test_foo.py::test_foo (function-level)
|
|
"""
|
|
# sglang layout: strip ____w__sglang__sglang__test__ prefix (encoded /__w/sglang/sglang/test/)
|
|
if not TEST_CASE_DIR_PREFIX or not test_name.startswith(TEST_CASE_DIR_PREFIX):
|
|
return test_name
|
|
# rest: encoded path after /test/ (e.g. registered__npu__xxx__test_foo.py)
|
|
rest = test_name[len(TEST_CASE_DIR_PREFIX) :]
|
|
# Restore test/ prefix (TEST_CASE_DIR_PREFIX ends with test__, __ encodes /)
|
|
result = "test/" + rest.replace("__", "/")
|
|
# Handle function-level marker: .../test_foo.py--test_bar -> .../test_foo.py::test_bar
|
|
result = result.replace("--", "::")
|
|
# File-level tests need .py suffix; avoid double .py when name already ends with .py
|
|
if "::" not in result and not result.endswith(".py"):
|
|
result = result + ".py"
|
|
return result
|
|
|
|
def get_covered_lines_from_file(self, cov_file: str, filename: str) -> set[int]:
|
|
"""
|
|
Get covered line numbers for a file from a single coverage SQLite file
|
|
"""
|
|
lines = set()
|
|
try:
|
|
conn = sqlite3.connect(cov_file)
|
|
cursor = conn.cursor()
|
|
|
|
# Find file ID (fuzzy path matching)
|
|
cursor.execute("SELECT id FROM file WHERE path LIKE ?", (f"%{filename}",))
|
|
row = cursor.fetchone()
|
|
if not row:
|
|
conn.close()
|
|
return lines
|
|
file_id = row[0]
|
|
|
|
# Get all arcs, calculate covered line numbers
|
|
cursor.execute(
|
|
"SELECT DISTINCT fromno, tono FROM arc WHERE file_id = ?", (file_id,)
|
|
)
|
|
for fromno, tono in cursor.fetchall():
|
|
if fromno > 0:
|
|
lines.add(fromno)
|
|
if tono > 0:
|
|
lines.add(tono)
|
|
|
|
conn.close()
|
|
except Exception as e:
|
|
print(f" Warning: Error reading {cov_file}: {e}")
|
|
return lines
|
|
|
|
def get_covered_files_from_file(self, cov_file: str) -> set[str]:
|
|
"""Get all covered files from a single coverage file"""
|
|
files = set()
|
|
try:
|
|
conn = sqlite3.connect(cov_file)
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT path FROM file")
|
|
for (path,) in cursor.fetchall():
|
|
# Product code paths contain PRODUCT_PREFIX
|
|
# (e.g. /__w/sglang/sglang/python/sglang/srt/xxx.py -> srt/xxx.py)
|
|
if PRODUCT_PREFIX in path:
|
|
rel_path = path.split(PRODUCT_PREFIX)[-1]
|
|
files.add(rel_path)
|
|
conn.close()
|
|
except Exception as e:
|
|
print(f" Warning: Error reading {cov_file}: {e}")
|
|
return files
|
|
|
|
def _get_function_def_lines(self, filepath: str) -> set[int]:
|
|
"""
|
|
Get function definition line numbers (def line only, not function body).
|
|
|
|
Args:
|
|
filepath: Source file path
|
|
|
|
Returns:
|
|
Set of line numbers where function definitions occur
|
|
"""
|
|
def_lines = set()
|
|
try:
|
|
with open(filepath, encoding="utf-8") as f:
|
|
source = f.read()
|
|
lines = source.splitlines()
|
|
|
|
tree = ast.parse(source, filename=filepath)
|
|
|
|
TARGET_DECORATORS = {"staticmethod", "classmethod", "property"}
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
# Add decorator lines (only @staticmethod, @classmethod, @property)
|
|
for decorator in node.decorator_list:
|
|
if (
|
|
isinstance(decorator, ast.Name)
|
|
and decorator.id in TARGET_DECORATORS
|
|
):
|
|
if hasattr(decorator, "lineno") and decorator.lineno:
|
|
def_lines.add(decorator.lineno)
|
|
# Handle multi-line decorator expressions
|
|
if (
|
|
hasattr(decorator, "end_lineno")
|
|
and decorator.end_lineno
|
|
):
|
|
for i in range(
|
|
decorator.lineno, decorator.end_lineno + 1
|
|
):
|
|
def_lines.add(i)
|
|
|
|
def_lines.add(node.lineno)
|
|
|
|
# Bracket counting to find header end
|
|
start_idx = node.lineno - 1
|
|
paren_count = lines[start_idx].count("(") - lines[start_idx].count(
|
|
")"
|
|
)
|
|
|
|
line_idx = start_idx
|
|
while paren_count > 0 and line_idx < len(lines):
|
|
line_idx += 1
|
|
paren_count += lines[line_idx].count("(") - lines[
|
|
line_idx
|
|
].count(")")
|
|
|
|
header_end = line_idx + 1 # Convert to 1-indexed
|
|
|
|
# Extend to return type annotation if present
|
|
if node.returns:
|
|
header_end = max(header_end, node.returns.end_lineno)
|
|
|
|
# Record all lines from def to header end
|
|
for i in range(node.lineno, header_end + 1):
|
|
def_lines.add(i)
|
|
except Exception:
|
|
pass
|
|
return def_lines
|
|
|
|
def _get_class_def_lines(self, filepath: str) -> set[int]:
|
|
"""
|
|
Get line numbers of all class definition lines.
|
|
|
|
Args:
|
|
filepath: Source file path
|
|
|
|
Returns:
|
|
Set of line numbers where class definitions occur
|
|
"""
|
|
class_lines = set()
|
|
try:
|
|
with open(filepath, encoding="utf-8") as f:
|
|
tree = ast.parse(f.read(), filename=filepath)
|
|
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.ClassDef):
|
|
class_lines.add(node.lineno)
|
|
except Exception:
|
|
pass
|
|
return class_lines
|
|
|
|
def _get_docstring_lines(self, filepath: str) -> set[int]:
|
|
"""
|
|
Get line numbers of all docstring lines (module, class, and function).
|
|
|
|
Docstrings are string literals that appear as the first statement
|
|
in a module, class, or function body.
|
|
|
|
Args:
|
|
filepath: Source file path
|
|
|
|
Returns:
|
|
Set of line numbers where docstrings occur
|
|
"""
|
|
docstring_lines = set()
|
|
try:
|
|
with open(filepath, encoding="utf-8") as f:
|
|
tree = ast.parse(f.read(), filename=filepath)
|
|
|
|
# Module-level docstring
|
|
if (
|
|
tree.body
|
|
and isinstance(tree.body[0], ast.Expr)
|
|
and isinstance(tree.body[0].value, ast.Constant)
|
|
):
|
|
docstring_lines.add(tree.body[0].lineno)
|
|
|
|
# Class and function docstrings
|
|
for node in ast.walk(tree):
|
|
if isinstance(
|
|
node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)
|
|
):
|
|
if (
|
|
node.body
|
|
and isinstance(node.body[0], ast.Expr)
|
|
and isinstance(node.body[0].value, ast.Constant)
|
|
):
|
|
docstring_lines.add(node.body[0].lineno)
|
|
except Exception:
|
|
pass
|
|
return docstring_lines
|
|
|
|
@staticmethod
|
|
def _get_blank_lines(filepath: str) -> set[int]:
|
|
"""
|
|
Get line numbers of all blank/whitespace-only lines in file.
|
|
|
|
Coverage arc data can record blank lines as control-flow nodes
|
|
(e.g., block boundaries after if/return statements). These lines
|
|
are not executable and must be filtered out to avoid false matches.
|
|
|
|
Args:
|
|
filepath: Source file path
|
|
|
|
Returns:
|
|
Set of line numbers that are blank or whitespace-only
|
|
"""
|
|
blank_lines = set()
|
|
try:
|
|
with open(filepath, encoding="utf-8") as f:
|
|
for line_no, line in enumerate(f, start=1):
|
|
if not line.strip():
|
|
blank_lines.add(line_no)
|
|
except Exception:
|
|
pass
|
|
return blank_lines
|
|
|
|
def _filter_noise_lines(self, filepath: str, lines: set[int]) -> set[int]:
|
|
"""
|
|
Filter out invalid noise lines from coverage data:
|
|
1. import/from...import statement lines
|
|
2. Function definition lines (def line only)
|
|
3. Class definition lines
|
|
4. Docstring lines
|
|
5. Blank/whitespace-only lines
|
|
|
|
Args:
|
|
filepath: Source file path
|
|
lines: Original set of covered line numbers
|
|
|
|
Returns:
|
|
Filtered set with noise lines removed
|
|
"""
|
|
if not lines:
|
|
return lines
|
|
|
|
# Use cache to avoid re-parsing the same file multiple times
|
|
if filepath not in self._noise_lines_cache:
|
|
import_lines = FunctionParser._get_import_lines(filepath)
|
|
def_lines = self._get_function_def_lines(filepath)
|
|
class_lines = self._get_class_def_lines(filepath)
|
|
docstring_lines = self._get_docstring_lines(filepath)
|
|
blank_lines = self._get_blank_lines(filepath)
|
|
self._noise_lines_cache[filepath] = (
|
|
import_lines | def_lines | class_lines | docstring_lines | blank_lines
|
|
)
|
|
|
|
return lines - self._noise_lines_cache[filepath]
|
|
|
|
def _resolve_source_file(self, filename: str) -> Path | None:
|
|
"""
|
|
Resolve source file path from relative filename.
|
|
|
|
Args:
|
|
filename: Relative file path (e.g., 'covstub/sglang/srt/models/qwen3_vl.py')
|
|
|
|
Returns:
|
|
Path object if found, None otherwise
|
|
"""
|
|
if not self.source_dir:
|
|
return None
|
|
|
|
source_path = self.source_dir / REPO_NAME / filename
|
|
return source_path if source_path.exists() else None
|
|
|
|
def build_test_case_map(self) -> dict:
|
|
"""Build test case -> covered files mapping (with line numbers)"""
|
|
print("Scanning test cases...")
|
|
test_cases = self.scan_test_cases()
|
|
print(f" Found {len(test_cases)} test cases")
|
|
|
|
for i, test_case in enumerate(test_cases):
|
|
print(f" [{i + 1}/{len(test_cases)}] Processing {test_case}...")
|
|
test_case_dir = self.coverage_data_dir / test_case
|
|
covdata_dir = test_case_dir / "covdata"
|
|
|
|
file_lines_map = defaultdict(set) # filepath -> set of lines
|
|
|
|
# Coverage data files (coverage.*) are stored directly under the test case dir
|
|
cov_dirs = [covdata_dir] if covdata_dir.exists() else [test_case_dir]
|
|
|
|
for cov_dir in cov_dirs:
|
|
for cov_file in cov_dir.glob("coverage.*"):
|
|
covered_files = self.get_covered_files_from_file(str(cov_file))
|
|
|
|
for filename in covered_files:
|
|
lines = self.get_covered_lines_from_file(
|
|
str(cov_file), filename
|
|
)
|
|
if lines:
|
|
# Filter noise lines if source_dir is available
|
|
if self.source_dir:
|
|
source_file = self._resolve_source_file(filename)
|
|
if source_file and source_file.exists():
|
|
lines = self._filter_noise_lines(
|
|
str(source_file), lines
|
|
)
|
|
# Skip files with no coverage after filtering
|
|
if lines:
|
|
file_lines_map[filename].update(lines)
|
|
|
|
normalized_name = self.normalize_test_name(test_case)
|
|
self.test_case_map[normalized_name] = {
|
|
"files": dict(file_lines_map),
|
|
"file_count": len(file_lines_map),
|
|
"line_count": sum(len(v) for v in file_lines_map.values()),
|
|
}
|
|
|
|
print(
|
|
f" -> {len(file_lines_map)} files, {sum(len(v) for v in file_lines_map.values())} lines"
|
|
)
|
|
|
|
return self.test_case_map
|
|
|
|
def save_map(self, output_path: str = "test_case_map.json"):
|
|
"""Save test case mapping to file"""
|
|
serializable_map = {}
|
|
for test_case, data in self.test_case_map.items():
|
|
serializable_map[test_case] = {
|
|
"files": {k: list(v) for k, v in data["files"].items()},
|
|
"file_count": data["file_count"],
|
|
"line_count": data["line_count"],
|
|
}
|
|
|
|
with open(output_path, "w", encoding="utf-8", newline="\n") as f:
|
|
json.dump(serializable_map, f, indent=2, ensure_ascii=False)
|
|
print(f"\nTest case mapping saved to: {output_path}")
|
|
|
|
def load_map(self, input_path: str = "test_case_map.json"):
|
|
"""Load test case mapping from file"""
|
|
with open(input_path, encoding="utf-8") as f:
|
|
serializable_map = json.load(f)
|
|
|
|
self.test_case_map = {}
|
|
for test_case, data in serializable_map.items():
|
|
self.test_case_map[test_case] = {
|
|
"files": {k: set(v) for k, v in data["files"].items()},
|
|
"file_count": data["file_count"],
|
|
"line_count": data["line_count"],
|
|
}
|
|
print(f"Loaded {len(self.test_case_map)} test case mappings from {input_path}")
|
|
return self.test_case_map
|
|
|
|
|
|
class CodeChangeDetector:
|
|
"""Code change detector"""
|
|
|
|
def __init__(self, source_dir: str):
|
|
self.source_dir = Path(source_dir)
|
|
self.file_hashes = {}
|
|
|
|
def _product_code_root(self) -> Path:
|
|
"""
|
|
Hash scanning uses this root so relative paths (e.g. srt/xxx.py) match
|
|
the keys in test_case_map.json (which are relative to python/sglang/).
|
|
"""
|
|
return self.source_dir / REPO_NAME
|
|
|
|
def compute_file_hash(self, filepath: str) -> str:
|
|
"""Calculate MD5 hash of file"""
|
|
hasher = hashlib.md5()
|
|
try:
|
|
with open(filepath, "rb") as f:
|
|
hasher.update(f.read())
|
|
return hasher.hexdigest()
|
|
except Exception as e:
|
|
print(f" Warning: Error computing file hash: {filepath}: {e}")
|
|
return ""
|
|
|
|
def scan_source_files(self) -> dict[str, str]:
|
|
"""Scan product code files, compute hashes"""
|
|
self.file_hashes = {}
|
|
root = self._product_code_root()
|
|
if not root.exists():
|
|
print(f" Warning: Product code root not found: {root}")
|
|
return self.file_hashes
|
|
for py_file in root.rglob("*.py"):
|
|
rel_path = py_file.relative_to(root).as_posix()
|
|
self.file_hashes[rel_path] = self.compute_file_hash(str(py_file))
|
|
return self.file_hashes
|
|
|
|
def detect_changes_by_comparison(self) -> dict[str, set[int]]:
|
|
"""Detect changes by file hash comparison (return all lines for changed files)"""
|
|
changed_files = {}
|
|
current_hashes = {}
|
|
|
|
root = self._product_code_root()
|
|
if not root.exists():
|
|
print(f" Warning: Product code root not found: {root}")
|
|
return changed_files
|
|
|
|
for py_file in root.rglob("*.py"):
|
|
rel_path = py_file.relative_to(root).as_posix()
|
|
current_hashes[rel_path] = self.compute_file_hash(str(py_file))
|
|
|
|
baseline_path = self.source_dir / ".file_hashes.json"
|
|
if baseline_path.exists():
|
|
with open(baseline_path) as f:
|
|
old_hashes = json.load(f)
|
|
|
|
for rel_path, current_hash in current_hashes.items():
|
|
old_hash = old_hashes.get(rel_path, "")
|
|
if current_hash != old_hash:
|
|
# File has changes, return all line numbers (conservative estimate)
|
|
changed_files[rel_path] = set(
|
|
range(1, 10000)
|
|
) # Conservative: assume all lines may have changed
|
|
else:
|
|
changed_files = {
|
|
rel_path: set(range(1, 10000)) for rel_path in current_hashes
|
|
}
|
|
with open(baseline_path, "w") as f:
|
|
json.dump(current_hashes, f)
|
|
|
|
return changed_files
|
|
|
|
def parse_git_diff(
|
|
self,
|
|
diff_output: str,
|
|
base_content_getter=None,
|
|
) -> dict[str, set[int]]:
|
|
"""
|
|
Parse git diff output, extract affected base (pre-change) line numbers.
|
|
|
|
Rules:
|
|
- Deleted lines: record the deleted base line itself, nothing more.
|
|
- Pure comment/docstring changes are excluded (needs base content):
|
|
a deletion group where every deleted line is a comment/docstring line
|
|
and the additions are comments or doc prose; an insertion inside a
|
|
docstring or consisting of comment lines only.
|
|
- Isolated blank-line deletion (neighbours not deleted): treated as a
|
|
one-line insertion -> candidate pair (line above, line below).
|
|
- Pure insertions and blank-deletion pairs are classified via ast of
|
|
the base file (needs base_content_getter):
|
|
1. modifies an existing function -> record the line above only;
|
|
2. sits between two function/class definitions -> excluded;
|
|
3. inserted text belongs to a newly added def/class -> excluded;
|
|
4. otherwise (module-level statements) -> record the line above only.
|
|
- Without base content (or non-parseable Python) pairs fall back to
|
|
counting both sides, bounded by the hunk's base range.
|
|
|
|
Args:
|
|
diff_output: diff content
|
|
base_content_getter: optional callable(repo-relative-path -> str | None)
|
|
returning the base file content for ast classification
|
|
|
|
Returns:
|
|
{filepath: {lineno, ...}} - set of affected base line numbers,
|
|
.py files under '{PRODUCT_PREFIX}' only, with the prefix stripped.
|
|
Renamed and deleted files are excluded: they are matched at file
|
|
level via detect_renames() (see parse_pr_diff_file/main).
|
|
"""
|
|
filter_prefix = PRODUCT_PREFIX
|
|
renamed_files, deleted_files = self.detect_renames(diff_output)
|
|
renamed_new_paths = set(renamed_files.values())
|
|
deleted_paths = set(deleted_files)
|
|
|
|
files, pending, del_groups = _parse_diff_base_lines(diff_output)
|
|
|
|
changed_files = {}
|
|
for path, lines in files.items():
|
|
# Renamed/deleted files go through file-level matching, skip line-level parsing
|
|
if path in renamed_new_paths or path in deleted_paths:
|
|
continue
|
|
# Filter: only keep product code (exclude test files, etc.)
|
|
if not path.startswith(filter_prefix):
|
|
continue
|
|
if not path.endswith(".py"):
|
|
continue
|
|
# Normalize path: remove the '{PRODUCT_PREFIX}' prefix
|
|
key = path[len(filter_prefix) :]
|
|
changed_files[key] = lines
|
|
pairs = pending.get(path) or []
|
|
groups = del_groups.get(path) or []
|
|
if pairs or groups:
|
|
base_text = base_content_getter(path) if base_content_getter else None
|
|
_classify_candidate_pairs(lines, pairs, groups, base_text, path)
|
|
|
|
# Drop files that end up with no affected code lines (e.g. pure comment changes)
|
|
return {k: v for k, v in changed_files.items() if v}
|
|
|
|
def detect_renames(self, diff_output: str) -> tuple[dict[str, str], list[str]]:
|
|
"""
|
|
Detect renamed and deleted files in git diff output (product code only,
|
|
under PRODUCT_PREFIX). Both are handled the same way: file-level matching
|
|
with the base path, excluded from line-level parsing.
|
|
|
|
Args:
|
|
diff_output: diff content
|
|
|
|
Returns:
|
|
Tuple of (rename_mapping, deleted_files)
|
|
- rename_mapping: {old_path: new_path}
|
|
- deleted_files: [path, ...] (base paths)
|
|
"""
|
|
renames = {}
|
|
deleted = []
|
|
current_old_path = None
|
|
current_new_path = None
|
|
header_old_path = None
|
|
|
|
for raw_line in diff_output.split("\n"):
|
|
line = raw_line.rstrip("\r")
|
|
|
|
# Detect rename marker
|
|
if line.startswith("rename from "):
|
|
current_old_path = line[12:].strip()
|
|
continue
|
|
if line.startswith("rename to "):
|
|
current_new_path = line[10:].strip()
|
|
# When we have both old and new path, record the rename
|
|
if current_old_path and current_new_path:
|
|
# Remove a/ or b/ prefix if present
|
|
old_path = (
|
|
current_old_path[2:]
|
|
if current_old_path.startswith("a/")
|
|
else current_old_path
|
|
)
|
|
new_path = (
|
|
current_new_path[2:]
|
|
if current_new_path.startswith("b/")
|
|
else current_new_path
|
|
)
|
|
# Only record product code renames (under PRODUCT_PREFIX)
|
|
if old_path.startswith(PRODUCT_PREFIX):
|
|
renames[old_path] = new_path
|
|
current_old_path = None
|
|
current_new_path = None
|
|
continue
|
|
|
|
# Detect deleted file via '--- a/path' + '+++ /dev/null'
|
|
if line.startswith("--- "):
|
|
header_old_path = line[4:].strip()
|
|
if header_old_path.startswith("a/"):
|
|
header_old_path = header_old_path[2:]
|
|
elif line.startswith("+++ "):
|
|
if (
|
|
line[4:].strip() == "/dev/null"
|
|
and header_old_path
|
|
and header_old_path.startswith(PRODUCT_PREFIX)
|
|
):
|
|
deleted.append(header_old_path)
|
|
header_old_path = None
|
|
|
|
return renames, deleted
|
|
|
|
def parse_pr_diff_file(
|
|
self, diff_file_path: str, base_content_getter=None
|
|
) -> tuple[dict[str, set[int]], dict[str, str], list[str]]:
|
|
"""
|
|
Parse changed line numbers, renames and deleted files from PR diff file.
|
|
|
|
Args:
|
|
diff_file_path: diff file path
|
|
base_content_getter: optional callable(repo-relative-path -> str | None)
|
|
returning the base file content for ast classification
|
|
|
|
Returns:
|
|
Tuple of (changed_files_with_lines, rename_mapping, deleted_files)
|
|
- changed_files_with_lines: {filepath: {lineno, ...}}
|
|
- rename_mapping: {old_path: new_path}
|
|
- deleted_files: [path, ...]
|
|
"""
|
|
try:
|
|
with open(diff_file_path, encoding="utf-8-sig") as f:
|
|
diff_content = f.read()
|
|
changed_files = self.parse_git_diff(
|
|
diff_content, base_content_getter=base_content_getter
|
|
)
|
|
renames, deleted_files = self.detect_renames(diff_content)
|
|
return changed_files, renames, deleted_files
|
|
except Exception as e:
|
|
print(f"Warning: Failed to read diff file: {e}")
|
|
return {}, {}, []
|
|
|
|
|
|
_HUNK_RE = re.compile(r"@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@")
|
|
_DEF_RE = re.compile(r"(async\s+def|def|class)\s")
|
|
|
|
|
|
def _parse_diff_base_lines(
|
|
diff_output: str,
|
|
) -> tuple[dict[str, set[int]], dict[str, list[tuple]], dict[str, list[tuple]]]:
|
|
"""Parse unified diff text into affected base (pre-change) line numbers.
|
|
|
|
Returns (files, pending, del_groups):
|
|
files[path] : set of base line numbers recorded directly
|
|
(blank lines inside contiguous deletion blocks)
|
|
pending[path] : candidate pairs needing base-content classification;
|
|
tuple = (a, b, kind, add_indent, introduces_def,
|
|
adds_all_comment, hunk_base_end)
|
|
kind='insert' -> pure insertion between a and b
|
|
kind='blank' -> isolated blank deletion at a+1 (b = a+2)
|
|
del_groups[path]: deletion groups needing comment/docstring filtering;
|
|
tuple = ([(base_line, deleted_text), ...], [added_text, ...])
|
|
"""
|
|
files, pending, del_groups = {}, {}, {}
|
|
current = None
|
|
base_no = None
|
|
hunk_base_end = 0
|
|
old_path = None # path from the last '--- a/...' line (used for deleted files)
|
|
group_del = [] # (base_line, text) of '-' lines in the current change group
|
|
group_add = [] # texts of '+' lines in the current change group
|
|
|
|
def flush_group():
|
|
if not group_del and not group_add:
|
|
return
|
|
if group_del:
|
|
del_set = {n for n, _ in group_del}
|
|
del_lines = []
|
|
for n, text in group_del:
|
|
if text.strip():
|
|
del_lines.append((n, text))
|
|
elif (n - 1) in del_set or (n + 1) in del_set:
|
|
# blank inside a contiguous deletion block: classify with
|
|
# the group (dropped too if the block is pure comment/docstring)
|
|
del_lines.append((n, text))
|
|
else:
|
|
pending[current].append(
|
|
(n - 1, n + 1, "blank", None, False, False, hunk_base_end)
|
|
)
|
|
if del_lines:
|
|
del_groups[current].append((del_lines, list(group_add)))
|
|
else:
|
|
# base_no is the next unprocessed base line = the line below the insertion
|
|
if base_no is None or base_no < 1:
|
|
# New file (hunk '@@ -0,0 ...'): there is no base version at all,
|
|
# so there is nothing to classify the insertion against. Skip the
|
|
# pair so callers never attempt to fetch a base file.
|
|
return
|
|
a = base_no - 1
|
|
indent = min(
|
|
((len(t) - len(t.lstrip())) for t in group_add if t.strip()), default=0
|
|
)
|
|
introduces_def = any(
|
|
t.strip().startswith("@") or _DEF_RE.match(t.strip())
|
|
for t in group_add
|
|
if t.strip()
|
|
)
|
|
adds_all_comment = all(
|
|
t.strip().startswith("#") for t in group_add if t.strip()
|
|
)
|
|
pending[current].append(
|
|
(
|
|
a,
|
|
a + 1,
|
|
"insert",
|
|
indent,
|
|
introduces_def,
|
|
adds_all_comment,
|
|
hunk_base_end,
|
|
)
|
|
)
|
|
|
|
for raw_line in diff_output.split("\n"):
|
|
line = raw_line.rstrip("\r")
|
|
if line.startswith("diff --git"):
|
|
flush_group()
|
|
group_del, group_add = [], []
|
|
current, base_no = None, None
|
|
continue
|
|
if line.startswith("--- "):
|
|
old_path = line[4:]
|
|
if old_path.startswith("a/"):
|
|
old_path = old_path[2:]
|
|
continue
|
|
if line.startswith("+++ "):
|
|
flush_group()
|
|
group_del, group_add = [], []
|
|
path = line[4:]
|
|
if path == "/dev/null":
|
|
# deleted file: keep the '--- a/...' path so deletions are recorded
|
|
path = old_path
|
|
old_path = None
|
|
if path is None or path == "/dev/null":
|
|
current = None
|
|
continue
|
|
if path.startswith("b/"):
|
|
path = path[2:]
|
|
current = path
|
|
files.setdefault(path, set())
|
|
pending.setdefault(path, [])
|
|
del_groups.setdefault(path, [])
|
|
continue
|
|
if line.startswith("@@"):
|
|
flush_group()
|
|
group_del, group_add = [], []
|
|
if current is None:
|
|
continue
|
|
m = _HUNK_RE.search(line)
|
|
base_no = int(m.group(1))
|
|
hunk_base_end = base_no + int(m.group(2) or "1") - 1
|
|
continue
|
|
if current is None or base_no is None:
|
|
continue
|
|
if line.startswith("-"):
|
|
group_del.append((base_no, line[1:]))
|
|
base_no += 1
|
|
elif line.startswith("+"):
|
|
group_add.append(line[1:])
|
|
elif line.startswith("\\"):
|
|
continue
|
|
else:
|
|
flush_group()
|
|
group_del, group_add = [], []
|
|
base_no += 1
|
|
flush_group()
|
|
return files, pending, del_groups
|
|
|
|
|
|
def _collect_defs(source: str) -> tuple[list, set, set, set]:
|
|
"""Parse Python source, return (ranges, end_lines, start_lines, blanks).
|
|
|
|
ranges : [(lineno, end_lineno, col_offset)] of every function/method
|
|
end_lines : line numbers where a function/class definition ends
|
|
start_lines : def/class lines and their decorator lines
|
|
blanks : blank line numbers
|
|
"""
|
|
tree = ast.parse(source)
|
|
ranges, end_lines, start_lines = [], set(), set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
|
|
end_lines.add(node.end_lineno)
|
|
start_lines.add(node.lineno)
|
|
for deco in node.decorator_list:
|
|
start_lines.add(deco.lineno)
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
ranges.append((node.lineno, node.end_lineno, node.col_offset))
|
|
blanks = {i for i, text in enumerate(source.splitlines(), 1) if not text.strip()}
|
|
return ranges, end_lines, start_lines, blanks
|
|
|
|
|
|
def _get_docstring_lines(source: str) -> set[int]:
|
|
"""Line numbers covered by docstrings (module/class/function docstring nodes)."""
|
|
tree = ast.parse(source)
|
|
lines = set()
|
|
for node in ast.walk(tree):
|
|
body = getattr(node, "body", None)
|
|
if (
|
|
isinstance(body, list)
|
|
and body
|
|
and isinstance(body[0], ast.Expr)
|
|
and isinstance(body[0].value, ast.Constant)
|
|
and isinstance(body[0].value.value, str)
|
|
):
|
|
lines.update(range(body[0].lineno, body[0].end_lineno + 1))
|
|
return lines
|
|
|
|
|
|
def _looks_like_code(texts: list) -> bool:
|
|
"""True if the added lines are real code: parseable as Python and not
|
|
solely string-literal expressions (docstring prose)."""
|
|
block = textwrap.dedent("\n".join(t for t in texts if t.strip()))
|
|
if not block.strip():
|
|
return False
|
|
try:
|
|
tree = ast.parse(block)
|
|
except (SyntaxError, ValueError):
|
|
return False
|
|
return any(
|
|
not (
|
|
isinstance(stmt, ast.Expr)
|
|
and isinstance(stmt.value, ast.Constant)
|
|
and isinstance(stmt.value.value, str)
|
|
)
|
|
for stmt in tree.body
|
|
)
|
|
|
|
|
|
def _innermost_func(ranges: list, n: int):
|
|
"""The innermost function whose body contains line n (None if module level)."""
|
|
best = None
|
|
for start, end, col in ranges:
|
|
if start <= n <= end and (best is None or start >= best[0]):
|
|
best = (start, end, col)
|
|
return best
|
|
|
|
|
|
def _between_definitions(
|
|
a: int, b: int, end_lines: set, start_lines: set, blanks: set
|
|
) -> bool:
|
|
"""True if the pair (a, b) sits between two definitions: the upper line is
|
|
the end of a function/class (if a itself is blank, walk up past consecutive
|
|
blank lines and check the nearest non-blank line instead) and the lower
|
|
line is the start of a function/class (def/class line or decorator)."""
|
|
if b not in start_lines:
|
|
return False
|
|
upper = a
|
|
while upper in blanks:
|
|
upper -= 1
|
|
return upper in end_lines
|
|
|
|
|
|
def _classify_candidate_pairs(
|
|
affected: set[int],
|
|
pairs: list[tuple],
|
|
del_groups: list[tuple],
|
|
base_text: str | None,
|
|
path: str,
|
|
) -> None:
|
|
"""Classify deletion groups and candidate pairs of one file using its base
|
|
content and update the affected line set in place.
|
|
|
|
Deletion groups: a group is dropped entirely when every deleted line is a
|
|
comment/docstring line in the base file AND the added lines are comments or
|
|
doc prose (not parseable Python), i.e. a pure comment/docstring change.
|
|
Candidate pairs: without base content (or non-parseable Python) both sides
|
|
of each pair are counted, bounded by the hunk."""
|
|
info = None
|
|
docstr_lines = set()
|
|
comment_lines = set()
|
|
if base_text is not None:
|
|
try:
|
|
info = _collect_defs(base_text)
|
|
docstr_lines = _get_docstring_lines(base_text)
|
|
comment_lines = {
|
|
i
|
|
for i, t in enumerate(base_text.splitlines(), 1)
|
|
if t.strip().startswith("#")
|
|
}
|
|
except (SyntaxError, ValueError):
|
|
info = None
|
|
if base_text is None:
|
|
print(
|
|
f" Warning: no base content for {path}, counting candidate pairs on both sides"
|
|
)
|
|
|
|
# Deleted non-blank lines: comment/docstring lines are never changes by
|
|
# themselves; a group made entirely of them is dropped unless its lines are
|
|
# replaced by real code (then they are kept as the only base anchors).
|
|
noise_lines = comment_lines | docstr_lines
|
|
for del_lines, add_texts in del_groups:
|
|
if info is None:
|
|
affected.update(n for n, _ in del_lines)
|
|
continue
|
|
code_dels = [n for n, _ in del_lines if n not in noise_lines]
|
|
if code_dels:
|
|
affected.update(code_dels)
|
|
dropped = [n for n, _ in del_lines if n in noise_lines]
|
|
if dropped:
|
|
print(
|
|
f" Skipped {path}:{dropped} (comment/docstring lines, not counted)"
|
|
)
|
|
continue
|
|
adds = [t for t in add_texts if t.strip()]
|
|
pure = (
|
|
not adds
|
|
or all(t.strip().startswith("#") for t in adds)
|
|
or not _looks_like_code(adds)
|
|
)
|
|
if pure:
|
|
print(
|
|
f" Skipped {path}:{[n for n, _ in del_lines]} (pure comment/docstring change, not counted)"
|
|
)
|
|
else:
|
|
# comment/docstring lines replaced by real code: keep as change anchors
|
|
affected.update(n for n, _ in del_lines)
|
|
|
|
for (
|
|
a,
|
|
b,
|
|
kind,
|
|
add_indent,
|
|
introduces_def,
|
|
adds_all_comment,
|
|
hunk_base_end,
|
|
) in pairs:
|
|
if info is None:
|
|
if a >= 1:
|
|
affected.add(a)
|
|
if b <= hunk_base_end:
|
|
affected.add(b)
|
|
continue
|
|
ranges, end_lines, start_lines, blanks = info
|
|
reason = None
|
|
if kind == "insert":
|
|
if a in docstr_lines:
|
|
reason = "inside a docstring"
|
|
elif adds_all_comment:
|
|
reason = "pure comment insertion"
|
|
else:
|
|
func = _innermost_func(ranges, a)
|
|
modifies = func is not None and (
|
|
b <= func[1] or (add_indent is not None and add_indent > func[2])
|
|
)
|
|
if not modifies:
|
|
if _between_definitions(a, b, end_lines, start_lines, blanks):
|
|
reason = "between function/class definitions"
|
|
elif introduces_def:
|
|
reason = "belongs to a newly added function/class"
|
|
# kind == 'blank': isolated blank deletion == one-line insertion
|
|
elif _between_definitions(a, b, end_lines, start_lines, blanks):
|
|
reason = "between function/class definitions"
|
|
if reason is None: # kept: record the line above only
|
|
if a >= 1:
|
|
affected.add(a)
|
|
else:
|
|
print(f" Skipped {path}:{a}-{b} ({reason}, not counted)")
|
|
|
|
|
|
class FunctionParser:
|
|
"""Python function parser - used to get line number ranges of functions and branches"""
|
|
|
|
@staticmethod
|
|
def get_function_ranges(filepath: str) -> dict[str, list[tuple[int, int]]]:
|
|
"""
|
|
Parse Python file, return function name -> [(start_line, end_line), ...] mapping
|
|
Supports multiple occurrences of the same function name (returns all matching ranges)
|
|
"""
|
|
function_ranges = defaultdict(list)
|
|
try:
|
|
with open(filepath, encoding="utf-8") as f:
|
|
tree = ast.parse(f.read(), filename=filepath)
|
|
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
function_ranges[node.name].append(
|
|
(node.lineno, node.end_lineno or node.lineno)
|
|
)
|
|
except Exception as e:
|
|
print(f" Warning: Failed to parse function definition {filepath}: {e}")
|
|
|
|
return function_ranges
|
|
|
|
@staticmethod
|
|
def _get_import_lines(filepath: str) -> set[int]:
|
|
"""
|
|
Get line numbers of all import statements in file
|
|
"""
|
|
import_lines = set()
|
|
try:
|
|
with open(filepath, encoding="utf-8") as f:
|
|
tree = ast.parse(f.read(), filename=filepath)
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, (ast.Import, ast.ImportFrom)):
|
|
import_lines.add(node.lineno)
|
|
if hasattr(node, "end_lineno") and node.end_lineno:
|
|
import_lines.update(range(node.lineno, node.end_lineno + 1))
|
|
except Exception:
|
|
pass
|
|
return import_lines
|
|
|
|
@staticmethod
|
|
def get_lines_functions(
|
|
filepath: str,
|
|
lines: set[int],
|
|
skip_imports: bool = False,
|
|
function_ranges: dict[str, list[tuple[int, int]]] | None = None,
|
|
) -> dict[int, str]:
|
|
"""
|
|
Get function name for each line
|
|
|
|
Args:
|
|
filepath: source file path
|
|
lines: set of line numbers to query
|
|
skip_imports: whether to skip import statement lines
|
|
function_ranges: pre-parsed function ranges to reuse, avoiding
|
|
re-parsing the file. When None, the file is parsed internally.
|
|
"""
|
|
line_to_function = {}
|
|
if not lines:
|
|
return line_to_function
|
|
|
|
if function_ranges is None:
|
|
function_ranges = FunctionParser.get_function_ranges(filepath)
|
|
if not function_ranges:
|
|
return line_to_function
|
|
|
|
# Flatten all function ranges into a single interval list sorted by
|
|
# start line, then match each queried line with one linear scan.
|
|
# This avoids per-function set expansion and the O(lines x functions)
|
|
# nested loop. Semantics are preserved: for nested functions the outer
|
|
# one has the smaller start line and is found first, matching the
|
|
# original ast.walk (parent-before-child) order.
|
|
intervals = []
|
|
for func_name, ranges in function_ranges.items():
|
|
for start, end in ranges:
|
|
intervals.append((start, end, func_name))
|
|
intervals.sort(key=lambda x: x[0])
|
|
|
|
for line in lines:
|
|
for start, end, func_name in intervals:
|
|
if start > line:
|
|
break
|
|
if line <= end:
|
|
line_to_function[line] = func_name
|
|
break
|
|
|
|
return line_to_function
|
|
|
|
|
|
class TestSelector:
|
|
"""Test selector - select test cases to run based on code changes (line granularity)"""
|
|
|
|
def __init__(self, test_case_map: dict):
|
|
self.test_case_map = test_case_map
|
|
|
|
def select_tests(
|
|
self,
|
|
changed_files_with_lines: dict[str, set[int]],
|
|
min_affected_lines: int = 1,
|
|
source_dir: str | None = None,
|
|
enable_line_match: bool = True,
|
|
enable_function_match: bool = True,
|
|
enable_file_match: bool = True,
|
|
enable_skip_imports: bool = False,
|
|
enable_dedup: bool = False,
|
|
) -> tuple[list[tuple[str, dict[str, set[int]], int]], str]:
|
|
"""
|
|
Select affected test cases based on changed files, supports 3 independent matching granularities:
|
|
- Line-level matching: precise intersection of changed lines and covered lines
|
|
- Function-level matching: entire function body range matching
|
|
- File-level matching: any covered line in file matching
|
|
|
|
Each granularity cascades: only when current granularity finds no tests, try the next.
|
|
|
|
Args:
|
|
changed_files_with_lines: changed files and their line numbers {filepath: {lineno, ...}}
|
|
min_affected_lines: minimum affected lines, below this value will not be selected
|
|
source_dir: source code directory, used for function/file-level expansion
|
|
enable_line_match: whether to enable line-level matching
|
|
enable_function_match: whether to enable function-level matching
|
|
enable_file_match: whether to enable file-level matching
|
|
enable_skip_imports: whether to skip import statement lines (only effective for function-level matching)
|
|
enable_dedup: whether to enable deduplication
|
|
|
|
Returns:
|
|
(selected_tests, expand_reason)
|
|
- selected_tests: [(test_case_name, {filepath: {covered_lines}}, total_affected_lines), ...]
|
|
- expand_reason: expansion reason
|
|
('' means no expansion, 'line'/'function'/'file' indicates the granularity used)
|
|
"""
|
|
selected = []
|
|
expand_reason = ""
|
|
|
|
# Normalize changed file paths: remove PRODUCT_PREFIX or REPO_NAME/ prefix
|
|
normalized_changed = {}
|
|
for f, lines in changed_files_with_lines.items():
|
|
if PRODUCT_PREFIX and f.startswith(PRODUCT_PREFIX):
|
|
normalized_changed[f[len(PRODUCT_PREFIX) :]] = lines
|
|
elif f.startswith(f"{REPO_NAME}/"):
|
|
normalized_changed[f[len(f"{REPO_NAME}/") :]] = lines
|
|
else:
|
|
normalized_changed[f] = lines
|
|
|
|
total_changed_lines = sum(len(lines) for lines in normalized_changed.values())
|
|
|
|
# ===== Line-level matching + Function-level matching (parallel execution, merge deduplication) =====
|
|
line_results = [] # [(test_case, affected_detail, total_lines)]
|
|
func_results = [] # [(test_case, affected_detail, total_lines)]
|
|
|
|
# ----- Stage 1: Line-level matching -----
|
|
if enable_line_match:
|
|
for test_case, data in self.test_case_map.items():
|
|
covered_files = data["files"] # {filepath: {lineno, ...}}
|
|
|
|
# Line-level matching: calculate which changed lines are covered by this test
|
|
affected_detail = {} # {filepath: set of covered changed lines}
|
|
all_intersected_lines = set() # union of intersections across all files
|
|
|
|
for changed_file, changed_lines in normalized_changed.items():
|
|
if changed_file in covered_files:
|
|
covered_lines = covered_files[changed_file]
|
|
# Calculate intersection of changed lines and covered lines
|
|
intersected_lines = changed_lines & covered_lines
|
|
if intersected_lines:
|
|
affected_detail[changed_file] = intersected_lines
|
|
all_intersected_lines.update(intersected_lines)
|
|
|
|
# Calculate overall coverage density: intersected lines / total changed lines
|
|
overall_density = (
|
|
len(all_intersected_lines) / total_changed_lines
|
|
if total_changed_lines
|
|
else 0
|
|
)
|
|
|
|
# Filter by coverage density and minimum affected lines
|
|
if (
|
|
all_intersected_lines
|
|
and overall_density >= COVERAGE_DENSITY_THRESHOLD
|
|
and len(all_intersected_lines) >= min_affected_lines
|
|
):
|
|
line_results.append(
|
|
(test_case, affected_detail, len(all_intersected_lines))
|
|
)
|
|
|
|
# Sort by affected lines (more first)
|
|
line_results.sort(key=lambda x: x[2], reverse=True)
|
|
|
|
# Line-level deduplication: for same covered lines, only select one test
|
|
if line_results and enable_dedup:
|
|
claimed_lines = set()
|
|
deduplicated = []
|
|
for test_case, affected_detail, total_lines in line_results:
|
|
# Collect all lines covered by this test
|
|
test_lines = set()
|
|
for lines in affected_detail.values():
|
|
test_lines.update(lines)
|
|
# Only keep tests with new lines
|
|
unclaimed = test_lines - claimed_lines
|
|
if unclaimed:
|
|
deduplicated.append(
|
|
(test_case, affected_detail, len(unclaimed))
|
|
)
|
|
claimed_lines.update(test_lines)
|
|
line_results = deduplicated
|
|
|
|
# ----- Stage 2: Function-level matching -----
|
|
if enable_function_match and source_dir:
|
|
# Collect functions that changed lines belong to
|
|
changed_functions = {} # {filepath: {func_name: Set[linenos]}}
|
|
changed_function_ranges = {} # {filepath: function_ranges} - parsed once per file
|
|
|
|
for changed_file, changed_lines in normalized_changed.items():
|
|
source_path = Path(source_dir) / REPO_NAME / changed_file
|
|
source_file = str(source_path) if source_path.exists() else None
|
|
|
|
if not source_file:
|
|
continue
|
|
|
|
# Parse function ranges ONCE per changed file and reuse the result
|
|
# for both line-to-function mapping and later range lookups
|
|
function_ranges = FunctionParser.get_function_ranges(source_file)
|
|
|
|
# Get function mapping for changed lines (reuses pre-parsed ranges)
|
|
line_to_function = FunctionParser.get_lines_functions(
|
|
source_file,
|
|
changed_lines,
|
|
skip_imports=enable_skip_imports,
|
|
function_ranges=function_ranges,
|
|
)
|
|
|
|
# Group by function name
|
|
func_to_lines = defaultdict(set)
|
|
for line, func_name in line_to_function.items():
|
|
func_to_lines[func_name].add(line)
|
|
|
|
if func_to_lines:
|
|
changed_functions[changed_file] = func_to_lines
|
|
changed_function_ranges[changed_file] = function_ranges
|
|
|
|
if changed_functions:
|
|
# Build function -> tests covering that function mapping
|
|
func_to_tests = defaultdict(list)
|
|
|
|
for test_case, data in self.test_case_map.items():
|
|
covered_files = data["files"]
|
|
|
|
for changed_file, func_to_lines in changed_functions.items():
|
|
if changed_file not in covered_files:
|
|
continue
|
|
|
|
covered_lines = covered_files[changed_file]
|
|
|
|
# Resolve source file once per changed file
|
|
source_path = Path(source_dir) / REPO_NAME / changed_file
|
|
source_file = str(source_path) if source_path.exists() else None
|
|
|
|
if not source_file:
|
|
continue
|
|
|
|
# Reuse function ranges parsed in the collection phase
|
|
func_ranges = changed_function_ranges.get(changed_file, {})
|
|
|
|
# Filter out import statement lines (for display), computed once
|
|
if enable_skip_imports:
|
|
import_lines = FunctionParser._get_import_lines(source_file)
|
|
display_changed_lines = (
|
|
normalized_changed.get(changed_file, set())
|
|
- import_lines
|
|
)
|
|
else:
|
|
display_changed_lines = normalized_changed.get(
|
|
changed_file, set()
|
|
)
|
|
|
|
for func_name in func_to_lines:
|
|
if func_name not in func_ranges:
|
|
continue
|
|
|
|
# Merge all matched function ranges
|
|
func_all_lines = set()
|
|
for func_start, func_end in func_ranges[func_name]:
|
|
func_all_lines.update(range(func_start, func_end + 1))
|
|
|
|
if not func_all_lines:
|
|
continue
|
|
|
|
# Check if this test covers any line of this function
|
|
covered_in_func = covered_lines & func_all_lines
|
|
if covered_in_func:
|
|
# Get intersection of test covered lines and actual changed lines (for display)
|
|
covered_changed_lines = (
|
|
covered_lines & display_changed_lines
|
|
)
|
|
func_to_tests[func_name].append(
|
|
(test_case, covered_in_func, covered_changed_lines)
|
|
)
|
|
|
|
# Select tests that cover other lines of changed functions (deduplication)
|
|
for changed_file, func_to_lines in changed_functions.items():
|
|
for func_name in func_to_lines:
|
|
if func_name in func_to_tests:
|
|
for (
|
|
test_case,
|
|
covered_in_func,
|
|
covered_changed_lines,
|
|
) in func_to_tests[func_name]:
|
|
existing = [s[0] for s in func_results]
|
|
if test_case not in existing and covered_in_func:
|
|
# Display changed lines coverage if available, otherwise function coverage
|
|
display_lines = (
|
|
covered_changed_lines
|
|
if covered_changed_lines
|
|
else set()
|
|
)
|
|
func_results.append(
|
|
(
|
|
test_case,
|
|
{changed_file: display_lines},
|
|
len(display_lines) or len(covered_in_func),
|
|
)
|
|
)
|
|
print(
|
|
f" [Function match] {test_case} covers function '{func_name}' in"
|
|
f" {changed_file}"
|
|
)
|
|
|
|
func_results.sort(key=lambda x: x[2], reverse=True)
|
|
|
|
# ===== Merge line-level and function-level results, deduplicate =====
|
|
if line_results or func_results:
|
|
# Deduplicate by test_case, keep line-level results (more precise)
|
|
seen = set()
|
|
for test_case, affected_detail, total_lines in line_results:
|
|
if test_case not in seen:
|
|
seen.add(test_case)
|
|
selected.append((test_case, affected_detail, total_lines))
|
|
|
|
# Add function-level exclusive results
|
|
for test_case, affected_detail, total_lines in func_results:
|
|
if test_case not in seen:
|
|
seen.add(test_case)
|
|
selected.append((test_case, affected_detail, total_lines))
|
|
|
|
# Sort by affected lines
|
|
selected.sort(key=lambda x: x[2], reverse=True)
|
|
|
|
if selected:
|
|
print(
|
|
f" Line match: {len(line_results)} tests, Function match: {len(func_results)} tests, "
|
|
f"Total: {len(selected)} tests"
|
|
)
|
|
return selected, "line+function"
|
|
|
|
# ===== Stage 3: File-level matching =====
|
|
if not selected and enable_file_match:
|
|
print(" Using file-level matching (renamed/deleted files)...")
|
|
expand_reason = "file"
|
|
|
|
# File-level matching: any test covering the changed file is selected
|
|
for test_case, data in self.test_case_map.items():
|
|
covered_files = data["files"]
|
|
|
|
for changed_file in normalized_changed:
|
|
if changed_file in covered_files:
|
|
covered_lines = covered_files[changed_file]
|
|
if covered_lines:
|
|
selected.append(
|
|
(
|
|
test_case,
|
|
{changed_file: covered_lines},
|
|
len(covered_lines),
|
|
)
|
|
)
|
|
break
|
|
|
|
# Deduplicate: same test case only selected once
|
|
if selected:
|
|
seen = set()
|
|
deduplicated = []
|
|
for s in selected:
|
|
if s[0] not in seen:
|
|
seen.add(s[0])
|
|
deduplicated.append(s)
|
|
selected = deduplicated
|
|
|
|
selected.sort(key=lambda x: x[2], reverse=True)
|
|
|
|
return selected, expand_reason
|
|
|
|
def print_selection(
|
|
self,
|
|
selected: list[tuple[str, dict[str, set[int]], int]],
|
|
changed_files: dict[str, set[int]],
|
|
min_affected_lines: int = 1,
|
|
expand_reason: str = "",
|
|
):
|
|
"""Print selection results"""
|
|
total_changed_lines = sum(len(v) for v in changed_files.values())
|
|
|
|
print("\n" + "=" * 70)
|
|
print(f"Code changes: {len(changed_files)} files, {total_changed_lines} lines")
|
|
|
|
# Display expansion reason
|
|
gran_names = {
|
|
"line": "Line match",
|
|
"function": "Function match",
|
|
"file": "File match",
|
|
"line+function": "Line+Function match",
|
|
}
|
|
gran_detail_titles = {
|
|
"line": "Details (Line match)",
|
|
"function": "Details (Function match)",
|
|
"file": "Details (File match)",
|
|
"line+function": "Details (Line+Function match)",
|
|
}
|
|
if expand_reason and expand_reason in gran_names:
|
|
print(f"Selected: {len(selected)} test cases ({gran_names[expand_reason]})")
|
|
else:
|
|
print(
|
|
f"Selected: {len(selected)} test cases (min affected: {min_affected_lines} lines)"
|
|
)
|
|
print("=" * 70)
|
|
|
|
if not selected:
|
|
print("\nNo test cases cover the changed code lines!")
|
|
print(f"Change details: {self._format_changed_files(changed_files)}")
|
|
return
|
|
|
|
print(f"\n{'#':<4} {'Test Case':<50} {'Affected Lines'}")
|
|
print("-" * 70)
|
|
|
|
for i, (test_case, affected_detail, total_lines) in enumerate(selected, 1):
|
|
# Build coverage line display
|
|
line_parts = []
|
|
for filepath, lines in sorted(affected_detail.items()):
|
|
line_parts.append(self._format_line_range(sorted(lines)))
|
|
line_display = f" ({', '.join(line_parts)})" if line_parts else ""
|
|
print(f"{i:<4} {test_case:<50} {total_lines}{line_display}")
|
|
|
|
print(f"\n{gran_detail_titles.get(expand_reason, 'Details')}:")
|
|
for test_case, affected_detail, total_lines in selected[:10]:
|
|
print(f"\n {test_case} ({total_lines} lines):")
|
|
for filepath, lines in sorted(affected_detail.items()):
|
|
line_str = self._format_line_range(sorted(lines))
|
|
print(f" - {filepath}: {line_str}")
|
|
|
|
@staticmethod
|
|
def _format_line_range(lines: list[int]) -> str:
|
|
"""Compress line number list into range representation"""
|
|
if not lines:
|
|
return ""
|
|
|
|
lines = sorted(set(lines))
|
|
ranges = []
|
|
start = lines[0]
|
|
end = lines[0]
|
|
|
|
for line in lines[1:]:
|
|
if line == end + 1:
|
|
end = line
|
|
else:
|
|
if start == end:
|
|
ranges.append(str(start))
|
|
else:
|
|
ranges.append(f"{start}-{end}")
|
|
start = end = line
|
|
|
|
if start == end:
|
|
ranges.append(str(start))
|
|
else:
|
|
ranges.append(f"{start}-{end}")
|
|
|
|
return ", ".join(ranges)
|
|
|
|
def _format_changed_files(self, changed_files: dict[str, set[int]]) -> str:
|
|
"""Format changed files"""
|
|
result = []
|
|
for f, lines in sorted(changed_files.items()):
|
|
if len(lines) > 10:
|
|
result.append(f"{f}: {len(lines)} lines")
|
|
else:
|
|
result.append(f"{f}: {sorted(lines)}")
|
|
return ", ".join(result[:5]) + ("..." if len(changed_files) > 5 else "")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Coverage-based precision test selector (line, function, file granularity)"
|
|
)
|
|
parser.add_argument(
|
|
"--github-pr", "-pr", help="GitHub PR, format: owner/repo#pr_number"
|
|
)
|
|
parser.add_argument(
|
|
"--source-dir",
|
|
"-s",
|
|
default="covstub",
|
|
help="Source code directory (default: covstub)",
|
|
)
|
|
parser.add_argument(
|
|
"--map-file",
|
|
"-m",
|
|
default="test_case_map.json",
|
|
help="Test case map file (default: test_case_map.json)",
|
|
)
|
|
parser.add_argument(
|
|
"--coverage-dir",
|
|
"-c",
|
|
default="coverage",
|
|
help="Coverage data directory (default: ./coverage)",
|
|
)
|
|
parser.add_argument(
|
|
"--build-map", "-b", action="store_true", help="Rebuild test case mapping"
|
|
)
|
|
parser.add_argument(
|
|
"--min-affected",
|
|
"-a",
|
|
type=int,
|
|
default=1,
|
|
help="Minimum affected lines threshold (default: 1)",
|
|
)
|
|
parser.add_argument(
|
|
"--dedup",
|
|
action="store_true",
|
|
help="Enable deduplication (keep only one test for same covered lines, default off)",
|
|
)
|
|
parser.add_argument(
|
|
"--skip-imports",
|
|
action="store_true",
|
|
help="Skip import statement lines (only effective for function-level matching, default off)",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Resolve relative paths against BASE_DIR (fixed structure), keep absolute paths as-is
|
|
def _resolve_abs(base: Path, p: str) -> Path:
|
|
path = Path(p)
|
|
return path if path.is_absolute() else base / path
|
|
|
|
coverage_dir = (
|
|
_resolve_abs(BASE_DIR, args.coverage_dir) if args.coverage_dir else None
|
|
)
|
|
source_dir = _resolve_abs(BASE_DIR, args.source_dir)
|
|
map_file = _resolve_abs(BASE_DIR, args.map_file)
|
|
|
|
# 1. Build or load test case mapping
|
|
selector = CoverageSelector(
|
|
str(coverage_dir) if coverage_dir else None, str(source_dir)
|
|
)
|
|
|
|
if args.build_map or not map_file.exists():
|
|
# Coverage data dir is required only when building the map
|
|
if not coverage_dir:
|
|
print(
|
|
"Error: --coverage-dir is required when building the test case map (no map file found)"
|
|
)
|
|
exit(1)
|
|
print("\n=== Building Test Case Mapping ===")
|
|
selector.build_test_case_map()
|
|
selector.save_map(str(map_file))
|
|
else:
|
|
print("\n=== Loading Test Case Mapping ===")
|
|
selector.load_map(str(map_file))
|
|
|
|
# If only need to generate map file, exit directly
|
|
if args.build_map and not args.github_pr:
|
|
print("\n=== Map file generated, done ===")
|
|
return
|
|
|
|
# 2. Parse code changes
|
|
print("\n=== Parsing Code Changes ===")
|
|
change_detector = CodeChangeDetector(str(source_dir))
|
|
|
|
diff_file = None
|
|
if args.github_pr:
|
|
# Fetch changes from GitHub PR
|
|
pr_spec = args.github_pr
|
|
repo = None
|
|
pr_num = None
|
|
|
|
# Parse owner/repo#pr_number format
|
|
if "#" in pr_spec:
|
|
parts = pr_spec.split("#")
|
|
repo = parts[0]
|
|
pr_num = parts[1]
|
|
else:
|
|
pr_num = pr_spec
|
|
# Try to get current repository
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "remote", "get-url", "origin"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if result.returncode == 0:
|
|
url = result.stdout.strip()
|
|
if "github.com" in url:
|
|
match = re.search(
|
|
r"github\.com[/:]([^/]+/[^/]+?)(?:\.git)?$", url
|
|
)
|
|
if match:
|
|
repo = match.group(1)
|
|
except Exception as e:
|
|
print(e)
|
|
pass
|
|
|
|
if not repo or not pr_num:
|
|
print("Error: Cannot parse PR info, please use owner/repo#pr_number format")
|
|
exit(1)
|
|
|
|
print(f"Fetching changes from GitHub PR: {repo}#{pr_num}")
|
|
|
|
github_token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN")
|
|
|
|
def _github_request(url: str) -> urllib.request.Request:
|
|
headers = {"Accept": "application/vnd.github.v3+json"}
|
|
if github_token:
|
|
headers["Authorization"] = f"Bearer {github_token}"
|
|
return urllib.request.Request(url, headers=headers)
|
|
|
|
# Create context that does not verify SSL certificates
|
|
ssl_context = ssl.create_default_context()
|
|
ssl_context.check_hostname = False
|
|
ssl_context.verify_mode = ssl.CERT_NONE
|
|
|
|
# Use cross-platform temp directory
|
|
diff_file = os.path.join(tempfile.gettempdir(), "pr.diff")
|
|
max_retries = 3
|
|
base_sha = None
|
|
|
|
for attempt in range(1, max_retries + 1):
|
|
print(f" Attempt {attempt}/{max_retries} to get PR diff via GitHub API...")
|
|
try:
|
|
pr_url = f"https://api.github.com/repos/{repo}/pulls/{pr_num}"
|
|
req = _github_request(pr_url)
|
|
with urllib.request.urlopen(
|
|
req, timeout=30, context=ssl_context
|
|
) as response:
|
|
pr_data = json.loads(response.read().decode())
|
|
diff_url = pr_data.get("diff_url")
|
|
base_sha = pr_data.get("base", {}).get("sha")
|
|
|
|
if not diff_url:
|
|
raise Exception("Cannot get diff URL")
|
|
|
|
# Download diff (use binary mode to avoid line ending conversion)
|
|
req = _github_request(diff_url)
|
|
with urllib.request.urlopen(
|
|
req, timeout=60, context=ssl_context
|
|
) as response:
|
|
diff_bytes = response.read()
|
|
with open(diff_file, "wb") as f:
|
|
f.write(diff_bytes)
|
|
print(" Using GitHub API to get diff")
|
|
break
|
|
except Exception as e:
|
|
print(f" Attempt {attempt} failed: {e}")
|
|
if attempt == max_retries:
|
|
print(f" All {max_retries} attempts failed, exiting")
|
|
exit(1)
|
|
time.sleep(1)
|
|
|
|
print(f" PR diff saved to: {diff_file}")
|
|
|
|
def _fetch_base_content(path: str) -> str | None:
|
|
"""Fetch base (pre-change) file content via GitHub contents API"""
|
|
content_url = f"https://api.github.com/repos/{repo}/contents/{urllib.parse.quote(path)}?ref={base_sha}"
|
|
try:
|
|
req = _github_request(content_url)
|
|
with urllib.request.urlopen(
|
|
req, timeout=30, context=ssl_context
|
|
) as response:
|
|
data = json.loads(response.read().decode())
|
|
if data.get("encoding") == "base64":
|
|
return base64.b64decode(data["content"]).decode("utf-8")
|
|
except Exception as e:
|
|
print(f" Warning: Failed to fetch base content for {path}: {e}")
|
|
return None
|
|
|
|
# Only usable when the PR base sha was fetched successfully
|
|
base_content_getter = _fetch_base_content if base_sha else None
|
|
else:
|
|
# Get from file comparison (default)
|
|
change_detector.scan_source_files()
|
|
changed_files_with_lines = change_detector.detect_changes_by_comparison()
|
|
print(f"Detected {len(changed_files_with_lines)} changed files")
|
|
|
|
# ===== Action 1: Extract new/deleted test files =====
|
|
new_test_files: list[str] = []
|
|
deleted_test_files: list[str] = []
|
|
if args.github_pr and diff_file:
|
|
new_test_files = _get_test_files_from_pr_diff(diff_file)
|
|
deleted_test_files = _get_deleted_test_files_from_pr(
|
|
diff_file, selector.test_case_map
|
|
)
|
|
|
|
# ===== Action 2: Detect Python product code changes -> Precision matching =====
|
|
selected: list[tuple[str, dict[str, set[int]], int]] = []
|
|
expand_reason = ""
|
|
changed_files_with_lines: dict[str, set[int]] = {}
|
|
|
|
renames: dict[str, str] = {}
|
|
deleted_files: list[str] = []
|
|
if args.github_pr and diff_file:
|
|
changed_files_with_lines, renames, deleted_files = (
|
|
change_detector.parse_pr_diff_file(
|
|
diff_file, base_content_getter=base_content_getter
|
|
)
|
|
)
|
|
print(f"Parsed {len(changed_files_with_lines)} changed files:")
|
|
for file_path, line_set in changed_files_with_lines.items():
|
|
print(f" {file_path}: {TestSelector._format_line_range(list(line_set))}")
|
|
|
|
# detect_renames already filters to PRODUCT_PREFIX only (product code renames)
|
|
if renames:
|
|
print(
|
|
f"\n=== Detected {len(renames)} Product Code Renamed File(s) - Using File-Level Matching ==="
|
|
)
|
|
for old_path, new_path in renames.items():
|
|
print(f" {old_path} -> {new_path}")
|
|
|
|
if deleted_files:
|
|
print(
|
|
f"\n=== Detected {len(deleted_files)} Product Code Deleted File(s) - Using File-Level Matching ==="
|
|
)
|
|
for path in deleted_files:
|
|
print(f" {path}")
|
|
|
|
if changed_files_with_lines or renames or deleted_files:
|
|
# Select test cases by precision matching
|
|
print("\n=== Selecting Affected Test Cases ===")
|
|
test_selector = TestSelector(selector.test_case_map)
|
|
|
|
# Renamed/deleted files are already excluded from changed_files by
|
|
# parse_git_diff; they are matched at file level below
|
|
normal_files = changed_files_with_lines
|
|
|
|
# Process normal files with precision matching
|
|
selected: list[tuple[str, dict, int]] = []
|
|
expand_reason = ""
|
|
if normal_files:
|
|
selected, expand_reason = test_selector.select_tests(
|
|
normal_files,
|
|
min_affected_lines=args.min_affected,
|
|
source_dir=str(source_dir),
|
|
enable_line_match=True,
|
|
enable_function_match=True,
|
|
enable_file_match=False, # File-level matching reserved for renamed/deleted files only
|
|
enable_skip_imports=args.skip_imports,
|
|
enable_dedup=args.dedup,
|
|
)
|
|
|
|
# Process renamed/deleted files: file-level matching with the base path
|
|
file_level_paths = [(p, f"{p} -> {n}") for p, n in renames.items()]
|
|
file_level_paths += [(p, p) for p in deleted_files]
|
|
for path, label in file_level_paths:
|
|
fl_selected, fl_expand = test_selector.select_tests(
|
|
{path: set()},
|
|
min_affected_lines=args.min_affected,
|
|
source_dir=str(source_dir),
|
|
enable_line_match=False, # Disable line match for file-level matching
|
|
enable_function_match=False, # Disable function match for file-level matching
|
|
enable_file_match=True, # Enable file match for renamed/deleted files
|
|
enable_skip_imports=args.skip_imports,
|
|
enable_dedup=args.dedup,
|
|
)
|
|
selected.extend(fl_selected)
|
|
expand_reason += fl_expand
|
|
# Print file-level matched test cases (even when empty, for diagnosis)
|
|
print(f"\n=== File-Level Matched Tests for {label} ===")
|
|
if fl_selected:
|
|
for test_name, _, _ in fl_selected:
|
|
print(f" {test_name}")
|
|
else:
|
|
print(
|
|
" (0 tests matched: no coverage data for this path in test_case_map)"
|
|
)
|
|
|
|
# Deduplicate
|
|
seen = set()
|
|
deduped = []
|
|
for item in selected:
|
|
if item[0] not in seen:
|
|
seen.add(item[0])
|
|
deduped.append(item)
|
|
selected = deduped
|
|
test_selector.print_selection(
|
|
selected,
|
|
changed_files_with_lines,
|
|
min_affected_lines=args.min_affected,
|
|
expand_reason=expand_reason,
|
|
)
|
|
else:
|
|
print("\n=== No product source code changes found ===")
|
|
|
|
# ===== Merge results =====
|
|
# Base set: precision matching results
|
|
base_selected = selected
|
|
|
|
# Add new test files
|
|
existing_test_names = {s[0] for s in base_selected}
|
|
for test_name in new_test_files:
|
|
if test_name not in existing_test_names:
|
|
base_selected.append((test_name, {}, 0))
|
|
existing_test_names.add(test_name)
|
|
|
|
if new_test_files:
|
|
print(f"\n=== New Test Files Added: {len(new_test_files)} ===")
|
|
print(f" {new_test_files}")
|
|
|
|
# Remove deleted test files
|
|
if deleted_test_files:
|
|
print(f"\n=== Deleted Test Files Removed: {len(deleted_test_files)} ===")
|
|
print(f" {deleted_test_files}")
|
|
deleted_set = set(deleted_test_files)
|
|
base_selected = [
|
|
(name, detail, count)
|
|
for name, detail, count in base_selected
|
|
if name not in deleted_set
|
|
and not any(name.startswith(d) for d in deleted_set)
|
|
]
|
|
|
|
# ===== Output results =====
|
|
test_names = [s[0] for s in base_selected]
|
|
if test_names:
|
|
print(f"\n=== Recommended Test Cases ({len(test_names)} tests) ===")
|
|
print(test_names)
|
|
else:
|
|
print("\n=== No Test Cases Recommended ===")
|
|
|
|
# Always write output file (even if empty), next to the script
|
|
output_file = BASE_DIR / "recommended_pytest_paths.txt"
|
|
with open(output_file, "w", encoding="utf-8") as f:
|
|
for test_name in test_names:
|
|
f.write(test_name + "\n")
|
|
print(f"\nResults saved to: {output_file}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|