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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291"""Integration tests for API authentication."""
import os
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from httpx import ASGITransport, AsyncClient
from src.main import app
@pytest.fixture
def enable_api_key_auth():
"""Enable API key authentication for tests."""
with patch("src.middleware.api_key_auth.settings") as mock_settings:
mock_settings.api_key_header_name = "X-API-Key"
mock_settings.environment = "production"
mock_settings.enable_api_key_auth = True
mock_settings.api_keys = MagicMock()
mock_settings.api_keys.get_secret_value.return_value = "test-key-123,test-key-456"
yield mock_settings
@pytest.mark.integration
@pytest.mark.asyncio
@pytest.mark.skipif(
os.getenv("ENABLE_API_KEY_AUTH", "").lower() == "false",
reason="Auth tests require auth enabled"
)
async def test_api_key_required_for_protected_endpoint(enable_api_key_auth):
"""Test that API key is required for protected endpoints."""
# Mock dependencies to avoid real service calls
with patch("src.api.documents.get_storage_service"), \
patch("src.api.documents.get_queue_service"), \
patch("src.api.documents.get_job_service"), \
patch("src.main.settings") as main_settings:
# Configure main settings
main_settings.enable_api_key_auth = True
main_settings.api_key_header_name = "X-API-Key"
main_settings.environment = "production"
main_settings.api_keys = MagicMock()
main_settings.api_keys.get_secret_value.return_value = "test-key-123"
# Recreate app with auth enabled (in real usage, this is set at startup)
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as client:
# Test without API key
response = await client.get("/api/v1/documents/some-id")
# Should be rejected (401)
assert response.status_code == 401
assert "API key" in response.json()["detail"]
@pytest.mark.integration
@pytest.mark.asyncio
async def test_valid_api_key_allows_access(enable_api_key_auth):
"""Test that valid API key allows access to protected endpoints."""
from src.dependencies import get_job_service
from src.middleware.api_key_auth import APIKeyAuthMiddleware
# Mock job service
mock_job_service = AsyncMock()
mock_job_service.get_job.return_value = {
"job_id": "test-123",
"status": "pii_scanning",
"filename": "test.pdf",
"created_at": "2025-01-01T00:00:00Z",
"updated_at": "2025-01-01T00:00:00Z"
}
# Override dependencies
app.dependency_overrides[get_job_service] = lambda: mock_job_service
try:
# Patch the middleware's validation method
def mock_match_key(self, provided_key: str) -> str | None:
return "test-label" if provided_key == "test-key-123" else None
with patch.object(APIKeyAuthMiddleware, '_match_key', mock_match_key):
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test",
headers={"X-API-Key": "test-key-123"}
) as client:
# Test with valid API key
response = await client.get("/api/v1/documents/test-123")
# Should succeed (200) or 404 if job not found, but NOT 401
assert response.status_code in [200, 404]
finally:
# Clean up overrides
app.dependency_overrides.clear()
@pytest.mark.integration
@pytest.mark.asyncio
async def test_health_endpoint_bypasses_api_key_auth(enable_api_key_auth):
"""Test that health endpoint works without API key."""
from src.dependencies import get_queue_service, get_storage_service
# Mock storage service
mock_storage = AsyncMock()
mock_storage.check_s3_access.return_value = True
# Mock queue service
mock_queue = AsyncMock()
mock_queue.check_redis_connection.return_value = True
mock_queue.check_queue_depth.return_value = 0
# Override dependencies
app.dependency_overrides[get_storage_service] = lambda: mock_storage
app.dependency_overrides[get_queue_service] = lambda: mock_queue
try:
with patch("src.main.settings") as main_settings:
main_settings.enable_api_key_auth = True
main_settings.environment = "production"
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as client:
# Test health endpoint without API key
response = await client.get("/health")
# Should succeed
assert response.status_code == 200
finally:
# Clean up overrides
app.dependency_overrides.clear()
@pytest.mark.integration
@pytest.mark.asyncio
async def test_multiple_api_keys_supported(enable_api_key_auth):
"""Test that multiple API keys can be configured and used."""
# Import the middleware to find its instance
from src.middleware.api_key_auth import APIKeyAuthMiddleware
with patch("src.dependencies.get_job_service") as mock_job_service_dep:
# Find the middleware instance in the app's middleware stack
for middleware in app.user_middleware:
if middleware.cls == APIKeyAuthMiddleware:
# Access the middleware instance through the app
if hasattr(middleware, 'kwargs') and 'app' in middleware.kwargs:
pass # Middleware not instantiated yet in test context
# Directly patch the method that validates keys
def mock_match_key(self, provided_key: str) -> str | None:
return provided_key if provided_key in {"key-1", "key-2", "key-3"} else None
with patch.object(APIKeyAuthMiddleware, '_match_key', mock_match_key):
mock_job_service = AsyncMock()
mock_job_service.get_job.return_value = {
"job_id": "test",
"status": "completed"
}
mock_job_service_dep.return_value = mock_job_service
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as client:
# Test with first key
response = await client.get(
"/api/v1/documents/test",
headers={"X-API-Key": "key-1"}
)
assert response.status_code != 401
# Test with second key
response = await client.get(
"/api/v1/documents/test",
headers={"X-API-Key": "key-2"}
)
assert response.status_code != 401
# Test with third key
response = await client.get(
"/api/v1/documents/test",
headers={"X-API-Key": "key-3"}
)
assert response.status_code != 401
@pytest.mark.integration
@pytest.mark.asyncio
async def test_docs_endpoint_is_public():
"""Test that /docs is publicly accessible (no Basic Auth)."""
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as client:
response = await client.get("/docs")
# Should succeed without any credentials
assert response.status_code == 200
assert "WWW-Authenticate" not in response.headers
@pytest.mark.integration
@pytest.mark.asyncio
async def test_openapi_json_is_public():
"""Test that /openapi.json is publicly accessible."""
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as client:
response = await client.get("/openapi.json")
# Should succeed and return valid OpenAPI schema
assert response.status_code == 200
assert response.json().get("openapi")
@pytest.mark.integration
@pytest.mark.asyncio
async def test_api_key_auth_unaffected_by_public_docs():
"""Test that API key auth on /api/* still works after docs became public."""
from src.middleware.api_key_auth import APIKeyAuthMiddleware
with patch("src.dependencies.get_job_service") as mock_job_service_dep:
def mock_match_key(self, provided_key: str) -> str | None:
return "test-label" if provided_key == "api-key-123" else None
with patch.object(APIKeyAuthMiddleware, '_match_key', mock_match_key):
mock_job_service = AsyncMock()
mock_job_service.get_job.return_value = {
"job_id": "test-id",
"status": "completed"
}
mock_job_service_dep.return_value = mock_job_service
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as client:
# /docs is public now — no auth header needed
response = await client.get("/docs")
assert response.status_code == 200
# /api/v1/* still requires API key
response = await client.get(
"/api/v1/documents/test-id",
headers={"X-API-Key": "api-key-123"}
)
# Should not be 401 (might be 404 if job not found, but auth passed)
assert response.status_code != 401
@pytest.mark.integration
@pytest.mark.asyncio
async def test_rate_limiting_still_works_with_auth():
"""Test that rate limiting middleware still functions with auth enabled."""
from src.dependencies import get_queue_service, get_storage_service
# Mock storage service
mock_storage = AsyncMock()
mock_storage.check_s3_access.return_value = True
# Mock queue service
mock_queue = AsyncMock()
mock_queue.check_redis_connection.return_value = True
mock_queue.check_queue_depth.return_value = 0
# Override dependencies
app.dependency_overrides[get_storage_service] = lambda: mock_storage
app.dependency_overrides[get_queue_service] = lambda: mock_queue
try:
with patch("src.main.settings") as main_settings:
main_settings.enable_api_key_auth = False # Disable for this test
main_settings.environment = "production"
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as client:
# Make request to health endpoint
response = await client.get("/health")
# Should succeed (middleware stack intact)
assert response.status_code == 200
finally:
# Clean up overrides
app.dependency_overrides.clear()