diff --git a/.github/workflows/precommit-check.yml b/.github/workflows/precommit-check.yml index d91861f44fdf..e81c809b9cfd 100644 --- a/.github/workflows/precommit-check.yml +++ b/.github/workflows/precommit-check.yml @@ -59,3 +59,13 @@ jobs: else python3 -u scripts/release_check.py fi + + # Informational only: list test-list entries whose parametrize IDs cannot + # be verified statically (runtime-computed argvalues/ids), grouped by + # reason, so they can be swept without rerunning the analysis locally. The + # pre-commit step above remains the gate for actual validation errors; + # this step never fails the job. + - name: Report unverifiable test-list param IDs + if: always() + continue-on-error: true + run: python3 -u scripts/check_test_list.py --validate --report-unverifiable diff --git a/jenkins/L0_Test.groovy b/jenkins/L0_Test.groovy index c942fff785cb..321cb817838a 100644 --- a/jenkins/L0_Test.groovy +++ b/jenkins/L0_Test.groovy @@ -3818,7 +3818,9 @@ def launchTestListCheck(pipeline) def llmPath = sh (script: "realpath .", returnStdout: true).trim() def llmSrc = "${llmPath}/TensorRT-LLM/src" trtllm_utils.llmExecStepWithRetry(pipeline, script: "pip3 install -r ${llmSrc}/requirements-dev.txt") - sh "NVIDIA_TRITON_SERVER_VERSION=26.05 LLM_ROOT=${llmSrc} LLM_BACKEND_ROOT=${llmSrc}/triton_backend python3 ${llmSrc}/scripts/check_test_list.py --l0 --qa --waive" + // --validate --parity: after --l0/--qa generate the collectable lists, assert every + // statically-verified parametrize ID is actually collectable (validate<->collection parity). + sh "NVIDIA_TRITON_SERVER_VERSION=26.05 LLM_ROOT=${llmSrc} LLM_BACKEND_ROOT=${llmSrc}/triton_backend python3 ${llmSrc}/scripts/check_test_list.py --l0 --qa --waive --validate --parity" } catch (InterruptedException e) { throw e } catch (Exception e) { diff --git a/scripts/check_test_list.py b/scripts/check_test_list.py index 97bb4332c1d9..2a5a44367052 100755 --- a/scripts/check_test_list.py +++ b/scripts/check_test_list.py @@ -16,6 +16,15 @@ --qa: Check only the QA tests under $LLM_ROOT/tests/integration/test_list/*.txt. --waive: Check only the tests in $LLM_ROOT/tests/integration/test_list/waives.txt. --validate: Run AST-based validation of test list entries against source files. +--report-unverifiable [PATH]: With --validate, write the list of active entries + whose parametrize IDs cannot be checked statically (runtime-computed + argvalues/ids) to PATH (default: stdout). Use to sweep them. +--strict-param-ids: With --validate, fail if any active entry has an + unverifiable parametrize ID. Off by default (sweep, then gate). +--parity: With --validate (run alongside --l0/--qa so the runtime lists exist), + fail if any statically-verified parametrize ID is not collectable by + pytest -- i.e. assert validate-accepts is a subset of collectable. + Catches resolver soundness bugs and stale entries. Note: All the perf tests will be excluded since they are generated dynamically. @@ -40,6 +49,11 @@ # Paths whose tests are generated dynamically — skip AST validation _EXCLUDED_PATH_PREFIXES = ("perf/", ) +# A parsed test-list entry: (rel_path, class_name|None, func_name, param_id|None). +_EntryTuple = tuple[str, str | None, str, str | None] +# Map of module-level constant name -> its bound value node. +_ModuleConsts = dict[str, ast.expr] + # ============================================================================= # AST-based test list validation # ============================================================================= @@ -97,7 +111,7 @@ def _is_parametrize_call(node) -> bool: return False -def _ast_constant_str(node) -> str | None: +def _ast_constant_str(node: ast.expr) -> str | None: """Return str() of an AST constant node, or None if not a simple literal.""" if isinstance(node, ast.Constant) and isinstance( node.value, (str, int, float, bool, type(None))): @@ -105,7 +119,59 @@ def _ast_constant_str(node) -> str | None: return None -def _get_parametrize_with_ids_ids(call: ast.Call) -> list[str] | None: +def _argnames_list(node: ast.expr) -> list[str] | None: + """Return the list of parametrize argnames, or None if not a static literal. + + Handles both the "a,b" comma-string form and the ["a", "b"] list/tuple form. + """ + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return [n.strip() for n in node.value.split(",")] + if isinstance(node, (ast.List, ast.Tuple)): + names = [] + for elt in node.elts: + if isinstance(elt, ast.Constant) and isinstance(elt.value, str): + names.append(elt.value.strip()) + else: + return None + return names + return None + + +def _resolve_const_node(node: ast.expr, + module_consts: _ModuleConsts | None, + _depth: int = 0) -> ast.expr: + """Follow a Name to the module-level literal it is bound to. + + Returns the resolved node (or the input node unchanged if it is not a Name + or cannot be resolved). Only names bound exactly once at module scope are in + ``module_consts`` (see build_ast_index), so this is sound. Bounded depth + guards against reference cycles. + """ + while (isinstance(node, ast.Name) and module_consts and _depth < 10 + and node.id in module_consts): + node = module_consts[node.id] + _depth += 1 + return node + + +def _argvalues_elts( + node: ast.expr, + module_consts: _ModuleConsts | None) -> list[ast.expr] | None: + """Return the element nodes of an argvalues expression, or None. + + Resolves a List/Tuple literal directly, or a Name transitively bound to + one. Anything else (a call result, comprehension, imported name, ...) is + not statically resolvable and yields None. + """ + node = _resolve_const_node(node, module_consts) + if isinstance(node, (ast.List, ast.Tuple)): + return list(node.elts) + return None + + +def _get_parametrize_with_ids_ids( + call: ast.Call, + module_consts: _ModuleConsts | None = None) -> list[str] | None: """Extract IDs from a parametrize_with_ids(argnames, argvalues) call. parametrize_with_ids generates IDs like "argname=value" joined with "-". @@ -114,24 +180,16 @@ def _get_parametrize_with_ids_ids(call: ast.Call) -> list[str] | None: return None argnames_node, argvalues_node = call.args[0], call.args[1] - if isinstance(argnames_node, ast.Constant) and isinstance( - argnames_node.value, str): - argname_list = [n.strip() for n in argnames_node.value.split(",")] - elif isinstance(argnames_node, (ast.List, ast.Tuple)): - argname_list = [] - for elt in argnames_node.elts: - if isinstance(elt, ast.Constant) and isinstance(elt.value, str): - argname_list.append(elt.value.strip()) - else: - return None - else: + argname_list = _argnames_list(argnames_node) + if argname_list is None: return None - if not isinstance(argvalues_node, ast.List): + elts = _argvalues_elts(argvalues_node, module_consts) + if elts is None: return None ids = [] - for elt in argvalues_node.elts: + for elt in elts: val_str = _ast_constant_str(elt) if val_str is not None: if len(argname_list) != 1: @@ -152,16 +210,18 @@ def _get_parametrize_with_ids_ids(call: ast.Call) -> list[str] | None: return ids -def _get_parametrize_ids(call: ast.Call) -> list[str] | None: +def _get_parametrize_ids(call: ast.Call, + module_consts=None) -> list[str] | None: """Extract the list of IDs from a parametrize call. Handles both pytest.mark.parametrize and parametrize_with_ids. - Tries ids= kwarg first, then falls back to string/int literal argvalues. + Tries ids= kwarg first, then falls back to string/int literal argvalues + (a module-level ``NAME = [...]`` reference is resolved via module_consts). Returns None if IDs cannot be determined statically. """ func = call.func if isinstance(func, ast.Name) and func.id == "parametrize_with_ids": - return _get_parametrize_with_ids_ids(call) + return _get_parametrize_with_ids_ids(call, module_consts) for kw in call.keywords: if kw.arg == "ids": @@ -178,22 +238,36 @@ def _get_parametrize_ids(call: ast.Call) -> list[str] | None: # No ids= — use string/int literal argvalues (or pytest.param(...)) if len(call.args) < 2: return None - argvalues = call.args[1] - if isinstance(argvalues, ast.List): - ids = [] - for elt in argvalues.elts: - val = _ast_constant_str(elt) - if val is not None: - ids.append(val) - elif isinstance(elt, ast.Call) and _is_pytest_param(elt): - inner = _pytest_param_id(elt) - if inner is None: - return None - ids.append(inner) - else: + argnames = _argnames_list(call.args[0]) + elts = _argvalues_elts(call.args[1], module_consts) + if elts is None: + return None + ids = [] + for elt in elts: + val = _ast_constant_str(elt) + if val is not None: + ids.append(val) + elif isinstance(elt, ast.Call) and _is_pytest_param(elt): + inner = _pytest_param_id(elt) + if inner is None: return None - return ids - return None + ids.append(inner) + elif isinstance(elt, (ast.Tuple, ast.List)): + # A multi-argument row: pytest joins each argument's scalar ID with + # "-". This is only sound when there are >1 argnames matching the + # row length and every value is a by-value scalar; a tuple bound to + # a single argname gets a positional "" ID instead, + # which we cannot reproduce statically. + if (argnames is None or len(argnames) < 2 + or len(elt.elts) != len(argnames)): + return None + parts = [_ast_constant_str(v) for v in elt.elts] + if any(p is None for p in parts): + return None + ids.append("-".join(parts)) + else: + return None + return ids def _is_pytest_param(node) -> bool: @@ -223,8 +297,10 @@ def _pytest_param_id(call: ast.Call) -> str | None: return "-".join(parts) if parts else None -def _compute_valid_param_ids(func_node, - class_decorators=None) -> set[str] | None: +def _compute_valid_param_ids( + func_node: ast.FunctionDef | ast.AsyncFunctionDef, + class_decorators: list[ast.expr] | None = None, + module_consts: _ModuleConsts | None = None) -> set[str] | None: """Return the set of valid parametrize IDs for a function node. Iterates method-level parametrize decorators in reverse source order @@ -236,34 +312,168 @@ def _compute_valid_param_ids(func_node, for decorator in reversed(func_node.decorator_list): if not _is_parametrize_call(decorator): continue - ids = _get_parametrize_ids(decorator) + ids = _get_parametrize_ids(decorator, module_consts) if ids is None: return None groups.append(ids) for decorator in reversed(class_decorators or []): if not _is_parametrize_call(decorator): continue - ids = _get_parametrize_ids(decorator) + ids = _get_parametrize_ids(decorator, module_consts) if ids is None: return None groups.append(ids) if not groups: return set() - return {"-".join(combo) for combo in product(*groups)} + # Build the cartesian product as a list so we can detect collisions: when + # two combinations yield the same ID, pytest disambiguates by appending + # "0"/"1"/... — an ID we cannot reproduce here — so treat the whole set as + # indeterminate rather than silently accepting the collapsed value. + combos = ["-".join(combo) for combo in product(*groups)] + if len(set(combos)) != len(combos): + return None + return set(combos) + + +def _classify_unverifiable(func_node: ast.FunctionDef | ast.AsyncFunctionDef, + class_decorators: list[ast.expr] | None, + module_consts: _ModuleConsts | None) -> str: + """Return a short reason code for why a param ID cannot be verified. + + Report-only; used to bucket unverifiable entries so they can be swept. + """ + decs = [ + d for d in list(func_node.decorator_list) + list(class_decorators or []) + if _is_parametrize_call(d) + ] + if not decs: + return "no-static-parametrize-decorator" + for d in decs: + if _get_parametrize_ids(d, module_consts) is not None: + continue + for kw in d.keywords: + if kw.arg == "ids" and not isinstance(kw.value, ast.List): + return "ids=callable-or-nonliteral" + argvalues = d.args[1] if len(d.args) >= 2 else None + if argvalues is None: + return "missing-argvalues" + resolved = _resolve_const_node(argvalues, module_consts) + if isinstance(resolved, ast.Name): + return f"argvalues=Name-unresolved:{resolved.id}" + if isinstance(resolved, (ast.ListComp, ast.GeneratorExp, ast.SetComp)): + return "argvalues=comprehension" + if isinstance(resolved, ast.Call): + return "argvalues=call-result" + if isinstance(resolved, (ast.List, ast.Tuple)): + return "argvalues-elements-non-literal" + return f"argvalues={type(resolved).__name__}" + # Every decorator resolved individually — the product had an ID collision. + return "param-id-collision" + + +def _iter_target_names(target: ast.expr) -> list[str]: + """Return the Name ids bound by an assignment/for/with target. + + Recurses into tuple/list/starred targets so unpacking binds are all caught. + """ + names: list[str] = [] + if isinstance(target, ast.Name): + names.append(target.id) + elif isinstance(target, (ast.Tuple, ast.List)): + for elt in target.elts: + names.extend(_iter_target_names(elt)) + elif isinstance(target, ast.Starred): + names.extend(_iter_target_names(target.value)) + return names + + +def _mutated_or_rebound_names(tree: ast.AST) -> set[str]: + """Return names that are rebound or mutated anywhere in ``tree``. + + Conservative (over-approximating) soundness guard for module_consts: a name + that is augmented-assigned, mutated in place (``NAME[i] = ...``, + ``NAME.attr = ...``, ``NAME.method(...)``, ``del NAME[i]``), rebound by a + non-Assign statement (for/with target, import alias, walrus, def/class), or + unbound by ``del NAME`` cannot be trusted to still equal its initial literal, + so it must not be resolved as a constant. The AST + walk is scope-insensitive, so this may also drop a name that is only shadowed + in a nested scope -- that costs resolver coverage, never soundness (an + unresolved name becomes unverifiable, not a wrong INVALID error). + """ + unsafe: set[str] = set() + + def _flag_inplace_target(target: ast.expr) -> None: + # A Subscript/Attribute store (NAME[i] = / NAME.attr =) mutates the base + # Name in place without rebinding it. + if not isinstance(target, (ast.Subscript, ast.Attribute)): + return + base: ast.expr = target + while isinstance(base, (ast.Subscript, ast.Attribute)): + base = base.value + if isinstance(base, ast.Name): + unsafe.add(base.id) + + for node in ast.walk(tree): + if isinstance(node, ast.AugAssign): + _flag_inplace_target(node.target) + if isinstance(node.target, ast.Name): + unsafe.add(node.target.id) + elif isinstance(node, ast.Assign): + for tgt in node.targets: + _flag_inplace_target(tgt) + elif isinstance(node, ast.AnnAssign): + _flag_inplace_target(node.target) + elif isinstance(node, ast.NamedExpr) and isinstance( + node.target, ast.Name): + unsafe.add(node.target.id) + elif isinstance(node, (ast.For, ast.AsyncFor)): + unsafe.update(_iter_target_names(node.target)) + elif isinstance(node, (ast.With, ast.AsyncWith)): + for item in node.items: + if item.optional_vars is not None: + unsafe.update(_iter_target_names(item.optional_vars)) + elif isinstance(node, (ast.Import, ast.ImportFrom)): + for alias in node.names: + unsafe.add(alias.asname or alias.name.split(".")[0]) + elif isinstance(node, + (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + # A def/class binds its name, shadowing any same-named const. + unsafe.add(node.name) + elif isinstance(node, ast.Delete): + for tgt in node.targets: + if isinstance(tgt, ast.Name): + unsafe.add(tgt.id) # del NAME unbinds it entirely. + else: + _flag_inplace_target( + tgt) # del NAME[i] / NAME.attr mutates. + elif (isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name)): + # NAME.method(...) may mutate NAME in place (append/extend/...). + unsafe.add(node.func.value.id) + return unsafe def build_ast_index(filepath: str): - """Return (classes, top_level_funcs, top_level_nodes, class_method_nodes). + """Return the AST index tuple for a single test source file. + + Tuple layout: (classes, top_level_funcs, top_level_nodes, + class_method_nodes, module_consts). classes: {class_name: {'methods': set[str], 'bases': list[str], 'decorators': list[ast.Call]}} - Returns (None, None, None, None) on error. + module_consts: {name: value_node} for module-level ``NAME = `` + bindings, kept only when NAME is assigned exactly once at top level and + is never otherwise rebound or mutated anywhere in the module (see + _mutated_or_rebound_names), so a statically resolved value cannot be + wrong. + Returns (None, None, None, None, None) on error. """ try: source = Path(filepath).read_text(encoding="utf-8") tree = ast.parse(source, filename=filepath) except (OSError, SyntaxError): - return None, None, None, None + return None, None, None, None, None classes = {} class_method_nodes = {} @@ -295,7 +505,28 @@ def build_ast_index(filepath: str): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) } top_level_funcs = set(top_level_nodes) - return classes, top_level_funcs, top_level_nodes, class_method_nodes + + # Module-level literal bindings, kept only when a name is assigned exactly + # once via a plain top-level ``NAME = `` AND is never otherwise rebound + # or mutated (augmented assign, NAME[i]=/NAME.attr= store, NAME.method(...), + # for/with target, import alias, walrus). Otherwise the value we resolve + # parametrize argvalues against could be stale. + assign_counts = defaultdict(int) + module_consts = {} + for node in tree.body: + if isinstance(node, ast.Assign): + for tgt in node.targets: + if isinstance(tgt, ast.Name): + assign_counts[tgt.id] += 1 + module_consts[tgt.id] = node.value + unsafe_names = _mutated_or_rebound_names(tree) + module_consts = { + name: val + for name, val in module_consts.items() + if assign_counts[name] == 1 and name not in unsafe_names + } + return (classes, top_level_funcs, top_level_nodes, class_method_nodes, + module_consts) def _matches_parameterized(func_name: str, methods: set) -> bool: @@ -315,7 +546,7 @@ def _has_method(class_name: str, method_name: str, ast_cache: dict, return False visited.add(class_name) - for classes, _, _, _ in ast_cache.values(): + for classes, _, _, _, _ in ast_cache.values(): if not classes or class_name not in classes: continue info = classes[class_name] @@ -376,8 +607,18 @@ def _ensure_dir_indexed(abs_path: str, ast_cache: dict) -> None: ast_cache[sibling] = build_ast_index(sibling) -def validate_test_lists(test_lists_dir: str, test_base_dir: str): - """Return list of error strings from AST-based validation.""" +def validate_test_lists( + test_lists_dir: str, test_base_dir: str +) -> tuple[list[str], list[tuple], list[_EntryTuple], list[_EntryTuple]]: + """Return (errors, unverifiable, accepted_param, rejected_param). + + errors: fatal validation problems (missing file/class/method, invalid or + truncated parametrize ID, waive not in active lists). + unverifiable: active entries whose parametrize ID cannot be checked + statically -- (rel_path, class, func, param_id, reason, refs) tuples. + accepted_param / rejected_param: active entries whose parametrize ID the + static resolver positively verified / rejected, for the parity check. + """ active_entries, active_malformed = collect_entries(test_lists_dir, include_waives=False) waive_entries, waive_malformed = collect_entries(test_lists_dir, @@ -393,6 +634,17 @@ def validate_test_lists(test_lists_dir: str, test_base_dir: str): ast_cache = {} errors = [] + # Active entries whose parametrize ID cannot be checked statically (runtime- + # computed argvalues/ids). Collected for reporting/sweeping, not fatal by + # default. Each item: (rel_path, class_name, func_name, param_id, reason, + # refs). + unverifiable = [] + # Active entries whose parametrize ID the static resolver positively verified + # (param_id in valid_ids) or positively rejected (INVALID PARAMETRIZE ID). + # Feed the validate<->collection parity check. Each item is an entry tuple + # (rel_path, class_name, func_name, param_id). + accepted_param = [] + rejected_param = [] active_malformed_set = set(active_malformed) for ref in active_malformed + [ @@ -447,7 +699,7 @@ def validate_test_lists(test_lists_dir: str, test_base_dir: str): if abs_path not in ast_cache: ast_cache[abs_path] = build_ast_index(abs_path) - classes, top_level, top_level_nodes, class_method_nodes = ( + classes, top_level, top_level_nodes, class_method_nodes, module_consts = ( ast_cache[abs_path]) if classes is None: errors.append(f"PARSE ERROR: {rel_path}") @@ -495,7 +747,8 @@ def validate_test_lists(test_lists_dir: str, test_base_dir: str): if param_id and func_node: class_decs = (classes[class_name]["decorators"] if class_name and class_name in classes else None) - valid_ids = _compute_valid_param_ids(func_node, class_decs) + valid_ids = _compute_valid_param_ids(func_node, class_decs, + module_consts) if valid_ids is not None and len(valid_ids) > 0: if param_id not in valid_ids: errors.append("INVALID PARAMETRIZE ID: {}{}::{}\n" @@ -507,6 +760,20 @@ def validate_test_lists(test_lists_dir: str, test_base_dir: str): sorted(valid_ids)[:8], "\n".join(f" -> {r}" for r in refs[:3]), )) + if not is_waive_only: + rejected_param.append( + (rel_path, class_name, func_name, param_id)) + elif not is_waive_only: + accepted_param.append( + (rel_path, class_name, func_name, param_id)) + elif not is_waive_only: + # Active entry whose ID we could not resolve statically. Record + # it for the sweep/gate rather than silently passing it (which + # is the coverage the runtime `pytest --co` stage uniquely had). + reason = _classify_unverifiable(func_node, class_decs, + module_consts) + unverifiable.append( + (rel_path, class_name, func_name, param_id, reason, refs)) if is_waive_only and not rel_path.startswith("unittest/"): param_suffix = f"[{param_id}]" if param_id else "" @@ -519,7 +786,114 @@ def validate_test_lists(test_lists_dir: str, test_base_dir: str): "\n".join(f" -> {r}" for r in refs[:3]), )) - return errors + return errors, unverifiable, accepted_param, rejected_param + + +def _format_unverifiable(unverifiable: list[tuple]) -> list[str]: + """Render unverifiable entries as sorted, one-per-line report strings.""" + lines = [] + for rel_path, class_name, func_name, param_id, reason, refs in sorted( + unverifiable, key=lambda x: (x[4], x[0], x[1] or "", x[2])): + cls = f"::{class_name}" if class_name else "" + ref = refs[0] if refs else "?" + lines.append( + f"{rel_path}{cls}::{func_name}[{param_id}] # {reason} ({ref})") + return lines + + +def write_unverifiable_report(unverifiable: list[tuple], dest: str) -> None: + """Write the unverifiable-param-id report to ``dest`` ('-' means stdout).""" + lines = _format_unverifiable(unverifiable) + # Group counts by reason for a quick triage summary at the top. + by_reason = defaultdict(int) + for *_, reason, _refs in unverifiable: + by_reason[reason] += 1 + header = [f"# {len(unverifiable)} unverifiable param IDs by reason:"] + header += [ + f"# {count:5d} {reason}" + for reason, count in sorted(by_reason.items(), key=lambda x: -x[1]) + ] + body = "\n".join(header + [""] + lines) + "\n" + if dest == "-": + print(body) + else: + with open(dest, "w", encoding="utf-8") as f: + f.write(body) + print(f"Wrote {len(unverifiable)} unverifiable entries to {dest}") + + +# ============================================================================= +# validate <-> collection parity +# ============================================================================= + + +def _format_entry_tuple(entry: _EntryTuple) -> str: + """Render an (rel_path, class, func, param_id) tuple as a test node id.""" + rel_path, class_name, func_name, param_id = entry + cls = f"::{class_name}" if class_name else "" + param = f"[{param_id}]" if param_id else "" + return f"{rel_path}{cls}::{func_name}{param}" + + +def _entry_sort_key(entry: _EntryTuple) -> tuple[str, str, str, str]: + rel_path, class_name, func_name, param_id = entry + return (rel_path, class_name or "", func_name, param_id or "") + + +def load_collectable_entries(llm_src: str) -> set[_EntryTuple] | None: + """Return the entry tuples the runtime stage proved collectable, or None. + + Reads the l0_test.txt / qa_test.txt lists written by verify_l0_test_lists / + verify_qa_test_lists. Those files are emitted only after `pytest --co` + confirmed every listed id collects, so a tuple's presence here means pytest + can collect it. Both these lines and the validator's entries are parsed by + the same parse_test_entry, so the tuple forms are directly comparable. + + Returns None if neither list exists (parity cannot be computed without the + runtime lists, e.g. --validate --parity run without --l0/--qa). + """ + collectable = set() + found_any = False + for name in ("l0_test.txt", "qa_test.txt"): + path = os.path.join(llm_src, name) + if not os.path.isfile(path): + continue + found_any = True + with open(path, encoding="utf-8") as f: + for line in f: + entry = parse_test_entry(line) + if entry is None or entry[0] == "MALFORMED": + continue + collectable.add(entry) + return collectable if found_any else None + + +def compute_parity( + accepted: list[_EntryTuple], + rejected: list[_EntryTuple], + collectable: set[_EntryTuple], +) -> tuple[list[_EntryTuple], list[_EntryTuple]]: + """Pure set logic for the validate <-> collection parity check. + + Args: + accepted: entry tuples whose parametrize ID the static resolver + positively verified (param_id in valid_ids). + rejected: entry tuples the static resolver rejected (INVALID PARAMETRIZE + ID). + collectable: set of entry tuples pytest proved collectable. + + Returns (false_confidence, false_alarm): + false_confidence: accepted entries that are NOT collectable -- the fast + check passed something the runtime stage would reject. This is the + gate-worthy class (violates accepted subset-of collectable). + false_alarm: rejected entries that ARE collectable -- the resolver was + wrong to reject; report to tighten it, but not fatal. + """ + false_confidence = sorted((t for t in accepted if t not in collectable), + key=_entry_sort_key) + false_alarm = sorted((t for t in rejected if t in collectable), + key=_entry_sort_key) + return false_confidence, false_alarm # ============================================================================= @@ -600,6 +974,13 @@ def verify_l0_test_lists(llm_src): def verify_qa_test_lists(llm_src): test_qa_path = f"{llm_src}/tests/integration/test_lists/qa" + # Start from a clean qa_test.txt so a stale file left by an earlier run + # (this opens it in append mode below) can't inject entries an older + # checkout collected into the current parity comparison. + try: + os.remove(f"{llm_src}/qa_test.txt") + except OSError: + pass # Remove dynamically generated perf tests subprocess.run(f"rm -f {test_qa_path}/*perf*", shell=True, check=True) test_def_files = subprocess.check_output( @@ -761,7 +1142,37 @@ def main(): help= f"Base directory for test source files for --validate (default: {_DEFAULT_TEST_BASE_DIR})", ) + parser.add_argument( + "--report-unverifiable", + nargs="?", + const="-", + default=None, + metavar="PATH", + help="With --validate: write active entries whose parametrize IDs " + "cannot be checked statically to PATH (default: stdout). For sweeping.", + ) + parser.add_argument( + "--strict-param-ids", + action="store_true", + help="With --validate: fail if any active entry has an unverifiable " + "parametrize ID. Off by default (sweep first, then gate).", + ) + parser.add_argument( + "--parity", + action="store_true", + help="With --validate (run alongside --l0/--qa): fail if any " + "statically-verified parametrize ID is not collectable by pytest, i.e. " + "assert validate-accepts is a subset of collectable. Catches resolver " + "soundness bugs and stale entries.", + ) args = parser.parse_args() + # Parity needs a fresh validation run (for the accepted/rejected buckets) and + # a fresh collectable list. Reject invocations that would otherwise compare + # against empty results or stale l0_test.txt / qa_test.txt artifacts. + if args.parity and not args.validate: + parser.error("--parity requires --validate") + if args.parity and not (args.l0 or args.qa): + parser.error("--parity requires --l0 or --qa in the same invocation") script_dir = os.path.dirname(os.path.realpath(__file__)) llm_src = os.path.abspath(os.path.join(script_dir, "../")) @@ -807,7 +1218,8 @@ def main(): if args.validate: print("-----------Starting AST test list validation...-----------", flush=True) - errors = validate_test_lists(args.test_lists_dir, args.test_base_dir) + errors, unverifiable, accepted_param, rejected_param = ( + validate_test_lists(args.test_lists_dir, args.test_base_dir)) if errors: print(f"Found {len(errors)} validation error(s):\n", file=sys.stderr) @@ -819,6 +1231,70 @@ def main(): entries, _ = collect_entries(args.test_lists_dir) print(f"OK: {len(entries)} unique test entries validated.") + # Instrumentation: active entries whose parametrize IDs cannot be + # checked statically. Informational (non-fatal) unless --strict-param-ids. + # Write the report whenever requested, even when empty, so automation + # can tell an empty result from a missing report file. + if args.report_unverifiable is not None: + write_unverifiable_report(unverifiable, args.report_unverifiable) + + if unverifiable: + print( + f"UNVERIFIABLE: {len(unverifiable)} active entr" + f"{'y has' if len(unverifiable) == 1 else 'ies have'} " + f"parametrize IDs that cannot be checked statically " + f"(runtime-computed argvalues/ids).", + file=sys.stderr) + if args.report_unverifiable is None: + print(" Re-run with --report-unverifiable to list them.", + file=sys.stderr) + if args.strict_param_ids: + print( + " --strict-param-ids is set: treating unverifiable " + "entries as errors.", + file=sys.stderr) + pass_flag = False + + # Parity: cross-check the statically-verified param IDs against the ids + # the runtime stage proved collectable (l0_test.txt / qa_test.txt). Only + # meaningful when those lists were generated in this run (--l0/--qa). + if args.parity: + collectable = load_collectable_entries(llm_src) + if collectable is None: + print( + "PARITY: skipped -- no l0_test.txt/qa_test.txt found " + "(run --parity alongside --l0/--qa).", + file=sys.stderr) + else: + false_confidence, false_alarm = compute_parity( + accepted_param, rejected_param, collectable) + if false_alarm: + n = len(false_alarm) + print( + f"PARITY false alarm: {n} " + f"{'entry' if n == 1 else 'entries'} rejected by " + f"--validate but collectable by pytest (resolver too " + f"strict -- please report to tighten it):", + file=sys.stderr) + for entry in false_alarm: + print(f" {_format_entry_tuple(entry)}", + file=sys.stderr) + if false_confidence: + n = len(false_confidence) + print( + f"PARITY VIOLATION: {n} " + f"{'entry' if n == 1 else 'entries'} passed --validate " + f"but pytest cannot collect (stale entry or resolver " + f"soundness bug):", + file=sys.stderr) + for entry in false_confidence: + print(f" {_format_entry_tuple(entry)}", + file=sys.stderr) + pass_flag = False + if not false_confidence and not false_alarm: + print(f"PARITY OK: {len(accepted_param)} " + f"statically-verified param IDs all collectable.") + invalid_json_file = os.path.join(llm_src, "invalid_tests.json") if os.path.isfile(invalid_json_file) and os.path.getsize( invalid_json_file) > 0: diff --git a/tests/integration/defs/sysinfo/get_sysinfo.py b/tests/integration/defs/sysinfo/get_sysinfo.py index 3643433c3e9f..b637b44e6112 100644 --- a/tests/integration/defs/sysinfo/get_sysinfo.py +++ b/tests/integration/defs/sysinfo/get_sysinfo.py @@ -108,6 +108,9 @@ def is_power(): return platform.processor() == "ppc64le" +# TODO(#17993): cherry-picked from PR #17993 to unblock this PR's CI (distro->'na' +# empty-render bug). Drop this and take main's version when resolving the rebase +# conflict after #17993 lands. def get_linux_distribution(): try: import distro diff --git a/tests/integration/test_lists/waives.txt b/tests/integration/test_lists/waives.txt index d664cbfd06d9..649bdf5f88ed 100644 --- a/tests/integration/test_lists/waives.txt +++ b/tests/integration/test_lists/waives.txt @@ -321,6 +321,12 @@ full:RTX_PRO_6000_Blackwell_Server_Edition/accuracy/test_llm_api_pytorch.py::Tes full:RTX_PRO_6000_Blackwell_Server_Edition/accuracy/test_llm_api_pytorch.py::TestQwen3_5_4B::test_dflash SKIP (https://nvbugs/6273850) full:RTX_PRO_6000_Blackwell_Server_Edition/accuracy/test_llm_api_pytorch.py::TestQwen3_5_4B::test_fp8 SKIP (https://nvbugs/6273850) full:sm100/unittest/bindings SKIP (Disable for Blackwell) +kv_cache/test_kv_cache_v2_scheduler.py::TestKVCacheV2Llama::test_chunked_prefill SKIP (https://nvbugs/6428002) +kv_cache/test_kv_cache_v2_scheduler.py::TestKVCacheV2Llama::test_eviction[cuda_graph] SKIP (https://nvbugs/6600098) +kv_cache/test_kv_cache_v2_scheduler.py::TestKVCacheV2Llama::test_eviction_with_block_reuse SKIP (https://nvbugs/6462303) +kv_cache/test_kv_cache_v2_scheduler.py::TestKVCacheV2Llama::test_overlap_scheduler[non_overlap] SKIP (https://nvbugs/6600098) +kv_cache/test_kv_cache_v2_scheduler.py::TestKVCacheV2Llama::test_overlap_scheduler[overlap] SKIP (https://nvbugs/6600098) +kv_cache/test_kv_cache_v2_scheduler.py::TestKVCacheV2Llama::test_token_budget_limited SKIP (https://nvbugs/6600098) kv_cache/test_kv_cache_v2_scheduler.py::TestKVCacheV2LoRA::test_lora_chunked_prefill SKIP (https://nvbugs/6640875) llmapi/test_llm_api_pytorch_bart.py::test_bart_pytorch_generate_encoder_decoder_end_to_end[bf16-kv-v1-cuda-graph-on-greedy-tp2-bart-large-cnn] SKIP (https://nvbugs/6463812) llmapi/test_llm_api_pytorch_moe_lora.py::test_mixtral_moe_routed_expert_fp8_multi_lora_varying_ranks[cudagraph] SKIP (https://nvbugs/6463829) diff --git a/tests/unittest/scripts/test_check_test_list.py b/tests/unittest/scripts/test_check_test_list.py new file mode 100644 index 000000000000..26ca1da1e9b4 --- /dev/null +++ b/tests/unittest/scripts/test_check_test_list.py @@ -0,0 +1,502 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Unit tests for scripts/check_test_list.py AST-based param-ID validation. + +These are pure-Python (no GPU, no built wheel) and are marked ``cpu_only`` so +they run on the CPU CI stage. +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.cpu_only + +REPO_ROOT = Path(__file__).resolve().parent.parent.parent.parent +SCRIPT_PATH = REPO_ROOT / "scripts" / "check_test_list.py" + + +@pytest.fixture(scope="module") +def mod(): + spec = importlib.util.spec_from_file_location("check_test_list", SCRIPT_PATH) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _index(mod, tmp_path, src: str, name: str = "mod_under_test.py"): + """Write ``src`` to a temp file and return build_ast_index's result.""" + p = tmp_path / name + p.write_text(src, encoding="utf-8") + return mod.build_ast_index(str(p)) + + +def _valid_ids(mod, tmp_path, src: str, func: str = "test_f"): + classes, _top, top_nodes, _cmn, consts = _index(mod, tmp_path, src) + return mod._compute_valid_param_ids(top_nodes[func], None, consts) + + +# -------------------------------------------------------------------------- +# _compute_valid_param_ids / _get_parametrize_ids +# -------------------------------------------------------------------------- + + +def test_literal_list_ids(mod, tmp_path): + src = 'import pytest\n@pytest.mark.parametrize("x", ["a", "b"])\ndef test_f(x): pass\n' + assert _valid_ids(mod, tmp_path, src) == {"a", "b"} + + +def test_explicit_ids_kwarg(mod, tmp_path): + # Non-literal argvalues, but explicit string ids= make it resolvable. + src = ( + "import pytest\n" + '@pytest.mark.parametrize("x", [object(), object()], ' + 'ids=["a", "b"])\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) == {"a", "b"} + + +def test_callable_ids_kwarg_punts(mod, tmp_path): + src = ( + "import pytest\n" + '@pytest.mark.parametrize("x", [1, 2], ids=lambda v: str(v))\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) is None + + +def test_pytest_param_id(mod, tmp_path): + src = ( + "import pytest\n" + '@pytest.mark.parametrize("a,b", ' + '[pytest.param(1, 2, id="p"), pytest.param(3, 4, id="q")])\n' + "def test_f(a, b): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) == {"p", "q"} + + +def test_multi_arg_tuple_rows(mod, tmp_path): + # pytest joins per-arg scalar IDs with "-". + src = ( + "import pytest\n" + '@pytest.mark.parametrize("a,b", [(1, "x"), (2, "y")])\n' + "def test_f(a, b): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) == {"1-x", "2-y"} + + +def test_single_argname_tuple_value_punts(mod, tmp_path): + # A tuple bound to ONE argname gets a positional "x0" ID we cannot + # reproduce statically — must punt, not guess. + src = 'import pytest\n@pytest.mark.parametrize("x", [(1, 2), (3, 4)])\ndef test_f(x): pass\n' + assert _valid_ids(mod, tmp_path, src) is None + + +def test_non_scalar_tuple_element_punts(mod, tmp_path): + src = ( + 'import pytest\n@pytest.mark.parametrize("a,b", [(1, object())])\ndef test_f(a, b): pass\n' + ) + assert _valid_ids(mod, tmp_path, src) is None + + +def test_id_collision_punts(mod, tmp_path): + # pytest would suffix the duplicate as "a0"/"a1"; we cannot, so punt. + src = 'import pytest\n@pytest.mark.parametrize("x", ["a", "a"])\ndef test_f(x): pass\n' + assert _valid_ids(mod, tmp_path, src) is None + + +def test_stacked_decorators_product(mod, tmp_path): + src = ( + "import pytest\n" + '@pytest.mark.parametrize("x", ["a", "b"])\n' + '@pytest.mark.parametrize("y", ["c", "d"])\n' + "def test_f(x, y): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) == {"c-a", "c-b", "d-a", "d-b"} + + +def test_no_parametrize_returns_empty(mod, tmp_path): + src = "def test_f(): pass\n" + assert _valid_ids(mod, tmp_path, src) == set() + + +# -------------------------------------------------------------------------- +# module-level constant resolution (build_ast_index + Name following) +# -------------------------------------------------------------------------- + + +def test_name_argvalues_resolved(mod, tmp_path): + src = ( + "import pytest\n" + 'PARAMS = ["a", "b"]\n' + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) == {"a", "b"} + + +def test_name_transitive_resolved(mod, tmp_path): + src = ( + "import pytest\n" + 'BASE = ["a", "b"]\n' + "PARAMS = BASE\n" + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) == {"a", "b"} + + +def test_reassigned_name_dropped(mod, tmp_path): + # Assigned twice -> unsound to resolve -> punt. + src = ( + "import pytest\n" + 'PARAMS = ["a"]\n' + 'PARAMS = ["b"]\n' + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) is None + + +def test_augassigned_name_dropped(mod, tmp_path): + src = ( + "import pytest\n" + 'PARAMS = ["a"]\n' + 'PARAMS += ["b"]\n' + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) is None + + +def test_call_argvalues_punts(mod, tmp_path): + src = ( + "import pytest\n" + 'def gen(): return ["a"]\n' + '@pytest.mark.parametrize("x", gen())\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) is None + + +def test_build_ast_index_module_consts(mod, tmp_path): + src = "SINGLE = [1]\nDOUBLE = [1]\nDOUBLE = [2]\nAUG = [1]\nAUG += [2]\n" + _c, _t, _tn, _cmn, consts = _index(mod, tmp_path, src) + assert "SINGLE" in consts + assert "DOUBLE" not in consts + assert "AUG" not in consts + + +# -------------------------------------------------------------------------- +# validate_test_lists end-to-end (errors vs unverifiable buckets) +# -------------------------------------------------------------------------- + + +def _make_layout(tmp_path, source: str, list_lines: list[str]): + """Create a defs source file + a test-list dir; return (lists_dir, base).""" + base = tmp_path / "tests" / "integration" / "defs" + base.mkdir(parents=True) + (base / "test_sample.py").write_text(source, encoding="utf-8") + lists_dir = tmp_path / "lists" + lists_dir.mkdir() + (lists_dir / "l0_sample.txt").write_text("\n".join(list_lines) + "\n", encoding="utf-8") + return str(lists_dir), str(base) + + +def test_validate_flags_invalid_id(mod, tmp_path): + source = 'import pytest\n@pytest.mark.parametrize("x", ["a", "b"])\ndef test_f(x): pass\n' + lists_dir, base = _make_layout(tmp_path, source, ["test_sample.py::test_f[zzz]"]) + errors, unverifiable, accepted, rejected = mod.validate_test_lists(lists_dir, base) + assert any("INVALID PARAMETRIZE ID" in e for e in errors) + assert unverifiable == [] + assert accepted == [] + assert rejected == [("test_sample.py", None, "test_f", "zzz")] + + +def test_validate_accepts_valid_id(mod, tmp_path): + source = ( + "import pytest\n" + 'PARAMS = ["a", "b"]\n' + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + lists_dir, base = _make_layout(tmp_path, source, ["test_sample.py::test_f[a]"]) + errors, unverifiable, accepted, rejected = mod.validate_test_lists(lists_dir, base) + assert errors == [] + assert unverifiable == [] + assert accepted == [("test_sample.py", None, "test_f", "a")] + assert rejected == [] + + +def test_validate_reports_unverifiable(mod, tmp_path): + source = ( + "import pytest\n" + '@pytest.mark.parametrize("x", [1, 2], ids=lambda v: str(v))\n' + "def test_f(x): pass\n" + ) + lists_dir, base = _make_layout(tmp_path, source, ["test_sample.py::test_f[1]"]) + errors, unverifiable, accepted, rejected = mod.validate_test_lists(lists_dir, base) + assert errors == [] + assert len(unverifiable) == 1 + rel, cls, func, pid, reason, _refs = unverifiable[0] + assert func == "test_f" and pid == "1" + assert reason == "ids=callable-or-nonliteral" + # An unverifiable entry is neither accepted nor rejected -- it has no static + # verdict, so it must not enter the parity buckets. + assert accepted == [] + assert rejected == [] + + +def test_write_unverifiable_report(mod, tmp_path): + unverifiable = [ + ("dir/test_a.py", None, "test_f", "p1", "argvalues=call-result", ["lists/l0.txt:3"]), + ] + out = tmp_path / "report.txt" + mod.write_unverifiable_report(unverifiable, str(out)) + text = out.read_text(encoding="utf-8") + assert "argvalues=call-result" in text + assert "dir/test_a.py::test_f[p1]" in text + + +# -------------------------------------------------------------------------- +# validate <-> collection parity (compute_parity / load_collectable_entries) +# -------------------------------------------------------------------------- + + +def test_compute_parity_all_collectable(mod): + accepted = [("test_a.py", None, "test_f", "a"), ("test_a.py", "TestC", "test_g", "b")] + collectable = set(accepted) + false_confidence, false_alarm = mod.compute_parity(accepted, [], collectable) + assert false_confidence == [] + assert false_alarm == [] + + +def test_compute_parity_false_confidence(mod): + # Accepted by --validate but pytest cannot collect it -> gate-worthy. + accepted = [("test_a.py", None, "test_f", "a"), ("test_a.py", None, "test_f", "ghost")] + collectable = {("test_a.py", None, "test_f", "a")} + false_confidence, false_alarm = mod.compute_parity(accepted, [], collectable) + assert false_confidence == [("test_a.py", None, "test_f", "ghost")] + assert false_alarm == [] + + +def test_compute_parity_false_alarm(mod): + # Rejected by --validate but pytest does collect it -> resolver too strict. + rejected = [("test_a.py", None, "test_f", "real")] + collectable = {("test_a.py", None, "test_f", "real")} + false_confidence, false_alarm = mod.compute_parity([], rejected, collectable) + assert false_confidence == [] + assert false_alarm == [("test_a.py", None, "test_f", "real")] + + +def test_compute_parity_rejected_not_collectable_is_silent(mod): + # Rejected and genuinely not collectable -> validator was right; no parity + # finding in either bucket. + rejected = [("test_a.py", None, "test_f", "zzz")] + false_confidence, false_alarm = mod.compute_parity([], rejected, set()) + assert false_confidence == [] + assert false_alarm == [] + + +def test_load_collectable_entries_reads_both_lists(mod, tmp_path): + (tmp_path / "l0_test.txt").write_text( + "accuracy/test_x.py::TestA::test_f[a]\n" + "# a comment line\n" + "full:GH200/accuracy/test_x.py::test_g[b] TIMEOUT 90\n", + encoding="utf-8", + ) + (tmp_path / "qa_test.txt").write_text("accuracy/test_y.py::test_h[c]\n", encoding="utf-8") + collectable = mod.load_collectable_entries(str(tmp_path)) + assert collectable == { + ("accuracy/test_x.py", "TestA", "test_f", "a"), + # full:GH200/ hardware prefix and the trailing TIMEOUT marker are + # normalized away by parse_test_entry, matching the validator's tuples. + ("accuracy/test_x.py", None, "test_g", "b"), + ("accuracy/test_y.py", None, "test_h", "c"), + } + + +def test_load_collectable_entries_missing_returns_none(mod, tmp_path): + assert mod.load_collectable_entries(str(tmp_path)) is None + + +def test_verify_qa_test_lists_clears_stale_file(mod, tmp_path, monkeypatch): + """A pre-existing qa_test.txt must not leak stale IDs into parity. + + verify_qa_test_lists opens qa_test.txt in append mode, so without the + up-front removal a stale entry left by an earlier checkout or a retried run + would survive and be treated as collectable by compute_parity -- letting a + now-invalid static acceptance falsely pass the parity gate. Stub the runtime + pytest --co / shell calls, keep the file I/O real, and assert the stale line + is gone and only the current QA def list's entries remain. + """ + llm_src = tmp_path + qa_dir = llm_src / "tests" / "integration" / "test_lists" / "qa" + qa_dir.mkdir(parents=True) + qa_def = qa_dir / "l0_a10.txt" + qa_def.write_text("accuracy/test_new.py::test_current[a] TIMEOUT 90\n", encoding="utf-8") + # Stale artifact from an earlier run, in the append target. + (llm_src / "qa_test.txt").write_text("accuracy/test_old.py::test_stale[b]\n", encoding="utf-8") + + # Stub the runtime pytest --co and the rm/ls shell-outs; the real removal + # and append file I/O is what we exercise. + monkeypatch.setattr(mod.subprocess, "run", lambda *a, **k: None) + monkeypatch.setattr(mod.subprocess, "check_output", lambda *a, **k: f"{qa_def}\n".encode()) + + mod.verify_qa_test_lists(str(llm_src)) + + result = (llm_src / "qa_test.txt").read_text().splitlines() + assert "accuracy/test_old.py::test_stale[b]" not in result + assert result == ["accuracy/test_new.py::test_current[a]"] + + +# -------------------------------------------------------------------------- +# module-const soundness against in-place mutation / rebinding +# -------------------------------------------------------------------------- + + +def test_subscript_mutated_name_punts(mod, tmp_path): + # NAME[i] = ... mutates in place without rebinding; must not be resolved. + src = ( + "import pytest\n" + 'PARAMS = ["a", "b"]\n' + 'PARAMS[0] = "z"\n' + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) is None + + +def test_method_mutated_name_punts(mod, tmp_path): + # NAME.append(...) may mutate in place; conservatively unresolved. + src = ( + "import pytest\n" + 'PARAMS = ["a"]\n' + 'PARAMS.append("b")\n' + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) is None + + +def test_attribute_store_mutated_name_punts(mod, tmp_path): + src = ( + "import pytest\n" + "PARAMS = obj\n" + "PARAMS.attr = 1\n" + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) is None + + +def test_for_rebound_name_punts(mod, tmp_path): + src = ( + "import pytest\n" + 'PARAMS = ["a"]\n' + "for PARAMS in other:\n pass\n" + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) is None + + +def test_function_rebound_name_punts(mod, tmp_path): + src = ( + "import pytest\n" + 'PARAMS = ["a"]\n' + "def PARAMS(): pass\n" + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) is None + + +def test_class_rebound_name_punts(mod, tmp_path): + src = ( + "import pytest\n" + 'PARAMS = ["a"]\n' + "class PARAMS: pass\n" + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) is None + + +def test_deleted_name_punts(mod, tmp_path): + src = ( + "import pytest\n" + 'PARAMS = ["a"]\n' + "del PARAMS\n" + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + assert _valid_ids(mod, tmp_path, src) is None + + +def test_mutated_or_rebound_names_helper(mod, tmp_path): + import ast + + src = ( + "import os\n" + "import numpy as np\n" + "A = [1]\n" + "B = [1]\nB[0] = 2\n" + "C = [1]\nC.append(2)\n" + "D = 0\nD += 1\n" + "for E in x:\n pass\n" + "with ctx() as F:\n pass\n" + "(G := 5)\n" + "def H(): pass\n" + "class I: pass\n" + "J = [1]\ndel J\n" + ) + unsafe = mod._mutated_or_rebound_names(ast.parse(src)) + # Rebound / mutated / imported / def / class / deleted names are unsafe; + # the clean const A is not. + assert {"B", "C", "D", "E", "F", "G", "H", "I", "J", "os", "np"} <= unsafe + assert "A" not in unsafe + + +def test_validate_mutated_name_is_unverifiable_not_error(mod, tmp_path): + # The soundness payoff: a mutated argvalues const must NOT produce a false + # INVALID PARAMETRIZE ID; it becomes unverifiable instead. + source = ( + "import pytest\n" + 'PARAMS = ["a"]\n' + 'PARAMS.append("b")\n' + '@pytest.mark.parametrize("x", PARAMS)\n' + "def test_f(x): pass\n" + ) + lists_dir, base = _make_layout(tmp_path, source, ["test_sample.py::test_f[b]"]) + errors, unverifiable, _accepted, rejected = mod.validate_test_lists(lists_dir, base) + assert errors == [] + assert rejected == [] + assert len(unverifiable) == 1 + assert unverifiable[0][:4] == ("test_sample.py", None, "test_f", "b") + + +def test_write_unverifiable_report_empty(mod, tmp_path): + # --report-unverifiable must produce a file even with an empty result, so + # automation can tell "empty" from "not run". + out = tmp_path / "report.txt" + mod.write_unverifiable_report([], str(out)) + assert out.exists() + assert "0 unverifiable" in out.read_text(encoding="utf-8")