7. Project: Integration Test Suite

Description

This is the Module 3 mini-project. You'll expand the test suite from Modules 1-2 with: integration tests that call the real LLM with budget controls, semantic similarity assertions, property-based tests with Hypothesis, flaky test management, and a three-environment setup (mock/sandbox/real). When you finish you'll have a complete suite: deterministic unit tests + integration tests that handle non-determinism robustly.

Project objectives

Upon completing this project you'll have:

1. Budget tracker that limits spending on API calls during tests
2. E2E tests with the real LLM and flexible assertions (maximum $0.50 per run)
3. At least 2 tests with a semantic similarity assertion
4. At least 2 tests with Hypothesis (property-based, with a mock)
5. Flaky management: retry on unstable tests, quarantine for the problematic ones
6. Environment setup: mock by default, integration when there's an API key
7. Basic CI/CD: GitHub Actions configuration

Project structure

your-project/
├── src/
│   └── app/
│       ├── __init__.py
│       ├── config.py
│       ├── parsers.py
│       ├── processors.py
│       ├── sentiment.py
│       └── main.py
├── tests/
│   ├── __init__.py
│   ├── helpers.py               # create_openai_chat_response, etc.
│   ├── semantic.py              # assert_semantically_similar, etc.
│   ├── conftest.py              # Shared fixtures: budget, clients
│   ├── unit/                    # M1-M2: contract, parsers, regression
│   │   ├── conftest.py
│   │   ├── contracts/
│   │   ├── parsers/
│   │   └── regression/
│   └── integration/             # M3: E2E, semantic, property
│       ├── __init__.py
│       ├── conftest.py          # Integration-specific fixtures
│       ├── test_e2e.py          # E2E with the real LLM
│       ├── test_semantic.py     # Semantic similarity assertions
│       └── test_property.py     # Property-based with Hypothesis
├── .github/
│   └── workflows/
│       └── tests.yml            # CI/CD config
├── pytest.ini
└── requirements.txt

Step 1: Update tests/semantic.py

# tests/semantic.py
import numpy as np
from functools import lru_cache
from typing import Optional

def cosine_similarity(a: list[float], b: list[float]) -> float:
    a_arr = np.array(a, dtype=float)
    b_arr = np.array(b, dtype=float)
    norm_a = np.linalg.norm(a_arr)
    norm_b = np.linalg.norm(b_arr)
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return float(np.dot(a_arr, b_arr) / (norm_a * norm_b))

@lru_cache(maxsize=1)
def _get_local_model():
    """Loads the sentence-transformers model once."""
    try:
        from sentence_transformers import SentenceTransformer
        return SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
    except ImportError:
        return None

_embedding_cache: dict[str, list[float]] = {}

def get_embedding_local(text: str) -> list[float]:
    """Local embedding with sentence-transformers. No API cost."""
    if text in _embedding_cache:
        return _embedding_cache[text]
    
    model = _get_local_model()
    if model is None:
        raise ImportError("sentence-transformers not installed. Run: pip install sentence-transformers")
    
    embedding = model.encode(text, normalize_embeddings=True)
    _embedding_cache[text] = embedding.tolist()
    return _embedding_cache[text]

def get_embedding_openai(text: str, client) -> list[float]:
    """Embedding via the OpenAI API."""
    if text in _embedding_cache:
        return _embedding_cache[text]
    
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text.strip()
    )
    embedding = response.data[0].embedding
    _embedding_cache[text] = embedding
    return embedding

def assert_semantically_similar(
    actual: str,
    expected: str,
    threshold: float = 0.80,
    client=None,
    message: str = None
) -> float:
    """
    Semantic assertion: fails if actual and expected are not similar enough.
    
    Args:
        actual: The real LLM output
        expected: The expected meaning (description of the expected content)
        threshold: Minimum acceptable similarity (default: 0.80)
        client: OpenAI client (if None, uses local sentence-transformers)
        message: Custom error message
    
    Returns:
        The computed similarity (useful for debugging)
    """
    if client is not None:
        emb_actual = get_embedding_openai(actual, client)
        emb_expected = get_embedding_openai(expected, client)
    else:
        emb_actual = get_embedding_local(actual)
        emb_expected = get_embedding_local(expected)
    
    similarity = cosine_similarity(emb_actual, emb_expected)
    
    default_msg = (
        f"Semantic similarity {similarity:.3f} < threshold {threshold}\n"
        f"  Actual:   '{actual[:100]}'\n"
        f"  Expected: '{expected[:100]}'"
    )
    
    assert similarity >= threshold, message or default_msg
    return similarity

Step 2: Update tests/conftest.py

# tests/conftest.py
import pytest
import os
import threading
from unittest.mock import MagicMock, AsyncMock
import json

from tests.helpers import create_openai_chat_response, create_sentiment_response

# ─── Budget Tracker ────────────────────────────────────────────────────────

class BudgetTracker:
    PRICES = {
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "gpt-4o":       {"input": 2.50, "output": 10.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) -> float:
        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, "cost": cost})
        return cost
    
    def check_budget(self):
        if self.spent >= self.max_usd:
            pytest.skip(f"E2E budget exceeded: ${self.spent:.4f} >= ${self.max_usd:.2f}")

@pytest.fixture(scope="session")
def e2e_budget():
    max_budget = float(os.getenv("E2E_BUDGET_USD", "0.50"))
    tracker = BudgetTracker(max_usd=max_budget)
    yield tracker
    print(f"\n💰 E2E Budget: ${tracker.spent:.4f} / ${tracker.max_usd:.2f} ({len(tracker.calls)} calls)")

# ─── Integration Client ─────────────────────────────────────────────────────

@pytest.fixture(scope="session")
def integration_client(e2e_budget):
    """
    Real OpenAI client for integration tests.
    Automatic skip if there's no API key or the budget was exceeded.
    """
    api_key = os.getenv("OPENAI_API_KEY")
    run_integration = os.getenv("RUN_INTEGRATION", "false").lower() == "true"
    
    if not api_key:
        pytest.skip("OPENAI_API_KEY not configured — skipping integration tests")
    
    if not run_integration and not os.getenv("FULL_VALIDATION"):
        pytest.skip("RUN_INTEGRATION not enabled — use RUN_INTEGRATION=true for integration")
    
    e2e_budget.check_budget()
    
    import openai
    return openai.OpenAI(api_key=api_key)

# ─── Mock Factories ─────────────────────────────────────────────────────────

@pytest.fixture
def make_sentiment_client():
    def _create(sentiment="neutral", score=0.5, explanation="Test analysis", keywords=None):
        if keywords is None:
            keywords = []
        client = MagicMock()
        client.chat.completions.create.return_value = create_sentiment_response(
            sentiment=sentiment, score=score, explanation=explanation, keywords=keywords
        )
        return client
    return _create

@pytest.fixture
def make_error_client():
    import openai
    def _create(error_type="generic"):
        client = MagicMock()
        errors = {
            "rate_limit": openai.RateLimitError("Rate limit", response=MagicMock(status_code=429), body={}),
            "timeout": openai.APITimeoutError(request=MagicMock()),
            "connection": openai.APIConnectionError(request=MagicMock()),
        }
        if error_type == "empty_response":
            client.chat.completions.create.return_value = create_openai_chat_response("")
        elif error_type in errors:
            client.chat.completions.create.side_effect = errors[error_type]
        else:
            client.chat.completions.create.side_effect = Exception(f"Error: {error_type}")
        return client
    return _create

Step 3: tests/integration/conftest.py

# tests/integration/conftest.py
import pytest
import os

@pytest.fixture(autouse=True)
def check_integration_available(request):
    """
    Auto-fixture: automatically verifies whether integration tests can run.
    Only acts on tests marked with @pytest.mark.integration.
    """
    if request.node.get_closest_marker("integration"):
        api_key = os.getenv("OPENAI_API_KEY")
        run_integration = (
            os.getenv("RUN_INTEGRATION", "false").lower() == "true" or
            os.getenv("FULL_VALIDATION", "false").lower() == "true"
        )
        
        if not api_key or not run_integration:
            pytest.skip(
                "Integration tests disabled. To enable:\n"
                "  export OPENAI_API_KEY=sk-...\n"
                "  export RUN_INTEGRATION=true"
            )

Step 4: tests/integration/test_e2e.py

# tests/integration/test_e2e.py
import pytest
import os
from app.sentiment import analyze_sentiment

@pytest.mark.integration
@pytest.mark.e2e
class TestSentimentE2E:
    """
    E2E tests for the sentiment analyzer with the real LLM.
    
    Assertions are flexible — they verify properties, not exact values.
    """
    
    @pytest.mark.timeout(30)
    def test_positive_sentiment_detection(self, integration_client, e2e_budget):
        """Clearly positive text → positive sentiment."""
        text = "I love this product, it's absolutely incredible and I recommend it."
        
        result = analyze_sentiment(text, client=integration_client)
        
        # Record the estimated cost
        e2e_budget.add_cost("gpt-4o-mini", prompt_tokens=100, completion_tokens=50)
        
        # Property assertions
        assert result["sentiment"] in ["positive", "negative", "neutral"]
        assert 0 <= result["score"] <= 1
        
        # For such positive text, the score must be high
        assert result["score"] >= 0.6, \
            f"For clearly positive text, score must be >=0.6. Got: {result['score']}"
    
    @pytest.mark.timeout(30)
    def test_negative_sentiment_detection(self, integration_client, e2e_budget):
        """Clearly negative text → negative sentiment."""
        text = "Terrible experience, the worst product I've bought. Horrible quality."
        
        result = analyze_sentiment(text, client=integration_client)
        e2e_budget.add_cost("gpt-4o-mini", prompt_tokens=100, completion_tokens=50)
        
        assert result["sentiment"] == "negative"
        assert result["score"] <= 0.4
    
    @pytest.mark.timeout(30)
    def test_neutral_factual_text(self, integration_client, e2e_budget):
        """Factual text with no emotional charge → neutral sentiment."""
        text = "The product arrived in a box. It has dimensions of 30x20x10 cm."
        
        result = analyze_sentiment(text, client=integration_client)
        e2e_budget.add_cost("gpt-4o-mini", prompt_tokens=80, completion_tokens=50)
        
        assert result["sentiment"] in ["neutral", "positive"]  # Can go either way
        assert isinstance(result["keywords"], list)
    
    @pytest.mark.timeout(30)
    def test_full_pipeline_no_crash(self, integration_client, e2e_budget):
        """The complete pipeline doesn't crash for any valid text."""
        texts = [
            "normal text",
            "How great!",
            "12345 numbers",
            "text in english is also fine",
            "Short text"
        ]
        
        for text in texts:
            result = analyze_sentiment(text, client=integration_client)
            e2e_budget.add_cost("gpt-4o-mini", prompt_tokens=60, completion_tokens=40)
            
            assert isinstance(result, dict), f"Pipeline didn't return a dict for: '{text}'"
            assert "sentiment" in result, f"Missing 'sentiment' key for: '{text}'"
    
    @pytest.mark.timeout(15)
    def test_short_text_handled(self, integration_client, e2e_budget):
        """Very short texts are handled without crashing."""
        result = analyze_sentiment("Good", client=integration_client)
        e2e_budget.add_cost("gpt-4o-mini", prompt_tokens=50, completion_tokens=40)
        
        assert result["sentiment"] in ["positive", "negative", "neutral"]

Step 5: tests/integration/test_semantic.py

# tests/integration/test_semantic.py
import pytest
from app.sentiment import analyze_sentiment
from tests.semantic import assert_semantically_similar

@pytest.mark.integration
class TestSentimentSemanticQuality:
    """
    Semantic quality tests using embeddings to compare meaning.
    
    They use local sentence-transformers (no additional API cost).
    """
    
    def test_positive_explanation_is_positive(self, integration_client, e2e_budget):
        """
        The explanation for positive text must be semantically positive.
        """
        result = analyze_sentiment(
            "Excellent quality, it exceeded all my expectations",
            client=integration_client
        )
        e2e_budget.add_cost("gpt-4o-mini", 100, 60)
        
        # Verify that the explanation is semantically positive
        assert_semantically_similar(
            actual=result.get("explanation", ""),
            expected="The text expresses satisfaction and positive appraisal",
            threshold=0.65,  # Flexible — many ways to express positivity
            client=None       # Local embeddings to save cost
        )
    
    def test_negative_explanation_is_negative(self, integration_client, e2e_budget):
        """
        The explanation for negative text must be semantically negative.
        """
        result = analyze_sentiment(
            "Terrible, disappointing, I wouldn't recommend it to anyone",
            client=integration_client
        )
        e2e_budget.add_cost("gpt-4o-mini", 100, 60)
        
        assert_semantically_similar(
            actual=result.get("explanation", ""),
            expected="The text uses negative language and expresses dissatisfaction",
            threshold=0.60,
            client=None
        )
    
    def test_explanations_for_opposite_sentiments_are_different(
        self, integration_client, e2e_budget
    ):
        """
        The explanations for opposite sentiments must be semantically different.
        """
        from tests.semantic import get_embedding_local, cosine_similarity
        
        result_pos = analyze_sentiment("Incredible, wonderful, perfect", client=integration_client)
        result_neg = analyze_sentiment("Terrible, horrible, disappointing", client=integration_client)
        e2e_budget.add_cost("gpt-4o-mini", 200, 120)
        
        if result_pos.get("explanation") and result_neg.get("explanation"):
            emb_pos = get_embedding_local(result_pos["explanation"])
            emb_neg = get_embedding_local(result_neg["explanation"])
            similarity = cosine_similarity(emb_pos, emb_neg)
            
            assert similarity < 0.70, (
                f"Explanations for opposite sentiments are too similar: {similarity:.3f}\n"
                f"Positive: {result_pos['explanation']}\n"
                f"Negative: {result_neg['explanation']}"
            )

Step 6: tests/integration/test_property.py

# tests/integration/test_property.py
import pytest
import json
from hypothesis import given, settings
import hypothesis.strategies as st
from unittest.mock import MagicMock
from tests.helpers import create_openai_chat_response

# Strategy for valid LLM outputs
valid_sentiment_output = st.fixed_dictionaries({
    "sentiment": st.sampled_from(["positive", "negative", "neutral"]),
    "score": st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
    "explanation": st.text(max_size=200),
    "keywords": st.lists(
        st.text(min_size=1, max_size=30, alphabet=st.characters(
            whitelist_categories=("L", "N", "Zs")
        )),
        max_size=5
    )
})

@given(llm_output=valid_sentiment_output)
@settings(max_examples=50)
def test_pipeline_contract_for_any_valid_llm_output(llm_output):
    """
    PROPERTY: For any valid LLM output (mocked),
    the pipeline produces a result that satisfies the contract.
    """
    from app.sentiment import analyze_sentiment
    
    mock_client = MagicMock()
    mock_client.chat.completions.create.return_value = create_openai_chat_response(
        json.dumps(llm_output, ensure_ascii=False)
    )
    
    result = analyze_sentiment("test text", client=mock_client)
    
    assert result["sentiment"] in ["positive", "negative", "neutral"]
    assert 0.0 <= result["score"] <= 1.0
    assert isinstance(result["keywords"], list)
    assert isinstance(result["explanation"], str)

@given(raw_score=st.floats(allow_nan=False, allow_infinity=False))
def test_processor_score_always_clamped(raw_score):
    """
    PROPERTY: For any input score, the processor always returns [0, 1].
    """
    from app.processors import process_sentiment_output
    
    result = process_sentiment_output({"sentiment": "neutral", "score": raw_score})
    
    assert 0.0 <= result["score"] <= 1.0, \
        f"Score {raw_score} → {result['score']} — not in [0, 1]"

@given(sentiment=st.text(max_size=100))
def test_processor_sentiment_always_valid(sentiment):
    """
    PROPERTY: For any input sentiment, the result is always one of the three valid ones.
    """
    from app.processors import process_sentiment_output
    
    result = process_sentiment_output({"sentiment": sentiment, "score": 0.5})
    
    assert result["sentiment"] in ["positive", "negative", "neutral"]

@given(raw_input=st.one_of(
    st.just(""),
    st.just("   "),
    st.just("\n\n"),
))
def test_parser_empty_always_raises(raw_input):
    """
    PROPERTY: Any empty/whitespace input always raises ValueError.
    """
    from app.parsers import parse_json_response
    
    with pytest.raises(ValueError):
        parse_json_response(raw_input)

@given(
    content=st.fixed_dictionaries({
        "x": st.integers(min_value=-1000, max_value=1000),
        "y": st.text(min_size=0, max_size=100)
    })
)
def test_parser_roundtrip(content):
    """
    PROPERTY: json.dumps → parse_json_response → same result.
    """
    from app.parsers import parse_json_response
    
    raw = json.dumps(content)
    result = parse_json_response(raw)
    
    assert result["x"] == content["x"]
    assert result["y"] == content["y"]

Step 7: Update pytest.ini

[pytest]
testpaths = tests
addopts = -v --tb=short

markers =
    unit: Unit tests (deterministic, no API calls)
    integration: Tests with the real LLM (require OPENAI_API_KEY and RUN_INTEGRATION=true)
    e2e: End-to-end tests of the complete flow
    contract: Prompt contract tests
    property: Property-based tests with Hypothesis
    semantic: Tests with semantic similarity assertions
    flaky: Tests with retry enabled due to LLM variance
    quarantine: Unstable tests — they don't run in regular CI
    smoke: Basic smoke tests

# By default: only unit tests (no integration, no quarantine)
# For integration: pytest -m integration
# For everything except quarantine: pytest -m "not quarantine"

Step 8: CI/CD with GitHub Actions

# .github/workflows/tests.yml
name: Tests

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
  schedule:
    - cron: "0 2 * * *"  # Nightly at 2am UTC

jobs:
  unit-tests:
    name: Unit Tests (always)
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: "3.11"
      - run: pip install -r requirements.txt
      - run: pytest -m "not integration and not quarantine" -v --tb=short
        # No API key, no cost, deterministic

  integration-tests:
    name: Integration Tests (main only)
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    env:
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
      RUN_INTEGRATION: "true"
      E2E_BUDGET_USD: "0.25"
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: "3.11"
      - run: pip install -r requirements.txt
      - run: pytest -m "integration and not quarantine" --timeout=60 -v --tb=short

  nightly-full:
    name: Full Test Suite (nightly)
    runs-on: ubuntu-latest
    if: github.event_name == 'schedule'
    env:
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
      RUN_INTEGRATION: "true"
      FULL_VALIDATION: "true"
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: "3.11"
      - run: pip install -r requirements.txt
      - run: pytest -v --timeout=120  # Includes quarantine for monitoring

Step 9: Final verification

# 1. Verify that the unit tests pass (without an API key)
pytest -m "not integration" -v --tb=short
# Expected: 50+ passed, 0 failed, 0 errors

# 2. With an API key: verify integration tests
export OPENAI_API_KEY=sk-...
export RUN_INTEGRATION=true
pytest -m integration -v --timeout=60 --tb=short
# Expected: the tests run (they don't get skipped), some may take 5-10s

# 3. Property-based tests
pytest -m property -v --tb=short
# Expected: tests run with multiple Hypothesis examples

# 4. Full coverage
pytest -m "not integration" --cov=app --cov-report=term-missing
# Expected: >80% coverage in parsers and processors

# 5. Timing: unit tests must be fast
time pytest -m "not integration" -q
# Expected: <15 seconds

Delivery checklist

Tests written

• At least 4 E2E tests with the real LLM in test_e2e.py
• At least 2 tests with semantic similarity in test_semantic.py
• At least 4 property-based tests in test_property.py (with a mock)
• Flaky management tests: at least 1 test with @pytest.mark.flaky
• Property tests for parsers and processors

Quality

• Budget tracker implemented and working
• Automatic skip when there's no API key
• Flexible assertions in integration tests (no exact equality)
• Property-based tests with clear, specific properties
• At least 1 test in quarantine with a documented reason

Infrastructure

• pytest.ini with all markers registered
• CI/CD configured: unit on every push, integration only on main
• README documentation on how to run each type

Additional exercises

Exercise 1: Add a second prompt

Add a classify_document(text) prompt that returns {category, confidence, tags}. Write:

• Contract test (unit, with a mock)
• E2E test (integration, with the real LLM)
• Semantic test for the category

# Contract test (M2 style):
def test_classify_contract(make_classification_client):
    client = make_classification_client(category="technology", confidence=0.9)
    result = classify_document("Article about Python", client=client)
    assert result["category"] in VALID_CATEGORIES
    assert 0 <= result["confidence"] <= 1

# E2E test (M3 style):
@pytest.mark.integration
def test_classify_e2e(integration_client, e2e_budget):
    result = classify_document("Python is a programming language", client=integration_client)
    e2e_budget.add_cost("gpt-4o-mini", 100, 50)
    assert result["category"] in VALID_CATEGORIES

# Semantic test:
@pytest.mark.integration
def test_classify_semantic_category(integration_client, e2e_budget):
    result = classify_document("Python is a programming language", client=integration_client)
    e2e_budget.add_cost("gpt-4o-mini", 100, 60)
    assert_semantically_similar(
        actual=result["category"],
        expected="technology or programming",
        threshold=0.65
    )

Exercise 2: Model drift test

Design a test that detects whether the model produces results very different from a saved reference set:

# tests/integration/test_drift.py
REFERENCE_CASES = [
    {"input": "I love it", "expected_sentiment": "positive"},
    {"input": "I hate it", "expected_sentiment": "negative"},
    {"input": "It arrived on Wednesday", "expected_sentiment": "neutral"},
]

@pytest.mark.integration
def test_model_drift(integration_client, e2e_budget):
    """Detects whether the model produces sentiments different from the reference ones."""
    mismatches = []
    
    for case in REFERENCE_CASES:
        result = analyze_sentiment(case["input"], client=integration_client)
        e2e_budget.add_cost("gpt-4o-mini", 60, 40)
        
        if result["sentiment"] != case["expected_sentiment"]:
            mismatches.append({
                "input": case["input"],
                "expected": case["expected_sentiment"],
                "got": result["sentiment"]
            })
    
    assert len(mismatches) == 0, (
        f"Model drift detected: {len(mismatches)}/{len(REFERENCE_CASES)} cases changed.\n"
        + "\n".join(f"  {m['input']}: {m['expected']} → {m['got']}" for m in mismatches)
    )

Project summary

Upon completing this project you have:

Cost per integration run: <$0.02 with gpt-4o-mini Total unit test time: <15 seconds Total time with integration: 1-3 minutes

Additional resources

1. pytest-rerunfailures — Retry for flaky tests
2. sentence-transformers — Local embeddings for semantic assertions
3. Hypothesis — Property-based testing
4. GitHub Actions — CI/CD configuration
5. OpenAI Usage — Monitor real API costs
6. Module 4: Guardrails — Next step