File size: 2,944 Bytes
7f0ad46 | 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 | from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from fastapi import status
from fastapi.testclient import TestClient
from app.api.dependencies import get_container
from app.main import create_app
from tests.integration.api.support import FakeProcessRemoteImage, FakeProcessUploadedImage, make_container
class ApiKeyIntegrationTests(unittest.TestCase):
def test_protected_endpoints_return_401_without_required_api_key(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
app = create_app()
app.dependency_overrides[get_container] = lambda: make_container(
root,
process_remote_image=FakeProcessRemoteImage(),
process_uploaded_image=FakeProcessUploadedImage(),
api_key="secret-key",
)
client = TestClient(app)
try:
response = client.get("/defaults")
self.assertEqual(status.HTTP_401_UNAUTHORIZED, response.status_code)
self.assertEqual("Invalid or missing API key.", response.json()["detail"])
finally:
client.close()
app.dependency_overrides.clear()
def test_protected_endpoints_accept_matching_api_key(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
app = create_app()
app.dependency_overrides[get_container] = lambda: make_container(
root,
process_remote_image=FakeProcessRemoteImage(),
process_uploaded_image=FakeProcessUploadedImage(),
api_key="secret-key",
)
client = TestClient(app)
try:
response = client.get("/defaults", headers={"X-API-Key": "secret-key"})
self.assertEqual(status.HTTP_200_OK, response.status_code)
finally:
client.close()
app.dependency_overrides.clear()
def test_health_is_public_even_when_api_key_is_configured(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
app = create_app()
app.dependency_overrides[get_container] = lambda: make_container(
root,
process_remote_image=FakeProcessRemoteImage(),
process_uploaded_image=FakeProcessUploadedImage(),
api_key="secret-key",
)
client = TestClient(app)
try:
response = client.get("/health")
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual({"status": "ok"}, response.json())
finally:
client.close()
app.dependency_overrides.clear()
if __name__ == "__main__":
unittest.main()
|