Style

Author: UgnineSirdis

tests/aio/test_credentials.py

@@ -9,9 +9,7 @@
 
 class TestServiceAccountCredentials(ydb.aio.iam.ServiceAccountCredentials):
     def _channel_factory(self):
-        return grpc.aio.insecure_channel(
-            self._iam_endpoint
-        )
+        return grpc.aio.insecure_channel(self._iam_endpoint)
 
     def get_expire_time(self):
         return self._expires_in - time.time()
@@ -25,7 +23,12 @@ def get_expire_time(self):
 @pytest.mark.asyncio
 async def test_yandex_service_account_credentials():
     server = tests.auth.test_credentials.IamTokenServiceTestServer()
-    credentials = TestServiceAccountCredentials(tests.auth.test_credentials.SERVICE_ACCOUNT_ID, tests.auth.test_credentials.ACCESS_KEY_ID, tests.auth.test_credentials.PRIVATE_KEY, server.get_endpoint())
+    credentials = TestServiceAccountCredentials(
+        tests.auth.test_credentials.SERVICE_ACCOUNT_ID,
+        tests.auth.test_credentials.ACCESS_KEY_ID,
+        tests.auth.test_credentials.PRIVATE_KEY,
+        server.get_endpoint(),
+    )
     t = (await credentials.auth_metadata())[0][1]
     assert t == "test_token"
     assert credentials.get_expire_time() <= 42
@@ -42,7 +45,12 @@ def serve(s):
     serve_thread = threading.Thread(target=serve, args=(server,))
     serve_thread.start()
 
-    credentials = TestNebiusServiceAccountCredentials(tests.auth.test_credentials.SERVICE_ACCOUNT_ID, tests.auth.test_credentials.ACCESS_KEY_ID, tests.auth.test_credentials.PRIVATE_KEY, server.endpoint())
+    credentials = TestNebiusServiceAccountCredentials(
+        tests.auth.test_credentials.SERVICE_ACCOUNT_ID,
+        tests.auth.test_credentials.ACCESS_KEY_ID,
+        tests.auth.test_credentials.PRIVATE_KEY,
+        server.endpoint(),
+    )
     t = (await credentials.auth_metadata())[0][1]
     assert t == "test_nebius_token"
     assert credentials.get_expire_time() <= 42

tests/auth/test_credentials.py

@@ -33,7 +33,9 @@ class IamTokenServiceForTest(iam_token_service_pb2_grpc.IamTokenServiceServicer)
     def Create(self, request, context):
         print("IAM token service request: {}".format(request))
         # Validate jwt:
-        decoded = jwt.decode(request.jwt, key=PUBLIC_KEY, algorithms=["PS256"], audience="https://iam.api.cloud.yandex.net/iam/v1/tokens")
+        decoded = jwt.decode(
+            request.jwt, key=PUBLIC_KEY, algorithms=["PS256"], audience="https://iam.api.cloud.yandex.net/iam/v1/tokens"
+        )
         assert decoded["iss"] == SERVICE_ACCOUNT_ID
         assert decoded["aud"] == "https://iam.api.cloud.yandex.net/iam/v1/tokens"
         assert abs(decoded["iat"] - time.time()) <= 60
@@ -61,9 +63,7 @@ def get_endpoint(self):
 
 class TestServiceAccountCredentials(ydb.iam.ServiceAccountCredentials):
     def _channel_factory(self):
-        return grpc.insecure_channel(
-            self._iam_endpoint
-        )
+        return grpc.insecure_channel(self._iam_endpoint)
 
     def get_expire_time(self):
         return self._expires_in - time.time()
@@ -93,7 +93,9 @@ def do_POST(self):
 
         assert len(parsed_request["subject_token"]) == 1
         jwt_token = parsed_request["subject_token"][0]
-        decoded = jwt.decode(jwt_token, key=PUBLIC_KEY, algorithms=["RS256"], audience="token-service.iam.new.nebiuscloud.net")
+        decoded = jwt.decode(
+            jwt_token, key=PUBLIC_KEY, algorithms=["RS256"], audience="token-service.iam.new.nebiuscloud.net"
+        )
         assert decoded["iss"] == SERVICE_ACCOUNT_ID
         assert decoded["sub"] == SERVICE_ACCOUNT_ID
         assert decoded["aud"] == "token-service.iam.new.nebiuscloud.net"
@@ -104,7 +106,7 @@ def do_POST(self):
             "access_token": "test_nebius_token",
             "issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
             "token_type": "Bearer",
-            "expires_in": 42
+            "expires_in": 42,
         }
 
         self.send_response(200)