Spaces:
Running
Running
File size: 4,164 Bytes
6f9a7df ab357e0 6f9a7df 5ca5e73 6f9a7df 5ca5e73 6f9a7df 5ca5e73 6f9a7df 5ca5e73 6f9a7df | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | """Tests for proxy-level API key authentication."""
import pytest
from fastapi.testclient import TestClient
from headroom.proxy.models import ProxyConfig
from headroom.proxy.server import _proxy_config_from_env, create_app
@pytest.fixture()
def api_client():
"""Create a TestClient with API key auth enabled."""
config = ProxyConfig(
proxy_api_key="test-secret-key",
require_proxy_api_key=True,
cache_enabled=False,
rate_limit_enabled=False,
log_requests=False,
)
with TestClient(create_app(config)) as client:
yield client
@pytest.fixture()
def open_client():
"""Create a TestClient with API key auth disabled."""
config = ProxyConfig(
cache_enabled=False,
rate_limit_enabled=False,
log_requests=False,
)
with TestClient(create_app(config)) as client:
yield client
class TestHealthEndpointsAlwaysAccessible:
"""Health endpoints must remain accessible without API key."""
def test_livez_no_key(self, api_client):
resp = api_client.get("/livez")
assert resp.status_code == 200
def test_healthz_no_key(self, api_client):
resp = api_client.get("/healthz")
assert resp.status_code == 200
def test_readyz_no_key(self, api_client):
resp = api_client.get("/readyz")
assert resp.status_code == 200
def test_health_no_key(self, api_client):
resp = api_client.get("/health")
assert resp.status_code == 200
def test_visualize_no_key(self, api_client):
resp = api_client.get("/visualize")
assert resp.status_code == 200
def test_dashboard_no_key(self, api_client):
resp = api_client.get("/dashboard")
assert resp.status_code == 200
class TestProtectedEndpointsRejectWithoutKey:
"""Functional endpoints must reject requests without a valid key."""
def test_stats_no_key(self, api_client):
resp = api_client.get("/stats")
assert resp.status_code == 401
def test_stats_wrong_key(self, api_client):
resp = api_client.get("/stats", headers={"Authorization": "Bearer wrong"})
assert resp.status_code == 401
def test_cache_clear_no_key(self, api_client):
resp = api_client.post("/cache/clear")
assert resp.status_code == 401
def test_quota_no_key(self, api_client):
resp = api_client.get("/quota")
assert resp.status_code == 401
class TestProtectedEndpointsAcceptValidKey:
"""Functional endpoints must accept requests with the correct key."""
def test_stats_with_key(self, api_client):
resp = api_client.get("/stats", headers={"Authorization": "Bearer test-secret-key"})
assert resp.status_code == 200
def test_quota_with_key(self, api_client):
resp = api_client.get("/quota", headers={"Authorization": "Bearer test-secret-key"})
assert resp.status_code == 200
def test_cache_clear_with_key(self, api_client):
resp = api_client.post("/cache/clear", headers={"Authorization": "Bearer test-secret-key"})
assert resp.status_code == 200
class TestAuthDisabledByDefault:
"""When no API key is configured, all endpoints are open."""
def test_stats_open(self, open_client):
resp = open_client.get("/stats")
assert resp.status_code == 200
def test_cache_clear_open(self, open_client):
resp = open_client.post("/cache/clear")
assert resp.status_code == 200
class TestEnvConfig:
"""The proxy should pick up API key auth from the environment."""
def test_env_enables_proxy_key(self, monkeypatch):
monkeypatch.setenv("HEADROOM_PROXY_API_KEY", "env-secret-key")
config = _proxy_config_from_env()
assert config.proxy_api_key == "env-secret-key"
assert config.require_proxy_api_key is True
class TestStartupValidation:
"""Proxy must fail fast if misconfigured."""
def test_raises_when_key_missing(self):
config = ProxyConfig(require_proxy_api_key=True, proxy_api_key=None)
with pytest.raises(ValueError, match="proxy_api_key is required"):
create_app(config)
|