Antoni Nuñez Romeu
145 lines
5.3 KiB
Python
145 lines
5.3 KiB
Python
"""E2E tests for API Key authentication flow.
|
|
|
|
Tests dual auth support (JWT Bearer + X-API-Key header),
|
|
API Key CRUD, scope enforcement, and role-based access control.
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
|
|
import httpx
|
|
|
|
BASE_URL = os.environ.get("PIP_BASE_URL", "http://localhost:8000")
|
|
ADMIN_CREDS = {"username": os.environ.get("PIP_ADMIN_USER", "synctest"), "password": os.environ.get("PIP_ADMIN_PASS", "Test123!")}
|
|
|
|
|
|
async def get_admin_token(client: httpx.AsyncClient) -> str:
|
|
r = await client.post("/api/v1/auth/token", json=ADMIN_CREDS)
|
|
assert r.status_code == 200, f"Login failed: {r.status_code} {r.text[:200]}"
|
|
return r.json()["access_token"]
|
|
|
|
|
|
def bearer_headers(token: str) -> dict:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
async def test_api_key_crud():
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as c:
|
|
token = await get_admin_token(c)
|
|
h = bearer_headers(token)
|
|
|
|
clients = await c.get("/api/v1/api-clients", headers=h)
|
|
assert clients.status_code == 200
|
|
client_id = next((ci["id"] for ci in clients.json().get("items", []) if ci["name"] == "Test Client"), None)
|
|
|
|
if not client_id:
|
|
resp = await c.post("/api/v1/api-clients", headers=h, json={
|
|
"name": "Test Client",
|
|
"description": "E2E test client",
|
|
"scopes": ["pharmacies:read", "medications:read", "inventory:read"],
|
|
})
|
|
assert resp.status_code == 201
|
|
client_id = resp.json()["id"]
|
|
|
|
key_resp = await c.post(f"/api/v1/api-clients/{client_id}/keys", headers=h, json={"name": "E2E CRUD Key"})
|
|
assert key_resp.status_code == 201, f"Key create failed: {key_resp.status_code} {key_resp.text[:200]}"
|
|
raw_key = key_resp.json()["key"]
|
|
key_prefix = key_resp.json()["key_prefix"]
|
|
assert len(raw_key) > 20
|
|
assert len(key_prefix) == 8
|
|
|
|
keys = await c.get(f"/api/v1/api-clients/{client_id}/keys", headers=h)
|
|
assert keys.status_code == 200
|
|
assert len(keys.json()) >= 1
|
|
|
|
print(" PASS: test_api_key_crud")
|
|
|
|
|
|
async def test_api_key_read_access():
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as c:
|
|
token = await get_admin_token(c)
|
|
h = bearer_headers(token)
|
|
|
|
clients = await c.get("/api/v1/api-clients", headers=h)
|
|
client_id = next((ci["id"] for ci in clients.json().get("items", []) if ci["name"] == "Test Client"), None)
|
|
kr = await c.post(f"/api/v1/api-clients/{client_id}/keys", headers=h, json={"name": "Read Access Key"})
|
|
raw_key = kr.json()["key"]
|
|
api_h = {"X-API-Key": raw_key}
|
|
|
|
r = await c.get("/api/v1/pharmacies", headers=api_h)
|
|
assert r.status_code == 200
|
|
assert r.json()["total"] >= 0
|
|
|
|
r2 = await c.get("/api/v1/medications", headers=api_h)
|
|
assert r2.status_code == 200
|
|
|
|
r3 = await c.get("/api/v1/inventory", headers=api_h)
|
|
assert r3.status_code == 200
|
|
|
|
print(" PASS: test_api_key_read_access")
|
|
|
|
|
|
async def test_api_key_write_blocked():
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as c:
|
|
token = await get_admin_token(c)
|
|
h = bearer_headers(token)
|
|
|
|
clients = await c.get("/api/v1/api-clients", headers=h)
|
|
client_id = next((ci["id"] for ci in clients.json().get("items", []) if ci["name"] == "Test Client"), None)
|
|
kr = await c.post(f"/api/v1/api-clients/{client_id}/keys", headers=h, json={"name": "Write Block Key"})
|
|
raw_key = kr.json()["key"]
|
|
api_h = {"X-API-Key": raw_key}
|
|
|
|
r = await c.post("/api/v1/pharmacies", headers=api_h, json={"code": "BLOCKED", "name": "Should Fail"})
|
|
assert r.status_code == 403, f"Expected 403, got {r.status_code}"
|
|
|
|
print(" PASS: test_api_key_write_blocked")
|
|
|
|
|
|
async def test_invalid_api_key():
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as c:
|
|
r = await c.get("/api/v1/pharmacies", headers={"X-API-Key": "invalidkey12345"})
|
|
assert r.status_code == 401
|
|
|
|
print(" PASS: test_invalid_api_key")
|
|
|
|
|
|
async def test_no_auth_rejected():
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as c:
|
|
r = await c.get("/api/v1/pharmacies")
|
|
assert r.status_code == 401
|
|
|
|
print(" PASS: test_no_auth_rejected")
|
|
|
|
|
|
async def test_dual_auth_bearer_priority():
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as c:
|
|
token = await get_admin_token(c)
|
|
h = bearer_headers(token)
|
|
|
|
clients = await c.get("/api/v1/api-clients", headers=h)
|
|
client_id = next((ci["id"] for ci in clients.json().get("items", []) if ci["name"] == "Test Client"), None)
|
|
kr = await c.post(f"/api/v1/api-clients/{client_id}/keys", headers=h, json={"name": "Dual Auth Key"})
|
|
raw_key = kr.json()["key"]
|
|
|
|
dual_h = {"Authorization": f"Bearer {token}", "X-API-Key": raw_key}
|
|
r = await c.get("/api/v1/pharmacies", headers=dual_h)
|
|
assert r.status_code == 200
|
|
|
|
print(" PASS: test_dual_auth_bearer_priority")
|
|
|
|
|
|
async def run_all():
|
|
print("=== API Key Auth E2E Tests ===")
|
|
await test_api_key_crud()
|
|
await test_api_key_read_access()
|
|
await test_api_key_write_blocked()
|
|
await test_invalid_api_key()
|
|
await test_no_auth_rejected()
|
|
await test_dual_auth_bearer_priority()
|
|
print("=== All API Key tests passed ===")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run_all())
|