2. End-to-End Tests

Description

End-to-end (E2E) tests run the complete flow: input → formatter → LLM → parse → output. No mocks on the critical path. This capsule covers how to structure E2E tests for AI apps, implement mandatory budget controls, which assertions to use when the output varies, and how to integrate E2E into your CI/CD pipeline without them becoming costly or fragile.

What an E2E test is in an AI context

An E2E test calls real components — including the LLM:

Unit test (M2):
  input → [MOCK LLM] → parse → output
  
E2E test (M3):
  input → formatter → [REAL LLM] → parse → processor → output
                           ↑
                    real API call → real cost → real time

What an E2E verifies that unit tests cannot:

• The prompt works with the real LLM (not just with the mock)
• The complete chain produces output usable by the end user
• Timeouts, rate limits, and real errors are handled well
• The model produces outputs that your parser handles correctly

The structure of a well-formed E2E test

import pytest
import os

@pytest.mark.integration
@pytest.mark.e2e
@pytest.mark.skipif(
    not os.getenv("OPENAI_API_KEY"),
    reason="OPENAI_API_KEY not configured — E2E tests require the real API"
)
def test_analyze_sentiment_e2e():
    """
    Complete flow: text → real LLM → parse → validated output.
    
    This test verifies that the complete system produces a usable
    result, not just that the structure is correct (that's already
    covered by the M2 contract tests).
    """
    # Arrange: real, controlled input
    text = "I just used this product for the first time and it's absolutely incredible."
    
    # Act: complete flow, no mocks
    result = analyze_sentiment(text)  # Calls the real LLM
    
    # Assert: flexible — not exact equality, but properties
    # Structure (redundant with contract tests, but useful for E2E)
    assert isinstance(result, dict), "The result must be a dict"
    assert "sentiment" in result
    assert "score" in result
    
    # Properties of the real LLM output
    assert result["sentiment"] in ["positive", "negative", "neutral"]
    assert 0.0 <= result["score"] <= 1.0
    
    # For clearly positive text, the score must not be very low
    # (this is a quality assertion, not just a structural one)
    assert result["score"] >= 0.5, \
        f"For positive text, the score should be >=0.5, it's {result['score']}"

Budget controls: complete implementation

Budget controls are mandatory in E2E tests. Without them, an accidental run can cost tens of dollars.

Basic implementation: session budget

# tests/conftest.py

import pytest
import os
import threading

class BudgetTracker:
    """Tracks the spending on API calls during the test session."""
    
    # Prices per 1M tokens (gpt-4o-mini, January 2025)
    PRICES = {
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "gpt-4o":       {"input": 2.50, "output": 10.00},
        "gpt-4":        {"input": 30.00, "output": 60.00},
    }
    
    def __init__(self, max_usd: float = 0.50):
        self.max_usd = max_usd
        self.spent = 0.0
        self._lock = threading.Lock()
        self.calls = []
    
    def add_cost(self, model: str, prompt_tokens: int, completion_tokens: int):
        prices = self.PRICES.get(model, self.PRICES["gpt-4o-mini"])
        cost = (
            prompt_tokens / 1_000_000 * prices["input"] +
            completion_tokens / 1_000_000 * prices["output"]
        )
        with self._lock:
            self.spent += cost
            self.calls.append({
                "model": model,
                "prompt_tokens": prompt_tokens,
                "completion_tokens": completion_tokens,
                "cost": cost
            })
        return cost
    
    def check_budget(self):
        if self.spent >= self.max_usd:
            raise pytest.skip.Exception(
                f"E2E budget exceeded: ${self.spent:.4f} >= ${self.max_usd:.2f}"
            )
    
    def report(self):
        return {
            "total_spent": self.spent,
            "max_budget": self.max_usd,
            "remaining": self.max_usd - self.spent,
            "total_calls": len(self.calls)
        }

@pytest.fixture(scope="session")
def e2e_budget():
    """Budget shared across the entire integration test session."""
    max_budget = float(os.getenv("E2E_BUDGET_USD", "0.50"))
    tracker = BudgetTracker(max_usd=max_budget)
    yield tracker
    # At the end of the session, report the spending
    report = tracker.report()
    print(f"\n💰 E2E Budget Report: ${report['total_spent']:.4f} / ${report['max_budget']:.2f}")
    print(f"   {report['total_calls']} API calls made")

@pytest.fixture
def openai_client_e2e(e2e_budget):
    """
    OpenAI client for E2E tests with budget tracking.
    
    Automatic skip if the budget is exceeded.
    """
    import openai
    
    e2e_budget.check_budget()  # Skip if budget exceeded
    
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        pytest.skip("OPENAI_API_KEY not configured")
    
    client = openai.OpenAI(api_key=api_key)
    return client, e2e_budget

Using the budget tracker in tests

@pytest.mark.integration
def test_sentiment_with_budget(openai_client_e2e):
    client, budget = openai_client_e2e
    
    # Make the LLM call
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Analyze: 'I love this product'"}],
        temperature=0.0,
        max_tokens=200
    )
    
    # Register the cost
    budget.add_cost(
        model="gpt-4o-mini",
        prompt_tokens=response.usage.prompt_tokens,
        completion_tokens=response.usage.completion_tokens
    )
    
    raw = response.choices[0].message.content
    result = parse_json_response(raw)
    processed = process_sentiment_output(result)
    
    assert processed["sentiment"] in ["positive", "negative", "neutral"]
    assert 0 <= processed["score"] <= 1

Robust assertions for E2E

The art of E2E tests is in the assertions. Too strict → fragile tests. Too weak → they don't detect problems.

Assertion levels

# ─── Level 1: Structure (mandatory minimum) ───
def assertions_level_1(result):
    """These assertions are identical to the M2 contract tests."""
    assert isinstance(result, dict)
    assert "sentiment" in result
    assert result["sentiment"] in ["positive", "negative", "neutral"]
    assert 0 <= result["score"] <= 1

# ─── Level 2: Quality properties ───
def assertions_level_2(result, input_text):
    """Assertions that verify the output makes sense."""
    # If the text has clearly positive words, the score must be high
    clearly_positive_words = ["incredible", "excellent", "fantastic", "perfect", "love"]
    if any(w in input_text.lower() for w in clearly_positive_words):
        assert result["score"] >= 0.6, \
            f"For positive text, the score must be >=0.6, it's {result['score']}"
    
    # The explanation must not be empty for texts with a clear sentiment
    if result["score"] < 0.3 or result["score"] > 0.7:
        assert len(result.get("explanation", "")) > 0

# ─── Level 3: Keywords from the input ───
def assertions_level_3(result, input_text):
    """Verify that the output is related to the input."""
    # At least one keyword from the input must appear in the keywords or explanation
    input_words = set(input_text.lower().split()) - {"the", "a", "an", "is", "of", "to"}
    output_text = (result.get("explanation", "") + " ".join(result.get("keywords", []))).lower()
    
    relevant_found = any(word in output_text for word in input_words if len(word) > 3)
    assert relevant_found, \
        "The output doesn't seem related to the input — possible hallucination"

Assertions by prompt type

# For SUMMARY prompts:
def assert_summary_quality(result: dict, original_text: str):
    # Reasonable length (not longer than the original)
    assert len(result["summary"]) <= len(original_text)
    # Minimally informative
    assert len(result["summary"].split()) >= 5
    # Reasonable confidence
    assert result["confidence"] >= 0.5, "Confidence too low for standard text"

# For CLASSIFICATION prompts:
def assert_classification_quality(result: dict):
    # High confidence if the category is clear
    if result["confidence"] > 0.9:
        assert result["category"] != "unknown"
    # Tags must be relevant (at least 1 if there's a category)
    if result["category"] != "unknown":
        assert len(result.get("tags", [])) >= 1

# For EXTRACTION prompts:
def assert_extraction_quality(result: dict, input_text: str):
    # The extracted entities must appear in the input
    for entity in result.get("entities", []):
        assert entity["text"].lower() in input_text.lower(), \
            f"Entity '{entity['text']}' not found in the input — possible hallucination"

Conditional skip: the basis of E2E in CI

E2E tests must be optional, not block the pipeline:

# Option 1: skipif based on an environment variable
import pytest
import os

requires_api_key = pytest.mark.skipif(
    not os.getenv("OPENAI_API_KEY"),
    reason="Requires OPENAI_API_KEY for E2E"
)

requires_integration = pytest.mark.skipif(
    os.getenv("SKIP_INTEGRATION", "true").lower() == "true",
    reason="Integration tests disabled (SKIP_INTEGRATION=true)"
)

# Usage:
@requires_api_key
@requires_integration
@pytest.mark.integration
def test_sentiment_e2e():
    ...

# Option 2: fixture that skips
@pytest.fixture
def integration_client():
    """Only creates the client if we have an API key and are in integration mode."""
    api_key = os.getenv("OPENAI_API_KEY")
    run_integration = os.getenv("RUN_INTEGRATION", "false").lower() == "true"
    
    if not api_key or not run_integration:
        pytest.skip("E2E tests require OPENAI_API_KEY and RUN_INTEGRATION=true")
    
    import openai
    return openai.OpenAI(api_key=api_key)

def test_with_conditional_fixture(integration_client):
    # If there's no key, this test is skipped automatically
    response = integration_client.chat.completions.create(...)

Economical model for E2E

Always use the cheapest model for tests:

# pytest.ini or conftest.py
E2E_MODEL = os.getenv("E2E_MODEL", "gpt-4o-mini")  # Default: the cheapest

# Never in E2E tests (unless there's a specific reason):
# ❌ "gpt-4"          → ~200x more expensive than gpt-4o-mini
# ❌ "gpt-4o"         → ~17x more expensive
# ✅ "gpt-4o-mini"    → Optimal quality/cost balance for testing

def test_e2e_with_economic_model(integration_client):
    response = integration_client.chat.completions.create(
        model="gpt-4o-mini",     # Always explicit
        messages=[...],
        temperature=0.0,          # No extra randomness
        max_tokens=200,           # Limit output tokens
    )

Comparison: E2E vs Unit vs standard Integration

Timeout: protecting against hung tests

# pip install pytest-timeout

@pytest.mark.integration
@pytest.mark.timeout(30)  # Maximum 30 seconds
def test_sentiment_e2e_with_timeout(integration_client):
    result = analyze_sentiment("test text", client=integration_client)
    assert "sentiment" in result

# Or configure a global timeout for all integration tests:
# pytest.ini
# [pytest]
# timeout = 30

E2E in CI/CD: recommended configuration

# .github/workflows/ci.yml

jobs:
  unit-tests:
    name: Unit Tests (always)
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Run unit tests
        run: pytest -m "not integration" -v --tb=short
        # Always runs, no API key, deterministic

  integration-tests:
    name: Integration Tests (main only)
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'  # Only on main
    env:
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
      RUN_INTEGRATION: "true"
      E2E_BUDGET_USD: "0.25"  # Maximum budget per run
    steps:
      - uses: actions/checkout@v3
      - name: Run integration tests
        run: pytest -m integration -v --tb=short --timeout=60
        # Only on main, with API key, with controlled budget

Exercises

Exercise 1: Write the complete E2E

Write a complete E2E test for the FastAPI app's /analyze endpoint. Include: conditional skip, budget tracking, property assertions.

import pytest
import os
from fastapi.testclient import TestClient
from app.main import app

@pytest.mark.integration
@pytest.mark.e2e
@pytest.mark.skipif(
    not os.getenv("OPENAI_API_KEY") or not os.getenv("RUN_INTEGRATION"),
    reason="Requires OPENAI_API_KEY and RUN_INTEGRATION=true"
)
@pytest.mark.timeout(30)
def test_analyze_endpoint_e2e():
    """E2E test of the /analyze endpoint using the real LLM."""
    client = TestClient(app)
    
    response = client.post("/analyze", json={
        "text": "I love this product, it's absolutely fantastic"
    })
    
    assert response.status_code == 200
    data = response.json()
    
    # Property assertions (not exact equality)
    assert data["sentiment"] in ["positive", "negative", "neutral"]
    assert 0 <= data["score"] <= 1
    assert isinstance(data["keywords"], list)
    
    # For clearly positive text:
    assert data["score"] >= 0.6, "Positive text must have score >= 0.6"
    assert data["sentiment"] == "positive", "Positive text must be classified as positive"

Exercise 2: Implement a budget fixture

Implement a simplified version of the budget tracker that skips if you spend more than $0.10:

@pytest.fixture(scope="session")
def simple_budget():
    budget = {"spent": 0.0, "max": 0.10}
    yield budget
    print(f"\nE2E spend: ${budget['spent']:.4f}")

def register_cost(budget: dict, tokens: int, model: str = "gpt-4o-mini"):
    cost_per_token = 0.00000015  # gpt-4o-mini approx
    budget["spent"] += tokens * cost_per_token
    if budget["spent"] >= budget["max"]:
        pytest.skip(f"Budget exceeded: ${budget['spent']:.4f}")

@pytest.mark.integration
def test_with_simple_budget(simple_budget, integration_client):
    response = integration_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Hello"}],
        max_tokens=50
    )
    register_cost(simple_budget, response.usage.total_tokens)
    assert response.choices[0].message.content

Exercise 3: Assertions for a classification prompt

For a document classification prompt (returns category, confidence, tags), design 5 robust assertions for E2E:

def assert_classification_e2e(result: dict, input_text: str):
    # 1. Structure (always)
    assert "category" in result and "confidence" in result and "tags" in result
    
    # 2. Allowed values
    VALID_CATEGORIES = ["technology", "science", "sports", "politics", "entertainment", "general"]
    assert result["category"] in VALID_CATEGORIES, f"Invalid category: {result['category']}"
    
    # 3. Confidence range
    assert 0.0 <= result["confidence"] <= 1.0
    
    # 4. Non-empty tags when there's a clear category
    if result["confidence"] >= 0.8:
        assert len(result["tags"]) >= 1, "High confidence must have at least one tag"
    
    # 5. Tags are relevant strings (not numeric or empty)
    for tag in result["tags"]:
        assert isinstance(tag, str) and len(tag.strip()) > 0

Exercise 4: E2E for an error flow

Write an E2E test that verifies the app correctly handles empty text:

@pytest.mark.integration
@pytest.mark.e2e
@pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="Requires an API key")
def test_empty_text_handled_e2e():
    """E2E: empty text must return a controlled error, not crash."""
    from fastapi.testclient import TestClient
    from app.main import app
    
    client = TestClient(app)
    
    response = client.post("/analyze", json={"text": ""})
    
    # FastAPI must validate the input before calling the LLM
    assert response.status_code == 422  # Unprocessable Entity (Pydantic validation)
    # The LLM must NOT be called — the error is pre-LLM

Exercise 5: Timeout and retry

Combine a 30s timeout with a retry of 2 attempts for an E2E test that can be slow:

@pytest.mark.integration
@pytest.mark.timeout(30)
@pytest.mark.flaky(reruns=2, reruns_delay=3)  # Requires pytest-rerunfailures
def test_slow_endpoint_with_retry():
    """Test that can be slow — with retry and timeout."""
    result = analyze_sentiment("test text")
    
    # Flexible assertions to avoid false negatives
    assert result["sentiment"] in ["positive", "negative", "neutral"]
    assert 0 <= result["score"] <= 1

Summary

• E2E tests = complete flow with the real LLM — they verify what mocks cannot
• Budget controls are mandatory — implement the tracker before writing tests
• Robust assertions: properties, ranges, keywords — not exact equality
• Conditional skip: by API key, by environment variable — E2E must not block CI
• Economical model: always gpt-4o-mini for tests, never gpt-4 without a reason
• Timeout: always @pytest.mark.timeout(30) — a hung LLM must not hang the pipeline

Additional resources

1. pytest-timeout — Per-test timeouts
2. pytest-rerunfailures — Automatic retry
3. OpenAI Usage dashboard — Monitor real costs
4. FastAPI TestClient — For endpoint E2E
5. GitHub Actions Secrets — For API keys in CI
6. pytest skipif — Conditional skip
7. OpenAI API Rate Limits — Understand the API limits