lesavka/scripts/ci/quality_gate.sh

#!/usr/bin/env bash
set -euo pipefail

ROOT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)
REPORT_DIR="${ROOT_DIR}/target/quality-gate"
COVERAGE_JSON="${REPORT_DIR}/coverage.json"
SUMMARY_TXT="${REPORT_DIR}/summary.txt"
METRICS_FILE="${REPORT_DIR}/metrics.prom"
BASELINE_JSON="${ROOT_DIR}/scripts/ci/quality_gate_baseline.json"
PUSHGATEWAY_URL=${QUALITY_GATE_PUSHGATEWAY_URL:-}

mkdir -p "${REPORT_DIR}"

cat >"${METRICS_FILE}" <<'METRICS'
# HELP platform_quality_gate_runs_total Number of quality gate runs by result.
# TYPE platform_quality_gate_runs_total counter
platform_quality_gate_runs_total{suite="lesavka",status="ok"} 0
platform_quality_gate_runs_total{suite="lesavka",status="failed"} 1
METRICS

fetch_remote_counter() {
  local status="$1"
  if [[ -z "${PUSHGATEWAY_URL}" ]]; then
    echo 0
    return 0
  fi

  curl --silent --show-error --fail "${PUSHGATEWAY_URL%/}/metrics" 2>/dev/null | awk -v suite="lesavka" -v status="${status}" '
    /^platform_quality_gate_runs_total\{/ {
      if (index($0, "job=\"platform-quality-ci\"") == 0) next
      if (index($0, "suite=\"" suite "\"") == 0) next
      if (index($0, "status=\"" status "\"") == 0) next
      print int($2)
      found = 1
      exit
    }
    END {
      if (!found) print 0
    }'
}

refresh_counter_metrics() {
  local outcome status_line ok_count failed_count tmp_file
  status_line="$(awk '/^platform_quality_gate_runs_total\{suite="lesavka"/ {print; exit}' "${METRICS_FILE}" 2>/dev/null || true)"
  outcome="$(printf '%s' "${status_line}" | sed -n 's/.*status="\([^"]*\)".*/\1/p')"
  [[ -n "${outcome}" ]] || outcome="failed"

  ok_count="$(fetch_remote_counter ok)"
  failed_count="$(fetch_remote_counter failed)"
  if [[ "${outcome}" == "ok" ]]; then
    ok_count=$((ok_count + 1))
  else
    failed_count=$((failed_count + 1))
  fi

  tmp_file="$(mktemp "${REPORT_DIR}/metrics.prom.XXXXXX")"
  {
    echo '# HELP platform_quality_gate_runs_total Number of quality gate runs by result.'
    echo '# TYPE platform_quality_gate_runs_total counter'
    echo "platform_quality_gate_runs_total{suite=\"lesavka\",status=\"ok\"} ${ok_count}"
    echo "platform_quality_gate_runs_total{suite=\"lesavka\",status=\"failed\"} ${failed_count}"
    awk '
      /^# HELP platform_quality_gate_runs_total / {next}
      /^# TYPE platform_quality_gate_runs_total / {next}
      /^platform_quality_gate_runs_total\{suite="lesavka"/ {next}
      {print}
    ' "${METRICS_FILE}"
  } >"${tmp_file}"
  mv "${tmp_file}" "${METRICS_FILE}"
}

publish_metrics() {
  if [[ -z "${PUSHGATEWAY_URL}" ]]; then
    echo "Skipping Pushgateway publish: QUALITY_GATE_PUSHGATEWAY_URL is not set"
    return 0
  fi

  curl --fail --silent --show-error \
    --data-binary @"${METRICS_FILE}" \
    "${PUSHGATEWAY_URL%/}/metrics/job/platform-quality-ci/suite/lesavka"
}

status=0
if cargo llvm-cov --workspace --all-targets --summary-only --json --output-path "${COVERAGE_JSON}"; then
  if python3 - "${COVERAGE_JSON}" "${BASELINE_JSON}" "${METRICS_FILE}" "${SUMMARY_TXT}" "${ROOT_DIR}" <<'PY'
import json
import pathlib
import sys
from datetime import datetime, timezone

coverage_path = pathlib.Path(sys.argv[1])
baseline_path = pathlib.Path(sys.argv[2])
metrics_path = pathlib.Path(sys.argv[3])
summary_path = pathlib.Path(sys.argv[4])
root = pathlib.Path(sys.argv[5])

with coverage_path.open('r', encoding='utf-8') as fh:
    report = json.load(fh)

coverage_data = report['data'][0]
coverage_totals = coverage_data['totals']
files = []
for entry in coverage_data['files']:
    filename = pathlib.Path(entry['filename'])
    rel = filename.relative_to(root).as_posix()
    if '/src/tests/' in rel:
        continue
    if '/src/' not in rel:
        continue
    loc = sum(1 for _ in filename.open('r', encoding='utf-8'))
    line_percent = float(entry['summary']['lines']['percent'])
    files.append({
        'path': rel,
        'loc': loc,
        'line_percent': line_percent,
    })

files.sort(key=lambda item: item['path'])

baseline = {'files': {}}
if baseline_path.exists():
    with baseline_path.open('r', encoding='utf-8') as fh:
        baseline = json.load(fh)

baseline_files = baseline.get('files', {})
regressions = []
current_by_path = {item['path']: item for item in files}
missing_from_baseline = [path for path in current_by_path if path not in baseline_files]

for path, current in current_by_path.items():
    baseline_entry = baseline_files.get(path)
    if baseline_entry is None:
        continue
    if current['loc'] > int(baseline_entry['loc']):
        regressions.append(f"{path}: loc grew from {baseline_entry['loc']} to {current['loc']}")
    if current['line_percent'] + 0.01 < float(baseline_entry['line_percent']):
        regressions.append(
            f"{path}: line coverage fell from {baseline_entry['line_percent']:.2f}% to {current['line_percent']:.2f}%"
        )

workspace_lines = float(coverage_totals['lines']['percent'])
files_at_95 = sum(1 for item in files if item['line_percent'] >= 95.0)
files_below_95 = len(files) - files_at_95
over_500 = sum(1 for item in files if item['loc'] > 500)

metrics = []
metrics.append('# HELP platform_quality_gate_runs_total Number of quality gate runs by result.')
metrics.append('# TYPE platform_quality_gate_runs_total counter')
status_label = 'ok' if not regressions and not missing_from_baseline else 'failed'
metrics.append(f'platform_quality_gate_runs_total{{suite="lesavka",status="{status_label}"}} 1')
metrics.append('# HELP platform_quality_gate_workspace_line_coverage_percent Workspace line coverage percent.')
metrics.append('# TYPE platform_quality_gate_workspace_line_coverage_percent gauge')
metrics.append(f'platform_quality_gate_workspace_line_coverage_percent{{suite="lesavka"}} {workspace_lines:.2f}')
metrics.append('# HELP platform_quality_gate_files_total Count of tracked source files in the quality gate.')
metrics.append('# TYPE platform_quality_gate_files_total gauge')
metrics.append(f'platform_quality_gate_files_total{{suite="lesavka"}} {len(files)}')
metrics.append('# HELP platform_quality_gate_files_at_or_above_95_total Count of files at or above the 95 percent line target.')
metrics.append('# TYPE platform_quality_gate_files_at_or_above_95_total gauge')
metrics.append(f'platform_quality_gate_files_at_or_above_95_total{{suite="lesavka"}} {files_at_95}')
metrics.append('# HELP platform_quality_gate_files_below_95_total Count of files below the 95 percent line target.')
metrics.append('# TYPE platform_quality_gate_files_below_95_total gauge')
metrics.append(f'platform_quality_gate_files_below_95_total{{suite="lesavka"}} {files_below_95}')
metrics.append('# HELP platform_quality_gate_source_lines_over_500_total Count of tracked source files over 500 LOC.')
metrics.append('# TYPE platform_quality_gate_source_lines_over_500_total gauge')
metrics.append(f'platform_quality_gate_source_lines_over_500_total{{suite="lesavka"}} {over_500}')
metrics.append('# HELP platform_quality_gate_file_line_coverage_percent Per-file line coverage percent.')
metrics.append('# TYPE platform_quality_gate_file_line_coverage_percent gauge')
metrics.append('# HELP platform_quality_gate_file_source_lines Per-file source line count.')
metrics.append('# TYPE platform_quality_gate_file_source_lines gauge')

def esc(value: str) -> str:
    return value.replace('\\', r'\\').replace('\n', r'\\n').replace('"', r'\"')

for item in files:
    label = esc(item['path'])
    metrics.append(
        f'platform_quality_gate_file_line_coverage_percent{{suite="lesavka",file="{label}"}} {item["line_percent"]:.2f}'
    )
    metrics.append(
        f'platform_quality_gate_file_source_lines{{suite="lesavka",file="{label}"}} {item["loc"]}'
    )

metrics_path.write_text('\n'.join(metrics) + '\n', encoding='utf-8')

lines = []
lines.append(f'quality gate report generated at {datetime.now(timezone.utc).isoformat()}')
lines.append(f'workspace line coverage: {workspace_lines:.2f}%')
lines.append(f'source files tracked: {len(files)}')
lines.append(f'files >= 95% line coverage: {files_at_95}')
lines.append(f'files < 95% line coverage: {files_below_95}')
lines.append(f'files over 500 LOC: {over_500}')
lines.append('')
lines.append('path | loc | line coverage | baseline loc | baseline coverage | status')
lines.append('-' * 86)
for item in files:
    baseline_entry = baseline_files.get(item['path'])
    if baseline_entry is None:
        baseline_loc = 'n/a'
        baseline_cov = 'n/a'
        status = 'new'
    else:
        baseline_loc = str(baseline_entry['loc'])
        baseline_cov = f"{float(baseline_entry['line_percent']):.2f}%"
        status = 'ok'
        if item['loc'] > int(baseline_entry['loc']) or item['line_percent'] + 0.01 < float(baseline_entry['line_percent']):
            status = 'regressed'
    lines.append(
        f"{item['path']} | {item['loc']} | {item['line_percent']:.2f}% | {baseline_loc} | {baseline_cov} | {status}"
    )

summary_path.write_text('\n'.join(lines) + '\n', encoding='utf-8')

print(summary_path.read_text(encoding='utf-8'))

if missing_from_baseline:
    print('missing baseline entries:', ', '.join(missing_from_baseline), file=sys.stderr)

if regressions or missing_from_baseline:
    for line in regressions:
        print(line, file=sys.stderr)
    raise SystemExit(1)
PY
  then
    :
  else
    status=$?
  fi
else
  status=$?
fi

publish_status=0
refresh_counter_metrics
if publish_metrics; then
  :
else
  publish_status=$?
fi

if [[ ${status} -eq 0 && ${publish_status} -ne 0 ]]; then
  status=${publish_status}
fi

exit ${status}