File: test_live_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 (104 lines) | stat: -rw-r--r-- 3,734 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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
from itertools import product
import pytest

from azure.identity.aio import (
    DefaultAzureCredential,
    CertificateCredential,
    ClientSecretCredential,
    AzureCliCredential,
    AzurePowerShellCredential,
    AzureDeveloperCliCredential,
)

from helpers import get_token_payload_contents, GET_TOKEN_METHODS

GRAPH_SCOPE = "https://graph.microsoft.com/.default"


async def get_token(credential, get_token_method, **kwargs):
    token = await getattr(credential, get_token_method)(GRAPH_SCOPE, **kwargs)
    assert token
    assert token.token
    assert token.expires_on
    return token


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "certificate_fixture,get_token_method", product(("live_pem_certificate", "live_pfx_certificate"), GET_TOKEN_METHODS)
)
async def test_certificate_credential(certificate_fixture, get_token_method, request):
    cert = request.getfixturevalue(certificate_fixture)

    tenant_id = cert["tenant_id"]
    client_id = cert["client_id"]

    credential = CertificateCredential(tenant_id, client_id, cert["cert_path"])
    await get_token(credential, get_token_method)

    credential = CertificateCredential(tenant_id, client_id, certificate_data=cert["cert_bytes"])
    token = await get_token(credential, get_token_method, enable_cae=True)
    parsed_payload = get_token_payload_contents(token.token)
    assert "xms_cc" in parsed_payload and "CP1" in parsed_payload["xms_cc"]

    if "password" in cert:
        credential = CertificateCredential(
            tenant_id, client_id, cert["cert_with_password_path"], password=cert["password"]
        )
        await get_token(credential, get_token_method)

        credential = CertificateCredential(
            tenant_id, client_id, certificate_data=cert["cert_with_password_bytes"], password=cert["password"]
        )
        await get_token(credential, get_token_method, enable_cae=True)


@pytest.mark.asyncio
@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
async def test_client_secret_credential(live_service_principal, get_token_method):
    credential = ClientSecretCredential(
        live_service_principal["tenant_id"],
        live_service_principal["client_id"],
        live_service_principal["client_secret"],
    )
    kwargs = {"enable_cae": True}
    if get_token_method == "get_token_info":
        kwargs = {"options": kwargs}
    token = await get_token(credential, get_token_method, **kwargs)
    parsed_payload = get_token_payload_contents(token.token)
    assert "xms_cc" in parsed_payload and "CP1" in parsed_payload["xms_cc"]


@pytest.mark.asyncio
@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
async def test_default_credential(live_service_principal, get_token_method):
    credential = DefaultAzureCredential()
    await get_token(credential, get_token_method)


@pytest.mark.manual
@pytest.mark.asyncio
@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
async def test_cli_credential(get_token_method):
    credential = AzureCliCredential()
    await get_token(credential, get_token_method)


@pytest.mark.manual
@pytest.mark.asyncio
@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
async def test_dev_cli_credential(get_token_method):
    credential = AzureDeveloperCliCredential()
    await get_token(credential, get_token_method)


@pytest.mark.manual
@pytest.mark.asyncio
@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
async def test_powershell_credential(get_token_method):
    credential = AzurePowerShellCredential()
    await get_token(credential, get_token_method)