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
|
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
from devtools_testutils.perfstress_tests import PerfStressTest
from azure.identity import ClientSecretCredential, TokenCachePersistenceOptions
from azure.identity.aio import ClientSecretCredential as AsyncClientSecretCredential
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
pass
class PersistentCacheRead(PerfStressTest):
def __init__(self, arguments):
super().__init__(arguments)
client_id = self.get_from_env("AZURE_CLIENT_ID")
tenant_id = self.get_from_env("AZURE_TENANT_ID")
secret = self.get_from_env("AZURE_CLIENT_SECRET")
cache_options = TokenCachePersistenceOptions(allow_unencrypted_storage=True)
self.credential = ClientSecretCredential(tenant_id, client_id, secret, cache_persistence_options=cache_options)
self.async_credential = AsyncClientSecretCredential(
tenant_id, client_id, secret, cache_persistence_options=cache_options
)
self.scope = "https://vault.azure.net/.default"
async def global_setup(self):
"""Cache an access token"""
await super().global_setup()
self.credential.get_token(self.scope)
await self.async_credential.get_token(self.scope)
def run_sync(self):
self.credential.get_token(self.scope)
async def run_async(self):
await self.async_credential.get_token(self.scope)
async def close(self):
await self.async_credential.close()
await super().close()
|
