From 1c2d15ab4b2e040eb81cc7f25cc13a418103bd2f Mon Sep 17 00:00:00 2001 From: Avin Agrawal Date: Mon, 17 May 2021 16:51:12 -0700 Subject: [PATCH 1/9] Add warning for qos deprecation and update contributing guide --- CONTRIBUTING.md | 11 +++++------ azext_iot/_params.py | 5 +++-- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 1e2322253..4b199d06c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -98,10 +98,10 @@ pip install -r path/to/source/dev_requirements ``` _Hub:_ -`pytest azext_iot/tests/test_iot_ext_unit.py` +`pytest azext_iot/tests/iothub/test_iot_ext_unit.py` _DPS:_ -`pytest azext_iot/tests/test_iot_dps_unit.py` +`pytest azext_iot/tests/dps/test_iot_dps_unit.py` ### Integration Tests @@ -127,7 +127,6 @@ You can either manually set the environment variables or use the `pytest.ini.exa AZURE_TEST_RUN_LIVE=True azext_iot_testrg="Resource Group that contains your IoT Hub" azext_iot_testhub="IoT Hub Name" - azext_iot_testhub_cs="IoT Hub Connection String" azext_iot_testdps="IoT Hub DPS Name" azext_iot_teststorageuri="Blob Container SAS Uri" azext_iot_identity_teststorageid="Storage Account ID" @@ -141,13 +140,13 @@ You can either manually set the environment variables or use the `pytest.ini.exa Execute the following command to run the IoT Hub integration tests: -`pytest azext_iot/tests/test_iot_ext_int.py` +`pytest azext_iot/tests/iothub/test_iot_ext_int.py` ##### Device Provisioning Service Execute the following command to run the IoT Hub DPS integration tests: -`pytest azext_iot/tests/test_iot_dps_int.py` +`pytest azext_iot/tests/dps/test_iot_dps_int.py` #### Unit and Integration Tests Single Command @@ -264,7 +263,7 @@ https://medium.com/@marcobelo/setting-up-python-black-on-visual-studio-code-5318 https://docs.python.org/3/library/pdb.html -1. `pip install pdb` +1. `pip install pdbpp` 2. If you need a breakpoint, put `import pdb; pdb.set_trace()` in your code 3. Run your command, it should break execution wherever you put the breakpoint. diff --git a/azext_iot/_params.py b/azext_iot/_params.py index 99f3cfcca..47ca46553 100644 --- a/azext_iot/_params.py +++ b/azext_iot/_params.py @@ -82,7 +82,8 @@ type=str, nargs="?", choices=["0", "1"], - help="Quality of Service. 0 = At most once, 1 = At least once. 2 (Exactly once) is not supported.", + help="Quality of Service. 0 = At most once, 1 = At least once. 2 (Exactly once) is not supported." + "This command parameter has been deprecated and will be removed in the next release." ) event_timeout_type = CLIArgumentType( @@ -557,7 +558,7 @@ def load_arguments(self, _): arg_type=get_enum_type(ProtocolType), help="Indicates device-to-cloud message protocol", ) - context.argument("qos", arg_type=qos_type) + context.argument("qos", arg_type=qos_type, deprecate_info=context.deprecate(expiration='0.10.12')) with self.argument_context("iot device simulate") as context: context.argument( From f5265b7b0d5f07fadb63db4abbf8de77e18ef0a2 Mon Sep 17 00:00:00 2001 From: Avin Agrawal Date: Mon, 17 May 2021 16:57:05 -0700 Subject: [PATCH 2/9] removing version from deprecate_info --- azext_iot/_params.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/azext_iot/_params.py b/azext_iot/_params.py index 47ca46553..f8e3982b4 100644 --- a/azext_iot/_params.py +++ b/azext_iot/_params.py @@ -558,7 +558,7 @@ def load_arguments(self, _): arg_type=get_enum_type(ProtocolType), help="Indicates device-to-cloud message protocol", ) - context.argument("qos", arg_type=qos_type, deprecate_info=context.deprecate(expiration='0.10.12')) + context.argument("qos", arg_type=qos_type, deprecate_info=context.deprecate()) with self.argument_context("iot device simulate") as context: context.argument( From bad3201fe379dd61148f2c67d8ac671a6f64b807 Mon Sep 17 00:00:00 2001 From: Avin Agrawal Date: Mon, 17 May 2021 17:06:33 -0700 Subject: [PATCH 3/9] Integration tests command update in Contributing guide --- CONTRIBUTING.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 4b199d06c..8301c0439 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -140,13 +140,13 @@ You can either manually set the environment variables or use the `pytest.ini.exa Execute the following command to run the IoT Hub integration tests: -`pytest azext_iot/tests/iothub/test_iot_ext_int.py` +`pytest azext_iot/tests/iothub/ -k "_int"` ##### Device Provisioning Service Execute the following command to run the IoT Hub DPS integration tests: -`pytest azext_iot/tests/dps/test_iot_dps_int.py` +`pytest azext_iot/tests/dps/ -k "_int"` #### Unit and Integration Tests Single Command From ad6da4e7177f060b46f3ad4b130cb75f3deb661e Mon Sep 17 00:00:00 2001 From: Avin Agrawal Date: Thu, 20 May 2021 01:55:02 -0700 Subject: [PATCH 4/9] Update help message for d2c command and mqtt simulation to indicate only sas auth is supported --- azext_iot/_help.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/azext_iot/_help.py b/azext_iot/_help.py index 0729b4d07..a4dcb7ae0 100644 --- a/azext_iot/_help.py +++ b/azext_iot/_help.py @@ -810,6 +810,7 @@ type: command short-summary: Send an mqtt device-to-cloud message. The command supports sending messages with application and system properties. + Note: The command only works for symmetric key auth (SAS) based devices examples: - name: Basic usage text: az iot device send-d2c-message -n {iothub_name} -d {device_id} @@ -831,7 +832,8 @@ While the device simulation is running, the device will automatically receive and acknowledge cloud-to-device (c2d) messages. For mqtt simulation, all c2d messages will be acknowledged with completion. For http simulation c2d acknowledgement is based on user - selection which can be complete, reject or abandon. + selection which can be complete, reject or abandon. Additionally, mqtt simulation is only + supported for symmetric key auth (SAS) based devices Note: The command by default will set content-type to application/json and content-encoding to utf-8. This can be overriden. From 8220bd0d437c41c375420f787a93ec27dadb615c Mon Sep 17 00:00:00 2001 From: Avin Agrawal Date: Fri, 28 May 2021 02:41:44 -0700 Subject: [PATCH 5/9] Checks and error messages when non SAS devices are used for MQTT operations --- azext_iot/operations/hub.py | 9 ++++++ azext_iot/tests/conftest.py | 34 +++++++++++++++++++++ azext_iot/tests/iothub/test_iot_ext_unit.py | 2 +- 3 files changed, 44 insertions(+), 1 deletion(-) diff --git a/azext_iot/operations/hub.py b/azext_iot/operations/hub.py index 541b7284c..820a8e560 100644 --- a/azext_iot/operations/hub.py +++ b/azext_iot/operations/hub.py @@ -2069,6 +2069,10 @@ def _iot_device_send_message( import ssl import os + device = _iot_device_show(target, device_id) + if device is not None and device.get("authentication", {}).get("type", "") != "sas": + raise CLIError('D2C send message command only supports symmetric key auth (SAS) based devices') + msgs = [] if properties: properties = validate_key_value_pairs(properties) @@ -2457,6 +2461,11 @@ def http_wrap(target, device_id, generator): try: if protocol_type == ProtocolType.mqtt.name: + + device = _iot_device_show(target, device_id) + if device is not None and device.get("authentication", {}).get("type", "") != "sas": + raise CLIError('MQTT simulation is only supported for symmetric key auth (SAS) based devices') + wrap = mqtt_client_wrap( target=target, device_id=device_id, diff --git a/azext_iot/tests/conftest.py b/azext_iot/tests/conftest.py index a60ed08a6..6ad39e2b2 100644 --- a/azext_iot/tests/conftest.py +++ b/azext_iot/tests/conftest.py @@ -26,6 +26,7 @@ path_iot_hub_monitor_events_entrypoint = ( "azext_iot.operations.hub._iot_hub_monitor_events" ) +path_iot_device_show = "azext_iot.operations.hub._iot_device_show" hub_entity = "myhub.azure-devices.net" instance_name = generate_generic_id() @@ -137,6 +138,39 @@ def fixture_monitor_events_entrypoint(mocker): return mocker.patch(path_iot_hub_monitor_events_entrypoint) +@pytest.fixture() +def fixture_iot_device_show(mocker): + device = mocker.patch(path_iot_device_show) + device.return_value = { + "authentication": { + "symmetricKey": { + "primaryKey": "test_pk", + "secondaryKey": "test_sk" + }, + "type": "sas", + "x509Thumbprint": { + "primaryThumbprint": None, + "secondaryThumbprint": None + } + }, + "capabilities": { + "iotEdge": False + }, + "cloudToDeviceMessageCount": 0, + "connectionState": "Disconnected", + "connectionStateUpdatedTime": "2021-05-27T00:36:11.2861732Z", + "deviceId": "Test_Device_1", + "etag": "ODgxNTgwOA==", + "generationId": "637534345627501371", + "hub": "test-iot-hub.azure-devices.net", + "lastActivityTime": "2021-05-27T00:18:16.3154299Z", + "status": "enabled", + "statusReason": None, + "statusUpdatedTime": "0001-01-01T00:00:00Z" + } + return device + + # TODO: To be deprecated asap. Leverage mocked_response fixture for this functionality. def build_mock_response( mocker=None, status_code=200, payload=None, headers=None, **kwargs diff --git a/azext_iot/tests/iothub/test_iot_ext_unit.py b/azext_iot/tests/iothub/test_iot_ext_unit.py index 3ae27c1cd..57672d8a1 100644 --- a/azext_iot/tests/iothub/test_iot_ext_unit.py +++ b/azext_iot/tests/iothub/test_iot_ext_unit.py @@ -2001,7 +2001,7 @@ def test_generate_sas_token(self): class TestDeviceSimulate: @pytest.fixture(params=[204]) - def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request): + def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request, fixture_iot_device_show): service_client = mocker.patch(path_service_client) service_client.return_value = build_mock_response(mocker, request.param, {}) return service_client From c08e13b313b287f6ca217ea01132c3be16a76098 Mon Sep 17 00:00:00 2001 From: Avin Agrawal Date: Fri, 28 May 2021 03:38:48 -0700 Subject: [PATCH 6/9] Indentation update in help file --- azext_iot/_help.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/azext_iot/_help.py b/azext_iot/_help.py index 0c798a614..a4c973d25 100644 --- a/azext_iot/_help.py +++ b/azext_iot/_help.py @@ -828,9 +828,11 @@ "iot device send-d2c-message" ] = """ type: command - short-summary: Send an mqtt device-to-cloud message. - The command supports sending messages with application and system properties. - Note: The command only works for symmetric key auth (SAS) based devices + short-summary: | + Send an mqtt device-to-cloud message. + The command supports sending messages with application and system properties. + + Note: The command only works for symmetric key auth (SAS) based devices examples: - name: Basic usage text: az iot device send-d2c-message -n {iothub_name} -d {device_id} From 7cb9ea44afdfd6bfd597e7d6d902b8bb8b30ccf3 Mon Sep 17 00:00:00 2001 From: Avin Agrawal Date: Fri, 28 May 2021 04:08:35 -0700 Subject: [PATCH 7/9] fix styling --- azext_iot/_help.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/azext_iot/_help.py b/azext_iot/_help.py index a4c973d25..1732e2b8a 100644 --- a/azext_iot/_help.py +++ b/azext_iot/_help.py @@ -829,9 +829,9 @@ ] = """ type: command short-summary: | - Send an mqtt device-to-cloud message. + Send an mqtt device-to-cloud message. The command supports sending messages with application and system properties. - + Note: The command only works for symmetric key auth (SAS) based devices examples: - name: Basic usage From 4a6e35ed02ab5895646f6eee5ed669c564f451ab Mon Sep 17 00:00:00 2001 From: Avin Agrawal Date: Fri, 28 May 2021 18:12:45 -0700 Subject: [PATCH 8/9] Adding enum and updating test --- azext_iot/common/shared.py | 10 ++++ azext_iot/operations/hub.py | 41 ++++++------- azext_iot/tests/conftest.py | 38 +++++++++++- azext_iot/tests/iothub/test_iot_ext_unit.py | 64 +++++++++++---------- 4 files changed, 102 insertions(+), 51 deletions(-) diff --git a/azext_iot/common/shared.py b/azext_iot/common/shared.py index 1513cc0f2..ad22e4c0e 100644 --- a/azext_iot/common/shared.py +++ b/azext_iot/common/shared.py @@ -56,6 +56,16 @@ class DeviceAuthType(Enum): x509_ca = "x509_ca" +class HubDeviceAuthType(Enum): + """ + Hub Device Authorization type. + """ + + sas = "sas" + selfSigned = "selfSigned" + certificateAuthority = "certificateAuthority" + + class KeyType(Enum): """ Shared private key. diff --git a/azext_iot/operations/hub.py b/azext_iot/operations/hub.py index 417e33e76..cb111ee27 100644 --- a/azext_iot/operations/hub.py +++ b/azext_iot/operations/hub.py @@ -28,6 +28,7 @@ SettleType, RenewKeyType, IoTHubStateType, + HubDeviceAuthType, ) from azext_iot.iothub.providers.discovery import IotHubDiscovery from azext_iot.common.utility import ( @@ -259,21 +260,21 @@ def _assemble_auth(auth_method, pk, sk): ) auth = None - if auth_method in [DeviceAuthType.shared_private_key.name, "sas"]: + if auth_method in [DeviceAuthType.shared_private_key.name, HubDeviceAuthType.sas.name]: auth = AuthenticationMechanism( - symmetric_key=SymmetricKey(primary_key=pk, secondary_key=sk), type="sas" + symmetric_key=SymmetricKey(primary_key=pk, secondary_key=sk), type=HubDeviceAuthType.sas.name ) - elif auth_method in [DeviceAuthType.x509_thumbprint.name, "selfSigned"]: + elif auth_method in [DeviceAuthType.x509_thumbprint.name, HubDeviceAuthType.selfSigned.name]: if not pk: raise ValueError("primary thumbprint required with selfSigned auth") auth = AuthenticationMechanism( x509_thumbprint=X509Thumbprint( primary_thumbprint=pk, secondary_thumbprint=sk ), - type="selfSigned", + type=HubDeviceAuthType.selfSigned.name, ) - elif auth_method in [DeviceAuthType.x509_ca.name, "certificateAuthority"]: - auth = AuthenticationMechanism(type="certificateAuthority") + elif auth_method in [DeviceAuthType.x509_ca.name, HubDeviceAuthType.certificateAuthority.name]: + auth = AuthenticationMechanism(type=HubDeviceAuthType.certificateAuthority.name) else: raise ValueError("Authorization method {} invalid.".format(auth_method)) return auth @@ -306,7 +307,7 @@ def update_iot_device_custom( auth_type = instance["authentication"]["type"] if auth_method is not None: if auth_method == DeviceAuthType.shared_private_key.name: - auth = "sas" + auth = HubDeviceAuthType.sas.name if (primary_key and not secondary_key) or ( not primary_key and secondary_key ): @@ -314,7 +315,7 @@ def update_iot_device_custom( instance["authentication"]["symmetricKey"]["primaryKey"] = primary_key instance["authentication"]["symmetricKey"]["secondaryKey"] = secondary_key elif auth_method == DeviceAuthType.x509_thumbprint.name: - auth = "selfSigned" + auth = HubDeviceAuthType.selfSigned.name if not any([primary_thumbprint, secondary_thumbprint]): raise CLIError( "primary or secondary Thumbprint required with selfSigned auth" @@ -328,13 +329,13 @@ def update_iot_device_custom( "secondaryThumbprint" ] = secondary_thumbprint elif auth_method == DeviceAuthType.x509_ca.name: - auth = "certificateAuthority" + auth = HubDeviceAuthType.certificateAuthority.name else: raise ValueError("Authorization method {} invalid.".format(auth_method)) instance["authentication"]["type"] = auth # if no new auth_method is provided, validate secondary auth arguments and update accordingly - elif auth_type == "sas": + elif auth_type == HubDeviceAuthType.sas.name: if any([primary_thumbprint, secondary_thumbprint]): raise ValueError( "Device authorization method {} does not support primary or secondary thumbprints.".format( @@ -346,7 +347,7 @@ def update_iot_device_custom( if secondary_key: instance["authentication"]["symmetricKey"]["secondaryKey"] = secondary_key - elif auth_type == "selfSigned": + elif auth_type == HubDeviceAuthType.selfSigned.name: if any([primary_key, secondary_key]): raise ValueError( "Device authorization method {} does not support primary or secondary keys.".format( @@ -476,7 +477,7 @@ def iot_device_key_regenerate( auth_type=auth_type_dataplane, ) device = _iot_device_show(target, device_id) - if device["authentication"]["type"] != "sas": + if device["authentication"]["type"] != HubDeviceAuthType.sas.name: raise CLIError("Device authentication should be of type sas") pk = device["authentication"]["symmetricKey"]["primaryKey"] @@ -867,15 +868,15 @@ def _handle_module_update_params(parameters): def _parse_auth(parameters): - valid_auth = ["sas", "selfSigned", "certificateAuthority"] + valid_auth = [HubDeviceAuthType.sas.name, HubDeviceAuthType.selfSigned.name, HubDeviceAuthType.certificateAuthority.name] auth = parameters["authentication"].get("type") if auth not in valid_auth: raise CLIError("authentication.type must be one of {}".format(valid_auth)) pk = sk = None - if auth == "sas": + if auth == HubDeviceAuthType.sas.name: pk = parameters["authentication"]["symmetricKey"]["primaryKey"] sk = parameters["authentication"]["symmetricKey"]["secondaryKey"] - elif auth == "selfSigned": + elif auth == HubDeviceAuthType.selfSigned.name: pk = parameters["authentication"]["x509Thumbprint"]["primaryThumbprint"] sk = parameters["authentication"]["x509Thumbprint"]["secondaryThumbprint"] if not any([pk, sk]): @@ -1866,7 +1867,7 @@ def iot_get_sas_token( ) return { - "sas": _iot_build_sas_token( + HubDeviceAuthType.sas.name: _iot_build_sas_token( cmd, hub_name, device_id, @@ -1964,13 +1965,13 @@ def _build_device_or_module_connection_string(entity, key_type="primary"): ) auth = entity["authentication"] auth_type = auth["type"].lower() - if auth_type == "sas": + if auth_type == HubDeviceAuthType.sas.name: key = "SharedAccessKey={}".format( auth["symmetricKey"]["primaryKey"] if key_type == "primary" else auth["symmetricKey"]["secondaryKey"] ) - elif auth_type in ["certificateauthority", "selfsigned"]: + elif auth_type in [HubDeviceAuthType.certificateAuthority.name, HubDeviceAuthType.selfSigned.name]: key = "x509=true" else: raise CLIError("Unable to form target connection string") @@ -2071,7 +2072,7 @@ def _iot_device_send_message( import os device = _iot_device_show(target, device_id) - if device is not None and device.get("authentication", {}).get("type", "") != "sas": + if device is not None and device.get("authentication", {}).get("type", "") != HubDeviceAuthType.sas.name: raise CLIError('D2C send message command only supports symmetric key auth (SAS) based devices') msgs = [] @@ -2462,7 +2463,7 @@ def http_wrap(target, device_id, generator): if protocol_type == ProtocolType.mqtt.name: device = _iot_device_show(target, device_id) - if device is not None and device.get("authentication", {}).get("type", "") != "sas": + if device is not None and device.get("authentication", {}).get("type", "") != HubDeviceAuthType.sas.name: raise CLIError('MQTT simulation is only supported for symmetric key auth (SAS) based devices') wrap = mqtt_client_wrap( diff --git a/azext_iot/tests/conftest.py b/azext_iot/tests/conftest.py index 6ad39e2b2..1a6ace508 100644 --- a/azext_iot/tests/conftest.py +++ b/azext_iot/tests/conftest.py @@ -14,6 +14,7 @@ from azure.cli.core.commands import AzCliCommand from azure.cli.core.mock import DummyCli from azext_iot.tests.generators import generate_generic_id +from azext_iot.common.shared import HubDeviceAuthType path_iot_hub_service_factory = "azext_iot._factory.iot_hub_service_factory" path_service_client = "msrest.service_client.ServiceClient.send" @@ -139,7 +140,7 @@ def fixture_monitor_events_entrypoint(mocker): @pytest.fixture() -def fixture_iot_device_show(mocker): +def fixture_iot_device_show_sas(mocker): device = mocker.patch(path_iot_device_show) device.return_value = { "authentication": { @@ -147,7 +148,40 @@ def fixture_iot_device_show(mocker): "primaryKey": "test_pk", "secondaryKey": "test_sk" }, - "type": "sas", + "type": HubDeviceAuthType.sas.name, + "x509Thumbprint": { + "primaryThumbprint": None, + "secondaryThumbprint": None + } + }, + "capabilities": { + "iotEdge": False + }, + "cloudToDeviceMessageCount": 0, + "connectionState": "Disconnected", + "connectionStateUpdatedTime": "2021-05-27T00:36:11.2861732Z", + "deviceId": "Test_Device_1", + "etag": "ODgxNTgwOA==", + "generationId": "637534345627501371", + "hub": "test-iot-hub.azure-devices.net", + "lastActivityTime": "2021-05-27T00:18:16.3154299Z", + "status": "enabled", + "statusReason": None, + "statusUpdatedTime": "0001-01-01T00:00:00Z" + } + return device + + +@pytest.fixture() +def fixture_self_signed_device_show_self_signed(mocker): + device = mocker.patch(path_iot_device_show) + device.return_value = { + "authentication": { + "symmetricKey": { + "primaryKey": "test_pk", + "secondaryKey": "test_sk" + }, + "type": HubDeviceAuthType.selfSigned.name, "x509Thumbprint": { "primaryThumbprint": None, "secondaryThumbprint": None diff --git a/azext_iot/tests/iothub/test_iot_ext_unit.py b/azext_iot/tests/iothub/test_iot_ext_unit.py index 57672d8a1..e14582b63 100644 --- a/azext_iot/tests/iothub/test_iot_ext_unit.py +++ b/azext_iot/tests/iothub/test_iot_ext_unit.py @@ -35,7 +35,7 @@ mock_target, generate_cs, ) - +from azext_iot.common.shared import HubDeviceAuthType device_id = "mydevice" child_device_id = "child_device1" @@ -131,13 +131,13 @@ def test_device_create(self, serviceclient, req): assert body["capabilities"]["iotEdge"] == req["ee"] if req["auth"] == "shared_private_key": - assert body["authentication"]["type"] == "sas" + assert body["authentication"]["type"] == HubDeviceAuthType.sas.name elif req["auth"] == "x509_ca": - assert body["authentication"]["type"] == "certificateAuthority" + assert body["authentication"]["type"] == HubDeviceAuthType.certificateAuthority.name assert not body["authentication"].get("x509Thumbprint") assert not body["authentication"].get("symmetricKey") elif req["auth"] == "x509_thumbprint": - assert body["authentication"]["type"] == "selfSigned" + assert body["authentication"]["type"] == HubDeviceAuthType.selfSigned.name x509tp = body["authentication"]["x509Thumbprint"] assert x509tp["primaryThumbprint"] if req["stp"] is None: @@ -192,7 +192,7 @@ def generate_device_show(**kvp): "authentication": { "symmetricKey": {"primaryKey": None, "secondaryKey": None}, "x509Thumbprint": {"primaryThumbprint": None, "secondaryThumbprint": None}, - "type": "sas", + "type": HubDeviceAuthType.sas.name, }, "capabilities": {"iotEdge": True}, "deviceId": device_id, @@ -244,7 +244,7 @@ def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request): generate_device_show( authentication={ "symmetricKey": {"primaryKey": "", "secondaryKey": ""}, - "type": "sas", + "type": HubDeviceAuthType.sas.name, } ) ), @@ -255,13 +255,13 @@ def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request): "primaryThumbprint": "123", "secondaryThumbprint": "321", }, - "type": "selfSigned", + "type": HubDeviceAuthType.selfSigned.name, } ) ), ( generate_device_show( - authentication={"type": "certificateAuthority"}, + authentication={"type": HubDeviceAuthType.certificateAuthority.name}, etag=generate_generic_id(), ) ), @@ -286,10 +286,10 @@ def test_device_update(self, fixture_cmd, serviceclient, req): assert body["status"] == req["status"] assert body["capabilities"]["iotEdge"] == req["capabilities"]["iotEdge"] assert req["authentication"]["type"] == body["authentication"]["type"] - if req["authentication"]["type"] == "certificateAuthority": + if req["authentication"]["type"] == HubDeviceAuthType.certificateAuthority.name: assert not body["authentication"].get("x509Thumbprint") assert not body["authentication"].get("symmetricKey") - elif req["authentication"]["type"] == "selfSigned": + elif req["authentication"]["type"] == HubDeviceAuthType.selfSigned.name: assert body["authentication"]["x509Thumbprint"]["primaryThumbprint"] assert body["authentication"]["x509Thumbprint"]["secondaryThumbprint"] @@ -320,7 +320,7 @@ def test_device_update(self, fixture_cmd, serviceclient, req): ( generate_device_show( authentication={ - "type": "selfSigned", + "type": HubDeviceAuthType.selfSigned.name, "symmetricKey": {"primaryKey": None, "secondaryKey": None}, "x509Thumbprint": { "primaryThumbprint": "123", @@ -337,7 +337,7 @@ def test_device_update(self, fixture_cmd, serviceclient, req): ( generate_device_show( authentication={ - "type": "certificateAuthority", + "type": HubDeviceAuthType.certificateAuthority.name, "symmetricKey": {"primaryKey": None, "secondaryKey": None}, "x509Thumbprint": { "primaryThumbprint": None, @@ -356,7 +356,7 @@ def test_device_update(self, fixture_cmd, serviceclient, req): ( generate_device_show( authentication={ - "type": "selfSigned", + "type": HubDeviceAuthType.selfSigned.name, "symmetricKey": {"primaryKey": None, "secondaryKey": None}, "x509Thumbprint": { "primaryThumbprint": "123", @@ -389,7 +389,7 @@ def test_iot_device_custom(self, fixture_cmd, serviceclient, req, arg): assert instance["statusReason"] == arg["status_reason"] if arg["auth_method"]: if arg["auth_method"] == "shared_private_key": - assert instance["authentication"]["type"] == "sas" + assert instance["authentication"]["type"] == HubDeviceAuthType.sas.name instance["authentication"]["symmetricKey"]["primaryKey"] == arg[ "primary_key" ] @@ -397,7 +397,7 @@ def test_iot_device_custom(self, fixture_cmd, serviceclient, req, arg): "secondary_key" ] if arg["auth_method"] == "x509_thumbprint": - assert instance["authentication"]["type"] == "selfSigned" + assert instance["authentication"]["type"] == HubDeviceAuthType.selfSigned.name if arg["primary_thumbprint"]: instance["authentication"]["x509Thumbprint"][ "primaryThumbprint" @@ -407,7 +407,7 @@ def test_iot_device_custom(self, fixture_cmd, serviceclient, req, arg): "secondaryThumbprint" ] = arg["secondary_thumbprint"] if arg["auth_method"] == "x509_ca": - assert instance["authentication"]["type"] == "certificateAuthority" + assert instance["authentication"]["type"] == HubDeviceAuthType.certificateAuthority.name @pytest.mark.parametrize( "req, arg, exp", @@ -437,7 +437,7 @@ def test_iot_device_custom(self, fixture_cmd, serviceclient, req, arg): ( generate_device_show( authentication={ - "type": "selfSigned", + "type": HubDeviceAuthType.selfSigned.name, "symmetricKey": {"primaryKey": None, "secondaryKey": None}, "x509Thumbprint": { "primaryThumbprint": "123", @@ -474,7 +474,7 @@ def test_iot_device_custom_invalid_args(self, serviceclient, req, arg, exp): "primaryThumbprint": "", "secondaryThumbprint": "", }, - "type": "selfSigned", + "type": HubDeviceAuthType.selfSigned.name, } ), CLIError, @@ -511,7 +511,7 @@ def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request): "authentication", { "symmetricKey": {"primaryKey": "123", "secondaryKey": "321"}, - "type": "sas", + "type": HubDeviceAuthType.sas.name, }, ) test_side_effect = [ @@ -737,13 +737,13 @@ def test_device_module_create(self, serviceclient, req): assert body["moduleId"] == req["module_id"] if req["auth"] == "shared_private_key": - assert body["authentication"]["type"] == "sas" + assert body["authentication"]["type"] == HubDeviceAuthType.sas.name elif req["auth"] == "x509_ca": - assert body["authentication"]["type"] == "certificateAuthority" + assert body["authentication"]["type"] == HubDeviceAuthType.certificateAuthority.name assert not body["authentication"].get("x509Thumbprint") assert not body["authentication"].get("symmetricKey") elif req["auth"] == "x509_thumbprint": - assert body["authentication"]["type"] == "selfSigned" + assert body["authentication"]["type"] == HubDeviceAuthType.selfSigned.name x509tp = body["authentication"]["x509Thumbprint"] assert x509tp["primaryThumbprint"] if req["stp"] is None: @@ -783,7 +783,7 @@ def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request): generate_device_module_show( authentication={ "symmetricKey": {"primaryKey": "", "secondaryKey": ""}, - "type": "sas", + "type": HubDeviceAuthType.sas.name, }, etag=generate_generic_id(), ) @@ -795,13 +795,13 @@ def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request): "primaryThumbprint": "123", "secondaryThumbprint": "321", }, - "type": "selfSigned", + "type": HubDeviceAuthType.selfSigned.name, } ) ), ( generate_device_module_show( - authentication={"type": "certificateAuthority"} + authentication={"type": HubDeviceAuthType.certificateAuthority.name} ) ), ], @@ -830,10 +830,10 @@ def test_device_module_update(self, serviceclient, req): assert body["moduleId"] == req["moduleId"] assert not body.get("capabilities") assert req["authentication"]["type"] == body["authentication"]["type"] - if req["authentication"]["type"] == "certificateAuthority": + if req["authentication"]["type"] == HubDeviceAuthType.certificateAuthority.name: assert not body["authentication"].get("x509Thumbprint") assert not body["authentication"].get("symmetricKey") - elif req["authentication"]["type"] == "selfSigned": + elif req["authentication"]["type"] == HubDeviceAuthType.selfSigned.name: assert body["authentication"]["x509Thumbprint"]["primaryThumbprint"] assert body["authentication"]["x509Thumbprint"]["secondaryThumbprint"] @@ -851,7 +851,7 @@ def test_device_module_update(self, serviceclient, req): "primaryThumbprint": "", "secondaryThumbprint": "", }, - "type": "selfSigned", + "type": HubDeviceAuthType.selfSigned.name, } ), CLIError, @@ -2001,7 +2001,7 @@ def test_generate_sas_token(self): class TestDeviceSimulate: @pytest.fixture(params=[204]) - def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request, fixture_iot_device_show): + def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request, fixture_iot_device_show_sas): service_client = mocker.patch(path_service_client) service_client.return_value = build_mock_response(mocker, request.param, {}) return service_client @@ -2123,6 +2123,12 @@ def test_device_simulate_mqtt_error(self, mqttclient_generic_error): fixture_cmd, device_id, hub_name=mock_target["entity"] ) + def test_device_simulate_mqtt_non_sas_device_error(self, fixture_ghcs, fixture_self_signed_device_show_self_signed): + with pytest.raises(CLIError): + subject.iot_simulate_device( + fixture_cmd, device_id, hub_name=mock_target["entity"] + ) + @pytest.mark.skipif( not validate_min_python_version(3, 5, exit_on_fail=False), From 0b799e6c3c62b1911bb23fe537e537ee29fdb147 Mon Sep 17 00:00:00 2001 From: Avin Agrawal Date: Tue, 1 Jun 2021 15:50:04 -0700 Subject: [PATCH 9/9] Addressed comments --- azext_iot/common/shared.py | 2 +- azext_iot/operations/hub.py | 42 ++++++++-------- azext_iot/tests/conftest.py | 6 +-- azext_iot/tests/iothub/test_iot_ext_unit.py | 56 ++++++++++----------- 4 files changed, 53 insertions(+), 53 deletions(-) diff --git a/azext_iot/common/shared.py b/azext_iot/common/shared.py index ad22e4c0e..ce89fc919 100644 --- a/azext_iot/common/shared.py +++ b/azext_iot/common/shared.py @@ -56,7 +56,7 @@ class DeviceAuthType(Enum): x509_ca = "x509_ca" -class HubDeviceAuthType(Enum): +class DeviceAuthApiType(Enum): """ Hub Device Authorization type. """ diff --git a/azext_iot/operations/hub.py b/azext_iot/operations/hub.py index cb111ee27..505c33cb5 100644 --- a/azext_iot/operations/hub.py +++ b/azext_iot/operations/hub.py @@ -28,7 +28,7 @@ SettleType, RenewKeyType, IoTHubStateType, - HubDeviceAuthType, + DeviceAuthApiType, ) from azext_iot.iothub.providers.discovery import IotHubDiscovery from azext_iot.common.utility import ( @@ -260,21 +260,21 @@ def _assemble_auth(auth_method, pk, sk): ) auth = None - if auth_method in [DeviceAuthType.shared_private_key.name, HubDeviceAuthType.sas.name]: + if auth_method in [DeviceAuthType.shared_private_key.name, DeviceAuthApiType.sas.value]: auth = AuthenticationMechanism( - symmetric_key=SymmetricKey(primary_key=pk, secondary_key=sk), type=HubDeviceAuthType.sas.name + symmetric_key=SymmetricKey(primary_key=pk, secondary_key=sk), type=DeviceAuthApiType.sas.value ) - elif auth_method in [DeviceAuthType.x509_thumbprint.name, HubDeviceAuthType.selfSigned.name]: + elif auth_method in [DeviceAuthType.x509_thumbprint.name, DeviceAuthApiType.selfSigned.value]: if not pk: raise ValueError("primary thumbprint required with selfSigned auth") auth = AuthenticationMechanism( x509_thumbprint=X509Thumbprint( primary_thumbprint=pk, secondary_thumbprint=sk ), - type=HubDeviceAuthType.selfSigned.name, + type=DeviceAuthApiType.selfSigned.value, ) - elif auth_method in [DeviceAuthType.x509_ca.name, HubDeviceAuthType.certificateAuthority.name]: - auth = AuthenticationMechanism(type=HubDeviceAuthType.certificateAuthority.name) + elif auth_method in [DeviceAuthType.x509_ca.name, DeviceAuthApiType.certificateAuthority.value]: + auth = AuthenticationMechanism(type=DeviceAuthApiType.certificateAuthority.value) else: raise ValueError("Authorization method {} invalid.".format(auth_method)) return auth @@ -307,7 +307,7 @@ def update_iot_device_custom( auth_type = instance["authentication"]["type"] if auth_method is not None: if auth_method == DeviceAuthType.shared_private_key.name: - auth = HubDeviceAuthType.sas.name + auth = DeviceAuthApiType.sas.value if (primary_key and not secondary_key) or ( not primary_key and secondary_key ): @@ -315,7 +315,7 @@ def update_iot_device_custom( instance["authentication"]["symmetricKey"]["primaryKey"] = primary_key instance["authentication"]["symmetricKey"]["secondaryKey"] = secondary_key elif auth_method == DeviceAuthType.x509_thumbprint.name: - auth = HubDeviceAuthType.selfSigned.name + auth = DeviceAuthApiType.selfSigned.value if not any([primary_thumbprint, secondary_thumbprint]): raise CLIError( "primary or secondary Thumbprint required with selfSigned auth" @@ -329,13 +329,13 @@ def update_iot_device_custom( "secondaryThumbprint" ] = secondary_thumbprint elif auth_method == DeviceAuthType.x509_ca.name: - auth = HubDeviceAuthType.certificateAuthority.name + auth = DeviceAuthApiType.certificateAuthority.value else: raise ValueError("Authorization method {} invalid.".format(auth_method)) instance["authentication"]["type"] = auth # if no new auth_method is provided, validate secondary auth arguments and update accordingly - elif auth_type == HubDeviceAuthType.sas.name: + elif auth_type == DeviceAuthApiType.sas.value: if any([primary_thumbprint, secondary_thumbprint]): raise ValueError( "Device authorization method {} does not support primary or secondary thumbprints.".format( @@ -347,7 +347,7 @@ def update_iot_device_custom( if secondary_key: instance["authentication"]["symmetricKey"]["secondaryKey"] = secondary_key - elif auth_type == HubDeviceAuthType.selfSigned.name: + elif auth_type == DeviceAuthApiType.selfSigned.value: if any([primary_key, secondary_key]): raise ValueError( "Device authorization method {} does not support primary or secondary keys.".format( @@ -477,7 +477,7 @@ def iot_device_key_regenerate( auth_type=auth_type_dataplane, ) device = _iot_device_show(target, device_id) - if device["authentication"]["type"] != HubDeviceAuthType.sas.name: + if device["authentication"]["type"] != DeviceAuthApiType.sas.value: raise CLIError("Device authentication should be of type sas") pk = device["authentication"]["symmetricKey"]["primaryKey"] @@ -868,15 +868,15 @@ def _handle_module_update_params(parameters): def _parse_auth(parameters): - valid_auth = [HubDeviceAuthType.sas.name, HubDeviceAuthType.selfSigned.name, HubDeviceAuthType.certificateAuthority.name] + valid_auth = [DeviceAuthApiType.sas.value, DeviceAuthApiType.selfSigned.value, DeviceAuthApiType.certificateAuthority.value] auth = parameters["authentication"].get("type") if auth not in valid_auth: raise CLIError("authentication.type must be one of {}".format(valid_auth)) pk = sk = None - if auth == HubDeviceAuthType.sas.name: + if auth == DeviceAuthApiType.sas.value: pk = parameters["authentication"]["symmetricKey"]["primaryKey"] sk = parameters["authentication"]["symmetricKey"]["secondaryKey"] - elif auth == HubDeviceAuthType.selfSigned.name: + elif auth == DeviceAuthApiType.selfSigned.value: pk = parameters["authentication"]["x509Thumbprint"]["primaryThumbprint"] sk = parameters["authentication"]["x509Thumbprint"]["secondaryThumbprint"] if not any([pk, sk]): @@ -1867,7 +1867,7 @@ def iot_get_sas_token( ) return { - HubDeviceAuthType.sas.name: _iot_build_sas_token( + DeviceAuthApiType.sas.value: _iot_build_sas_token( cmd, hub_name, device_id, @@ -1965,13 +1965,13 @@ def _build_device_or_module_connection_string(entity, key_type="primary"): ) auth = entity["authentication"] auth_type = auth["type"].lower() - if auth_type == HubDeviceAuthType.sas.name: + if auth_type == DeviceAuthApiType.sas.value.lower(): key = "SharedAccessKey={}".format( auth["symmetricKey"]["primaryKey"] if key_type == "primary" else auth["symmetricKey"]["secondaryKey"] ) - elif auth_type in [HubDeviceAuthType.certificateAuthority.name, HubDeviceAuthType.selfSigned.name]: + elif auth_type in [DeviceAuthApiType.certificateAuthority.value.lower(), DeviceAuthApiType.selfSigned.value.lower()]: key = "x509=true" else: raise CLIError("Unable to form target connection string") @@ -2072,7 +2072,7 @@ def _iot_device_send_message( import os device = _iot_device_show(target, device_id) - if device is not None and device.get("authentication", {}).get("type", "") != HubDeviceAuthType.sas.name: + if device and device.get("authentication", {}).get("type", "") != DeviceAuthApiType.sas.value: raise CLIError('D2C send message command only supports symmetric key auth (SAS) based devices') msgs = [] @@ -2463,7 +2463,7 @@ def http_wrap(target, device_id, generator): if protocol_type == ProtocolType.mqtt.name: device = _iot_device_show(target, device_id) - if device is not None and device.get("authentication", {}).get("type", "") != HubDeviceAuthType.sas.name: + if device and device.get("authentication", {}).get("type", "") != DeviceAuthApiType.sas.value: raise CLIError('MQTT simulation is only supported for symmetric key auth (SAS) based devices') wrap = mqtt_client_wrap( diff --git a/azext_iot/tests/conftest.py b/azext_iot/tests/conftest.py index 1a6ace508..8f7e44721 100644 --- a/azext_iot/tests/conftest.py +++ b/azext_iot/tests/conftest.py @@ -14,7 +14,7 @@ from azure.cli.core.commands import AzCliCommand from azure.cli.core.mock import DummyCli from azext_iot.tests.generators import generate_generic_id -from azext_iot.common.shared import HubDeviceAuthType +from azext_iot.common.shared import DeviceAuthApiType path_iot_hub_service_factory = "azext_iot._factory.iot_hub_service_factory" path_service_client = "msrest.service_client.ServiceClient.send" @@ -148,7 +148,7 @@ def fixture_iot_device_show_sas(mocker): "primaryKey": "test_pk", "secondaryKey": "test_sk" }, - "type": HubDeviceAuthType.sas.name, + "type": DeviceAuthApiType.sas.value, "x509Thumbprint": { "primaryThumbprint": None, "secondaryThumbprint": None @@ -181,7 +181,7 @@ def fixture_self_signed_device_show_self_signed(mocker): "primaryKey": "test_pk", "secondaryKey": "test_sk" }, - "type": HubDeviceAuthType.selfSigned.name, + "type": DeviceAuthApiType.selfSigned.value, "x509Thumbprint": { "primaryThumbprint": None, "secondaryThumbprint": None diff --git a/azext_iot/tests/iothub/test_iot_ext_unit.py b/azext_iot/tests/iothub/test_iot_ext_unit.py index e14582b63..86412732e 100644 --- a/azext_iot/tests/iothub/test_iot_ext_unit.py +++ b/azext_iot/tests/iothub/test_iot_ext_unit.py @@ -35,7 +35,7 @@ mock_target, generate_cs, ) -from azext_iot.common.shared import HubDeviceAuthType +from azext_iot.common.shared import DeviceAuthApiType device_id = "mydevice" child_device_id = "child_device1" @@ -131,13 +131,13 @@ def test_device_create(self, serviceclient, req): assert body["capabilities"]["iotEdge"] == req["ee"] if req["auth"] == "shared_private_key": - assert body["authentication"]["type"] == HubDeviceAuthType.sas.name + assert body["authentication"]["type"] == DeviceAuthApiType.sas.value elif req["auth"] == "x509_ca": - assert body["authentication"]["type"] == HubDeviceAuthType.certificateAuthority.name + assert body["authentication"]["type"] == DeviceAuthApiType.certificateAuthority.value assert not body["authentication"].get("x509Thumbprint") assert not body["authentication"].get("symmetricKey") elif req["auth"] == "x509_thumbprint": - assert body["authentication"]["type"] == HubDeviceAuthType.selfSigned.name + assert body["authentication"]["type"] == DeviceAuthApiType.selfSigned.value x509tp = body["authentication"]["x509Thumbprint"] assert x509tp["primaryThumbprint"] if req["stp"] is None: @@ -192,7 +192,7 @@ def generate_device_show(**kvp): "authentication": { "symmetricKey": {"primaryKey": None, "secondaryKey": None}, "x509Thumbprint": {"primaryThumbprint": None, "secondaryThumbprint": None}, - "type": HubDeviceAuthType.sas.name, + "type": DeviceAuthApiType.sas.value, }, "capabilities": {"iotEdge": True}, "deviceId": device_id, @@ -244,7 +244,7 @@ def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request): generate_device_show( authentication={ "symmetricKey": {"primaryKey": "", "secondaryKey": ""}, - "type": HubDeviceAuthType.sas.name, + "type": DeviceAuthApiType.sas.value, } ) ), @@ -255,13 +255,13 @@ def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request): "primaryThumbprint": "123", "secondaryThumbprint": "321", }, - "type": HubDeviceAuthType.selfSigned.name, + "type": DeviceAuthApiType.selfSigned.value, } ) ), ( generate_device_show( - authentication={"type": HubDeviceAuthType.certificateAuthority.name}, + authentication={"type": DeviceAuthApiType.certificateAuthority.value}, etag=generate_generic_id(), ) ), @@ -286,10 +286,10 @@ def test_device_update(self, fixture_cmd, serviceclient, req): assert body["status"] == req["status"] assert body["capabilities"]["iotEdge"] == req["capabilities"]["iotEdge"] assert req["authentication"]["type"] == body["authentication"]["type"] - if req["authentication"]["type"] == HubDeviceAuthType.certificateAuthority.name: + if req["authentication"]["type"] == DeviceAuthApiType.certificateAuthority.value: assert not body["authentication"].get("x509Thumbprint") assert not body["authentication"].get("symmetricKey") - elif req["authentication"]["type"] == HubDeviceAuthType.selfSigned.name: + elif req["authentication"]["type"] == DeviceAuthApiType.selfSigned.value: assert body["authentication"]["x509Thumbprint"]["primaryThumbprint"] assert body["authentication"]["x509Thumbprint"]["secondaryThumbprint"] @@ -320,7 +320,7 @@ def test_device_update(self, fixture_cmd, serviceclient, req): ( generate_device_show( authentication={ - "type": HubDeviceAuthType.selfSigned.name, + "type": DeviceAuthApiType.selfSigned.value, "symmetricKey": {"primaryKey": None, "secondaryKey": None}, "x509Thumbprint": { "primaryThumbprint": "123", @@ -337,7 +337,7 @@ def test_device_update(self, fixture_cmd, serviceclient, req): ( generate_device_show( authentication={ - "type": HubDeviceAuthType.certificateAuthority.name, + "type": DeviceAuthApiType.certificateAuthority.value, "symmetricKey": {"primaryKey": None, "secondaryKey": None}, "x509Thumbprint": { "primaryThumbprint": None, @@ -356,7 +356,7 @@ def test_device_update(self, fixture_cmd, serviceclient, req): ( generate_device_show( authentication={ - "type": HubDeviceAuthType.selfSigned.name, + "type": DeviceAuthApiType.selfSigned.value, "symmetricKey": {"primaryKey": None, "secondaryKey": None}, "x509Thumbprint": { "primaryThumbprint": "123", @@ -389,7 +389,7 @@ def test_iot_device_custom(self, fixture_cmd, serviceclient, req, arg): assert instance["statusReason"] == arg["status_reason"] if arg["auth_method"]: if arg["auth_method"] == "shared_private_key": - assert instance["authentication"]["type"] == HubDeviceAuthType.sas.name + assert instance["authentication"]["type"] == DeviceAuthApiType.sas.value instance["authentication"]["symmetricKey"]["primaryKey"] == arg[ "primary_key" ] @@ -397,7 +397,7 @@ def test_iot_device_custom(self, fixture_cmd, serviceclient, req, arg): "secondary_key" ] if arg["auth_method"] == "x509_thumbprint": - assert instance["authentication"]["type"] == HubDeviceAuthType.selfSigned.name + assert instance["authentication"]["type"] == DeviceAuthApiType.selfSigned.value if arg["primary_thumbprint"]: instance["authentication"]["x509Thumbprint"][ "primaryThumbprint" @@ -407,7 +407,7 @@ def test_iot_device_custom(self, fixture_cmd, serviceclient, req, arg): "secondaryThumbprint" ] = arg["secondary_thumbprint"] if arg["auth_method"] == "x509_ca": - assert instance["authentication"]["type"] == HubDeviceAuthType.certificateAuthority.name + assert instance["authentication"]["type"] == DeviceAuthApiType.certificateAuthority.value @pytest.mark.parametrize( "req, arg, exp", @@ -437,7 +437,7 @@ def test_iot_device_custom(self, fixture_cmd, serviceclient, req, arg): ( generate_device_show( authentication={ - "type": HubDeviceAuthType.selfSigned.name, + "type": DeviceAuthApiType.selfSigned.value, "symmetricKey": {"primaryKey": None, "secondaryKey": None}, "x509Thumbprint": { "primaryThumbprint": "123", @@ -474,7 +474,7 @@ def test_iot_device_custom_invalid_args(self, serviceclient, req, arg, exp): "primaryThumbprint": "", "secondaryThumbprint": "", }, - "type": HubDeviceAuthType.selfSigned.name, + "type": DeviceAuthApiType.selfSigned.value, } ), CLIError, @@ -511,7 +511,7 @@ def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request): "authentication", { "symmetricKey": {"primaryKey": "123", "secondaryKey": "321"}, - "type": HubDeviceAuthType.sas.name, + "type": DeviceAuthApiType.sas.value, }, ) test_side_effect = [ @@ -737,13 +737,13 @@ def test_device_module_create(self, serviceclient, req): assert body["moduleId"] == req["module_id"] if req["auth"] == "shared_private_key": - assert body["authentication"]["type"] == HubDeviceAuthType.sas.name + assert body["authentication"]["type"] == DeviceAuthApiType.sas.value elif req["auth"] == "x509_ca": - assert body["authentication"]["type"] == HubDeviceAuthType.certificateAuthority.name + assert body["authentication"]["type"] == DeviceAuthApiType.certificateAuthority.value assert not body["authentication"].get("x509Thumbprint") assert not body["authentication"].get("symmetricKey") elif req["auth"] == "x509_thumbprint": - assert body["authentication"]["type"] == HubDeviceAuthType.selfSigned.name + assert body["authentication"]["type"] == DeviceAuthApiType.selfSigned.value x509tp = body["authentication"]["x509Thumbprint"] assert x509tp["primaryThumbprint"] if req["stp"] is None: @@ -783,7 +783,7 @@ def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request): generate_device_module_show( authentication={ "symmetricKey": {"primaryKey": "", "secondaryKey": ""}, - "type": HubDeviceAuthType.sas.name, + "type": DeviceAuthApiType.sas.value, }, etag=generate_generic_id(), ) @@ -795,13 +795,13 @@ def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request): "primaryThumbprint": "123", "secondaryThumbprint": "321", }, - "type": HubDeviceAuthType.selfSigned.name, + "type": DeviceAuthApiType.selfSigned.value, } ) ), ( generate_device_module_show( - authentication={"type": HubDeviceAuthType.certificateAuthority.name} + authentication={"type": DeviceAuthApiType.certificateAuthority.value} ) ), ], @@ -830,10 +830,10 @@ def test_device_module_update(self, serviceclient, req): assert body["moduleId"] == req["moduleId"] assert not body.get("capabilities") assert req["authentication"]["type"] == body["authentication"]["type"] - if req["authentication"]["type"] == HubDeviceAuthType.certificateAuthority.name: + if req["authentication"]["type"] == DeviceAuthApiType.certificateAuthority.value: assert not body["authentication"].get("x509Thumbprint") assert not body["authentication"].get("symmetricKey") - elif req["authentication"]["type"] == HubDeviceAuthType.selfSigned.name: + elif req["authentication"]["type"] == DeviceAuthApiType.selfSigned.value: assert body["authentication"]["x509Thumbprint"]["primaryThumbprint"] assert body["authentication"]["x509Thumbprint"]["secondaryThumbprint"] @@ -851,7 +851,7 @@ def test_device_module_update(self, serviceclient, req): "primaryThumbprint": "", "secondaryThumbprint": "", }, - "type": HubDeviceAuthType.selfSigned.name, + "type": DeviceAuthApiType.selfSigned.value, } ), CLIError,