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
|
import pytest
from azure.iot.deviceprovisioning.aio import DeviceProvisioningClient
from conftest import GLOBAL_PROVISIONING_HOST, ProvisioningServicePreparer
from devtools_testutils import AzureRecordedTestCase
from devtools_testutils.aio import recorded_by_proxy_async
from utility.common import generate_enrollment_group, sign_string
class TestDeviceRegistration(AzureRecordedTestCase):
def create_provisioning_service_client(self, endpoint):
credential = self.get_credential(DeviceProvisioningClient, is_async=True)
client = DeviceProvisioningClient(endpoint=endpoint, credential=credential)
return client
async def create_symmetric_device_registration(
self, endpoint, id_scope, group_id, device_id
):
from azure.iot.device import ProvisioningDeviceClient
client = self.create_provisioning_service_client(endpoint)
# get enrollment key
enrollment_group_attestation = (
await client.enrollment_group.get_attestation_mechanism(id=group_id)
)
primary_key = enrollment_group_attestation["symmetricKey"]["primaryKey"]
symmetric_key = sign_string(primary_key, device_id)
sdk = ProvisioningDeviceClient.create_from_symmetric_key(
provisioning_host=GLOBAL_PROVISIONING_HOST,
registration_id=device_id,
id_scope=id_scope,
symmetric_key=symmetric_key,
)
return sdk.register()
@pytest.mark.asyncio
@ProvisioningServicePreparer()
@recorded_by_proxy_async
async def test_device_registration_lifecycle(
self, iothub_dps_endpoint, iothub_dps_idscope
):
client = self.create_provisioning_service_client(
iothub_dps_endpoint
)
# create new enrollment group
enrollment_group_id = self.create_random_name("reg_enroll_grp_")
device_id = self.create_random_name("device-")
enrollment_group = generate_enrollment_group(
id=enrollment_group_id,
allocation_policy="static",
attestation_type="symmetricKey",
)
enrollment_group = await client.enrollment_group.create_or_update(
id=enrollment_group_id, enrollment_group=enrollment_group
)
# query - should have zero registrations
device_registrations = await client.device_registration_state.query(
id=enrollment_group_id
)
assert len([r async for r in device_registrations]) == 0
# create device registration
if self.is_live:
registration = await self.create_symmetric_device_registration(
endpoint=iothub_dps_endpoint,
id_scope=iothub_dps_idscope,
group_id=enrollment_group_id,
device_id=device_id,
)
assert device_id == registration.registration_state.device_id
# get registration
registration_response = await client.device_registration_state.get(id=device_id)
assert registration_response["status"] == "assigned"
registration_id = registration_response["registrationId"]
# query registration
registration_query_response = await client.device_registration_state.query(
id=enrollment_group_id
)
registrations = [reg async for reg in registration_query_response]
assert len(registrations) == 1
assert registrations[0]["registrationId"] == registration_id
# delete registration
await client.device_registration_state.delete(id=registration_id)
# confirm delete
registration_query_response = await client.device_registration_state.query(
id=enrollment_group_id
)
assert len([reg async for reg in registration_query_response]) == 0
# delete enrollment group
await client.enrollment_group.delete(id=enrollment_group_id)
|
