diff --git a/.github/workflows/_npu-analyze-failure.yml b/.github/workflows/_npu-analyze-failure.yml new file mode 100644 index 000000000..f801f4557 --- /dev/null +++ b/.github/workflows/_npu-analyze-failure.yml @@ -0,0 +1,96 @@ +# Reusable workflow: cross-reference CI test failures with coverage-based +# test recommendations. Called by pr-test-npu.yml's analyze-failure-report job. +name: Analyze Failure Report + +on: + workflow_call: + inputs: + log_artifact_pattern: + type: string + required: true + description: 'Glob pattern for test log artifacts (e.g. selected-test-logs-*)' + recommendations_content: + type: string + required: false + default: '' + description: 'Content of recommendations from upstream job output' + +defaults: + run: + shell: bash -el {0} + +jobs: + analyze: + name: cross-reference failures with recommendations + continue-on-error: true # Non-blocking on parse/read failure. + runs-on: ubuntu-latest + steps: + - name: Checkout sglang repo + uses: actions/checkout@v4 + + # ----- Step 1: Locate recommendations ----- + - name: Locate recommendations file + id: locate-recs + run: | + mkdir -p ./test-logs + + if [ -n "${{ inputs.recommendations_content }}" ]; then + echo "has_recommendations=true" >> $GITHUB_OUTPUT + echo "Using workflow output recommendations" + echo "${{ inputs.recommendations_content }}" > ./test-logs/_recommended.txt + echo "RECS_FILE=./test-logs/_recommended.txt" >> $GITHUB_ENV + echo "RECS_SOURCE=output" >> $GITHUB_ENV + COUNT=$(wc -l < ./test-logs/_recommended.txt) + echo "Recommendations count: ${COUNT}" + else + echo "has_recommendations=false" >> $GITHUB_OUTPUT + echo "::notice::No recommended test cases detected, skipping subsequent analysis tasks." + fi + + # ----- Step 2: Download test logs ----- + - name: Download test logs + uses: actions/download-artifact@v7 + with: + pattern: ${{ inputs.log_artifact_pattern }} + path: ./test-logs + merge-multiple: true + continue-on-error: true + + - name: Show downloaded logs structure + run: | + echo "Downloaded log files:" + find ./test-logs -type f | head -50 || echo " (no logs found)" + + # ----- Step 3: Run analysis ----- + - name: Ensure regex is available + if: steps.locate-recs.outputs.has_recommendations != 'false' + run: | + set -euo pipefail + if ! python3 -c "import regex" >/dev/null 2>&1; then + echo "regex not found for $(python3 -V); bootstrapping pip and installing regex" + curl -fsSL https://bootstrap.pypa.io/get-pip.py -o get-pip.py + python3 get-pip.py --break-system-packages + python3 -m pip install --break-system-packages regex + fi + python3 -c "import regex; print('regex ok:', regex.__file__)" + + - name: Run failure analysis + if: steps.locate-recs.outputs.has_recommendations != 'false' + id: analysis + continue-on-error: true + run: | + python3 scripts/ci/npu/precise-test/analyze_failure_report.py \ + --log-dir ./test-logs \ + --recommendations-file "${RECS_FILE}" \ + --recommendations-source "${RECS_SOURCE:-none}" \ + --output ./test-logs/failure_report.md + + # ----- Step 4: Upload report ----- + - name: Upload failure analysis report + if: always() + uses: actions/upload-artifact@v7 + with: + name: failure-analysis-report + path: ./test-logs/failure_report.md + if-no-files-found: ignore + retention-days: 14 diff --git a/.github/workflows/_npu-pr-test-stage.yml b/.github/workflows/_npu-pr-test-stage.yml index 42cd4b0e4..81ec0903a 100644 --- a/.github/workflows/_npu-pr-test-stage.yml +++ b/.github/workflows/_npu-pr-test-stage.yml @@ -1,5 +1,7 @@ name: PR Test Stage for NPU -# Reusable workflow for one CUDA test stage. Caller pr-test-npu.yml forwards +# Reusable workflow for one NPU test stage. Caller pr-test-npu.yml forwards +# job parameters (runner, image, partitions) and optionally delegates test +# execution to the coverage runner (use_coverage_runner=true). on: workflow_call: @@ -33,17 +35,25 @@ on: type: string default: '{"size":1,"arr":[0]}' ref: - description: 'Git ref (branch, tag, or SHA) to test. If not provided, uses the default branch.' + description: 'Git ref (branch, tag, or SHA) to test. If not provided, uses the event commit SHA.' type: string default: '' skip_pr_test_health_check: - description: 'Git ref (branch, tag, or SHA) to test. If not provided, uses the default branch.' + description: 'Set to true to skip the PR test health check (fast-fail gate).' type: string default: 'false' is_nightly_pipeline_job: description: 'Run the test suite with --nightly (collects nightly-registered tests) and --continue-on-error' type: boolean default: false + use_coverage_runner: + description: 'If true, delegate test execution to scripts/ci/npu/precise-test/run_tests_with_coverage.sh instead of running via run_suite.py directly' + type: boolean + default: false + upload_test_logs: + description: 'If true, upload the test logs as a GitHub Actions artifact (only enabled by pr-test-npu)' + type: boolean + default: false github-token: description: 'GitHub token for API calls' type: string @@ -72,7 +82,9 @@ jobs: - name: Checkout code uses: actions/checkout@v4 with: - ref: ${{ inputs.ref || github.ref }} + # Pin to the event SHA (not the branch name) so every job in this run + # checks out the same commit even if the branch advances mid-run. + ref: ${{ inputs.ref || github.sha }} - name: Mark repository safe run: | @@ -228,10 +240,12 @@ jobs: timeout-minutes: ${{ fromJson(inputs.run_timeout_minutes) }} env: CONTINUE_ON_ERROR_FLAG: ${{ inputs.continue_on_error == 'true' && '--continue-on-error' || '' }} + TEST_LOG_DIR: /tmp/test-logs shell: bash run: | # Fail fast on any command error, undefined variable, or pipe failure. set -euo pipefail + mkdir -p "${TEST_LOG_DIR}" # Install missing python deps (skip if already importable). PYTHON_FOR_SGLANG="python" @@ -266,8 +280,38 @@ jobs: if [ "${{ inputs.is_nightly_pipeline_job }}" = "true" ]; then NIGHTLY_FLAG="--nightly --continue-on-error" fi + + # Optional: delegate execution to scripts/ci/npu/precise-test/run_tests_with_coverage.sh when the + # caller sets use_coverage_runner=true. list_tests.py (same directory) enumerates the + # selected test files; the original run_suite.py execution below is preserved + # unchanged for the default (false) path. + LOG_FILE="${TEST_LOG_DIR}/${{ inputs.self_name }}-part${{ matrix.partition }}.log" + + if [[ "${{ inputs.use_coverage_runner }}" == "true" ]]; then + pip install coverage==7.8.* + python3 ../scripts/ci/npu/precise-test/list_tests.py --hw npu --suite ${{ inputs.self_name }} \ + --auto-partition-id ${{ matrix.partition }} \ + --auto-partition-size ${{ fromJson(inputs.partitions).size }} \ + -o /tmp/selected_tests.txt + echo "Selected test files:" + cat /tmp/selected_tests.txt + if [ -s /tmp/selected_tests.txt ]; then + bash ../scripts/ci/npu/precise-test/run_tests_with_coverage.sh $(cat /tmp/selected_tests.txt) + fi + exit 0 + fi + python3 run_suite.py --hw npu --suite ${{ inputs.self_name }} \ --auto-partition-id ${{ matrix.partition }} \ --auto-partition-size ${{ fromJson(inputs.partitions).size }} \ ${{ inputs.timeout_per_file && format('--timeout-per-file {0}', inputs.timeout_per_file) || '' }} \ - $NIGHTLY_FLAG $CONTINUE_ON_ERROR_FLAG + $NIGHTLY_FLAG $CONTINUE_ON_ERROR_FLAG 2>&1 | tee "${LOG_FILE}" + + - name: Upload test logs + if: always() && inputs.use_coverage_runner != true && inputs.upload_test_logs == true + uses: actions/upload-artifact@v7 + with: + name: selected-test-logs-${{ inputs.self_name }}-part${{ matrix.partition }} + path: /tmp/test-logs/ + if-no-files-found: ignore + retention-days: 14 diff --git a/.github/workflows/_npu-single-node-test-stage.yml b/.github/workflows/_npu-single-node-test-stage.yml index 1b04aec65..2a6b434e3 100644 --- a/.github/workflows/_npu-single-node-test-stage.yml +++ b/.github/workflows/_npu-single-node-test-stage.yml @@ -43,7 +43,7 @@ on: default: '{}' description: 'JSON run metadata {branch_label, workflow_name, create_time}, recorded once at workflow start' skip_pr_test_health_check: - description: 'Git ref (branch, tag, or SHA) to test. If not provided, uses the default branch.' + description: 'Set to true to skip the PR test health check (fast-fail gate).' type: string default: 'false' is_nightly_pipeline_job: @@ -54,6 +54,10 @@ on: description: 'Partition config to parallelize a suite across jobs, e.g. {"size":3,"arr":[0,1,2]}. Each element of arr is a partition id; the template expands one matrix job per element. size is passed to run_suite.py --auto-partition-size.' type: string default: '{"size":1,"arr":[0]}' + ref: + description: 'Git ref (branch, tag, or SHA) to test. If not provided, uses the event SHA.' + type: string + default: '' github-token: description: 'GitHub token for API calls' type: string @@ -63,6 +67,14 @@ on: type: string default: '300' description: 'timeout-minutes for the Run test step' + use_coverage_runner: + description: 'If true, delegate test execution to scripts/ci/npu/precise-test/run_tests_with_coverage.sh instead of running via run_suite.py directly' + type: boolean + default: false + upload_test_logs: + description: 'If true, upload the test logs as a GitHub Actions artifact (only enabled by pr-test-npu)' + type: boolean + default: false env: SKIP_PR_TEST_HEALTH_CHECK: ${{ inputs.skip_pr_test_health_check }} @@ -84,6 +96,10 @@ jobs: steps: - name: Checkout code uses: actions/checkout@v4 + with: + # Pin to the event SHA (not the branch name) so every job in this run + # checks out the same commit even if the branch advances mid-run. + ref: ${{ inputs.ref || github.sha }} - name: Check PR test health uses: actions/github-script@v8 @@ -227,6 +243,9 @@ jobs: # Fail fast on any command error, undefined variable, or pipe failure. set -euo pipefail + TEST_LOG_DIR="/tmp/test-logs" + mkdir -p "${TEST_LOG_DIR}" + sglang_source_path=$(pwd) echo "Source code path: ${sglang_source_path}" ln -sf ${sglang_source_path} /root/sglang @@ -390,14 +409,29 @@ jobs: fi # Nightly additionally persists the full suite log to ${log_path}/${tc_name}.log - # under the structured path; PR runs only tee to /tmp/test_output.log. + # under the structured path; PR runs tee to ${TEST_LOG_DIR}/${tc_name}.log. # Use an array so the extra target is omitted when empty: passing an empty # string arg makes GNU tee fail (exit code 1) and flips the pipeline exit code. - LOG_TEE_TARGETS=("/tmp/test_output.log") + LOG_TEE_TARGETS=("${TEST_LOG_DIR}/${tc_name}.log") if [ "${{ inputs.is_nightly_pipeline_job }}" = "true" ]; then LOG_TEE_TARGETS+=("${log_path}/${tc_name}.log") fi + # Optional: delegate execution to scripts/ci/npu/precise-test/run_tests_with_coverage.sh when the + # caller sets use_coverage_runner=true. list_tests.py (same directory) enumerates the + # selected test files; the original run_suite.py execution below is preserved + # unchanged for the default (false) path. + if [[ "${{ inputs.use_coverage_runner }}" == "true" ]]; then + pip install coverage==7.8.* + ${PYTHON_FOR_SGLANG} -u ../scripts/ci/npu/precise-test/list_tests.py --hw npu --suite ${test_suite} ${PARTITION_ARGS} -o /tmp/selected_tests.txt + echo "Selected test files:" + cat /tmp/selected_tests.txt + if [ -s /tmp/selected_tests.txt ]; then + bash ../scripts/ci/npu/precise-test/run_tests_with_coverage.sh $(cat /tmp/selected_tests.txt) + fi + exit 0 + fi + # Run NPU test suite with run_suite.py. # Capture mode (--enable-retry --max-attempts 1) keeps test subprocesses off the # tee pipe, so leftover server processes cannot block the pipeline on exit. @@ -418,11 +452,11 @@ jobs: echo "## ${tc_name} ${status_icon} ${test_status}" >> $GITHUB_STEP_SUMMARY echo "" >> $GITHUB_STEP_SUMMARY if [ "${{ inputs.is_nightly_pipeline_job }}" != "true" ]; then - metric_count=$(grep -c '\[METRIC\]' /tmp/test_output.log 2>/dev/null || echo 0) + metric_count=$(grep -c '\[METRIC\]' "${TEST_LOG_DIR}/${tc_name}.log" 2>/dev/null || echo 0) if [ "${metric_count}" -gt 0 ]; then echo "| Metric | Value | Pass |" >> $GITHUB_STEP_SUMMARY echo "|--------|-------|------|" >> $GITHUB_STEP_SUMMARY - grep '\[METRIC\]' /tmp/test_output.log | while IFS= read -r line; do + grep '\[METRIC\]' "${TEST_LOG_DIR}/${tc_name}.log" | while IFS= read -r line; do metric_name=$(echo "$line" | sed -E 's/.*\[METRIC\] ([^=]+)=.*/\1/') metric_value=$(echo "$line" | sed -E 's/.*\[METRIC\] [^=]+=([^ ]+).*/\1/') echo "| ${metric_name} | ${metric_value} | ${status_icon} |" >> $GITHUB_STEP_SUMMARY @@ -433,3 +467,12 @@ jobs: echo "" >> $GITHUB_STEP_SUMMARY fi exit ${test_exit_code} + + - name: Upload test logs + if: always() && inputs.use_coverage_runner != true && inputs.upload_test_logs == true + uses: actions/upload-artifact@v7 + with: + name: selected-test-logs-${{ inputs.test_suite }}-part${{ matrix.partition }} + path: /tmp/test-logs/ + if-no-files-found: ignore + retention-days: 14 diff --git a/.github/workflows/coverage-collection-npu.yml b/.github/workflows/coverage-collection-npu.yml new file mode 100644 index 000000000..52ff15cd4 --- /dev/null +++ b/.github/workflows/coverage-collection-npu.yml @@ -0,0 +1,27 @@ +# Coverage data collection pipeline for the NPU precision test selection. +# Delegates to pr-test-npu.yml with coverage_mode=true, which switches: +# - Runners to the coverage pool (shared /root/.cache with setup-covstub) +# - use_coverage_runner=true on all test jobs +# - Skips pr-gate, base-a (910b), multimodal-gen, recommend, analyze +# - Runs setup-covstub to assemble coverage data and publish it to the +# shared disk at a fixed path (/root/.cache/tests/precise-test/coverage) +# pr-test-npu-finish is the final gate in the called workflow. +# +# Manually triggered only (workflow_dispatch): PRs run the standard +# pr-test-npu.yml instead; this pipeline builds the full coverage baseline +# on demand. +name: Coverage Collection (NPU) + +on: + workflow_dispatch: + +concurrency: + group: npu-coverage-collection-${{ github.ref }} + cancel-in-progress: true + +jobs: + coverage: + uses: ./.github/workflows/pr-test-npu.yml + with: + coverage_mode: true + secrets: inherit diff --git a/.github/workflows/pr-test-npu.yml b/.github/workflows/pr-test-npu.yml index c676a973a..fcf079a03 100644 --- a/.github/workflows/pr-test-npu.yml +++ b/.github/workflows/pr-test-npu.yml @@ -7,10 +7,16 @@ on: - cron: '0 12 * * *' # Run daily at 12:00 UTC pull_request: workflow_dispatch: + inputs: + coverage_mode: + description: 'Run in coverage collection mode (uses coverage runners, enables coverage instrumentation, runs setup-covstub)' + required: false + type: boolean + default: false workflow_call: inputs: ref: - description: 'Git ref (branch, tag, or SHA) to test. If not provided, uses the default branch.' + description: 'Git ref (branch, tag, or SHA) to test. If not provided, uses the event commit SHA.' required: false type: string default: '' @@ -19,9 +25,14 @@ on: required: false type: boolean default: false + coverage_mode: + description: 'Run in coverage collection mode (uses coverage runners, enables coverage instrumentation, runs setup-covstub)' + required: false + type: boolean + default: false concurrency: - group: pr-test-npu-${{ inputs.ref || github.ref }} + group: pr-test-npu-${{ inputs.coverage_mode == true && 'coverage-' || '' }}${{ inputs.ref || github.ref }} cancel-in-progress: ${{ github.event_name != 'workflow_call' }} jobs: @@ -36,7 +47,9 @@ jobs: - name: Checkout code uses: actions/checkout@v4 with: - ref: ${{ inputs.ref || github.ref }} + # Pin to the event SHA (not the branch name) so every job in this run + # checks out the same commit even if the branch advances mid-run. + ref: ${{ inputs.ref || github.sha }} - name: Determine run mode id: run-mode @@ -77,7 +90,7 @@ jobs: # ==================== PR Gate ==================== # pr-gate: needs: check-changes - if: needs.check-changes.outputs.changes_exist == 'true' + if: ${{ inputs.coverage_mode != true && needs.check-changes.outputs.changes_exist == 'true' }} uses: ./.github/workflows/pr-gate.yml secrets: inherit @@ -96,7 +109,7 @@ jobs: base-a-test-1-npu-a2: needs: [check-changes, pr-gate, set-image-config] - if: ${{ !failure() && !cancelled() && needs.check-changes.outputs.main_package == 'true' }} + if: ${{ !failure() && !cancelled() && inputs.coverage_mode != true && needs.check-changes.outputs.main_package == 'true' }} uses: ./.github/workflows/_npu-pr-test-stage.yml with: self_name: base-a-test-1-npu-a2 @@ -108,68 +121,78 @@ jobs: base-b-test-1-npu-a3: needs: [check-changes, pr-gate, set-image-config, base-a-test-1-npu-a2] - if: ${{ !failure() && !cancelled() && needs.check-changes.outputs.main_package == 'true' }} + if: ${{ !failure() && !cancelled() && (inputs.coverage_mode == true || needs.check-changes.outputs.main_package == 'true') }} uses: ./.github/workflows/_npu-pr-test-stage.yml with: self_name: base-b-test-1-npu-a3 - runner_config: linux-aarch64-a3-2- + runner_config: ${{ inputs.coverage_mode == true && 'linux-aarch64-a3-800t-2' || 'linux-aarch64-a3-2-' }} image: ${{ needs.set-image-config.outputs.CANN_image_a3 }} run_timeout_minutes: '60' timeout_per_file: '3600' + use_coverage_runner: ${{ inputs.coverage_mode == true }} + upload_test_logs: ${{ inputs.coverage_mode != true }} secrets: inherit base-b-test-2-npu-a3: needs: [check-changes, pr-gate, set-image-config, base-a-test-1-npu-a2] - if: ${{ !failure() && !cancelled() && needs.check-changes.outputs.main_package == 'true' }} + if: ${{ !failure() && !cancelled() && (inputs.coverage_mode == true || needs.check-changes.outputs.main_package == 'true') }} uses: ./.github/workflows/_npu-pr-test-stage.yml with: self_name: base-b-test-2-npu-a3 - runner_config: linux-aarch64-a3-2- + runner_config: ${{ inputs.coverage_mode == true && 'linux-aarch64-a3-800t-2' || 'linux-aarch64-a3-2-' }} image: ${{ needs.set-image-config.outputs.CANN_image_a3 }} run_timeout_minutes: '60' timeout_per_file: '3600' + use_coverage_runner: ${{ inputs.coverage_mode == true }} + upload_test_logs: ${{ inputs.coverage_mode != true }} secrets: inherit base-b-test-4-npu-a3: needs: [check-changes, pr-gate, set-image-config, base-a-test-1-npu-a2] - if: ${{ !failure() && !cancelled() && needs.check-changes.outputs.main_package == 'true' }} + if: ${{ !failure() && !cancelled() && (inputs.coverage_mode == true || needs.check-changes.outputs.main_package == 'true') }} uses: ./.github/workflows/_npu-pr-test-stage.yml with: self_name: base-b-test-4-npu-a3 - runner_config: linux-aarch64-a3-4- + runner_config: ${{ inputs.coverage_mode == true && 'linux-aarch64-a3-800t-4' || 'linux-aarch64-a3-4-' }} image: ${{ needs.set-image-config.outputs.CANN_image_a3 }} run_timeout_minutes: '120' timeout_per_file: '3600' partitions: '{"size":2,"arr":[0, 1]}' + use_coverage_runner: ${{ inputs.coverage_mode == true }} + upload_test_logs: ${{ inputs.coverage_mode != true }} secrets: inherit base-b-test-8-npu-a3: needs: [check-changes, pr-gate, set-image-config, base-a-test-1-npu-a2] - if: ${{ !failure() && !cancelled() && needs.check-changes.outputs.main_package == 'true' }} + if: ${{ !failure() && !cancelled() && (inputs.coverage_mode == true || needs.check-changes.outputs.main_package == 'true') }} uses: ./.github/workflows/_npu-pr-test-stage.yml with: self_name: base-b-test-8-npu-a3 - runner_config: linux-aarch64-a3-8- + runner_config: ${{ inputs.coverage_mode == true && 'linux-aarch64-a3-800t-8' || 'linux-aarch64-a3-8-' }} image: ${{ needs.set-image-config.outputs.CANN_image_a3 }} run_timeout_minutes: '60' timeout_per_file: '3600' + use_coverage_runner: ${{ inputs.coverage_mode == true }} + upload_test_logs: ${{ inputs.coverage_mode != true }} secrets: inherit base-b-test-16-npu-a3: needs: [check-changes, pr-gate, set-image-config, base-a-test-1-npu-a2] - if: ${{ !failure() && !cancelled() && needs.check-changes.outputs.main_package == 'true' }} + if: ${{ !failure() && !cancelled() && (inputs.coverage_mode == true || needs.check-changes.outputs.main_package == 'true') }} uses: ./.github/workflows/_npu-pr-test-stage.yml with: self_name: base-b-test-16-npu-a3 - runner_config: linux-aarch64-a3-16- + runner_config: ${{ inputs.coverage_mode == true && 'linux-aarch64-a3-800t-16' || 'linux-aarch64-a3-16-' }} image: ${{ needs.set-image-config.outputs.CANN_image_a3 }} run_timeout_minutes: '120' timeout_per_file: '3600' + use_coverage_runner: ${{ inputs.coverage_mode == true }} + upload_test_logs: ${{ inputs.coverage_mode != true }} secrets: inherit multimodal-gen-test-1-npu-a3: needs: [check-changes, pr-gate, set-image-config, base-a-test-1-npu-a2] - if: ${{ !failure() && !cancelled() && needs.check-changes.outputs.multimodal_gen == 'true' }} + if: ${{ !failure() && !cancelled() && inputs.coverage_mode != true && needs.check-changes.outputs.multimodal_gen == 'true' }} runs-on: linux-aarch64-a3-800t-2 strategy: fail-fast: false @@ -233,7 +256,7 @@ jobs: multimodal-gen-test-4-npu-a3: needs: [check-changes, pr-gate, set-image-config, base-a-test-1-npu-a2] - if: ${{ !failure() && !cancelled() && needs.check-changes.outputs.multimodal_gen == 'true' }} + if: ${{ !failure() && !cancelled() && inputs.coverage_mode != true && needs.check-changes.outputs.multimodal_gen == 'true' }} runs-on: linux-aarch64-a3-800t-4 strategy: fail-fast: false @@ -298,34 +321,38 @@ jobs: base-c-test-acc-2-npu-a3: name: base-c-test-acc-2-npu-a3 needs: [ check-changes, pr-gate, set-image-config, base-b-test-1-npu-a3, base-b-test-2-npu-a3, base-b-test-4-npu-a3, base-b-test-8-npu-a3, base-b-test-16-npu-a3 ] - if: ${{ !failure() && !cancelled() && needs.check-changes.outputs.main_package == 'true' }} + if: ${{ !failure() && !cancelled() && (inputs.coverage_mode == true || needs.check-changes.outputs.main_package == 'true') }} uses: ./.github/workflows/_npu-single-node-test-stage.yml with: - runner: linux-aarch64-a3-2- + runner: ${{ inputs.coverage_mode == true && 'linux-aarch64-a3-800t-2' || 'linux-aarch64-a3-2-' }} test_type: 'accuracy' test_suite: base-c-test-acc-2-npu-a3 image: ${{ needs.set-image-config.outputs.CANN_image_a3 }} install_sglang_deps: true device_type_for_deps: 'a3' partitions: '{"size":2,"arr":[0,1]}' + use_coverage_runner: ${{ inputs.coverage_mode == true }} + upload_test_logs: ${{ inputs.coverage_mode != true }} base-c-test-acc-16-npu-a3: name: base-c-test-acc-16-npu-a3 needs: [ check-changes, pr-gate, set-image-config, base-b-test-1-npu-a3, base-b-test-2-npu-a3, base-b-test-4-npu-a3, base-b-test-8-npu-a3, base-b-test-16-npu-a3 ] - if: ${{ !failure() && !cancelled() && needs.check-changes.outputs.main_package == 'true' }} + if: ${{ !failure() && !cancelled() && (inputs.coverage_mode == true || needs.check-changes.outputs.main_package == 'true') }} uses: ./.github/workflows/_npu-single-node-test-stage.yml with: - runner: linux-aarch64-a3-16- + runner: ${{ inputs.coverage_mode == true && 'linux-aarch64-a3-800t-16' || 'linux-aarch64-a3-16-' }} test_type: 'accuracy' test_suite: base-c-test-acc-16-npu-a3 image: ${{ needs.set-image-config.outputs.CANN_image_a3 }} install_sglang_deps: true device_type_for_deps: 'a3' + use_coverage_runner: ${{ inputs.coverage_mode == true }} + upload_test_logs: ${{ inputs.coverage_mode != true }} base-c-test-perf-2-npu-a3: name: base-c-test-perf-2-npu-a3 needs: [ check-changes, pr-gate, set-image-config, base-c-test-acc-2-npu-a3, base-c-test-acc-16-npu-a3 ] - if: ${{ !failure() && !cancelled() && needs.check-changes.outputs.main_package == 'true' }} + if: ${{ !failure() && !cancelled() && (inputs.coverage_mode == true || needs.check-changes.outputs.main_package == 'true') }} uses: ./.github/workflows/_npu-single-node-test-stage.yml with: runner: linux-aarch64-a3-800t-2 @@ -334,11 +361,13 @@ jobs: image: ${{ needs.set-image-config.outputs.CANN_image_a3 }} install_sglang_deps: true device_type_for_deps: 'a3' + use_coverage_runner: ${{ inputs.coverage_mode == true }} + upload_test_logs: ${{ inputs.coverage_mode != true }} base-c-test-perf-16-npu-a3: name: base-c-test-perf-16-npu-a3 needs: [ check-changes, pr-gate, set-image-config, base-c-test-acc-2-npu-a3, base-c-test-acc-16-npu-a3 ] - if: ${{ !failure() && !cancelled() && needs.check-changes.outputs.main_package == 'true' }} + if: ${{ !failure() && !cancelled() && (inputs.coverage_mode == true || needs.check-changes.outputs.main_package == 'true') }} uses: ./.github/workflows/_npu-single-node-test-stage.yml with: runner: linux-aarch64-a3-800t-16 @@ -348,8 +377,262 @@ jobs: install_sglang_deps: true device_type_for_deps: 'a3' test_timeout_minutes: '180' + use_coverage_runner: ${{ inputs.coverage_mode == true }} + upload_test_logs: ${{ inputs.coverage_mode != true }} + # ==================== Recommend Tests from Coverage ==================== # + # Recommends PR-affected test cases from the pre-built coverage baseline. + # continue-on-error so it never blocks the PR gate. + recommend-tests-from-coverage: + name: Recommend tests from coverage + runs-on: ubuntu-latest + continue-on-error: true + if: ${{ !cancelled() && github.event_name == 'pull_request' && inputs.coverage_mode != true }} + outputs: + coverage_paths: ${{ steps.export-paths.outputs.coverage_paths }} + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + steps: + - name: Checkout sglang scripts + uses: actions/checkout@v4 + with: + sparse-checkout: | + scripts + sparse-checkout-cone-mode: false + + - name: Download test case map + run: | + set -euo pipefail + curl -fsSL \ + --retry 3 \ + --retry-delay 5 \ + --retry-all-errors \ + "https://sglang-npu.obs.cn-southwest-2.myhuaweicloud.com/coverage/test_case_map.json" \ + -o test_case_map.json + test -s test_case_map.json + echo "Downloaded test_case_map.json" + du -h test_case_map.json + + - name: Download coverage package and extract covstub + run: | + set -euo pipefail + curl -fsSL \ + --retry 3 \ + --retry-delay 5 \ + --retry-all-errors \ + "https://sglang-npu.obs.cn-southwest-2.myhuaweicloud.com/coverage/outputs.tar.gz" \ + -o outputs.tar.gz + test -s outputs.tar.gz + echo "Downloaded outputs.tar.gz" + du -h outputs.tar.gz + tar xzf outputs.tar.gz outputs/covstub + test -d outputs/covstub + echo "Extracted outputs/covstub" + find outputs/covstub -maxdepth 2 -type d | head -20 || true + + - name: Recommend tests for current PR + run: | + set -euo pipefail + if ! python3 -c "import regex" >/dev/null 2>&1; then + echo "regex not found for $(python3 -V); bootstrapping pip and installing regex" + curl -fsSL https://bootstrap.pypa.io/get-pip.py -o get-pip.py + python3 get-pip.py --break-system-packages + python3 -m pip install --break-system-packages regex + fi + python3 -c "import regex; print('regex ok:', regex.__file__)" + PR_SPEC="${{ github.repository }}#${{ github.event.pull_request.number }}" + echo "Selecting tests for PR: ${PR_SPEC}" + # test_selector.py resolves relative paths against the script's own dir, + # so pass absolute paths for files downloaded to the workspace root. + python3 scripts/ci/npu/precise-test/test_selector.py \ + --github-pr "${PR_SPEC}" \ + --source-dir "$PWD/outputs/covstub" \ + --map-file "$PWD/test_case_map.json" + + - name: Export coverage paths output + id: export-paths + if: always() + run: | + FILE="scripts/ci/npu/precise-test/recommended_pytest_paths.txt" + if [ -f "$FILE" ]; then + { + echo "coverage_paths<> "$GITHUB_OUTPUT" + echo "Exported $(wc -l < "$FILE") recommended paths" + else + echo "coverage_paths=" >> "$GITHUB_OUTPUT" + echo "::notice::recommended_pytest_paths.txt not found" + fi + + # ==================== Analyze Failure Report ==================== # + # Cross-references test failures with coverage recommendations. + # !cancelled() only: skipped jobs in needs poison success(). + analyze-failure-report: + name: Analyze failure report + needs: + [ + base-a-test-1-npu-a2, + base-b-test-1-npu-a3, + base-b-test-2-npu-a3, + base-b-test-4-npu-a3, + base-b-test-8-npu-a3, + base-b-test-16-npu-a3, + + multimodal-gen-test-1-npu-a3, + multimodal-gen-test-4-npu-a3, + + base-c-test-acc-2-npu-a3, + base-c-test-acc-16-npu-a3, + base-c-test-perf-2-npu-a3, + base-c-test-perf-16-npu-a3, + + recommend-tests-from-coverage, + ] + if: ${{ !cancelled() && inputs.coverage_mode != true }} + uses: ./.github/workflows/_npu-analyze-failure.yml + with: + log_artifact_pattern: selected-test-logs-* + recommendations_content: ${{ needs.recommend-tests-from-coverage.outputs.coverage_paths || '' }} + + # ==================== Coverage Assembly & Publish ==================== # + # Assembles coverage data and publishes the baseline to the shared disk. + # !failure() && !cancelled() (not success()): skipped upstream poisons success(). + setup-covstub: + needs: + [ + set-image-config, + base-b-test-1-npu-a3, + base-b-test-2-npu-a3, + base-b-test-4-npu-a3, + base-b-test-8-npu-a3, + base-b-test-16-npu-a3, + base-c-test-acc-2-npu-a3, + base-c-test-acc-16-npu-a3, + base-c-test-perf-2-npu-a3, + base-c-test-perf-16-npu-a3, + ] + if: ${{ !failure() && !cancelled() && inputs.coverage_mode == true }} + runs-on: linux-aarch64-a3-800t-2 + container: + image: ${{ needs.set-image-config.outputs.CANN_image_a3 }} + steps: + - name: Checkout code + uses: actions/checkout@v4 + with: + # Pin to the event SHA (not the branch name) so every job in this run + # checks out the same commit even if the branch advances mid-run. + ref: ${{ inputs.ref || github.sha }} + + - name: Copy sglang source to covstub + run: | + RUN_DIR="/root/.cache/tests/precise-test/${{ github.run_id }}-attempt-${{ github.run_attempt }}" + COVSTUB="${RUN_DIR}/outputs/covstub" + mkdir -p "${COVSTUB}" + cp -r python/sglang "${COVSTUB}/" + + - name: Build test case map + shell: bash + run: | + set -euo pipefail + + RUN_DIR="/root/.cache/tests/precise-test/${{ github.run_id }}-attempt-${{ github.run_attempt }}" + # Date tag derivation must match scripts/ci/npu/precise-test/run_tests_with_coverage.sh. + # The custom k8s runner only injects explicitly-set env vars, so + # GITHUB_RUN_STARTED_AT may be absent (set -u would abort); fall back + # to local date exactly like run_tests_with_coverage.sh does, so both + # sides derive the same tag. + if [ -n "${GITHUB_RUN_STARTED_AT:-}" ]; then + COV_DATE_TAG="${GITHUB_RUN_STARTED_AT:0:10}" + COV_DATE_TAG="${COV_DATE_TAG//-/}" + else + COV_DATE_TAG="$(date +%Y%m%d)" + fi + COV_ROOT="${RUN_DIR}/outputs/sglang@${COV_DATE_TAG}" + COVSTUB="${RUN_DIR}/outputs/covstub" + # Placed outside outputs/ so it is NOT included in outputs.tar.gz. + MAP_FILE="${RUN_DIR}/test_case_map.json" + + echo "Coverage dir: ${COV_ROOT}" + echo "Source dir: ${COVSTUB}" + + # Coverage data may be absent when all test jobs were skipped + # (e.g. by the changes filter). Skip map building in that case. + if [ ! -d "${COV_ROOT}" ]; then + echo "::warning::Coverage dir not found: ${COV_ROOT}, skip building test case map" + exit 0 + fi + + COVERAGE_FILE_COUNT=$(find "${COV_ROOT}" -name 'coverage.*' -type f | wc -l) + echo "Found ${COVERAGE_FILE_COUNT} coverage files" + if [ "${COVERAGE_FILE_COUNT}" -eq 0 ]; then + echo "::warning::No coverage files found under ${COV_ROOT}, skip building test case map" + exit 0 + fi + + if ! python3 -c "import regex" 2>/dev/null; then + pip install regex \ + -i http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple \ + --trusted-host cache-service.nginx-pypi-cache.svc.cluster.local \ + --retries 3 --timeout 60 \ + || pip install regex -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple + fi + + python3 scripts/ci/npu/precise-test/test_selector.py \ + --build-map \ + --coverage-dir "${COV_ROOT}" \ + --source-dir "${COVSTUB}" \ + --map-file "${MAP_FILE}" + test -s "${MAP_FILE}" + du -h "${MAP_FILE}" + + - name: Archive outputs + run: | + RUN_DIR="/root/.cache/tests/precise-test/${{ github.run_id }}-attempt-${{ github.run_attempt }}" + tar -czf "${RUN_DIR}/outputs.tar.gz" -C "${RUN_DIR}" outputs + + - name: Publish outputs and map to shared disk + shell: bash + run: | + set -euo pipefail + + RUN_DIR="/root/.cache/tests/precise-test/${{ github.run_id }}-attempt-${{ github.run_attempt }}" + # Fixed publish dir with no RUN_DIR layer: other pipelines have a + # different GITHUB_RUN_ID and must be able to read the baseline. + # Must match PUBLISH_DIR in the recommend-tests-from-coverage job. + PUBLISH_DIR="/root/.cache/tests/precise-test/coverage" + + # Skip publish when no coverage data was collected this run (map + # build was skipped), so we never overwrite the historical + # baseline on the shared disk with an incomplete one. + if [ ! -s "${RUN_DIR}/test_case_map.json" ]; then + echo "::warning::test_case_map.json missing (no coverage data this run), skip publish" + exit 0 + fi + + mkdir -p "${PUBLISH_DIR}" + + # mv (rename) within the same shared filesystem is atomic, so a + # concurrent reader sees either the old or the new baseline, + # never a partial file. + mv "${RUN_DIR}/outputs.tar.gz" "${PUBLISH_DIR}/outputs.tar.gz" + mv "${RUN_DIR}/test_case_map.json" "${PUBLISH_DIR}/test_case_map.json" + + echo "Published baseline to ${PUBLISH_DIR}" + du -h "${PUBLISH_DIR}/outputs.tar.gz" "${PUBLISH_DIR}/test_case_map.json" + + - name: Clean up per-run directory + # Skipped on publish failure, keeping RUN_DIR for manual recovery. + shell: bash + run: | + set -euo pipefail + RUN_DIR="/root/.cache/tests/precise-test/${{ github.run_id }}-attempt-${{ github.run_attempt }}" + rm -rf "${RUN_DIR}" + echo "Removed ${RUN_DIR}" + pr-test-npu-finish: needs: [ @@ -369,6 +652,8 @@ jobs: base-c-test-acc-16-npu-a3, base-c-test-perf-2-npu-a3, base-c-test-perf-16-npu-a3, + + setup-covstub, ] if: always() runs-on: ubuntu-latest diff --git a/scripts/ci/npu/precise-test/analyze_failure_report.py b/scripts/ci/npu/precise-test/analyze_failure_report.py new file mode 100755 index 000000000..ab760c407 --- /dev/null +++ b/scripts/ci/npu/precise-test/analyze_failure_report.py @@ -0,0 +1,525 @@ +#!/usr/bin/env python3 +""" +analyze_failure_report.py + +Cross-reference CI test failures with test recommendations. + +Pipeline: + 1. Scan each log file for failed tests using three methods: + a. TIMINGS JSON block (machine-readable, from ci_utils.py) + b. ci_utils.py "✗ FAILED:" summary section (structured text) + c. pytest "short test summary info" block (for pytest-style logs) + 2. Read recommended_pytest_paths.txt + 3. Match: exact match + file-level match + 4. Generate a Markdown report + +Usage: + python analyze_failure_report.py --log-dir LOG_DIR --recommendations-file RECOMMENDED.txt [--output report.md] +""" + +import argparse +import contextlib +import json +import sys +from pathlib import Path + +import regex as re + +# ============================================================ +# Utility: strip CI log noise +# ============================================================ + + +def strip_ansi(text): + """Remove ANSI color codes like \x1b[31m, \x1b[0m, etc.""" + return re.sub(r"\x1b\[[0-9;]*m", "", text) + + +def strip_timestamp(line): + """Remove GitHub Actions timestamp prefix: YYYY-MM-DDTHH:MM:SS.fffffffZ""" + return re.sub(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z\s+", "", line) + + +def clean_line(line): + """Strip BOM, ANSI codes, and timestamp from one log line.""" + line = line.lstrip("\ufeff") # UTF-8 BOM marker + return strip_ansi(strip_timestamp(line)).strip() + + +# ============================================================ +# Step 1: Extract FAILED and ERROR tests from log files +# ============================================================ + + +# Match pytest-style FAILED/ERROR lines with the test/ prefix (sglang convention). +FAILED_PATTERN = re.compile(r"^(?:FAILED|ERROR)\s+(test/\S+?\.py(?:::\S+?)?)\s") +SUMMARY_SEPARATOR_PATTERN = re.compile(r"^=+\s") +CPU_LOG_PATH_PATTERN = re.compile(r"(?:^|-)cpu-\d+card(?:-|$)", re.IGNORECASE) +CPU_FAILURE_LABEL = "cpu-ut" + +# ci_utils.py summary: "✗ FAILED:" section lines like " /path/to/test/registered/test_xxx.py (exit code 1)". +# Paths are absolute (from os.path.abspath in run_suite.py's glob). +CI_UTILS_FAILED_PATTERN = re.compile(r"^[✗X]\s*FAILED:\s*$") +CI_UTILS_FAILED_LINE_PATTERN = re.compile(r"^\s{2,}(\S+\.py)\s*\(") + +# TIMINGS block: machine-readable JSON lines with "passed": false. +TIMINGS_BEGIN_PATTERN = re.compile(r"^=+\s*TIMINGS\s+BEGIN\s*=+") +TIMINGS_END_PATTERN = re.compile(r"^=+\s*TIMINGS\s+END\s*=+") + + +def _extract_from_timings(lines): + """Extract failed test file paths from the TIMINGS JSON block (ci_utils.py).""" + failed = [] + in_timings = False + for line in lines: + text = clean_line(line) + if TIMINGS_BEGIN_PATTERN.search(text): + in_timings = True + continue + if not in_timings: + continue + if TIMINGS_END_PATTERN.search(text): + break + try: + entry = json.loads(text) + if not entry.get("passed", True) and entry.get("file"): + failed.append(entry["file"]) + except (json.JSONDecodeError, ValueError): + continue + return failed + + +def _extract_from_ci_utils_summary(lines): + """Extract failed test file paths from ci_utils.py's '✗ FAILED:' summary section.""" + failed = [] + in_failed_section = False + for line in lines: + text = clean_line(line) + if CI_UTILS_FAILED_PATTERN.match(text): + in_failed_section = True + continue + if not in_failed_section: + continue + if SUMMARY_SEPARATOR_PATTERN.match(text): + break + match = CI_UTILS_FAILED_LINE_PATTERN.match(line) + if match: + # Paths in this section are absolute (e.g. "/__w/sglang/sglang/test/registered/test_xxx.py"). + # Strip everything up to and including the "/sglang/" marker to get + # the repo-relative form (e.g. "test/registered/test_xxx.py"). + path = match.group(1) + marker = "/sglang/" + idx = path.rfind(marker) + if idx >= 0: + path = path[idx + len(marker) :] + failed.append(path) + return failed + + +def extract_failed_from_log(log_path): + """Extract failed test paths from one log file. + + Tries three methods in order of reliability: + 1. TIMINGS JSON block (machine-readable, ci_utils.py) + 2. ci_utils.py '✗ FAILED:' summary section (human-readable but structured) + 3. pytest 'short test summary info' block (for pytest-style logs) + """ + try: + lines = log_path.read_text(encoding="utf-8", errors="replace").splitlines() + except Exception as exc: + print(f"::warning:: Cannot read {log_path}: {exc}") + return [] + + # Method 1: TIMINGS block (most reliable, machine-readable). + failed = _extract_from_timings(lines) + if failed: + return failed + + # Method 2: ci_utils.py summary section. + failed = _extract_from_ci_utils_summary(lines) + if failed: + return failed + + # Method 3: pytest-style "short test summary info" block. + failed = [] + in_summary = False + for line in lines: + text = clean_line(line) + + if "short test summary info" in text: + in_summary = True + continue + + if not in_summary: + continue + + if SUMMARY_SEPARATOR_PATTERN.match(text): + in_summary = False + continue + + match = FAILED_PATTERN.match(text) + if match: + failed.append(match.group(1)) + + return failed + + +def is_cpu_log(log_path): + """Return whether a log belongs to a CPU selected-test artifact.""" + if log_path.stem.lower().endswith("-cpu-ut"): + return True + + return any(CPU_LOG_PATH_PATTERN.search(part) for part in log_path.parent.parts) + + +def extract_failed_from_logs(log_dir): + """ + Scan CPU logs first and represent all CPU failures as one ``cpu-ut`` item. + Scan all remaining logs with the existing pytest node-ID behavior. + """ + base = Path(log_dir) + if not base.is_dir(): + print(f"::warning:: Log directory not found: {log_dir}") + return [] + + # Scan .log files (from NPU test stages) and .txt files (legacy/mock). + candidates = [] + candidates.extend(base.rglob("*.log")) + candidates.extend(base.rglob("*.txt")) + candidates = [ + candidate + for candidate in sorted(candidates) + if candidate.suffix != ".txt" or "run-selected-tests" in candidate.name + ] + + cpu_candidates = [] + regular_candidates = [] + for candidate in candidates: + target = cpu_candidates if is_cpu_log(candidate) else regular_candidates + target.append(candidate) + + all_failed = [] + seen = set() + + cpu_failed = False + for candidate in cpu_candidates: + if extract_failed_from_log(candidate): + cpu_failed = True + + if cpu_failed: + seen.add(CPU_FAILURE_LABEL) + all_failed.append(CPU_FAILURE_LABEL) + + for candidate in regular_candidates: + for test_path in extract_failed_from_log(candidate): + if test_path not in seen: + seen.add(test_path) + all_failed.append(test_path) + + return all_failed + + +# ============================================================ +# Step 2: Read recommendations +# ============================================================ + + +def read_recommended(recommendations_file): + """ + recommended_pytest_paths.txt contains one pytest path per line, e.g.: + test/ops/test_matmul.py::test_bf16 + test/layers/test_attention.py + """ + path = Path(recommendations_file) + if not path.exists(): + print(f"::warning:: Recommendations file not found: {recommendations_file}") + return [] + raw = path.read_text(encoding="utf-8").lstrip("\ufeff") + return [ + line.strip() + for line in raw.splitlines() + if line.strip() and not line.startswith("ERROR") + ] + + +# ============================================================ +# Step 3: Match +# ============================================================ + + +def normalize_test_path(test_path): + """Return a stable comparison key for a pytest path or node ID.""" + normalized = test_path.strip().replace("\\", "/").removeprefix("./") + file_path, separator, test_name = normalized.partition("::") + file_path = file_path.removesuffix(".py") + if separator: + test_name = test_name.partition("[")[0] + return f"{file_path}{separator}{test_name}" if separator else file_path + + +def match_failed_vs_recommended(failed, recommended): + """ + Two-level matching: + Level 1 - File-level: recommended "test/foo.py" (no function) + matches failed "test/foo.py::anything" + Level 2 - Exact: "test/foo.py::test_bar" in both lists + + Returns {"hit": [...], "miss": [...], "untested": [...]} + hit: failed AND recommended + miss: failed but NOT recommended + untested: recommended but NOT in failed list + """ + recommended_files = { + normalize_test_path(item) for item in recommended if "::" not in item + } + recommended_functions = { + normalize_test_path(item) for item in recommended if "::" in item + } + + hit = [] + miss = [] + + normalized_failed = {item: normalize_test_path(item) for item in failed} + for original, normalized in normalized_failed.items(): + failed_file = normalized.split("::", 1)[0] + if failed_file in recommended_files or normalized in recommended_functions: + hit.append(original) + else: + miss.append(original) + + # Recommended but not failed + failed_functions = set(normalized_failed.values()) + failed_files = {item.split("::", 1)[0] for item in failed_functions} + untested = [] + for item in recommended: + normalized = normalize_test_path(item) + has_failure = ( + normalized in failed_functions + if "::" in item + else normalized in failed_files + ) + if not has_failure: + untested.append(item) + + return {"hit": hit, "miss": miss, "untested": untested} + + +# ============================================================ +# Step 4: Generate Markdown report +# ============================================================ + + +def generate_report( + failed, recommended, matched, log_dir, recommendations_source="none" +): + """Produce a Markdown summary table.""" + hit = matched["hit"] + miss = matched["miss"] + untested = matched["untested"] + + out = [] + out.append("# Test Failure vs Recommendation Report") + out.append("") + out.append(f"**Log source**: `{log_dir}`") + out.append("") + + # Recommendation source indicator + if recommendations_source == "output": + out.append( + "> **[Source: Workflow Output]** Recommended cases are passed from coverage recommendations outputs" + ) + elif recommendations_source == "committed": + out.append( + "> **[Source: Local File]** Recommended test cases come from a txt file in the repository" + ) + else: + out.append("> **[Source: None]** No recommended test cases found") + out.append("") + + # ================================================================ + # Section 1: Full Failed Test List + # ================================================================ + out.append("---") + out.append("") + out.append(f"## Failed Test Cases（ {len(failed)} total）") + out.append("") + if failed: + for i, t in enumerate(failed, 1): + tag = ( + " **[Matched Recommendation]**" + if t in hit + else " **[Not Matched Recommendation]**" + ) + out.append(f"{i}. `{t}`{tag}") + out.append("") + else: + out.append("> No failed test cases") + out.append("") + + # ================================================================ + # Section 2: Full Recommended Test List + # ================================================================ + out.append("---") + out.append("") + out.append(f"## Recommended Test Cases（ {len(recommended)} total）") + out.append("") + if recommended: + normalized_failed = {normalize_test_path(item) for item in failed} + failed_file_set = {item.split("::", 1)[0] for item in normalized_failed} + for i, item in enumerate(recommended, 1): + normalized = normalize_test_path(item) + has_failure = ( + normalized in normalized_failed + if "::" in item + else normalized in failed_file_set + ) + tag = " **[Already Failed]**" if has_failure else "" + out.append(f"{i}. `{item}`{tag}") + out.append("") + else: + out.append("> No recommended test cases") + out.append("") + + # ================================================================ + # Section 3: Core Conclusion + # ================================================================ + out.append("---") + out.append("") + out.append("## Core Conclusion") + out.append("") + if not failed: + out.append( + "> No failed cases in this CI run; no need to compare against the recommendation list." + ) + elif len(miss) == 0: + out.append("> **All failed test cases are within the recommended scope.**") + else: + total_failed = len(failed) + out.append( + f"> ** {len(miss)}/{total_failed} failed cases are outside the recommended scope.**" + ) + out.append("") + + # ================================================================ + # Section 4: Detail table + # ================================================================ + out.append("| Category | Count |") + out.append("|---|---|") + out.append(f"| Failed & Matched Recommendation | {len(hit)} |") + out.append(f"| Failed but Not Matched Recommendation | {len(miss)} |") + out.append(f"| Recommended but Not Failed | {len(untested)} |") + out.append("") + + if hit: + out.append("## Failed & Matched Recommendation") + out.append("") + out.append("| # | Failed test |") + out.append("|---|---|") + for i, t in enumerate(hit, 1): + out.append(f"| {i} | `{t}` |") + out.append("") + + if miss: + out.append("## Failed but Not Matched Recommendation") + out.append("") + out.append( + "> Possible causes: uncovered modules, environment issues, flaky tests." + ) + out.append("") + for t in miss: + out.append(f"- `{t}`") + out.append("") + + if untested: + out.append("## Recommended but Not Failed") + out.append("") + out.append( + "> These test cases were recommended but did not fail this run (passed or not executed)." + ) + out.append("") + for t in untested: + out.append(f"- `{t}`") + out.append("") + + if not hit and not miss: + out.append("## No failed cases") + out.append("") + + out.append("---") + out.append("*Generated by analyze_failure_report.py*") + return "\n".join(out) + + +def main(): + parser = argparse.ArgumentParser( + description="Cross-reference CI test failures with test recommendations" + ) + parser.add_argument( + "--log-dir", required=True, help="Directory containing CI .log files" + ) + parser.add_argument( + "--recommendations-file", + help="Path to recommended_pytest_paths.txt", + ) + parser.add_argument( + "--output", + default="failure_report.md", + help="Output Markdown report path (default: failure_report.md)", + ) + parser.add_argument( + "--recommendations-source", + default="none", + choices=["committed", "output", "none"], + help="Where recommendations came from", + ) + args = parser.parse_args() + + if not args.recommendations_file: + parser.error("--recommendations-file is required") + + # For Windows console: force UTF-8 if possible + if sys.platform == "win32": + with contextlib.suppress(Exception): + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + + print("=" * 50) + print("Step 1: Extract failed tests from CI logs") + print("=" * 50) + failed = extract_failed_from_logs(args.log_dir) + print(f"Failed: {len(failed)}") + + print() + print("=" * 50) + print("Step 2: Read recommendations") + print("=" * 50) + recommended = read_recommended(args.recommendations_file) + print(f"Recommended: {len(recommended)}") + + print() + print("=" * 50) + print("Step 3: Match") + print("=" * 50) + matched = match_failed_vs_recommended(failed, recommended) + print(f"Hit (failed + recommended): {len(matched['hit'])}") + print(f"Miss (failed, not recommended): {len(matched['miss'])}") + print(f"Untested (recommended, no failure): {len(matched['untested'])}") + + report = generate_report( + failed, recommended, matched, args.log_dir, args.recommendations_source + ) + output_path = Path(args.output) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(report, encoding="utf-8") + print() + print(f"Report => {output_path}") + print() + + # Print report to stdout (safe fallback for Windows encoding) + try: + print(report) + except UnicodeEncodeError: + print(report.encode("ascii", errors="replace").decode("ascii")) + + +if __name__ == "__main__": + main() diff --git a/scripts/ci/npu/precise-test/coveragerc b/scripts/ci/npu/precise-test/coveragerc new file mode 100644 index 000000000..05fabd93b --- /dev/null +++ b/scripts/ci/npu/precise-test/coveragerc @@ -0,0 +1,16 @@ +[run] +branch = True +relative_files = False +parallel = True +concurrency = thread,multiprocessing +sigterm = False +disable_warnings = no-data-collected,module-not-python + +include = + */python/sglang/* + +omit = + */.local/* + /usr/* + */sglang/test/* + */sglang/benchmark/* diff --git a/scripts/ci/npu/precise-test/list_tests.py b/scripts/ci/npu/precise-test/list_tests.py new file mode 100755 index 000000000..f8b47a70a --- /dev/null +++ b/scripts/ci/npu/precise-test/list_tests.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +""" +List the test files selected for a CI suite without running them. + +Standalone extraction of `test/run_suite.py --list-tests-output`: discover +the registered tests under test/registered/, filter them by hardware +backend / suite (PR per-commit tests only; nightly-registered tests are +excluded), then write the selected test file paths (one per line) to the +output file. + +Usage: + python3 list_tests.py --hw npu --suite base-b-test-1-npu-a3 \ + [--auto-partition-id N --auto-partition-size M] \ + -o /tmp/selected_tests.txt +""" + +import argparse +import glob +import os +import sys +from pathlib import Path + +# Repo layout: this script lives at /scripts/ci/npu/precise-test/. +SCRIPT_DIR = Path(__file__).resolve().parent +REPO_ROOT = SCRIPT_DIR.parents[3] + +# ci_register.py is stdlib-only; import it directly (bypassing the sglang +# package __init__, which pulls in torch) so this script runs anywhere. +sys.path.insert(0, str(REPO_ROOT / "python" / "sglang" / "test" / "ci")) + +from ci_register import HWBackend, auto_partition, collect_tests # noqa: E402 + +HW_MAPPING = { + "cpu": HWBackend.CPU, + "cuda": HWBackend.CUDA, + "amd": HWBackend.AMD, + "musa": HWBackend.MUSA, + "npu": HWBackend.NPU, + "xpu": HWBackend.XPU, + "mlx": HWBackend.MLX, +} + + +def main(): + parser = argparse.ArgumentParser( + description=( + "Write the test files selected for a CI suite (one per line) " + "without running them." + ) + ) + parser.add_argument( + "--hw", + type=str, + choices=HW_MAPPING.keys(), + required=True, + help="Hardware backend to select tests for.", + ) + parser.add_argument( + "--suite", + type=str, + required=True, + help=( + "Test suite to select. Accepts a comma-separated list of suites; " + "their tests are unioned." + ), + ) + parser.add_argument( + "--auto-partition-id", + type=int, + help="Use auto load balancing. The part id.", + ) + parser.add_argument( + "--auto-partition-size", + type=int, + help="Use auto load balancing. The number of parts.", + ) + parser.add_argument( + "--output", + "-o", + type=str, + required=True, + help="Write selected test file paths (one per line) to this file.", + ) + args = parser.parse_args() + + # Validate auto-partition arguments (same rules as run_suite.py). + if (args.auto_partition_id is not None) != (args.auto_partition_size is not None): + parser.error( + "--auto-partition-id and --auto-partition-size must be specified together." + ) + if args.auto_partition_size is not None: + if args.auto_partition_size <= 0: + parser.error("--auto-partition-size must be positive.") + if not 0 <= args.auto_partition_id < args.auto_partition_size: + parser.error( + f"--auto-partition-id must be in range [0, {args.auto_partition_size}), " + f"but got {args.auto_partition_id}" + ) + + hw = HW_MAPPING[args.hw] + suites = {s.strip() for s in args.suite.split(",") if s.strip()} + + # Registered tests under /test/registered/ + files = [ + f + for f in glob.glob( + str(REPO_ROOT / "test" / "registered" / "**" / "*.py"), recursive=True + ) + # conftest.py / __init__.py are pytest+package structure, never + # registered tests, and must not be listed as one. + if os.path.basename(f) not in ("conftest.py", "__init__.py") + ] + all_tests = collect_tests(files) + + # Same filter as run_suite.py PR mode: backend + suite, per-commit + # (non-nightly) tests only, enabled only. + ci_tests = [ + t + for t in all_tests + if t.backend == hw + and t.effective_suite in suites + and not t.nightly + and t.disabled is None + ] + + # Shard the selected tests across runners (LPT, same as run_suite.py). + # NPU workflows rely on this to split one suite across matrix jobs. + if args.auto_partition_size: + ci_tests = auto_partition( + ci_tests, args.auto_partition_id, args.auto_partition_size + ) + + with open(args.output, "w") as f: + for t in ci_tests: + f.write(t.filename + "\n") + + +if __name__ == "__main__": + main() diff --git a/scripts/ci/npu/precise-test/run_tests_with_coverage.sh b/scripts/ci/npu/precise-test/run_tests_with_coverage.sh new file mode 100755 index 000000000..681239be9 --- /dev/null +++ b/scripts/ci/npu/precise-test/run_tests_with_coverage.sh @@ -0,0 +1,126 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +CACHE_ROOT="/root/.cache/tests/precise-test" + +# Per-CI-run, per-attempt unique directory (no cross-run residue, no overwrite on re-run) +RUN_ID="${GITHUB_RUN_ID:-local}" +RUN_ATTEMPT="${GITHUB_RUN_ATTEMPT:-1}" +RUN_DIR="${RUN_ID}-attempt-${RUN_ATTEMPT}" + +# Date tag for grouping coverage data. In CI, prefer GITHUB_RUN_STARTED_AT +# (same value across all jobs in one run, immune to midnight rollover). +# Fall back to local date for non-CI execution. +if [ -n "${GITHUB_RUN_STARTED_AT:-}" ]; then + COV_DATE_TAG="${GITHUB_RUN_STARTED_AT:0:10}" + COV_DATE_TAG="${COV_DATE_TAG//-/}" +else + COV_DATE_TAG="$(date +%Y%m%d)" +fi +COV_ROOT="${CACHE_ROOT}/${RUN_DIR}/outputs/sglang@${COV_DATE_TAG}" + +mkdir -p "${COV_ROOT}" + +targets=("$@") +if [ "${#targets[@]}" -eq 0 ]; then + echo "Usage: $0 [test ...]" + exit 1 +fi + +overall_status=0 + +results=() + +# Derive a filesystem-safe directory name from a test target: +# 1. strip the trailing ".py" +# 2. flatten path separators: / -> __ +# 3. flatten pytest separators: :: -> -- +# 4. replace any remaining unsafe character with "_" +# Each test gets its own COVERAGE_FILE so results never collide. +setup_coverage() { + local target="$1" + local name="${target%.py}" + name="${name//\//__}" + name="${name//::/--}" + name="${name//[^a-zA-Z0-9_.-]/_}" + local covdir="${COV_ROOT}/${name}" + mkdir -p "${covdir}" + export COVERAGE_FILE="${covdir}/coverage" +} + +run_one() { + local target="$1" + + echo "=== Running: ${target} ===" + setup_coverage "${target}" + + set +e + python -m coverage run --rcfile="${SCRIPT_DIR}/coveragerc" -m pytest -sv --color=yes "${target}" 2>&1 + local status=$? + set -e + + if [ "${status}" -ne 0 ]; then + echo "1" > "$(dirname "${COVERAGE_FILE}")/FAILED" + echo "=== FAILED: ${target} ===" + overall_status=1 + results+=("${target}|FAILED") + else + echo "=== PASSED: ${target} ===" + results+=("${target}|PASSED") + fi +} + +for target in "${targets[@]}"; do + run_one "${target}" +done + +# ==================== +# Test result summary +# ==================== +passed_list=() +failed_list=() + +for entry in "${results[@]}"; do + test_name="${entry%%|*}" + test_status="${entry##*|}" + if [ "${test_status}" = "PASSED" ]; then + passed_list+=("${test_name}") + else + failed_list+=("${test_name}") + fi +done + +passed_count="${#passed_list[@]}" +failed_count="${#failed_list[@]}" +total_count=$((passed_count + failed_count)) + +echo +echo "============================================================" +echo "Test Summary: total ${total_count}, passed ${passed_count}, failed ${failed_count}" +echo "============================================================" + +if [ "${passed_count}" -gt 0 ]; then + echo "✓ PASSED:" + for t in "${passed_list[@]}"; do + echo " ${t}" + done +fi + +if [ "${failed_count}" -gt 0 ]; then + echo + echo "✗ FAILED:" + for t in "${failed_list[@]}"; do + echo " ${t}" + done +fi + +echo "============================================================" + +if [ "${failed_count}" -gt 0 ]; then + echo "ERROR: Some tests failed." +fi +echo "Coverage: ${COV_ROOT}/" + +exit "${overall_status}" diff --git a/scripts/ci/npu/precise-test/test_selector.py b/scripts/ci/npu/precise-test/test_selector.py new file mode 100644 index 000000000..0506b8380 --- /dev/null +++ b/scripts/ci/npu/precise-test/test_selector.py @@ -0,0 +1,1939 @@ +""" +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()