ATC Automatisierung auf BTP: Code-Qualität kontinuierlich überwachen

import xml.etree.ElementTree as ET
import sys
import json
from dataclasses import dataclass, asdict
from typing import List

@dataclass
class Finding:
    check_id: str
    check_title: str
    message: str
    priority: int
    object_name: str
    location: str
    has_quickfix: bool

def parse_atc_results(xml_file: str) -> dict:
    tree = ET.parse(xml_file)
    root = tree.getroot()

    ns = {
        'atc': 'http://www.sap.com/adt/atc',
        'adtcore': 'http://www.sap.com/adt/core'
    }

    findings: List[Finding] = []
    priority_counts = {1: 0, 2: 0, 3: 0}

    for obj in root.findall('.//atc:object', ns):
        obj_name = obj.get('{http://www.sap.com/adt/core}name', 'Unknown')

        for finding in obj.findall('.//atc:finding', ns):
            priority = int(finding.get('priority', 3))
            priority_counts[priority] = priority_counts.get(priority, 0) + 1

            quickfixes = finding.find('atc:quickfixes', ns)
            has_quickfix = quickfixes is not None and int(quickfixes.get('count', 0)) > 0

            findings.append(Finding(
                check_id=finding.get('checkId', ''),
                check_title=finding.get('checkTitle', ''),
                message=finding.get('messageTitle', ''),
                priority=priority,
                object_name=obj_name,
                location=finding.get('location', ''),
                has_quickfix=has_quickfix
            ))

    return {
        'total': len(findings),
        'errors': priority_counts.get(1, 0),
        'warnings': priority_counts.get(2, 0),
        'info': priority_counts.get(3, 0),
        'findings': [asdict(f) for f in findings]
    }

def print_report(results: dict, error_threshold: int = 0, warning_threshold: int = 5):
    print("=" * 60)
    print("ATC QUALITY REPORT")
    print("=" * 60)
    print(f"Total Findings:  {results['total']}")
    print(f"  Errors (P1):   {results['errors']}")
    print(f"  Warnings (P2): {results['warnings']}")
    print(f"  Info (P3):     {results['info']}")
    print("=" * 60)

    # Findings nach Priorität gruppiert ausgeben
    for priority in [1, 2, 3]:
        priority_findings = [f for f in results['findings'] if f['priority'] == priority]
        if priority_findings:
            print(f"\nPriority {priority} Findings:")
            print("-" * 40)
            for f in priority_findings:
                print(f"  [{f['check_id']}] {f['object_name']}")
                print(f"    {f['message']}")
                quickfix = "✓ Quickfix available" if f['has_quickfix'] else ""
                print(f"    {f['location']} {quickfix}")

    print("\n" + "=" * 60)

    # Thresholds prüfen
    exit_code = 0
    if results['errors'] > error_threshold:
        print(f"::error::Found {results['errors']} error(s) (threshold: {error_threshold})")
        exit_code = 1

    if results['warnings'] > warning_threshold:
        print(f"::warning::Found {results['warnings']} warning(s) (threshold: {warning_threshold})")

    return exit_code

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print("Usage: python parse_atc_results.py <xml_file> [--json]")
        sys.exit(1)

    results = parse_atc_results(sys.argv[1])

    if '--json' in sys.argv:
        print(json.dumps(results, indent=2))
    else:
        exit_code = print_report(results)
        sys.exit(exit_code)

name: ATC Quality Gate

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]

env:
  PACKAGE: Z_FLIGHT_BOOKING
  CHECK_VARIANT: ABAP_CLOUD_READINESS
  ERROR_THRESHOLD: 0
  WARNING_THRESHOLD: 5

jobs:
  atc-check:
    name: ATC Quality Check
    runs-on: ubuntu-latest

    steps:
      - name: Checkout
        uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Create ATC Configuration
        run: |
          cat > atc_config.xml << 'EOF'
          <?xml version="1.0" encoding="UTF-8"?>
          <atc:run xmlns:atc="http://www.sap.com/adt/atc"
                   xmlns:adtcore="http://www.sap.com/adt/core"
                   maximumVerdicts="500">
            <objectSets>
              <objectSet kind="inclusive">
                <adtcore:objectReferences>
                  <adtcore:objectReference
                    adtcore:uri="/sap/bc/adt/vit/wb/object_type/devck/object_name/${{ env.PACKAGE }}"/>
                </adtcore:objectReferences>
              </objectSet>
            </objectSets>
            <checkVariant>
              <atc:name>${{ env.CHECK_VARIANT }}</atc:name>
            </checkVariant>
          </atc:run>
          EOF

      - name: Start ATC Run
        id: start-atc
        run: |
          RESPONSE=$(curl -s -D - \
            -X POST \
            "${{ secrets.ABAP_ENDPOINT }}/sap/bc/adt/atc/runs" \
            -u "${{ secrets.ABAP_USER }}:${{ secrets.ABAP_PASSWORD }}" \
            -H "Content-Type: application/vnd.sap.atc.run.v1+xml" \
            -H "Accept: application/vnd.sap.atc.run.v1+xml" \
            -d @atc_config.xml)

          RUN_ID=$(echo "$RESPONSE" | grep -i "location:" | sed 's/.*runs\///' | tr -d '\r\n')

          if [ -z "$RUN_ID" ]; then
            echo "::error::Failed to start ATC run"
            exit 1
          fi

          echo "run_id=$RUN_ID" >> $GITHUB_OUTPUT
          echo "Started ATC run: $RUN_ID"

      - name: Wait for ATC Completion
        run: |
          RUN_ID="${{ steps.start-atc.outputs.run_id }}"
          MAX_ATTEMPTS=60

          for i in $(seq 1 $MAX_ATTEMPTS); do
            STATUS=$(curl -s \
              -X GET \
              "${{ secrets.ABAP_ENDPOINT }}/sap/bc/adt/atc/runs/$RUN_ID" \
              -u "${{ secrets.ABAP_USER }}:${{ secrets.ABAP_PASSWORD }}" \
              -H "Accept: application/vnd.sap.atc.run.v1+xml" \
              | grep -oP 'status="\K[^"]+' || echo "unknown")

            echo "Status: $STATUS (Attempt $i/$MAX_ATTEMPTS)"

            if [ "$STATUS" = "finished" ]; then
              break
            fi

            sleep 5
          done

          if [ "$STATUS" != "finished" ]; then
            echo "::error::ATC run timed out"
            exit 1
          fi

      - name: Fetch ATC Results
        run: |
          RUN_ID="${{ steps.start-atc.outputs.run_id }}"

          curl -s \
            -X GET \
            "${{ secrets.ABAP_ENDPOINT }}/sap/bc/adt/atc/runs/$RUN_ID/results" \
            -u "${{ secrets.ABAP_USER }}:${{ secrets.ABAP_PASSWORD }}" \
            -H "Accept: application/vnd.sap.atc.worklist.v1+xml" \
            -o atc_results.xml

      - name: Parse and Evaluate Results
        id: parse
        run: |
          python << 'PYTHON_SCRIPT'
          import xml.etree.ElementTree as ET
          import os

          tree = ET.parse('atc_results.xml')
          root = tree.getroot()
          ns = {'atc': 'http://www.sap.com/adt/atc', 'adtcore': 'http://www.sap.com/adt/core'}

          errors = warnings = info = 0

          for finding in root.findall('.//atc:finding', ns):
              priority = int(finding.get('priority', 3))
              if priority == 1:
                  errors += 1
              elif priority == 2:
                  warnings += 1
              else:
                  info += 1

          with open(os.environ['GITHUB_OUTPUT'], 'a') as f:
              f.write(f"errors={errors}\n")
              f.write(f"warnings={warnings}\n")
              f.write(f"info={info}\n")
              f.write(f"total={errors + warnings + info}\n")

          print(f"Errors: {errors}, Warnings: {warnings}, Info: {info}")

          # Quality Gate prüfen
          if errors > int(os.environ.get('ERROR_THRESHOLD', 0)):
              print(f"::error::Quality Gate failed: {errors} error(s) found")
              exit(1)
          PYTHON_SCRIPT

      - name: Upload ATC Results
        uses: actions/upload-artifact@v4
        with:
          name: atc-results
          path: atc_results.xml

      - name: Create Summary
        if: always()
        run: |
          cat >> $GITHUB_STEP_SUMMARY << EOF
          ## ATC Quality Gate Results

          | Metric | Value | Threshold |
          |--------|-------|-----------|
          | Errors (P1) | ${{ steps.parse.outputs.errors }} | ${{ env.ERROR_THRESHOLD }} |
          | Warnings (P2) | ${{ steps.parse.outputs.warnings }} | ${{ env.WARNING_THRESHOLD }} |
          | Info (P3) | ${{ steps.parse.outputs.info }} | - |
          | **Total** | ${{ steps.parse.outputs.total }} | |

          **Status:** ${{ steps.parse.outputs.errors == 0 && '✅ Passed' || '❌ Failed' }}
          EOF

stages:
  - quality

variables:
  PACKAGE: Z_FLIGHT_BOOKING
  CHECK_VARIANT: ABAP_CLOUD_READINESS
  ERROR_THRESHOLD: "0"

atc-quality-gate:
  stage: quality
  image: python:3.11-slim
  before_script:
    - apt-get update && apt-get install -y curl
  script:
    # ATC Run starten
    - |
      RESPONSE=$(curl -s -D - \
        -X POST \
        "${ABAP_ENDPOINT}/sap/bc/adt/atc/runs" \
        -u "${ABAP_USER}:${ABAP_PASSWORD}" \
        -H "Content-Type: application/vnd.sap.atc.run.v1+xml" \
        -H "Accept: application/vnd.sap.atc.run.v1+xml" \
        -d "<atc:run xmlns:atc=\"http://www.sap.com/adt/atc\"
                    xmlns:adtcore=\"http://www.sap.com/adt/core\"
                    maximumVerdicts=\"500\">
              <objectSets>
                <objectSet kind=\"inclusive\">
                  <adtcore:objectReferences>
                    <adtcore:objectReference
                      adtcore:uri=\"/sap/bc/adt/vit/wb/object_type/devck/object_name/${PACKAGE}\"/>
                  </adtcore:objectReferences>
                </objectSet>
              </objectSets>
              <checkVariant>
                <atc:name>${CHECK_VARIANT}</atc:name>
              </checkVariant>
            </atc:run>")

      RUN_ID=$(echo "$RESPONSE" | grep -i "location:" | sed 's/.*runs\///' | tr -d '\r\n')
      echo "ATC Run ID: $RUN_ID"

    # Auf Abschluss warten
    - |
      for i in $(seq 1 60); do
        STATUS=$(curl -s \
          "${ABAP_ENDPOINT}/sap/bc/adt/atc/runs/${RUN_ID}" \
          -u "${ABAP_USER}:${ABAP_PASSWORD}" \
          -H "Accept: application/vnd.sap.atc.run.v1+xml" \
          | grep -oP 'status="\K[^"]+' || echo "unknown")

        echo "Attempt $i: Status = $STATUS"
        [ "$STATUS" = "finished" ] && break
        sleep 5
      done

    # Ergebnisse abrufen
    - |
      curl -s \
        "${ABAP_ENDPOINT}/sap/bc/adt/atc/runs/${RUN_ID}/results" \
        -u "${ABAP_USER}:${ABAP_PASSWORD}" \
        -H "Accept: application/vnd.sap.atc.worklist.v1+xml" \
        -o atc_results.xml

    # Ergebnisse prüfen
    - |
      python3 << 'EOF'
      import xml.etree.ElementTree as ET
      import sys
      import os

      tree = ET.parse('atc_results.xml')
      root = tree.getroot()
      ns = {'atc': 'http://www.sap.com/adt/atc'}

      errors = sum(1 for f in root.findall('.//atc:finding', ns) if f.get('priority') == '1')
      print(f"Found {errors} error(s)")

      if errors > int(os.environ.get('ERROR_THRESHOLD', 0)):
          sys.exit(1)
      EOF
  artifacts:
    paths:
      - atc_results.xml
    reports:
      junit: atc_results.xml
    when: always
  rules:
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"
    - if: $CI_COMMIT_BRANCH == "main"