Module 8: Capstone Project — Deployed AI System

4. Post-Deploy Validation — Verify That It Really Works

Description

In this capsule you'll implement post-deploy validation that goes beyond "the app responds 200 OK." You'll implement health checks that verify dependencies, smoke tests that send real prompts and verify that inference works, and automated validation scripts. An endpoint that returns 200 but whose inference is broken is not a successful deploy.

Context: The previous capsule automated the deployment. But a pipeline that deploys without verifying is dangerous — you can have a system "in production" that doesn't work. This capsule closes that gap. By the end, you'll know with certainty whether your deploy is successful or needs a rollback.


The Difference Between "Online" and "Working"

The problem with the basic health check

# INSUFFICIENT health check
@app.get("/health")
async def health():
    return {"status": "ok"}

# This endpoint returns 200 even if:
# - The OpenAI API key is invalid
# - The configured model doesn't exist
# - The vector database is not accessible
# - The Redis cache is down
# - The prompt template has a syntax error

A health check that only verifies the FastAPI process is running is like a doctor who only checks that the patient is breathing. You need deeper tests.

The three levels of validation

Level 1: LIVENESS — Is the process running?
├── HTTP 200 on /health
├── Verifies: the container didn't crash
└── Doesn't verify: anything functional

Level 2: READINESS — Can it receive traffic?
├── Dependencies connected (valid API keys, accessible DB)
├── Verifies: the system is ready to process
└── Doesn't verify: that the logic works correctly

Level 3: SMOKE TEST — Does inference work end-to-end?
├── Sends a real prompt, receives a valid response
├── Verifies: the complete flow works
└── It's the definitive validation

Level 1: Health Check with Dependencies

Health check that verifies the whole system

# src/health.py
import time
from fastapi import APIRouter
from src.config import get_settings

router = APIRouter()

async def check_openai_connection() -> dict:
    """Verifies that the OpenAI API key is valid."""
    settings = get_settings()
    try:
        from openai import AsyncOpenAI
        client = AsyncOpenAI(api_key=settings.openai_api_key)
        models = await client.models.list()
        return {"status": "connected", "models_available": True}
    except Exception as e:
        return {"status": "error", "detail": str(e)[:100]}

async def check_vector_db() -> dict:
    """Verifies that the vector database is accessible."""
    try:
        # Adapt to your vector DB (ChromaDB, Pinecone, etc.)
        import chromadb
        client = chromadb.Client()
        collections = client.list_collections()
        return {"status": "connected", "collections": len(collections)}
    except Exception as e:
        return {"status": "error", "detail": str(e)[:100]}

@router.get("/health")
async def health_check():
    """Liveness check — is the process running?"""
    return {
        "status": "healthy",
        "timestamp": time.time(),
        "version": get_settings().version,
    }

@router.get("/health/ready")
async def readiness_check():
    """Readiness check — are the dependencies ready?"""
    settings = get_settings()
    checks = {}

    checks["openai"] = await check_openai_connection()
    checks["config"] = {
        "status": "ok",
        "environment": settings.environment,
        "model": settings.openai_model,
    }

    all_healthy = all(
        c.get("status") in ("connected", "ok")
        for c in checks.values()
    )

    return {
        "status": "ready" if all_healthy else "degraded",
        "checks": checks,
        "timestamp": time.time(),
    }

Register it in the app

# src/main.py
from fastapi import FastAPI
from src.health import router as health_router

app = FastAPI(title="AI System")
app.include_router(health_router)

Local verification

# Liveness (should be fast, <100ms)
$ curl http://localhost:8000/health
{"status":"healthy","timestamp":1741456800.0,"version":"1.0.0"}

# Readiness (may take 1-2s due to the checks)
$ curl http://localhost:8000/health/ready
{
  "status": "ready",
  "checks": {
    "openai": {"status": "connected", "models_available": true},
    "config": {"status": "ok", "environment": "development", "model": "gpt-4o-mini"}
  },
  "timestamp": 1741456801.0
}

Level 2: Inference Smoke Tests

What a smoke test is

A smoke test sends a real request to the deployed system and verifies that the response is correct. It's not a unit test — it's an end-to-end verification against production.

Smoke test flow:
1. Send a known prompt to the inference endpoint
2. Verify that the response has the expected structure
3. Verify that the response contains coherent content
4. Measure latency and verify that it's within the target
5. PASS or FAIL

Smoke test script

# scripts/smoke_test.py
"""
Smoke tests to validate a deployment.
Run after each deploy:
    python scripts/smoke_test.py https://your-app.platform.app
"""
import sys
import time
import json
import urllib.request
import urllib.error

class SmokeTestRunner:
    def __init__(self, base_url: str):
        self.base_url = base_url.rstrip("/")
        self.results = []

    def run_test(self, name: str, method: str, path: str,
                 body: dict = None, expected_status: int = 200,
                 validate_body: callable = None,
                 max_latency_ms: int = 5000) -> bool:
        """Runs an individual smoke test."""
        url = f"{self.base_url}{path}"
        start = time.time()

        try:
            data = json.dumps(body).encode() if body else None
            headers = {"Content-Type": "application/json"} if body else {}
            req = urllib.request.Request(url, data=data, headers=headers, method=method)
            response = urllib.request.urlopen(req, timeout=30)
            status = response.status
            response_body = json.loads(response.read().decode())

        except urllib.error.HTTPError as e:
            status = e.code
            response_body = {"error": str(e)}
        except Exception as e:
            status = 0
            response_body = {"error": str(e)}

        latency_ms = (time.time() - start) * 1000

        passed = True
        errors = []

        if status != expected_status:
            passed = False
            errors.append(f"Expected status {expected_status}, got {status}")

        if latency_ms > max_latency_ms:
            passed = False
            errors.append(f"Latency {latency_ms:.0f}ms > {max_latency_ms}ms target")

        if validate_body and status == expected_status:
            try:
                body_valid = validate_body(response_body)
                if not body_valid:
                    passed = False
                    errors.append("Body validation failed")
            except Exception as e:
                passed = False
                errors.append(f"Body validation error: {e}")

        result = {
            "name": name,
            "passed": passed,
            "status": status,
            "latency_ms": round(latency_ms),
            "errors": errors,
        }
        self.results.append(result)

        icon = "PASS" if passed else "FAIL"
        print(f"  [{icon}] {name}{status}{latency_ms:.0f}ms")
        if errors:
            for e in errors:
                print(f"         {e}")

        return passed

    def report(self) -> bool:
        """Prints a summary and returns True if all passed."""
        total = len(self.results)
        passed = sum(1 for r in self.results if r["passed"])
        failed = total - passed

        print(f"\nResults: {passed}/{total} passed, {failed} failed")

        if failed > 0:
            print("\nFailed tests:")
            for r in self.results:
                if not r["passed"]:
                    print(f"  - {r['name']}: {', '.join(r['errors'])}")

        return failed == 0


def run_smoke_tests(base_url: str) -> bool:
    """Runs all smoke tests."""
    runner = SmokeTestRunner(base_url)

    print(f"Running smoke tests against: {base_url}\n")

    # Test 1: Liveness
    runner.run_test(
        name="Liveness check",
        method="GET",
        path="/health",
        max_latency_ms=1000,
        validate_body=lambda b: b.get("status") == "healthy",
    )

    # Test 2: Readiness
    runner.run_test(
        name="Readiness check",
        method="GET",
        path="/health/ready",
        max_latency_ms=3000,
        validate_body=lambda b: b.get("status") in ("ready", "degraded"),
    )

    # Test 3: Inference - basic response
    runner.run_test(
        name="Inference - basic prompt",
        method="POST",
        path="/api/inference",
        body={"prompt": "Respond with exactly one word: hello"},
        max_latency_ms=10000,
        validate_body=lambda b: "response" in b and len(b["response"]) > 0,
    )

    # Test 4: Inference - response structure
    runner.run_test(
        name="Inference - response structure",
        method="POST",
        path="/api/inference",
        body={"prompt": "What is 2+2?"},
        max_latency_ms=10000,
        validate_body=lambda b: all(k in b for k in ["response", "model"]),
    )

    # Test 5: Error handling - empty prompt
    runner.run_test(
        name="Error handling - empty prompt",
        method="POST",
        path="/api/inference",
        body={"prompt": ""},
        expected_status=422,
        max_latency_ms=1000,
    )

    return runner.report()


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python scripts/smoke_test.py <base_url>")
        print("Example: python scripts/smoke_test.py https://my-app.railway.app")
        sys.exit(1)

    url = sys.argv[1]
    success = run_smoke_tests(url)
    sys.exit(0 if success else 1)

Execution

# Local
$ python scripts/smoke_test.py http://localhost:8000

Running smoke tests against: http://localhost:8000

  [PASS] Liveness check — 200 — 12ms
  [PASS] Readiness check — 200 — 1245ms
  [PASS] Inference - basic prompt — 200 — 2340ms
  [PASS] Inference - response structure — 200 — 1890ms
  [PASS] Error handling - empty prompt — 422 — 8ms

Results: 5/5 passed, 0 failed

# Production
$ python scripts/smoke_test.py https://my-app.railway.app

Running smoke tests against: https://my-app.railway.app

  [PASS] Liveness check — 200 — 89ms
  [PASS] Readiness check — 200 — 1567ms
  [PASS] Inference - basic prompt — 200 — 3210ms
  [PASS] Inference - response structure — 200 — 2890ms
  [PASS] Error handling - empty prompt — 422 — 45ms

Results: 5/5 passed, 0 failed

Integration in the CI/CD Pipeline

Smoke tests as a validation job

# In .github/workflows/deploy.yml
validate:
  needs: deploy
  runs-on: ubuntu-latest
  timeout-minutes: 5
  steps:
    - uses: actions/checkout@v4

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

    - name: Wait for deployment
      run: sleep 60

    - name: Run smoke tests
      run: python scripts/smoke_test.py "${{ vars.PRODUCTION_URL }}"

    - name: Upload results
      if: always()
      run: echo "Smoke test completed at $(date -u)"

Bash validation script (lightweight alternative)

#!/bin/bash
# scripts/validate-deploy.sh
# Usage: ./scripts/validate-deploy.sh https://my-app.railway.app

set -e

BASE_URL="${1:?Usage: $0 <base_url>}"
PASSED=0
FAILED=0

check() {
    local name="$1"
    local expected_status="$2"
    local method="$3"
    local path="$4"
    local body="$5"

    if [ -n "$body" ]; then
        response=$(curl -s -w "\n%{http_code}\n%{time_total}" \
            -X "$method" "$BASE_URL$path" \
            -H "Content-Type: application/json" \
            -d "$body")
    else
        response=$(curl -s -w "\n%{http_code}\n%{time_total}" \
            -X "$method" "$BASE_URL$path")
    fi

    http_code=$(echo "$response" | tail -2 | head -1)
    latency=$(echo "$response" | tail -1)
    body_resp=$(echo "$response" | head -n -2)

    latency_ms=$(echo "$latency * 1000" | bc | cut -d. -f1)

    if [ "$http_code" = "$expected_status" ]; then
        echo "  [PASS] $name$http_code${latency_ms}ms"
        PASSED=$((PASSED + 1))
    else
        echo "  [FAIL] $name — expected $expected_status, got $http_code${latency_ms}ms"
        echo "         Response: $body_resp"
        FAILED=$((FAILED + 1))
    fi
}

echo "Validating deployment: $BASE_URL"
echo ""

check "Health check" "200" "GET" "/health"
check "Readiness check" "200" "GET" "/health/ready"
check "Inference" "200" "POST" "/api/inference" \
    '{"prompt": "Say OK"}'
check "Error handling" "422" "POST" "/api/inference" \
    '{"prompt": ""}'

echo ""
echo "Results: $PASSED passed, $FAILED failed"

if [ "$FAILED" -gt 0 ]; then
    echo "DEPLOYMENT VALIDATION FAILED"
    exit 1
fi

echo "DEPLOYMENT VALIDATED SUCCESSFULLY"
chmod +x scripts/validate-deploy.sh
./scripts/validate-deploy.sh https://my-app.railway.app

Continuous Validation (Scheduled)

Periodic health checks with GitHub Actions

# .github/workflows/health-monitor.yml
name: Health Monitor

on:
  schedule:
    - cron: '*/30 * * * *'  # Every 30 minutes
  workflow_dispatch:

jobs:
  health-check:
    runs-on: ubuntu-latest
    timeout-minutes: 2
    steps:
      - name: Check production health
        run: |
          status=$(curl -s -o /dev/null -w "%{http_code}" \
            "${{ vars.PRODUCTION_URL }}/health")
          if [ "$status" != "200" ]; then
            echo "ALERT: Production health check failed: $status"
            exit 1
          fi
          echo "Production healthy at $(date -u)"

Free external monitoring

ToolFree TierIntervalAlerts
UptimeRobot50 monitors5 minEmail, Slack
Better Stack10 monitors3 minEmail, Slack
Freshping50 monitors1 minEmail
Recommended configuration in UptimeRobot:
├── Monitor 1: GET /health (every 5 min)
├── Monitor 2: GET /health/ready (every 15 min)
├── Alert: Email + Slack when it fails 2 times in a row
└── Status page: public for stakeholders (optional)

Troubleshooting

Problem 1: "The inference smoke test always fails on timeout"

Cause: The first request after a deploy has a long cold start (the platform is starting the container).

Solution:

# Warm-up request before the smoke tests
def warmup(base_url: str, retries: int = 5, delay: int = 10):
    """Sends warm-up requests until the service responds."""
    for i in range(retries):
        try:
            req = urllib.request.Request(f"{base_url}/health")
            response = urllib.request.urlopen(req, timeout=15)
            if response.status == 200:
                print(f"  Service ready after {i+1} attempts")
                return True
        except Exception:
            pass
        print(f"  Warming up... attempt {i+1}/{retries}")
        time.sleep(delay)
    return False

# In run_smoke_tests:
if not warmup(base_url):
    print("Service not available after warm-up")
    return False

Problem 2: "Health check passes but readiness shows OpenAI as an error"

Cause: The API key isn't configured on the production platform, or it's expired.

Solution:

# Verify the key in production
curl -s https://api.openai.com/v1/models \
    -H "Authorization: Bearer $OPENAI_API_KEY" | python -m json.tool | head -5

# If it gives a 401 error: the key is invalid
# If it gives a connection error: the platform blocks outbound requests

# Verify on the platform
# Railway: railway variables | grep OPENAI
# Fly.io: flyctl secrets list

Problem 3: "The smoke tests pass the first time but fail intermittently"

Cause: OpenAI API rate limiting, cold starts on the free tier, or insufficient memory.

Solution:

# Add retry logic to the smoke test runner
def run_test_with_retry(self, max_retries=2, **kwargs):
    for attempt in range(max_retries + 1):
        if self.run_test(**kwargs):
            return True
        if attempt < max_retries:
            print(f"         Retrying in 5s...")
            time.sleep(5)
    return False

Problem 4: "The validation script works locally but fails in CI"

Cause: Network differences between GitHub Actions runners and your local machine.

Solution:

# Give more time in CI
- name: Wait for deployment
  run: sleep 90  # 90 seconds instead of 45

# Increase timeouts
- name: Run smoke tests
  run: python scripts/smoke_test.py "${{ vars.PRODUCTION_URL }}"
  timeout-minutes: 5

Hands-On Exercises

Exercise 1: Health check with dependencies

Implement the three-level health check (liveness, readiness) in your FastAPI app.

See solution
# src/health.py
import time
from fastapi import APIRouter

router = APIRouter(tags=["health"])

@router.get("/health")
async def liveness():
    return {"status": "healthy", "timestamp": time.time()}

@router.get("/health/ready")
async def readiness():
    from src.config import get_settings
    settings = get_settings()

    checks = {}

    # Check OpenAI
    try:
        from openai import AsyncOpenAI
        client = AsyncOpenAI(api_key=settings.openai_api_key)
        await client.models.list()
        checks["openai"] = {"status": "connected"}
    except Exception as e:
        checks["openai"] = {"status": "error", "detail": str(e)[:80]}

    # Check config
    errors = settings.validate_for_production()
    checks["config"] = {
        "status": "ok" if not errors else "error",
        "environment": settings.environment,
        "errors": errors,
    }

    all_ok = all(c["status"] in ("connected", "ok") for c in checks.values())
    return {
        "status": "ready" if all_ok else "degraded",
        "checks": checks,
    }

Verify:

curl -s http://localhost:8000/health | python -m json.tool
curl -s http://localhost:8000/health/ready | python -m json.tool

Exercise 2: Complete smoke test script

Create the scripts/smoke_test.py file with at least 4 tests (liveness, readiness, inference, error handling).

See solution

Use the complete script shown in this capsule's "Smoke test script" section. Make sure to:

  1. Adapt it to your inference endpoint (it could be /api/inference, /api/chat, /api/query, etc.)
  2. Adjust the validate_body for your app's response structure
  3. Adjust the max_latency_ms for realistic targets for your system
# Local test
python scripts/smoke_test.py http://localhost:8000

# Production test
python scripts/smoke_test.py https://your-app.railway.app

If you don't have the inference endpoint yet, use at least health and readiness. Add the inference tests when you implement the endpoint.

Exercise 3: Bash validation script

Create scripts/validate-deploy.sh as a lightweight alternative to the Python smoke test.

See solution

Use the bash script shown in this capsule's "Bash validation script" section. Steps:

# Create the script
touch scripts/validate-deploy.sh
chmod +x scripts/validate-deploy.sh

# Copy the content of the capsule's bash script

# Run locally
./scripts/validate-deploy.sh http://localhost:8000

# Run against production
./scripts/validate-deploy.sh https://your-app.railway.app

The bash script is useful because it doesn't require Python installed — it can run on any CI runner.

Exercise 4: Configure external monitoring

Register your production URL in UptimeRobot (or an alternative) and configure alerts.

See solution
  1. Go to uptimerobot.com and create a free account
  2. New Monitor:
    • Type: HTTP(s)
    • Friendly Name: "My AI App - Health"
    • URL: https://your-app.railway.app/health
    • Monitoring Interval: 5 minutes
  3. Configure Alert Contacts:
    • Email: your email
    • (Optional) Webhook: Slack incoming webhook
  4. Second Monitor:
    • Type: HTTP(s) — Keyword
    • URL: https://your-app.railway.app/health/ready
    • Keyword: "ready"
    • Monitoring Interval: 15 minutes

Result: you'll receive an email if your service is down for more than 5-10 minutes. It's the simplest form of external monitoring.


Summary

  • A health check that only returns 200 is not enough — it needs to verify dependencies
  • Three levels: liveness (alive?), readiness (ready?), smoke test (does inference work?)
  • Smoke tests send real prompts and verify responses — they're the definitive validation
  • The smoke test script should be executable both locally and against production
  • Integrate into CI/CD: the pipeline should fail if the smoke tests fail
  • Continuous monitoring: use UptimeRobot or similar for free periodic verification
  • A deploy without post-deploy validation is a blind deploy — you don't know if it works until a user complains

Additional Resources

  1. Kubernetes Health Checks — Liveness vs Readiness — Health check concepts (applies outside K8s)
  2. UptimeRobot — Free uptime monitoring
  3. Better Stack Uptime — Monitoring with status pages
  4. Testing in Production — Charity Majors — Why and how to test in production
  5. Smoke Testing — Martin Fowler — Definition and best practices
  6. Health Check Patterns — Microsoft — Health check monitoring patterns