import httpx
import json
import logging
from typing import Optional, Any
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass
class APIResponse:
status_code: int
body: Any
headers: dict
elapsed_ms: float
def assert_status(self, expected: int):
assert self.status_code == expected, \
f"Expected {expected}, got {self.status_code}. Body: {self.body}"
return self
def assert_field(self, field: str, expected=None):
assert field in self.body, f"Field '{field}' missing from response: {self.body}"
if expected is not None:
assert self.body[field] == expected, \
f"Field '{field}': expected {expected!r}, got {self.body[field]!r}"
return self
def assert_schema(self, required_fields: list):
missing = [f for f in required_fields if f not in self.body]
assert not missing, f"Missing required fields: {missing}. Got: {list(self.body.keys())}"
return self
class APIClient:
def __init__(self, base_url: str, default_headers: dict = None, timeout: float = 30.0):
self.base_url = base_url.rstrip('/')
self._default_headers = default_headers or {"Content-Type": "application/json"}
self._timeout = timeout
self._client = httpx.Client(timeout=timeout)
def set_auth(self, token: str):
self._default_headers["Authorization"] = f"Bearer {token}"
return self
def request(self, method: str, path: str, **kwargs) -> APIResponse:
url = f"{self.base_url}/{path.lstrip('/')}"
headers = {**self._default_headers, **kwargs.pop("headers", {})}
logger.debug(f"{method} {url}")
response = self._client.request(method, url, headers=headers, **kwargs)
try:
body = response.json()
except Exception:
body = response.text
elapsed = response.elapsed.total_seconds() * 1000
return APIResponse(response.status_code, body, dict(response.headers), elapsed)
def get(self, path: str, **kwargs) -> APIResponse:
return self.request("GET", path, **kwargs)
def post(self, path: str, json: dict = None, **kwargs) -> APIResponse:
return self.request("POST", path, json=json, **kwargs)
def put(self, path: str, json: dict = None, **kwargs) -> APIResponse:
return self.request("PUT", path, json=json, **kwargs)
def patch(self, path: str, json: dict = None, **kwargs) -> APIResponse:
return self.request("PATCH", path, json=json, **kwargs)
def delete(self, path: str, **kwargs) -> APIResponse:
return self.request("DELETE", path, **kwargs)
from dataclasses import dataclass, field
from typing import Optional
import uuid
from datetime import datetime
@dataclass
class OrderItemData:
product_id: str = None
quantity: int = 1
unit_price: float = 29.99
def __post_init__(self):
if self.product_id is None:
self.product_id = f"prod-{uuid.uuid4().hex[:8]}"
@dataclass
class OrderData:
customer_id: str = None
items: list = field(default_factory=list)
shipping_address: str = "123 Test St, Springfield, US 12345"
status: str = "draft"
notes: Optional[str] = None
def __post_init__(self):
if self.customer_id is None:
self.customer_id = f"cust-{uuid.uuid4().hex[:8]}"
if not self.items:
self.items = [OrderItemData()]
class OrderFactory:
def __init__(self, api_client: APIClient):
self.api = api_client
self._created_ids = []
def build(self, **overrides) -> dict:
"""Build order payload without creating it."""
data = OrderData(**overrides)
return {
"customer_id": data.customer_id,
"items": [
{"product_id": i.product_id, "quantity": i.quantity}
for i in data.items
],
"shipping_address": data.shipping_address,
"notes": data.notes,
}
def create(self, **overrides) -> dict:
"""Create an order via API and track for cleanup."""
payload = self.build(**overrides)
response = self.api.post("/api/v1/orders", json=payload)
response.assert_status(201)
order = response.body
self._created_ids.append(order["order_id"])
return order
def create_paid(self, **overrides) -> dict:
"""Create an order in PAID state."""
order = self.create(**overrides)
self.api.post(f"/api/v1/test/orders/{order['order_id']}/pay")
return {**order, "status": "paid"}
def cleanup(self):
"""Delete all created orders (called in teardown)."""
for order_id in self._created_ids:
self.api.delete(f"/api/v1/orders/{order_id}")
self._created_ids.clear()
import pytest
import os
from framework.clients.api_client import APIClient
from framework.factories.order_factory import OrderFactory
@pytest.fixture(scope="session")
def base_url() -> str:
return os.environ.get("API_BASE_URL", "http://localhost:8080")
@pytest.fixture(scope="session")
def api_client(base_url) -> APIClient:
return APIClient(base_url)
@pytest.fixture
def auth_client(api_client) -> APIClient:
"""Authenticated API client. Re-authenticates per test for isolation."""
response = api_client.post("/auth/login", json={
"email": os.environ["TEST_USER_EMAIL"],
"password": os.environ["TEST_USER_PASSWORD"],
})
token = response.body["access_token"]
client = APIClient(api_client.base_url)
client.set_auth(token)
return client
@pytest.fixture
def order_factory(auth_client) -> OrderFactory:
factory = OrderFactory(auth_client)
yield factory
factory.cleanup()
@pytest.fixture
def created_order(order_factory) -> dict:
"""Convenience fixture: create a standard test order."""
return order_factory.create()
@pytest.fixture
def paid_order(order_factory) -> dict:
"""Convenience fixture: create a paid test order."""
return order_factory.create_paid()
import json
from pathlib import Path
from jinja2 import Template
from datetime import datetime
def generate_report(results: list, output_path: str):
summary = {
"total": len(results),
"passed": sum(1 for r in results if r["outcome"] == "passed"),
"failed": sum(1 for r in results if r["outcome"] == "failed"),
"skipped": sum(1 for r in results if r["outcome"] == "skipped"),
"duration_s": sum(r.get("duration", 0) for r in results),
"generated_at": datetime.now().isoformat(),
}
summary["pass_rate"] = summary["passed"] / summary["total"] if summary["total"] else 0
failures = [r for r in results if r["outcome"] == "failed"]
slowest = sorted(results, key=lambda r: r.get("duration", 0), reverse=True)[:10]
Path(output_path).with_suffix(".json").write_text(
json.dumps({"summary": summary, "failures": failures, "slowest": slowest})
)