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
| Tool | Free Tier | Interval | Alerts |
|---|---|---|---|
| UptimeRobot | 50 monitors | 5 min | Email, Slack |
| Better Stack | 10 monitors | 3 min | Email, Slack |
| Freshping | 50 monitors | 1 min |
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:
- Adapt it to your inference endpoint (it could be
/api/inference,/api/chat,/api/query, etc.) - Adjust the
validate_bodyfor your app's response structure - Adjust the
max_latency_msfor 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
- Go to uptimerobot.com and create a free account
- New Monitor:
- Type: HTTP(s)
- Friendly Name: "My AI App - Health"
- URL:
https://your-app.railway.app/health - Monitoring Interval: 5 minutes
- Configure Alert Contacts:
- Email: your email
- (Optional) Webhook: Slack incoming webhook
- 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
- Kubernetes Health Checks — Liveness vs Readiness — Health check concepts (applies outside K8s)
- UptimeRobot — Free uptime monitoring
- Better Stack Uptime — Monitoring with status pages
- Testing in Production — Charity Majors — Why and how to test in production
- Smoke Testing — Martin Fowler — Definition and best practices
- Health Check Patterns — Microsoft — Health check monitoring patterns