🛠️ apply pre-commit fixes

Author: getsantry[bot]

src/sentry/web/frontend/oauth_token.py

@@ -17,11 +17,7 @@
 from sentry import options
 from sentry.locks import locks
 from sentry.models.apiapplication import ApiApplication, ApiApplicationStatus
-from sentry.models.apidevicecode import (
-    DEFAULT_INTERVAL,
-    ApiDeviceCode,
-    DeviceCodeStatus,
-)
+from sentry.models.apidevicecode import DEFAULT_INTERVAL, ApiDeviceCode, DeviceCodeStatus
 from sentry.models.apigrant import ApiGrant, ExpiredGrantError, InvalidGrantError
 from sentry.models.apitoken import ApiToken
 from sentry.ratelimits import backend as ratelimiter
@@ -128,9 +124,7 @@ def post(self, request: Request) -> HttpResponse:
 
         # Validate grant_type first (needed to determine auth requirements)
         if not grant_type:
-            return self.error(
-                request=request, name="invalid_request", reason="missing grant_type"
-            )
+            return self.error(request=request, name="invalid_request", reason="missing grant_type")
         if grant_type not in [
             GrantTypes.AUTHORIZATION,
             GrantTypes.REFRESH,
@@ -139,9 +133,7 @@ def post(self, request: Request) -> HttpResponse:
             return self.error(request=request, name="unsupported_grant_type")
 
         # Determine client credentials from header or body (mutually exclusive).
-        (client_id, client_secret), cred_error = self._extract_basic_auth_credentials(
-            request
-        )
+        (client_id, client_secret), cred_error = self._extract_basic_auth_credentials(request)
         if cred_error is not None:
             return cred_error
 
@@ -214,9 +206,7 @@ def post(self, request: Request) -> HttpResponse:
                     "oauth_token.post.invalid",
                     sample_rate=1.0,
                 )
-                logger.warning(
-                    "Invalid client_id / secret pair", extra={"client_id": client_id}
-                )
+                logger.warning("Invalid client_id / secret pair", extra={"client_id": client_id})
                 return self.error(
                     request=request,
                     name="invalid_client",
@@ -242,9 +232,7 @@ def post(self, request: Request) -> HttpResponse:
                     # Use unguarded_write because deleting the grant triggers SET_NULL on
                     # SentryAppInstallation.api_grant, which is a cross-model write
                     with unguarded_write(using=router.db_for_write(ApiGrant)):
-                        ApiGrant.objects.filter(
-                            application=application, code=code
-                        ).delete()
+                        ApiGrant.objects.filter(application=application, code=code).delete()
             # For device_code, invalidate the device code
             elif grant_type == GrantTypes.DEVICE_CODE:
                 device_code_value = request.POST.get("device_code")
@@ -281,17 +269,11 @@ def post(self, request: Request) -> HttpResponse:
             )
 
         if grant_type == GrantTypes.AUTHORIZATION:
-            token_data = self.get_access_tokens(
-                request=request, application=application
-            )
+            token_data = self.get_access_tokens(request=request, application=application)
         elif grant_type == GrantTypes.DEVICE_CODE:
-            return self.handle_device_code_grant(
-                request=request, application=application
-            )
+            return self.handle_device_code_grant(request=request, application=application)
         else:
-            token_data = self.get_refresh_token(
-                request=request, application=application
-            )
+            token_data = self.get_refresh_token(request=request, application=application)
         if "error" in token_data:
             return self.error(
                 request=request,
@@ -356,28 +338,22 @@ def _extract_basic_auth_credentials(
                 # avoid excessive memory use on decode.
                 b64 = param.strip()
                 if len(b64) > MAX_BASIC_AUTH_B64_LEN:
-                    logger.warning(
-                        "Invalid Basic auth header: too long", extra={"client_id": None}
-                    )
+                    logger.warning("Invalid Basic auth header: too long", extra={"client_id": None})
                     return (None, None), self.error(
                         request=request,
                         name="invalid_client",
                         reason="invalid basic auth (too long)",
                         status=401,
                     )
                 try:
-                    decoded = base64.b64decode(
-                        b64.encode("ascii"), validate=True
-                    ).decode("utf-8")
+                    decoded = base64.b64decode(b64.encode("ascii"), validate=True).decode("utf-8")
                     # format: client_id:client_secret (client_secret may be empty)
                     if ":" not in decoded:
                         raise ValueError("missing colon in basic credentials")
                     client_id, client_secret = decoded.split(":", 1)
                     return (client_id, client_secret), None
                 except Exception:
-                    logger.warning(
-                        "Invalid Basic auth header", extra={"client_id": None}
-                    )
+                    logger.warning("Invalid Basic auth header", extra={"client_id": None})
                     return (None, None), self.error(
                         request=request,
                         name="invalid_client",

tests/sentry/web/frontend/test_oauth_token.py

@@ -28,9 +28,7 @@ def test_no_get(self) -> None:
     def test_missing_grant_type(self) -> None:
         self.login_as(self.user)
 
-        resp = self.client.post(
-            self.path, {"client_id": "abcd", "client_secret": "abcd"}
-        )
+        resp = self.client.post(self.path, {"client_id": "abcd", "client_secret": "abcd"})
 
         assert resp.status_code == 400
         assert json.loads(resp.content) == {"error": "invalid_request"}
@@ -289,9 +287,7 @@ def test_grant_lock(self) -> None:
 
         # Simulate a concurrent request by using an existing grant
         # that has its grant lock taken out.
-        lock = locks.get(
-            ApiGrant.get_lock_key(self.grant.id), duration=10, name="api_grant"
-        )
+        lock = locks.get(ApiGrant.get_lock_key(self.grant.id), duration=10, name="api_grant")
         lock.acquire()
 
         # Attempt to create a token with the same grant
@@ -344,9 +340,7 @@ def test_inactive_application_rejects_token_creation(self) -> None:
         assert not ApiGrant.objects.filter(id=self.grant.id).exists()
 
         # Verify no token was created
-        assert not ApiToken.objects.filter(
-            application=self.application, user=self.user
-        ).exists()
+        assert not ApiToken.objects.filter(application=self.application, user=self.user).exists()
 
     def test_invalid_redirect_uri(self) -> None:
         self.login_as(self.user)
@@ -462,9 +456,7 @@ def test_expires_in_value(self) -> None:
         assert expires_in > 0, "expires_in should be positive (seconds until expiry)"
         # Allow for a few seconds of test execution time, but should be close to 30 days
         expected_seconds = 30 * 24 * 60 * 60  # 2,592,000 seconds
-        assert expires_in >= expected_seconds - 60, (
-            "expires_in should be close to 30 days"
-        )
+        assert expires_in >= expected_seconds - 60, "expires_in should be close to 30 days"
         assert expires_in <= expected_seconds, "expires_in should not exceed 30 days"
 
     def test_valid_params_id_token(self) -> None:
@@ -1371,9 +1363,7 @@ def test_inactive_application_rejects_device_code_grant(self) -> None:
         assert not ApiDeviceCode.objects.filter(id=self.device_code.id).exists()
 
         # No token should be created
-        assert not ApiToken.objects.filter(
-            application=self.application, user=self.user
-        ).exists()
+        assert not ApiToken.objects.filter(application=self.application, user=self.user).exists()
 
     def test_public_client_success(self) -> None:
         """Public clients (without client_secret) should be able to exchange device codes.