104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
import uuid
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase, override_settings
|
|
from django.urls import path
|
|
from rest_framework.response import Response
|
|
from rest_framework.test import APIClient
|
|
|
|
from apps.api.authentication import ApiKeyAuthentication
|
|
from apps.api.views import ExternalApiView
|
|
from apps.users.models import ApiKey
|
|
|
|
|
|
class AuthenticatedEchoView(ExternalApiView):
|
|
def get(self, request):
|
|
return Response(
|
|
{
|
|
"user_id": request.user.id,
|
|
"api_key_id": request.auth.id,
|
|
}
|
|
)
|
|
|
|
|
|
urlpatterns = [
|
|
path("api/test-auth/", AuthenticatedEchoView.as_view()),
|
|
]
|
|
|
|
|
|
@override_settings(ROOT_URLCONF=__name__)
|
|
class ApiKeyAuthenticationTests(TestCase):
|
|
url = "/api/test-auth/"
|
|
|
|
def setUp(self):
|
|
suffix = uuid.uuid4().hex[:8]
|
|
self.user = get_user_model().objects.create_user(
|
|
username=f"api-user-{suffix}",
|
|
email=f"api-user-{suffix}@example.com",
|
|
password="password",
|
|
)
|
|
self.api_key, self.raw_key = ApiKey.create_for_user(self.user, name="test")
|
|
self.client = APIClient()
|
|
|
|
def auth_header(self, raw_key: str | None = None) -> dict:
|
|
return {"HTTP_AUTHORIZATION": f"Bearer {raw_key or self.raw_key}"}
|
|
|
|
def test_external_api_view_only_uses_api_key_authentication(self):
|
|
self.assertEqual(AuthenticatedEchoView.authentication_classes, (ApiKeyAuthentication,))
|
|
|
|
def test_valid_bearer_key_authenticates_user_and_api_key(self):
|
|
response = self.client.get(self.url, **self.auth_header())
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["user_id"], self.user.id)
|
|
self.assertEqual(response.data["api_key_id"], self.api_key.id)
|
|
|
|
self.api_key.refresh_from_db()
|
|
self.assertIsNotNone(self.api_key.last_used_at)
|
|
|
|
def test_missing_api_key_returns_401(self):
|
|
response = self.client.get(self.url)
|
|
|
|
self.assertEqual(response.status_code, 401)
|
|
self.assertEqual(response["WWW-Authenticate"], "Bearer")
|
|
self.assertEqual(response.data["error"]["code"], "unauthorized")
|
|
|
|
def test_invalid_api_key_returns_401(self):
|
|
response = self.client.get(self.url, **self.auth_header("sk_cmhub_invalid"))
|
|
|
|
self.assertEqual(response.status_code, 401)
|
|
self.assertEqual(response["WWW-Authenticate"], "Bearer")
|
|
self.assertEqual(response.data["error"]["code"], "unauthorized")
|
|
|
|
def test_malformed_authorization_header_returns_401(self):
|
|
response = self.client.get(self.url, HTTP_AUTHORIZATION=f"Token {self.raw_key}")
|
|
|
|
self.assertEqual(response.status_code, 401)
|
|
self.assertEqual(response.data["error"]["code"], "unauthorized")
|
|
|
|
def test_revoked_api_key_returns_403(self):
|
|
self.api_key.status = ApiKey.Status.REVOKED
|
|
self.api_key.save(update_fields=("status", "updated_at"))
|
|
|
|
response = self.client.get(self.url, **self.auth_header())
|
|
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertEqual(response.data["error"]["code"], "account_disabled")
|
|
|
|
def test_disabled_user_returns_403(self):
|
|
self.user.status = self.user.Status.DISABLED
|
|
self.user.save(update_fields=("status",))
|
|
|
|
response = self.client.get(self.url, **self.auth_header())
|
|
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertEqual(response.data["error"]["code"], "account_disabled")
|
|
|
|
def test_web_session_login_is_not_accepted_for_external_api(self):
|
|
self.client.force_login(self.user)
|
|
|
|
response = self.client.get(self.url)
|
|
|
|
self.assertEqual(response.status_code, 401)
|
|
self.assertEqual(response.data["error"]["code"], "unauthorized")
|