120 lines
4.3 KiB
Python
120 lines
4.3 KiB
Python
"""
|
|
Pure HTTP Contract Tests - Base Class
|
|
|
|
Framework-agnostic: works against ANY backend implementation.
|
|
"""
|
|
|
|
import unittest
|
|
import httpx
|
|
|
|
from .config import config
|
|
|
|
|
|
class ContractTestCase(unittest.TestCase):
|
|
"""
|
|
Base class for pure HTTP contract tests.
|
|
|
|
Features:
|
|
- Framework-agnostic (works with Django, FastAPI, Node, etc.)
|
|
- Pure HTTP via httpx library
|
|
- No database access - all data through API
|
|
- API Key authentication
|
|
"""
|
|
|
|
_base_url = None
|
|
_api_key = None
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
"""Set up once per test class"""
|
|
super().setUpClass()
|
|
cls._base_url = config.get("CONTRACT_TEST_URL", "").rstrip("/")
|
|
if not cls._base_url:
|
|
raise ValueError("CONTRACT_TEST_URL required in environment")
|
|
|
|
cls._api_key = config.get("CONTRACT_TEST_API_KEY", "")
|
|
if not cls._api_key:
|
|
raise ValueError("CONTRACT_TEST_API_KEY required in environment")
|
|
|
|
@property
|
|
def base_url(self):
|
|
return self._base_url
|
|
|
|
@property
|
|
def api_key(self):
|
|
return self._api_key
|
|
|
|
def _auth_headers(self):
|
|
"""Get authorization headers"""
|
|
return {"Authorization": f"Api-Key {self.api_key}"}
|
|
|
|
# =========================================================================
|
|
# HTTP helpers
|
|
# =========================================================================
|
|
|
|
def get(self, path: str, params: dict = None, **kwargs):
|
|
"""GET request"""
|
|
url = f"{self.base_url}{path}"
|
|
headers = {**self._auth_headers(), **kwargs.pop("headers", {})}
|
|
response = httpx.get(url, params=params, headers=headers, timeout=30, **kwargs)
|
|
return self._wrap_response(response)
|
|
|
|
def post(self, path: str, data: dict = None, **kwargs):
|
|
"""POST request with JSON"""
|
|
url = f"{self.base_url}{path}"
|
|
headers = {**self._auth_headers(), **kwargs.pop("headers", {})}
|
|
response = httpx.post(url, json=data, headers=headers, timeout=30, **kwargs)
|
|
return self._wrap_response(response)
|
|
|
|
def put(self, path: str, data: dict = None, **kwargs):
|
|
"""PUT request with JSON"""
|
|
url = f"{self.base_url}{path}"
|
|
headers = {**self._auth_headers(), **kwargs.pop("headers", {})}
|
|
response = httpx.put(url, json=data, headers=headers, timeout=30, **kwargs)
|
|
return self._wrap_response(response)
|
|
|
|
def patch(self, path: str, data: dict = None, **kwargs):
|
|
"""PATCH request with JSON"""
|
|
url = f"{self.base_url}{path}"
|
|
headers = {**self._auth_headers(), **kwargs.pop("headers", {})}
|
|
response = httpx.patch(url, json=data, headers=headers, timeout=30, **kwargs)
|
|
return self._wrap_response(response)
|
|
|
|
def delete(self, path: str, **kwargs):
|
|
"""DELETE request"""
|
|
url = f"{self.base_url}{path}"
|
|
headers = {**self._auth_headers(), **kwargs.pop("headers", {})}
|
|
response = httpx.delete(url, headers=headers, timeout=30, **kwargs)
|
|
return self._wrap_response(response)
|
|
|
|
def _wrap_response(self, response):
|
|
"""Add .data attribute for consistency with DRF responses"""
|
|
try:
|
|
response.data = response.json()
|
|
except Exception:
|
|
response.data = None
|
|
return response
|
|
|
|
# =========================================================================
|
|
# Assertion helpers
|
|
# =========================================================================
|
|
|
|
def assert_status(self, response, expected_status: int):
|
|
"""Assert response has expected status code"""
|
|
self.assertEqual(
|
|
response.status_code,
|
|
expected_status,
|
|
f"Expected {expected_status}, got {response.status_code}. "
|
|
f"Response: {response.data if hasattr(response, 'data') else response.content[:500]}"
|
|
)
|
|
|
|
def assert_has_fields(self, data: dict, *fields: str):
|
|
"""Assert dictionary has all specified fields"""
|
|
missing = [f for f in fields if f not in data]
|
|
self.assertEqual(missing, [], f"Missing fields: {missing}. Got: {list(data.keys())}")
|
|
|
|
def assert_is_list(self, data, min_length: int = 0):
|
|
"""Assert data is a list with minimum length"""
|
|
self.assertIsInstance(data, list)
|
|
self.assertGreaterEqual(len(data), min_length)
|