File: test_client_init_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 (80 lines) | stat: -rw-r--r-- 3,117 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
import pytest
from azure.core.credentials import AzureNamedKeyCredential, AzureSasCredential
from azure.core.utils import parse_connection_string
from azure.iot.deviceprovisioning import generate_sas_token
from azure.iot.deviceprovisioning.aio import DeviceProvisioningClient
from conftest import ProvisioningServicePreparer
from devtools_testutils import AzureRecordedTestCase
from devtools_testutils.aio import recorded_by_proxy_async


class TestIndividualEnrollments(AzureRecordedTestCase):
    def create_provisioning_service_client_default_credentials(self, endpoint):
        credential = self.get_credential(DeviceProvisioningClient, is_async=True)
        client = DeviceProvisioningClient(endpoint=endpoint, credential=credential)
        return client

    def create_provisioning_service_client_namedkey(self, endpoint, name, key):
        credential = AzureNamedKeyCredential(name=name, key=key)
        client = DeviceProvisioningClient(endpoint=endpoint, credential=credential)
        return client

    def create_provisioning_service_client_sas(self, endpoint, sas_token):
        credential = AzureSasCredential(signature=sas_token)
        client = DeviceProvisioningClient(endpoint=endpoint, credential=credential)
        return client

    @pytest.mark.asyncio
    @ProvisioningServicePreparer()
    @recorded_by_proxy_async
    async def test_default_credentials(self, iothub_dps_endpoint):
        client = self.create_provisioning_service_client_default_credentials(
            iothub_dps_endpoint
        )

        enrollments = await client.enrollment.query(
            query_specification={"query": "SELECT *"}
        )
        assert len([enrollment async for enrollment in enrollments]) == 0

    @pytest.mark.asyncio
    @ProvisioningServicePreparer()
    @recorded_by_proxy_async
    async def test_named_key(
        self, iothub_dps_endpoint, iothub_dps_conn_str
    ):
        # get name and key from token
        cs_parts = parse_connection_string(iothub_dps_conn_str)
        name = cs_parts["sharedaccesskeyname"]
        key = cs_parts["sharedaccesskey"]
        client = self.create_provisioning_service_client_namedkey(
            iothub_dps_endpoint, name, key
        )

        enrollments = await client.enrollment.query(
            query_specification={"query": "SELECT *"}
        )

        assert len([enrollment async for enrollment in enrollments]) == 0

    @pytest.mark.asyncio
    @ProvisioningServicePreparer()
    @recorded_by_proxy_async
    async def test_sas_creds(
        self, iothub_dps_endpoint, iothub_dps_conn_str
    ):
        cs_parts = parse_connection_string(iothub_dps_conn_str)

        name = cs_parts["sharedaccesskeyname"]
        key = cs_parts["sharedaccesskey"]

        sas = generate_sas_token(iothub_dps_endpoint, name, key)
        client = self.create_provisioning_service_client_sas(
            iothub_dps_endpoint, sas
        )

        enrollments = await client.enrollment.query(
            query_specification={"query": "SELECT *"}
        )

        assert len([enrollment async for enrollment in enrollments]) == 0