File: test_cloud_shell_async.py

package info (click to toggle)
python-azure 20250603%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 851,724 kB
  • sloc: python: 7,362,925; ansic: 804; javascript: 287; makefile: 195; sh: 145; xml: 109
file content (26 lines) | stat: -rw-r--r-- 1,192 bytes parent folder | download | duplicates (2)
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
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
import pytest

from azure.core import AsyncPipelineClient
from azure.core.pipeline.policies import ContentDecodePolicy, HttpLoggingPolicy, AsyncRedirectPolicy, AsyncRetryPolicy
from azure.identity.aio import ManagedIdentityCredential


@pytest.mark.cloudshell
@pytest.mark.asyncio
async def test_cloud_shell_live(cloud_shell):
    credential = ManagedIdentityCredential()
    token = await credential.get_token("https://vault.azure.net")

    # Validate the token by sending a request to the Key Vault. The request is manual because azure-keyvault-secrets
    # can't authenticate in Cloud Shell; the MSI endpoint there doesn't support AADv2 scopes.
    policies = [ContentDecodePolicy(), AsyncRedirectPolicy(), AsyncRetryPolicy(), HttpLoggingPolicy()]
    client = AsyncPipelineClient(cloud_shell["vault_url"], policies=policies)
    list_secrets = client.get(
        "secrets", headers={"Authorization": "Bearer " + token.token}, params={"api-version": "7.0"}
    )
    async with client:
        await client._pipeline.run(list_secrets)