Load OAuth 2.0 token exchange credentials provider from config file

tests/aio/test_credentials.py

@@ -2,6 +2,9 @@
 import time
 import grpc
 import threading
+import tempfile
+import os
+import json
 
 import tests.auth.test_credentials
 import tests.oauth2_token_exchange
@@ -11,23 +14,23 @@
 import ydb.oauth2_token_exchange.token_source
 
 
-class TestServiceAccountCredentials(ydb.aio.iam.ServiceAccountCredentials):
+class ServiceAccountCredentialsForTest(ydb.aio.iam.ServiceAccountCredentials):
     def _channel_factory(self):
         return grpc.aio.insecure_channel(self._iam_endpoint)
 
     def get_expire_time(self):
         return self._expires_in - time.time()
 
 
-class TestOauth2TokenExchangeCredentials(ydb.aio.oauth2_token_exchange.Oauth2TokenExchangeCredentials):
+class Oauth2TokenExchangeCredentialsForTest(ydb.aio.oauth2_token_exchange.Oauth2TokenExchangeCredentials):
     def get_expire_time(self):
         return self._expires_in - time.time()
 
 
 @pytest.mark.asyncio
 async def test_yandex_service_account_credentials():
     server = tests.auth.test_credentials.IamTokenServiceTestServer()
-    credentials = TestServiceAccountCredentials(
+    credentials = ServiceAccountCredentialsForTest(
         tests.auth.test_credentials.SERVICE_ACCOUNT_ID,
         tests.auth.test_credentials.ACCESS_KEY_ID,
         tests.auth.test_credentials.PRIVATE_KEY,
@@ -49,7 +52,7 @@ def serve(s):
     serve_thread = threading.Thread(target=serve, args=(server,))
     serve_thread.start()
 
-    credentials = TestOauth2TokenExchangeCredentials(
+    credentials = Oauth2TokenExchangeCredentialsForTest(
         server.endpoint(),
         ydb.oauth2_token_exchange.token_source.FixedTokenSource("test_src_token", "test_token_type"),
         audience=["a1", "a2"],
@@ -60,3 +63,51 @@ def serve(s):
     assert credentials.get_expire_time() <= 42
 
     serve_thread.join()
+
+
+@pytest.mark.asyncio
+async def test_oauth2_token_exchange_credentials_file():
+    server = tests.oauth2_token_exchange.test_token_exchange.Oauth2TokenExchangeServiceForTest(40124)
+
+    def serve(s):
+        s.handle_request()
+
+    serve_thread = threading.Thread(target=serve, args=(server,))
+    serve_thread.start()
+
+    cfg = {
+        "subject-credentials": {
+            "type": "FIXED",
+            "token": "test_src_token",
+            "token-type": "test_token_type",
+        },
+        "aud": [
+            "a1",
+            "a2",
+        ],
+        "scope": [
+            "s1",
+            "s2",
+        ],
+    }
+
+    temp_cfg_file = tempfile.NamedTemporaryFile(delete=False)
+    cfg_file_name = temp_cfg_file.name
+
+    try:
+        temp_cfg_file.write(json.dumps(cfg, indent=4).encode("utf-8"))
+        temp_cfg_file.close()
+
+        credentials = Oauth2TokenExchangeCredentialsForTest.from_file(
+            cfg_file=cfg_file_name, iam_endpoint=server.endpoint()
+        )
+
+        t = (await credentials.auth_metadata())[0][1]
+        assert t == "Bearer test_dst_token"
+        assert credentials.get_expire_time() <= 42
+
+        serve_thread.join()
+        os.remove(cfg_file_name)
+    except Exception:
+        os.remove(cfg_file_name)
+        raise

tests/auth/test_credentials.py

@@ -57,7 +57,7 @@ def get_endpoint(self):
         return "localhost:54321"
 
 
-class TestServiceAccountCredentials(ydb.iam.ServiceAccountCredentials):
+class ServiceAccountCredentialsForTest(ydb.iam.ServiceAccountCredentials):
     def _channel_factory(self):
         return grpc.insecure_channel(self._iam_endpoint)
 
@@ -67,7 +67,9 @@ def get_expire_time(self):
 
 def test_yandex_service_account_credentials():
     server = IamTokenServiceTestServer()
-    credentials = TestServiceAccountCredentials(SERVICE_ACCOUNT_ID, ACCESS_KEY_ID, PRIVATE_KEY, server.get_endpoint())
+    credentials = ServiceAccountCredentialsForTest(
+        SERVICE_ACCOUNT_ID, ACCESS_KEY_ID, PRIVATE_KEY, server.get_endpoint()
+    )
     t = credentials.get_auth_token()
     assert t == "test_token"
     assert credentials.get_expire_time() <= 42