diff --git a/tests/unittest/tools/test_test_to_stage_mapping.py b/tests/unittest/tools/test_test_to_stage_mapping.py index 18c982bab4ec..19fb16b7a11e 100644 --- a/tests/unittest/tools/test_test_to_stage_mapping.py +++ b/tests/unittest/tools/test_test_to_stage_mapping.py @@ -4,6 +4,7 @@ import subprocess import sys from collections import defaultdict +from types import SimpleNamespace import pytest @@ -25,6 +26,27 @@ MIN_PATTERN_LENGTH = 3 # Minimum length for search patterns +def _stage_backed_tests(stage_query: StageQuery) -> list[str]: + """Return tests from YAML files that are wired to a Jenkins stage.""" + return sorted(test for test, mappings in stage_query.test_map.items() + if all(yml in stage_query.yaml_to_stages + for yml, _stage, _backend in mappings)) + + +def test_stage_backed_tests_exclude_mixed_mappings() -> None: + """A test in any unwired YAML is not a live-stage sampling candidate.""" + stage_query = SimpleNamespace( + test_map={ + 'mixed': [('l0_wired.yml', 'pre_merge', 'pytorch'), + ('perf.yml', 'post_merge', 'pytorch')], + 'wired': [('l0_wired.yml', 'pre_merge', 'pytorch')], + }, + yaml_to_stages={'l0_wired.yml': ['L0-PyTorch']}, + ) + + assert _stage_backed_tests(stage_query) == ['wired'] + + @pytest.fixture(scope="module") def stage_query(): """Fixture that provides a StageQuery instance.""" @@ -32,30 +54,25 @@ def stage_query(): @pytest.fixture(scope="module") -def sample_test_cases(stage_query): - """Fixture that provides sample test cases from actual data.""" - random.seed(0) # Ensure deterministic test results - all_tests = list(stage_query.test_map.keys()) +def sample_test_cases(stage_query: StageQuery) -> list[str]: + """Fixture that samples tests backed by a live Jenkins stage.""" + all_tests = _stage_backed_tests(stage_query) if not all_tests: raise RuntimeError( - "No tests found in test mapping. This indicates a configuration " - "issue - either the test database YAML files are missing/empty " - "or the StageQuery is not parsing them correctly. Please check " - "that the test database directory exists and contains valid YAML " - "files with test definitions.") + "No tests are backed by a live Jenkins stage. Check that the " + "Groovy stage map and test database reference the same YAML files.") # Return up to MAX_SAMPLES tests randomly selected if len(all_tests) <= MAX_SAMPLES: return all_tests - return random.sample(all_tests, MAX_SAMPLES) + return random.Random(0).sample(all_tests, MAX_SAMPLES) @pytest.fixture(scope="module") -def sample_stages(stage_query): +def sample_stages(stage_query: StageQuery) -> list[str]: """Fixture that provides sample stages from actual data.""" - random.seed(0) # Ensure deterministic test results - all_stages = list(stage_query.stage_to_yaml.keys()) + all_stages = sorted(stage_query.stage_to_yaml) if not all_stages: raise RuntimeError( "No stages found in stage mapping. This indicates a configuration " @@ -68,7 +85,7 @@ def sample_stages(stage_query): if len(all_stages) <= MAX_SAMPLES: return all_stages - return random.sample(all_stages, MAX_SAMPLES) + return random.Random(0).sample(all_stages, MAX_SAMPLES) def test_data_availability(stage_query): @@ -82,6 +99,19 @@ def test_data_availability(stage_query): print(f"Max samples configured: {MAX_SAMPLES}") +def test_all_stage_backed_tests_map(stage_query: StageQuery) -> None: + """Every test in a Jenkins-wired YAML must resolve to a live stage.""" + stage_backed_tests = _stage_backed_tests(stage_query) + assert stage_backed_tests, "No tests are backed by a live Jenkins stage" + + unmapped = [ + test for test in stage_backed_tests + if not stage_query.tests_to_stages([test]) + ] + assert not unmapped, \ + f"Stage-backed tests should map to at least one stage: {unmapped}" + + def test_documented_stage_examples_are_live(stage_query): """Documented --stages examples must name stages that still exist in CI.""" sources = [ @@ -158,21 +188,17 @@ def test_known_stage_without_tests_is_reported(tmp_path): assert 'no tests mapped to: Empty-PyTorch-1' in proc.stderr.decode() -@pytest.mark.skip(reason="https://nvbugs/5547275") @pytest.mark.parametrize("direction", ["test_to_stage", "stage_to_test", "roundtrip"]) def test_bidirectional_mapping_consistency(stage_query, sample_test_cases, sample_stages, direction): """Test mapping consistency in both directions with roundtrip validation.""" - if direction == "test_to_stage": if not sample_test_cases: pytest.skip("No test cases available") for test_case in sample_test_cases: stages = stage_query.tests_to_stages([test_case]) - assert stages, \ - f"Test '{test_case}' should map to at least one stage" # Verify all returned stages are valid for stage in stages: