Modern Software Development Workflows

Modern Software Development Workflows

Implements end-to-end software delivery workflows that connect version control through production deployment. When loaded, the model designs CI/CD pipelines with proper stage gating, configures automated testing strategies across unit/integration/e2e layers, enforces code quality gates before merge, integrates observability into every release, and applies safe deployment patterns (blue-green, canary, feature flags) to ship software with confidence and rollback capability.

TL;DR Checklist

• Verify CI pipeline has at least 4 stages: lint → test → build → deploy with explicit failure gates
• Confirm branch protection rules block merges when required status checks fail
• Check that unit tests run in parallel with a coverage threshold of ≥ 80% enforced as a quality gate
• Validate integration tests target a disposable staging environment, not production or shared test DBs
• Ensure deployment uses a safe pattern (blue-green, canary, or feature flags) with automated rollback
• Verify observability metrics and health checks are injected into every release artifact before promotion
• Confirm all secrets flow through a vault (e.g., SOPS, HashiCorp Vault) — never via environment variables in CI logs

When to Use

Use this skill when:

• Designing or refactoring a CI/CD pipeline from scratch for a new application repository
• A team needs to establish quality gates (linting, testing, security scanning) as merge-blocking requirements
• Migrating from manual deployments or ad-hoc shell scripts to automated, auditable release pipelines
• Integrating observability (metrics, health checks, tracing) into the build artifacts before they reach production
• Implementing a safe deployment strategy (blue-green, canary, feature flags) to reduce blast radius of releases
• Conducting a DevOps maturity assessment and identifying gaps in the current software delivery lifecycle
• Setting up automated rollback triggers based on post-deployment health metric thresholds

When NOT to Use

Avoid this skill for:

• Implementing SOLID or DRY principles — that belongs in coding-software-engineering-principles
• Writing unit test methodology or TDD discipline — use coding-test-driven-development instead
• Analyzing code for security vulnerabilities — use coding-security-review instead
• Simple single-file scripts or proof-of-concept projects where a full pipeline adds more overhead than value

Core Workflow

1. Audit Current Delivery Pipeline — Inventory every step from code commit to production: version control branching model, CI triggers, test execution, artifact storage, deployment targets, and rollback mechanisms. Map existing tools (GitHub Actions, GitLab CI, Jenkins) and identify gaps where manual steps or missing quality gates create delivery risk. Checkpoint: Produce a pipeline map listing each stage, its duration, failure rate, and whether it has an automated rollback path.

2. Design Branching Model and Merge Policy — Select a branching strategy (trunk-based development with feature flags for fast flow, or GitFlow for regulated environments). Configure branch protection rules: require pull request reviews from at least one approved reviewer, enforce required status checks (lint passes, unit tests pass, build succeeds), and disable force pushes to protected branches. Set up automatic squashing of merge commits to keep history linear on main. Checkpoint: Push a test commit to the protected branch and verify that all required checks block the merge when any stage fails.

3. Configure CI Pipeline Stages — Define the pipeline in .github/workflows/ci.yml (or equivalent) with explicit stages: lint (ruff check, mypy --strict), unit test (pytest tests/unit/ with coverage enforcement), integration test (pytest tests/integration/ against disposable infrastructure), build artifact creation (Docker image with SBOM), and security scanning (trivy, gitleaks). Each stage must fail-fast — subsequent stages skip on upstream failure using if: success() conditions. Checkpoint: Trigger a pipeline run with an intentional lint error and verify that test and build stages are skipped, not failed.

4. Establish Code Quality Gates — Define pass/fail thresholds as gate criteria that block merging: code coverage ≥ 80% (tracked via pytest-cov), zero critical/severe vulnerabilities in dependency scanning (trivy), zero secrets leaked (gitleaks), type checking passes (mypy --strict with no ignored errors). Configure SonarQube or equivalent for static analysis quality gate enforcement — block merges on new code covering less than the threshold, technical debt ratio under 5%, and zero bugs. Checkpoint: Run sonar-scanner locally and verify it fails when coverage drops below threshold or vulnerabilities are introduced.

5. Set Up CD with Safe Deployment Pattern — Implement deployment in stages: first deploy to a staging environment for automated smoke tests, then promote to production using blue-green (two identical environments with traffic switch via load balancer) or canary (route 5-10% of traffic to new version, monitor error rates and latency, gradually increase). Configure feature flags using LaunchDarkly or Unleash so features ship behind toggles independent of deployments. Define automated rollback triggers: if error rate exceeds 1% for more than 2 minutes post-deploy, automatically revert to the previous revision. Checkpoint: Deploy a canary release and verify that elevated error rates trigger an automatic rollback within the defined SLA window.

6. Inject Observability into Release Artifacts — Ensure every built artifact includes health check endpoints (/health returning HTTP 200 with service status, /ready for readiness probe), structured JSON logging (not plain text), OpenTelemetry tracing instrumentation, and Prometheus metrics exposure at /metrics. Register the service in the monitoring stack so dashboards auto-create. Configure alert rules that fire on SLO violations: error budget burn rate > 2x baseline, p99 latency exceeding threshold, or availability below 99.9%. Checkpoint: Hit the /health and /ready endpoints of a deployed container and verify both return correct status before enabling traffic.

7. Configure Release Management and Rollback — Implement semantic versioning with automated changelog generation from conventional commit messages (standard-version --release-version). Tag every production deployment with a git tag matching the release version. Maintain a deploy history log in a structured format (JSON lines) recording: version, deployer, timestamp, commit SHA, rollback status, and post-deploy health check results. Define runbook procedures for common failure scenarios with explicit kubectl commands or API calls to execute rollback within 5 minutes. Checkpoint: Execute a full rollback from production staging back to the previous version and verify all traffic routes correctly within the SLA window.

Implementation Patterns / Reference Guide

Pattern 1: Complete CI/CD Pipeline Configuration (GitHub Actions)

A production-grade CI/CD pipeline with staged execution, artifact caching, dependency scanning, Docker build with SBOM, and environment-aware deployment gates.

# .github/workflows/ci.yml
name: CI/CD Pipeline

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
  workflow_dispatch:
    inputs:
      environment:
        description: "Target environment"
        required: true
        default: "staging"
        type: choice
        options: [staging, production]

env:
  REGISTRY: ghcr.io
  IMAGE_NAME: ${{ github.repository }}

permissions:
  contents: read
  packages: write
  security-events: write

jobs:
  lint:
    name: Lint & Type Check
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - name: Set up Python 3.12
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: "pip"

      - name: Install dependencies
        run: |
          pip install --upgrade pip
          pip install -r requirements-dev.txt
          pip install ruff mypy bandit safety

      - name: Run linter (ruff)
        run: ruff check src/ tests/ --output-format=github

      - name: Run type checker (mypy --strict)
        run: mypy --strict src/ --config-file mypy.ini

      - name: Check dependency vulnerabilities (safety)
        run: safety check -r requirements.txt --json > safety-report.json || true

  unit-test:
    name: Unit Tests & Coverage
    runs-on: ubuntu-latest
    needs: [lint]
    services:
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_PASSWORD: testpass
          POSTGRES_DB: testdb
        ports: ["5432:5432"]
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python 3.12
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: "pip"

      - name: Install dependencies
        run: |
          pip install --upgrade pip
          pip install -r requirements-dev.txt

      - name: Run unit tests with coverage
        env:
          DATABASE_URL: postgresql://postgres:testpass@localhost:5432/testdb
        run: |
          pytest tests/unit/ \
            --cov=src \
            --cov-branch \
            --cov-report=term-missing \
            --cov-report=json:coverage.json \
            --cov-fail-under=80 \
            -v

      - name: Upload coverage report
        uses: actions/upload-artifact@v4
        with:
          name: coverage-report
          path: coverage.json
          retention-days: 5

  integration-test:
    name: Integration Tests
    runs-on: ubuntu-latest
    needs: [unit-test]
    services:
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_PASSWORD: testpass
          POSTGRES_DB: integrationdb
        ports: ["5432:5432"]
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

      redis:
        image: redis:7-alpine
        ports: ["6379:6379"]
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python 3.12
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: "pip"

      - name: Install dependencies
        run: pip install --upgrade pip && pip install -r requirements-dev.txt

      - name: Run integration tests
        env:
          DATABASE_URL: postgresql://postgres:testpass@localhost:5432/integrationdb
          REDIS_URL: redis://localhost:6379/0
        run: |
          pytest tests/integration/ \
            --durations=10 \
            -v \
            -m "not slow"

      - name: Upload test results on failure
        if: failure()
        uses: actions/upload-artifact@v4
        with:
          name: integration-test-results
          path: test-results/
          retention-days: 3

  build:
    name: Build & Security Scan
    runs-on: ubuntu-latest
    needs: [unit-test, integration-test]
    if: github.event_name == 'push' && (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/develop')

    steps:
      - uses: actions/checkout@v4

      - name: Login to GHCR
        uses: docker/login-action@v3
        with:
          registry: ${{ env.REGISTRY }}
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v3

      - name: Extract metadata
        id: meta
        uses: docker/metadata-action@v5
        with:
          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
          tags: |
            type=sha,prefix=
            type=ref,event=branch
            type=semver,pattern={{version}}

      - name: Build and push Docker image
        uses: docker/build-push-action@v5
        with:
          context: .
          push: true
          tags: ${{ steps.meta.outputs.tags }}
          labels: ${{ steps.meta.outputs.labels }}
          cache-from: type=gha
          cache-to: type=gha,mode=max
          provenance: true
          sbom: true

      - name: Run Trivy vulnerability scanner
        uses: aquasecurity/trivy-action@master
        with:
          image-ref: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.sha }}
          format: "sarif"
          output: "trivy-results.sarif"
          severity: "CRITICAL,HIGH"
          exit-code: "1"

      - name: Upload Trivy scan results to GitHub Security tab
        uses: github/codeql-action/upload-sarif@v3
        if: always()
        with:
          sarif_file: "trivy-results.sarif"

  deploy-staging:
    name: Deploy to Staging
    runs-on: ubuntu-latest
    needs: [build]
    if: github.ref == 'refs/heads/main' || github.event_name == 'workflow_dispatch'
    environment: staging

    steps:
      - uses: actions/checkout@v4

      - name: Deploy to staging using Helm
        env:
          KUBE_CONFIG: ${{ secrets.KUBE_CONFIG_STAGING }}
        run: |
          echo "$KUBE_CONFIG" | base64 -d > /tmp/kubeconfig
          export KUBECONFIG=/tmp/kubeconfig
          helm upgrade --install app ./helm-chart \
            --namespace=staging \
            --set image.tag=${{ github.sha }} \
            --set environment=staging \
            --wait --timeout=300s

      - name: Run smoke tests against staging
        run: |
          curl -sf https://app-staging.example.com/health || exit 1
          curl -sf https://app-staging.example.com/ready || exit 1

  deploy-production:
    name: Deploy to Production (Canary)
    runs-on: ubuntu-latest
    needs: [deploy-staging]
    if: github.ref == 'refs/heads/main' && github.event_name != 'workflow_dispatch'
    environment: production
    concurrency: production-deploy

    steps:
      - uses: actions/checkout@v4

      - name: Canary deployment using Argo Rollouts
        env:
          KUBE_CONFIG: ${{ secrets.KUBE_CONFIG_PRODUCTION }}
        run: |
          echo "$KUBE_CONFIG" | base64 -d > /tmp/kubeconfig
          export KUBECONFIG=/tmp/kubeconfig
          kubectl apply -f argo-rollouts/canary-deployment.yaml

      - name: Wait for canary analysis
        run: |
          sleep 120
          kubectl get rollouts app-rollout -n production -o jsonpath='{.status.phase}'

      - name: Promote canary to full rollout
        if: success()
        run: |
          export KUBECONFIG=/tmp/kubeconfig
          kubectl argo rollouts promote app-rollout -n production

      - name: Record deployment in deploy log
        run: |
          echo "{\"version\":\"${{ github.sha }}\",\"deployer\":\"ci\",\"timestamp\":\"$(date -u +%Y-%m-%dT%H:%M:%SZ)\",\"status\":\"success\"}" >> /tmp/deploy-log.json

Pattern 2: Comprehensive Testing Strategy Framework with Coverage Enforcement

A complete testing framework covering unit, integration, and smoke test layers with pytest configuration, fixture management, coverage thresholds, and a test execution runner.

"""testing/strategy.py — Multi-layer test execution framework with quality gates."""

from __future__ import annotations

import sys
import subprocess
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Protocol

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class CoverageThreshold:
    """Coverage thresholds per test layer that must be met to pass quality gate.

    Attributes:
        lines_pct: Minimum percentage of total lines covered (0.0–1.0).
        branches_pct: Minimum percentage of branches covered (0.0–1.0).
        missing_functions: List of function names excluded from coverage minimums.
    """
    lines_pct: float = 0.80
    branches_pct: float = 0.60
    missing_functions: list[str] = field(default_factory=list)


@dataclass(frozen=True)
class TestResult:
    """Immutable result of a test layer execution."""
    layer: str
    passed: bool
    total: int
    failed: int
    skipped: int
    duration_seconds: float
    coverage_lines_pct: float | None = None
    errors: list[str] = field(default_factory=list)

    @property
    def summary(self) -> str:
        status = "PASS" if self.passed else "FAIL"
        return (
            f"[{status}] {self.layer}: "
            f"{self.failed} failed, {self.skipped} skipped, "
            f"{self.duration_seconds:.1f}s, "
            f"coverage: {self.coverage_lines_pct:.1f}%{' ' if self.coverage_lines_pct else ''}"
        )


class TestLayer(Protocol):
    """Protocol for a test layer (unit, integration, smoke) that can be executed."""

    def execute(self) -> TestResult: ...


class UnitTestRunner:
    """Executes unit tests with coverage enforcement and parallel execution.

    Uses pytest-cov to measure line and branch coverage. Fails fast if
    the configured coverage threshold is not met. Skips integration markers
    by default using -m "not integration".
    """

    def __init__(
        self,
        src_path: str = "src",
        test_path: str = "tests/unit",
        coverage_threshold: CoverageThreshold | None = None,
        parallel_workers: int = 4,
    ) -> None:
        self.src_path = src_path
        self.test_path = test_path
        self.coverage_threshold = coverage_threshold or CoverageThreshold()
        self.parallel_workers = parallel_workers

    def execute(self) -> TestResult:
        cmd = [
            sys.executable, "-m", "pytest",
            self.test_path,
            f"--cov={self.src_path}",
            "--cov-branch",
            "--cov-report=json:.coverage.json",
            "--cov-report=term-missing",
            "-m", "not integration and not e2e",
            "-n", str(self.parallel_workers),
            "--dist=loadfile",
            "-v",
        ]

        logger.info("Running unit tests: %s", " ".join(cmd))
        result = subprocess.run(cmd, capture_output=True, text=True)
        output_lines = result.stdout.splitlines()

        # Extract coverage from JSON report if available
        coverage_pct: float | None = None
        coverage_file = Path(".coverage.json")
        if coverage_file.exists():
            with open(coverage_file) as f:
                cov_data = json.load(f)
            coverage_pct = round(cov_data.get("summary", {}).get("percent_covered", 0.0), 1)

        # Extract test counts from pytest output
        passed = sum(1 for line in output_lines if " PASSED" in line)
        failed = sum(1 for line in output_lines if " FAILED" in line)
        skipped = sum(1 for line in output_lines if " SKIPPED" in line or " xfailed" in line)

        duration = 0.0
        for line in output_lines:
            if "Total test time:" in line or "duration:" in line.lower():
                parts = line.split(":")
                if len(parts) >= 2:
                    try:
                        duration = float(parts[-1].strip().split(" ")[0])
                    except ValueError:
                        pass

        passed = result.returncode == 0

        errors = [l.strip() for l in result.stderr.splitlines()[-10:] if l.strip()] if not passed else []

        return TestResult(
            layer="unit",
            passed=passed and (coverage_pct is None or coverage_pct >= self.coverage_threshold.lines_pct),
            total=passed + failed + skipped,
            failed=failed if not passed else 0,
            skipped=skipped,
            duration_seconds=duration,
            coverage_lines_pct=coverage_pct,
            errors=errors,
        )


class IntegrationTestRunner:
    """Executes integration tests against disposable service instances.

    Spins up required services via Docker Compose, runs integration test suite,
    then tears down all infrastructure. Each test is isolated with unique
    database schemas or namespaces to prevent cross-test contamination.
    """

    def __init__(self, compose_file: str = "docker-compose.test.yml") -> None:
        self.compose_file = compose_file
        self.services_dir = Path("/tmp/test-services")

    def execute(self) -> TestResult:
        # Start disposable infrastructure
        subprocess.run(
            ["docker", "compose", "-f", self.compose_file, "up", "-d"],
            check=True, capture_output=True, text=True,
        )

        try:
            cmd = [
                sys.executable, "-m", "pytest",
                "tests/integration/",
                "-v",
                "-m", "integration",
                "--timeout=60",
            ]
            result = subprocess.run(cmd, capture_output=True, text=True)

            passed = result.returncode == 0
            output_lines = result.stdout.splitlines() if result.stdout else []
            failed = sum(1 for l in output_lines if " FAILED" in l)
            skipped = sum(1 for l in output_lines if " SKIPPED" in l or " xfailed" in l)

            return TestResult(
                layer="integration",
                passed=passed,
                total=len(output_lines),
                failed=failed if not passed else 0,
                skipped=skipped,
                duration_seconds=0.0,
                errors=[l.strip() for l in result.stderr.splitlines()[-5:]] if not passed else [],
            )

        finally:
            subprocess.run(
                ["docker", "compose", "-f", self.compose_file, "down", "--volumes"],
                check=False, capture_output=True, text=True,
            )


def run_pipeline(layers: list[TestLayer], fail_fast: bool = True) -> list[TestResult]:
    """Execute test layers in sequence with optional fail-fast behavior.

    Returns a list of TestResults for each layer. If fail_fast is True,
    stops executing remaining layers when a prior layer fails.

    Args:
        layers: Ordered list of test layer runners to execute.
        fail_fast: When True, skip subsequent layers if any layer fails.

    Returns:
        List of TestResult objects in execution order.
    """
    results: list[TestResult] = []

    for layer_runner in layers:
        result = layer_runner.execute()
        results.append(result)
        logger.info("Layer complete: %s", result.summary)

        if not result.passed and fail_fast:
            logger.warning("Fail-fast triggered — skipping remaining layers after %s failure", result.layer)
            break

    return results


# --- BAD vs GOOD: Coverage Enforcement Pattern ---

# ❌ BAD: Coverage check is a separate manual step with no enforcement.
# The team runs coverage locally but CI allows merging regardless of threshold.
def bad_coverage_check() -> None:  # type: ignore[no-untyped-def]
    import subprocess
    subprocess.run(["pytest", "--cov=src", "tests/unit/"])
    print("Coverage run complete — please verify it's above 80%")  # No enforcement!


# ✅ GOOD: Coverage threshold is enforced programmatically with clear failure mode.
# pytest-cov's --cov-fail-under exits non-zero when below the threshold,
# causing CI to block the merge automatically.
def good_coverage_check(threshold: float = 0.80) -> None:
    """Run unit tests and enforce minimum coverage threshold.

    Args:
        threshold: Minimum required coverage percentage (e.g., 0.80 for 80%).

    Raises:
        RuntimeError: If coverage falls below the enforced threshold.
    """
    import subprocess

    result = subprocess.run(
        [sys.executable, "-m", "pytest", "tests/unit/",
         f"--cov=src", "--cov-branch",
         f"--cov-fail-under={int(threshold * 100)}", "-v"],
        capture_output=True, text=True,
    )

    if result.returncode != 0:
        coverage_line = [l for l in result.stdout.splitlines() if "TOTAL" in l]
        actual = coverage_line[0].split()[-1] if coverage_line else "unknown"
        raise RuntimeError(
            f"Coverage gate FAILED: {actual} is below the {int(threshold * 100)}% threshold.\n"
            f"Add tests or update src/ to improve coverage."
        )

    print(f"Coverage gate PASSED: >= {int(threshold * 100)}%")

Pattern 3: Code Quality Gate Enforcement with Multi-Tool Scanning

Enforces quality gates across multiple tools (ruff, mypy, bandit, safety, trivy) with a unified pass/fail report and configurable severity thresholds.

"""quality/gates.py — Unified code quality gate enforcement for CI merge protection."""

from __future__ import annotations

import json
import logging
import subprocess
import sys
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Protocol

logger = logging.getLogger(__name__)


class Severity(str, Enum):
    """Severity levels for quality gate findings."""
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"


@dataclass(frozen=True)
class GateConfig:
    """Configuration for a single quality gate.

    Attributes:
        tool: Name of the scanning tool (ruff, mypy, bandit, safety, trivy).
        severity_threshold: Minimum severity to consider as a violation.
        allow_fixme: If True, findings marked with "noqa" or "# type: ignore" are exempt.
    """
    tool: str
    severity_threshold: Severity = Severity.HIGH
    allow_fixme: bool = True


@dataclass(frozen=True)
class GateResult:
    """Result from a single quality gate evaluation."""
    config: GateConfig
    passed: bool
    findings_count: int
    critical_count: int
    high_count: int
    details: list[str] = field(default_factory=list)

    def __str__(self) -> str:
        status = "PASS" if self.passed else "FAIL"
        return (
            f"[{status}] {self.config.tool}: "
            f"{self.findings_count} findings "
            f"({self.critical_count} critical, {self.high_count} high)"
        )


class QualityTool(Protocol):
    """Protocol for a quality scanning tool."""

    def scan(self, config: GateConfig) -> GateResult: ...


class RuffScanner:
    """Runs ruff linter and checks for violations above the severity threshold.

    Uses ruff's --output-format=json output to parse individual findings
    with their severity levels (E = error/warning, F = fatal).
    """

    def __init__(self, src_dirs: list[str] | None = None) -> None:
        self.src_dirs = src_dirs or ["src/", "tests/"]

    def scan(self, config: GateConfig) -> GateResult:
        if config.tool != "ruff":
            return GateResult(
                config=config, passed=True, findings_count=0,
                critical_count=0, high_count=0,
                details=["Scanning skipped: tool mismatch"],
            )

        cmd = [sys.executable, "-m", "ruff", "check"] + self.src_dirs + [
            "--output-format=json",
        ]
        result = subprocess.run(cmd, capture_output=True, text=True)

        findings: list[dict] = []
        if result.stdout.strip():
            try:
                findings = json.loads(result.stdout)
            except json.JSONDecodeError:
                return GateResult(
                    config=config, passed=False, findings_count=1,
                    critical_count=1, high_count=0,
                    details=[f"Failed to parse ruff JSON output: {result.stderr[:200]}"],
                )

        severity_map = {"F": Severity.CRITICAL, "E": Severity.HIGH, "W": Severity.MEDIUM}
        threshold_order = {Severity.CRITICAL: 0, Severity.HIGH: 1, Severity.MEDIUM: 2, Severity.LOW: 3}
        max_severity_idx = threshold_order.get(config.severity_threshold, 1)

        critical_count = 0
        high_count = 0
        details: list[str] = []

        for finding in findings:
            code = finding.get("code", "")
            rule_type = code[0] if code else ""
            severity = severity_map.get(rule_type, Severity.LOW)
            idx = threshold_order.get(severity, 3)

            if config.allow_fixme and "noqa" in finding.get("message", "").lower():
                continue

            details.append(f"{finding['filename']}:{finding.get('line_no', '?')}:{finding.get('column', '?')} [{code}] {finding.get('message', '')}")

            if idx <= max_severity_idx:
                if severity == Severity.CRITICAL:
                    critical_count += 1
                elif severity == Severity.HIGH:
                    high_count += 1

        total_violations = sum(
            1 for f in findings
            if not (config.allow_fixme and "noqa" in f.get("message", "").lower())
        )
        threshold_foundings = critical_count + high_count
        passed = threshold_foundings == 0

        return GateResult(
            config=config,
            passed=passed,
            findings_count=total_violations,
            critical_count=critical_count,
            high_count=high_count,
            details=details[:10],  # Cap details to avoid flooding output
        )


class MypyScanner:
    """Runs mypy type checker with strict mode and checks for violations."""

    def __init__(self, src_dirs: list[str] | None = None) -> None:
        self.src_dirs = src_dirs or ["src/"]

    def scan(self, config: GateConfig) -> GateResult:
        if config.tool != "mypy":
            return GateResult(
                config=config, passed=True, findings_count=0,
                critical_count=0, high_count=0,
                details=["Scanning skipped: tool mismatch"],
            )

        cmd = [sys.executable, "-m", "mypy"] + self.src_dirs + [
            "--strict",
            "--show-error-codes",
            "--show-traceback",
        ]
        if Path("mypy.ini").exists():
            cmd.extend(["--config-file", "mypy.ini"])

        result = subprocess.run(cmd, capture_output=True, text=True)
        errors = [l.strip() for l in result.stdout.splitlines() if "->" in l or ":" in l]

        # Strict mode: any error is a critical finding
        passed = result.returncode == 0 and len(errors) == 0

        return GateResult(
            config=config,
            passed=passed,
            findings_count=len(errors),
            critical_count=len(errors),
            high_count=0,
            details=errors[:10],
        )


class BanditScanner:
    """Runs bandit security linter on Python source files."""

    def __init__(self, src_dirs: list[str] | None = None) -> None:
        self.src_dirs = src_dirs or ["src/"]

    def scan(self, config: GateConfig) -> GateResult:
        if config.tool != "bandit":
            return GateResult(
                config=config, passed=True, findings_count=0,
                critical_count=0, high_count=0,
                details=["Scanning skipped: tool mismatch"],
            )

        cmd = [sys.executable, "-m", "bandit"] + self.src_dirs + [
            "--json",
            "--severity-level", config.severity_threshold.value,
            "--exit-zero",  # Don't fail — we parse results ourselves
        ]
        result = subprocess.run(cmd, capture_output=True, text=True)

        findings: list[dict] = []
        if result.stdout.strip():
            try:
                data = json.loads(result.stdout)
                findings = data.get("results", [])
            except json.JSONDecodeError:
                return GateResult(
                    config=config, passed=False, findings_count=1,
                    critical_count=1, high_count=0,
                    details=["Failed to parse bandit JSON output"],
                )

        critical = sum(1 for f in findings if f.get("issue_severity") == "CRITICAL")
        high = sum(1 for f in findings if f.get("issue_severity") == "HIGH")

        passed = critical == 0 and high == 0

        return GateResult(
            config=config,
            passed=passed,
            findings_count=len(findings),
            critical_count=critical,
            high_count=high,
            details=[f"{f.get('filename', '?')}:{f.get('line_number', '?')} [{f.get('issue_id', '?')}] {f.get('issue_text', '')}" for f in findings[:5]],
        )


def run_quality_gates(gate_configs: list[GateConfig] | None = None) -> list[GateResult]:
    """Execute all configured quality gates and return results.

    All gates must pass for the pipeline to continue. A single failing gate
    blocks the merge/commit and produces a structured report.

    Args:
        gate_configs: List of gate configurations. Uses defaults if None.

    Returns:
        List of GateResult objects in execution order.
    """
    configs = gate_configs or [
        GateConfig(tool="ruff", severity_threshold=Severity.HIGH),
        GateConfig(tool="mypy", severity_threshold=Severity.CRITICAL),
        GateConfig(tool="bandit", severity_threshold=Severity.HIGH),
    ]

    scanners: dict[str, QualityTool] = {
        "ruff": RuffScanner(),
        "mypy": MypyScanner(),
        "bandit": BanditScanner(),
    }

    results: list[GateResult] = []

    for config in configs:
        scanner = scanners.get(config.tool)
        if not scanner:
            logger.warning("No scanner registered for tool: %s", config.tool)
            continue

        result = scanner.scan(config)
        results.append(result)
        logger.info("Gate result: %s", result)

    all_passed = all(r.passed for r in results)
    failed_gates = [r for r in results if not r.passed]

    if failed_gates:
        logger.error(
            "%d/%d quality gates FAILED: %s",
            len(failed_gates), len(results),
            ", ".join(r.config.tool for r in failed_gates),
        )
    else:
        logger.info("All %d quality gates PASSED", len(results))

    return results


# --- Usage example ---
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
    results = run_quality_gates()

    exit_code = 0 if all(r.passed for r in results) else 1
    for r in results:
        print(str(r))

    sys.exit(exit_code)

Pattern 4: Deployment Health Monitor with Automated Rollback Triggers

Monitors post-deployment health metrics and triggers automated rollback when SLOs are violated. Integrates with Prometheus alerting rules for real-time detection.

"""deploy/health_monitor.py — Post-deployment health monitoring with automated rollback."""

from __future__ import annotations

import logging
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Protocol

logger = logging.getLogger(__name__)


class RollbackStatus(str, Enum):
    """Status of a rollback operation."""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    NOT_NEEDED = "not_needed"


@dataclass(frozen=True)
class SLOViolation:
    """Recorded SLO violation that triggered rollback consideration.

    Attributes:
        metric: The Prometheus metric name that violated its threshold.
        current_value: The observed value at violation time.
        threshold: The configured alerting threshold.
        duration_seconds: How long the violation persisted before triggering.
        window_minutes: The evaluation window for the SLO check.
    """
    metric: str
    current_value: float
    threshold: float
    duration_seconds: int
    window_minutes: int


@dataclass(frozen=True)
class HealthCheckResult:
    """Result of a single health check against a deployed service.

    Attributes:
        endpoint: The URL or path checked.
        http_status: HTTP status code returned (0 if unreachable).
        response_time_ms: Time taken for the request in milliseconds.
        passed: True when all health criteria are met.
    """
    endpoint: str
    http_status: int
    response_time_ms: float
    passed: bool


class HealthChecker(Protocol):
    """Protocol for checking service health endpoints."""

    def check_health(self, service_name: str) -> list[HealthCheckResult]: ...
    def check_slo_metrics(self, service_name: str, window_minutes: int = 5) -> list[SLOViolation]: ...


class PrometheusHealthChecker:
    """Checks service health by querying Prometheus for metrics and HTTP endpoints.

    Uses the Prometheus API to evaluate SLO conditions: error budget burn rate,
    p99 latency, and availability percentage. Also performs HTTP-level health
    check probing against the deployed /health endpoint.
    """

    def __init__(self, prometheus_url: str, timeout_seconds: float = 10.0) -> None:
        self.prometheus_url = prometheus_url.rstrip("/")
        self.timeout_seconds = timeout_seconds

    def _query_prometheus(self, query: str) -> dict | list[dict]:  # type: ignore[type-arg]
        """Execute a PromQL query against the Prometheus API."""
        import urllib.request
        import json

        url = f"{self.prometheus_url}/api/v1/query?query={urllib.parse.quote(query)}"
        req = urllib.request.Request(url)
        with urllib.request.urlopen(req, timeout=self.timeout_seconds) as resp:
            data = json.loads(resp.read())

        if data.get("status") != "success":
            raise RuntimeError(f"Prometheus query failed: {data}")

        return data["data"]

    def check_health(self, service_name: str) -> list[HealthCheckResult]:
        """Check HTTP health and readiness endpoints for the service.

        Args:
            service_name: The Kubernetes service name to probe.

        Returns:
            List of health check results for each endpoint checked.
        """
        import urllib.request
        import urllib.error

        endpoints = ["/health", "/ready"]
        base_url = f"http://{service_name}"
        results: list[HealthCheckResult] = []

        for endpoint in endpoints:
            url = f"{base_url}{endpoint}"
            start = time.monotonic()
            try:
                req = urllib.request.Request(url, method="GET")
                with urllib.request.urlopen(req, timeout=self.timeout_seconds) as resp:
                    status = resp.status
                    elapsed_ms = (time.monotonic() - start) * 1000

                results.append(HealthCheckResult(
                    endpoint=url,
                    http_status=status,
                    response_time_ms=round(elapsed_ms, 2),
                    passed=status == 200 and elapsed_ms < 500,
                ))
            except (urllib.error.URLError, TimeoutError) as e:
                results.append(HealthCheckResult(
                    endpoint=url,
                    http_status=0,
                    response_time_ms=round((time.monotonic() - start) * 1000, 2),
                    passed=False,
                ))
                logger.warning("Health check failed for %s: %s", url, e)

        return results

    def check_slo_metrics(self, service_name: str, window_minutes: int = 5) -> list[SLOViolation]:
        """Check SLO metrics via PromQL queries.

        Monitors error budget burn rate, p99 latency, and availability.
        Triggers on sustained violations for the evaluation window.

        Args:
            service_name: Service identifier matching Prometheus labels.
            window_minutes: Time window for SLO evaluation in minutes.

        Returns:
            List of SLO violations found (empty if all SLOs are met).
        """
        now = int(time.time())
        queries = {
            "error_rate": (
                f'sum(rate(http_requests_total{{service="{service_name}",'
                f'code=~"5.."}}[{window_minutes}m])) / '
                f'sum(rate(http_requests_total{{service="{service_name}"}}[{windo

…(truncated)