Grey Box Testing: Best of Both Worlds

# Grey box tester knows:
# - AuthService validates credentials
# - UserService manages user data
# - EmailService sends notifications
# - Redis caches user sessions

def test_user_registration_flow():
    """Test complete registration with knowledge of system architecture"""

    # 1. Register new user (UserService)
    response = api.post("/register", {
        "email": "newuser@example.com",
        "password": "SecurePass123"
    })
    assert response.status_code == 201
    user_id = response.json()["id"]

    # 2. Verify user record in database
 (as discussed in [Test Environment Setup: Complete Configuration Guide](/blog/test-environment-setup))    # (Grey box: knows database structure)
    user = db.query("SELECT * FROM users WHERE id = ?", user_id)
    assert user["email"] == "newuser@example.com"
    assert user["status"] == "pending_verification"

    # 3. Check welcome email sent (EmailService)
    # (Grey box: knows email queue exists)
    email_queue = redis.lrange("email_queue", 0, -1)
    assert any("newuser@example.com" in email for email in email_queue)

    # 4. Verify session not created yet (Redis)
    # (Grey box: knows session storage mechanism)
    session_key = f"session:{user_id}"
    assert redis.get(session_key) is None

def test_order_processing_data_integrity():
    """Test order processing with database knowledge"""

    # Create test order
    order_data = {
        "user_id": 100,
        "items": [
            {"product_id": 1, "quantity": 2, "price": 50.00},
            {"product_id": 2, "quantity": 1, "price": 30.00}
        ],
        "discount_code": "SAVE10"
    }

    response = api.post("/orders", order_data)
    order_id = response.json()["order_id"]

    # Grey box validation: Check database state
    # 1. Order record created correctly
    order = db.query("SELECT * FROM orders WHERE id = ?", order_id)
    assert order["user_id"] == 100
    assert order["status"] == "pending"
    assert order["total"] == 117.00  # (130 - 13 discount)

    # 2. Order items saved
    items = db.query("SELECT * FROM order_items WHERE order_id = ?", order_id)
    assert len(items) == 2

    # 3. Inventory decremented
    product1_stock = db.query("SELECT stock FROM products WHERE id = 1")
    # Verify stock reduced by 2

    # 4. Discount code usage tracked
    discount_usage = db.query(
        "SELECT * FROM discount_usages WHERE code = ? AND user_id = ?",
        "SAVE10", 100
    )
    assert discount_usage is not None

    # 5. Audit log created
    logs = db.query("SELECT * FROM audit_logs WHERE entity_id = ?", order_id)
    assert len(logs) > 0

def test_api_with_architectural_knowledge():
    """Test REST API with knowledge of backend structure"""

    # Grey box knowledge:
    # - API uses JWT for authentication
    # - Rate limiting: 100 requests/minute per user
    # - Response cached for 5 minutes
    # - Background jobs for heavy operations

    # 1. Authentication flow
    auth_response = api.post("/auth/login", {
        "email": "user@example.com",
        "password": "password"
    })

    token = auth_response.json()["access_token"]

    # Validate JWT structure (grey box: knows token format)
    import jwt
    decoded = jwt.decode(token, options={"verify_signature": False})
    assert decoded["user_id"] is not None
    assert decoded["exp"] > time.time()  # Not expired

    # 2. Test rate limiting
    # (Grey box: knows limit is 100/minute)
    headers = {"Authorization": f"Bearer {token}"}

    for i in range(100):
        response = api.get("/api/data", headers=headers)
        assert response.status_code == 200

    # 101st request should be rate-limited
    response = api.get("/api/data", headers=headers)
    assert response.status_code == 429  # Too Many Requests

    # 3. Test caching behavior
    # (Grey box: knows 5-minute cache)
    response1 = api.get("/api/products/1", headers=headers)
    cache_header1 = response1.headers.get("X-Cache")

    # Immediate second request should hit cache
    response2 = api.get("/api/products/1", headers=headers)
    cache_header2 = response2.headers.get("X-Cache")
    assert cache_header2 == "HIT"

    # 4. Test async job creation
    # (Grey box: knows heavy operations queued)
    response = api.post("/api/reports/generate", {
        "type": "annual",
        "format": "pdf"
    }, headers=headers)

    assert response.status_code == 202  # Accepted
    job_id = response.json()["job_id"]

    # Check job queue (grey box: knows queue location)
    job = redis.hget("jobs", job_id)
    assert job is not None
    assert json.loads(job)["status"] == "queued"

def test_security_with_architectural_knowledge():
    """Security testing with system design knowledge"""

    # Grey box knowledge:
    # - Passwords hashed with bcrypt
    # - CSRF tokens required for state-changing operations
    # - File uploads stored in S3
    # - SQL parameterization used

    # 1. Test password storage
    # (Grey box: knows hashing algorithm)
    api.post("/register", {
        "email": "security@example.com",
        "password": "TestPass123"
    })

    # Check database - password should be hashed
    user = db.query(
        "SELECT password_hash FROM users WHERE email = ?",
        "security@example.com"
    )
    # Should be bcrypt hash (starts with $2b$ or $2a$)
    assert user["password_hash"].startswith("$2")
    assert len(user["password_hash"]) == 60  # bcrypt hash length

    # 2. Test CSRF protection
    # (Grey box: knows CSRF implementation)
    session = requests.Session()
    session.post("/login", {
        "email": "security@example.com",
        "password": "TestPass123"
    })

    # Request without CSRF token should fail
    response = session.post("/api/sensitive-action", {
        "action": "delete_account"
    })
    assert response.status_code == 403  # Forbidden

    # Request with CSRF token should succeed
    csrf_token = session.get("/api/csrf-token").json()["token"]
    response = session.post("/api/sensitive-action", {
        "action": "delete_account",
        "csrf_token": csrf_token
    })
    assert response.status_code in [200, 204]

    # 3. Test SQL injection prevention
    # (Grey box: knows parameterization is used)
    # Attempt SQL injection
    response = api.get("/api/users?name='; DROP TABLE users--")

    # Should not execute SQL, should be safe
    assert response.status_code in [200, 400]

    # Verify users table still exists
    result = db.query("SELECT COUNT(*) as count FROM users")
    assert result["count"] >= 0  # Table still exists

    # 4. Test file upload security
    # (Grey box: knows S3 upload flow)
    # Attempt to upload malicious file
    files = {
        'file': ('malicious.php', '<?php system($_GET["cmd"]); ?>', 'application/x-php')
    }

    response = api.post("/api/upload", files=files)

    # Verify file extension validation
    if response.status_code == 200:
        upload_url = response.json()["url"]
        # URL should be S3, not local server
        assert "s3.amazonaws.com" in upload_url or "cloudfront" in upload_url
        # PHP should not be executable from upload location

def test_user_permissions_matrix():
    """Test permission matrix with role knowledge"""

    # Grey box knowledge: Role-based access control
    # Roles: Admin, Manager, User, Guest
    # Resources: Users, Orders, Reports, Settings

    permission_matrix = {
        "Admin": {
            "users": ["create", "read", "update", "delete"],
            "orders": ["create", "read", "update", "delete"],
            "reports": ["read", "generate"],
            "settings": ["read", "update"]
        },
        "Manager": {
            "users": ["read"],
            "orders": ["read", "update"],
            "reports": ["read", "generate"],
            "settings": ["read"]
        },
        "User": {
            "users": ["read_own"],
            "orders": ["create", "read_own"],
            "reports": [],
            "settings": []
        },
        "Guest": {
            "users": [],
            "orders": [],
            "reports": [],
            "settings": []
        }
    }

    for role, permissions in permission_matrix.items():
        user_token = create_user_with_role(role)

        for resource, actions in permissions.items():
            # Test allowed actions
            for action in actions:
                response = perform_action(resource, action, user_token)
                assert response.status_code in [200, 201], \
                    f"{role} should have {action} access to {resource}"

            # Test forbidden actions
            all_actions = ["create", "read", "update", "delete"]
            forbidden_actions = set(all_actions) - set(actions)

            for action in forbidden_actions:
                response = perform_action(resource, action, user_token)
                assert response.status_code == 403, \
                    f"{role} should NOT have {action} access to {resource}"

def test_order_state_transitions_with_data():
    """Test order states with database validation"""

    # Grey box: knows order states and database schema
    # States: draft → submitted → processing → shipped → delivered

    order_id = create_draft_order()

    # State: draft
    verify_order_state(order_id, "draft")
    verify_timestamps(order_id, created=True, submitted=False)

    # Transition: draft → submitted
    submit_order(order_id)
    verify_order_state(order_id, "submitted")
    verify_timestamps(order_id, created=True, submitted=True)
    verify_inventory_reserved(order_id)

    # Transition: submitted → processing
    process_order(order_id)
    verify_order_state(order_id, "processing")
    verify_payment_captured(order_id)
    verify_shipping_label_created(order_id)

    # Transition: processing → shipped
    ship_order(order_id)
    verify_order_state(order_id, "shipped")
    verify_tracking_number_assigned(order_id)
    verify_customer_notified(order_id, "shipped")

    # Transition: shipped → delivered
    deliver_order(order_id)
    verify_order_state(order_id, "delivered")
    verify_timestamps(order_id, delivered=True)
    verify_customer_notified(order_id, "delivered")

def verify_order_state(order_id, expected_state):
    """Verify order state in database"""
    order = db.query("SELECT state FROM orders WHERE id = ?", order_id)
    assert order["state"] == expected_state

def verify_timestamps(order_id, **checks):
    """Verify order timestamps"""
    order = db.query(
        "SELECT created_at, submitted_at, delivered_at FROM orders WHERE id = ?",
        order_id
    )

    if checks.get("created"):
        assert order["created_at"] is not None

    if checks.get("submitted"):
        assert order["submitted_at"] is not None

    if checks.get("delivered"):
        assert order["delivered_at"] is not None

def test_data_migration_regression():
    """Test data integrity after migration"""

    # Grey box: knows database schema before and after migration

    # Take snapshot before migration
    snapshot_before = {
        "user_count": db.query("SELECT COUNT(*) as count FROM users")["count"],
        "order_count": db.query("SELECT COUNT(*) as count FROM orders")["count"],
        "total_revenue": db.query("SELECT SUM(total) as sum FROM orders")["sum"]
    }

    # Run migration
    run_migration("v2.0_add_shipping_zones")

    # Verify data integrity after migration
    snapshot_after = {
        "user_count": db.query("SELECT COUNT(*) as count FROM users")["count"],
        "order_count": db.query("SELECT COUNT(*) as count FROM orders")["count"],
        "total_revenue": db.query("SELECT SUM(total) as sum FROM orders")["sum"]
    }

    # Core data should remain unchanged
    assert snapshot_before["user_count"] == snapshot_after["user_count"]
    assert snapshot_before["order_count"] == snapshot_after["order_count"]
    assert snapshot_before["total_revenue"] == snapshot_after["total_revenue"]

    # New schema elements should exist
    result = db.query("""
        SELECT column_name
        FROM information_schema.columns
        WHERE table_name = 'orders' AND column_name = 'shipping_zone_id'
    """)
    assert result is not None

    # Test new functionality
    order_id = create_order_with_shipping_zone("US-WEST")
    order = db.query("SELECT shipping_zone_id FROM orders WHERE id = ?", order_id)
    assert order["shipping_zone_id"] is not None

# Black box approach: Tests only API responses
def test_user_creation_black_box():
    response = api.post("/users", {"name": "John", "email": "john@example.com"})
    assert response.status_code == 201

# Grey box approach: Tests API + data + side effects
def test_user_creation_grey_box():
    response = api.post("/users", {"name": "John", "email": "john@example.com"})
    assert response.status_code == 201

    user_id = response.json()["id"]

    # Verify database record
    user = db.query("SELECT * FROM users WHERE id = ?", user_id)
    assert user["name"] == "John"
    assert user["email_verified"] == False

    # Verify welcome email queued
    emails = get_queued_emails()
    assert any(e["to"] == "john@example.com" for e in emails)

    # Verify default preferences created
    prefs = db.query("SELECT * FROM user_preferences WHERE user_id = ?", user_id)
    assert prefs is not None

    # Verify audit log
    logs = db.query("SELECT * FROM audit_logs WHERE user_id = ? AND action = ?",
                    user_id, "user_created")
    assert len(logs) == 1

def test_payment_processing_with_debugging():
    """Grey box testing with built-in debugging"""

    order_id = create_test_order(total=100.00)

    # Attempt payment
    response = api.post(f"/orders/{order_id}/pay", {
        "card_number": "4111111111111111",
        "expiry": "12/25",
        "cvv": "123"
    })

    if response.status_code != 200:
        # Grey box: knows where to look for debugging
        # 1. Check payment gateway logs
        gateway_logs = db.query(
            "SELECT * FROM payment_gateway_logs WHERE order_id = ? ORDER BY created_at DESC",
            order_id
        )
        print(f"Gateway response: {gateway_logs[0]['response']}")

        # 2. Check order state
        order = db.query("SELECT * FROM orders WHERE id = ?", order_id)
        print(f"Order state: {order['state']}")

        # 3. Check retry queue
        retries = redis.lrange(f"payment_retries:{order_id}", 0, -1)
        print(f"Retry attempts: {len(retries)}")

        # 4. Check error logs
        errors = db.query(
            "SELECT * FROM error_logs WHERE context LIKE ?",
            f"%order_id:{order_id}%"
        )
        if errors:
            print(f"Errors: {errors}")

    assert response.status_code == 200

def setup_realistic_test_data():
    """Create realistic test data using database knowledge"""

    # Grey box: knows database schema and relationships

    # Create user with complete profile
    user_id = db.execute("""
        INSERT INTO users (name, email, created_at, status)
        VALUES (?, ?, ?, ?)
    """, "Test User", "test@example.com", datetime.now(), "active")

    # Create address history
    addresses = [
        ("123 Main St", "New York", "NY", "10001"),
        ("456 Oak Ave", "Los Angeles", "CA", "90001")
    ]
    for street, city, state, zip_code in addresses:
        db.execute("""
            INSERT INTO addresses (user_id, street, city, state, zip)
            VALUES (?, ?, ?, ?, ?)
        """, user_id, street, city, state, zip_code)

    # Create order history
    for i in range(5):
        order_id = db.execute("""
            INSERT INTO orders (user_id, total, status, created_at)
            VALUES (?, ?, ?, ?)
        """, user_id, 50.00 + i * 10, "completed", datetime.now() - timedelta(days=30-i))

        # Create order items
        db.execute("""
            INSERT INTO order_items (order_id, product_id, quantity, price)
            VALUES (?, ?, ?, ?)
        """, order_id, 1, 2, 25.00)

    # Create payment methods
    db.execute("""
        INSERT INTO payment_methods (user_id, type, last_four, expiry)
        VALUES (?, ?, ?, ?)
    """, user_id, "credit_card", "1111", "12/25")

    return user_id

# test_architecture.yml
system_architecture:
  authentication:
    method: JWT
    token_expiry: 3600  # seconds
    refresh_enabled: true

  database:
    type: PostgreSQL
    connection_pool: 20
    key_tables:
      - users (id, email, password_hash, created_at)
      - orders (id, user_id, total, status, created_at)
      - sessions (id, user_id, token, expires_at)

  caching:
    provider: Redis
    ttl: 300  # seconds
    cached_endpoints:
      - /api/products
      - /api/categories

  message_queue:
    provider: RabbitMQ
    queues:
      - email_notifications
      - payment_processing
      - inventory_updates

  external_services:
    payment_gateway: Stripe
    email_provider: SendGrid
    storage: AWS S3

class UserAPITestSuite:
    """Balanced grey box testing approach"""

    def test_user_registration_black_box_view(self):
        """Test from end-user perspective"""
        # Pure black box: only test API contract
        response = api.post("/register", {
            "email": "user@example.com",
            "password": "SecurePass123"
        })

        assert response.status_code == 201
        assert "id" in response.json()
        assert "email" in response.json()

    def test_user_registration_grey_box_validation(self):
        """Test with architectural knowledge"""
        response = api.post("/register", {
            "email": "user@example.com",
            "password": "SecurePass123"
        })

        user_id = response.json()["id"]

        # Grey box: verify internal state
        user = db.query("SELECT * FROM users WHERE id = ?", user_id)
        assert user["email_verified"] == False
        assert user["status"] == "pending"

        # Verify side effects
        assert email_sent_to("user@example.com", subject="Verify your email")

    def avoid_white_box_implementation_details(self):
        """Don't test implementation details"""
        # ❌ Bad: Testing internal functions (white box)
        # from app.services import UserService
        # assert UserService._hash_password("test") starts_with "$2b$"

        # ✓ Good: Test behavior and observable state (grey box)
        response = api.post("/register", {"email": "test@example.com", "password": "test"})
        user = db.query("SELECT password_hash FROM users WHERE email = ?", "test@example.com")
        assert user["password_hash"] != "test"  # Password is hashed, not stored plainly

# conftest.py - Grey box testing utilities

import pytest
from database import Database
from redis_client import Redis
from api_client import APIClient

@pytest.fixture
def db():
    """Database connection for grey box testing"""
    db = Database()
    yield db
    db.rollback()  # Clean up after each test

@pytest.fixture
def redis():
    """Redis connection for cache verification"""
    redis = Redis()
    yield redis
    redis.flushdb()  # Clean cache after each test

@pytest.fixture
def api():
    """API client for black box testing"""
    return APIClient(base_url="http://localhost:3000")

@pytest.fixture
def grey_box_context(db, redis, api):
    """Combined context for grey box testing"""
    return {
        "db": db,
        "cache": redis,
        "api": api
    }

# Usage in tests
def test_with_grey_box_context(grey_box_context):
    ctx = grey_box_context

    # Black box: API call
    response = ctx["api"].post("/users", {"name": "Test"})

    # Grey box: Verify internal state
    user = ctx["db"].query("SELECT * FROM users WHERE id = ?", response.json()["id"])
    assert user is not None

    # Grey box: Verify caching
    cache_key = f"user:{user['id']}"
    assert ctx["cache"].exists(cache_key)

class ECommerceCheckoutGreyBoxTest:
    """Complete grey box testing for checkout flow"""

    def test_checkout_complete_flow(self, grey_box_context):
        """Test checkout with architectural knowledge"""
        ctx = grey_box_context

        # Setup: Create user and add items to cart
        user_id = self.create_test_user(ctx)
        cart_id = self.add_items_to_cart(ctx, user_id, [
            {"product_id": 1, "quantity": 2},
            {"product_id": 2, "quantity": 1}
        ])

        # Black box: Initiate checkout
        response = ctx["api"].post("/checkout", {
            "cart_id": cart_id,
            "shipping_address": {
                "street": "123 Main St",
                "city": "New York",
                "state": "NY",
                "zip": "10001"
            },
            "payment_method": {
                "card_number": "4111111111111111",
                "expiry": "12/25",
                "cvv": "123"
            }
        })

        assert response.status_code == 200
        order_id = response.json()["order_id"]

        # Grey box: Verify order created correctly
        order = ctx["db"].query("SELECT * FROM orders WHERE id = ?", order_id)
        assert order["user_id"] == user_id
        assert order["status"] == "processing"

        # Grey box: Verify inventory updated
        product1_stock = ctx["db"].query(
            "SELECT stock FROM products WHERE id = 1"
        )["stock"]
        # Stock should be decreased

        # Grey box: Verify payment processed
        payment = ctx["db"].query(
            "SELECT * FROM payments WHERE order_id = ?", order_id
        )
        assert payment["status"] == "completed"
        assert payment["amount"] == order["total"]

        # Grey box: Verify shipping label created
        shipping = ctx["db"].query(
            "SELECT * FROM shipping_labels WHERE order_id = ?", order_id
        )
        assert shipping is not None
        assert shipping["tracking_number"] is not None

        # Grey box: Verify cart cleared
        cart_items = ctx["db"].query(
            "SELECT * FROM cart_items WHERE cart_id = ?", cart_id
        )
        assert len(cart_items) == 0

        # Grey box: Verify confirmation email queued
        email_queue = ctx["cache"].lrange("email_queue", 0, -1)
        confirmation_email = [
            e for e in email_queue
            if f"order_id:{order_id}" in e
        ]
        assert len(confirmation_email) == 1

        # Grey box: Verify analytics event logged
        events = ctx["db"].query(
            "SELECT * FROM analytics_events WHERE order_id = ?", order_id
        )
        assert any(e["event_type"] == "checkout_completed" for e in events)