File: test_cloud_shell_async.py

package info (click to toggle)
python-azure 20201208%2Bgit-6
  • links: PTS, VCS
  • area: main
  • in suites: bullseye
  • size: 1,437,920 kB
  • sloc: python: 4,287,452; javascript: 269; makefile: 198; sh: 187; xml: 106
file content (26 lines) | stat: -rw-r--r-- 1,186 bytes parent folder | download
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 = 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)