Adds subscription option for DT endpoint create. (#201)

* Tweak to CI unit test execution. Execute tests based on pattern.
* Upped timeout values for endpoints and removed kwarg from command
  layer.
* Re-added "--source" for relationship source twin, for those that want to
  use "--source" & "--target" conventions.

.azure-devops/templates/run-tests.yml

@@ -28,22 +28,9 @@ steps:
       displayName: 'Execute all Tests'
 
   - ${{ if eq(parameters.runUnitTestsOnly, 'true') }}:
-    - script: pytest -v azext_iot/tests/test_iot_ext_unit.py --junitxml=junit/test-iothub-unit-results.xml
-      displayName: 'Execute IoT Hub unit tests'
-    - script: pytest -v azext_iot/tests/test_iot_dps_unit.py --junitxml=junit/test-dps-unit-results.xml
-      displayName: 'Execute DPS unit tests'
-    - script: pytest -v azext_iot/tests/test_iot_utility_unit.py --junitxml=junit/test-utility-unit-results.xml
-      displayName: 'Execute Utility unit tests'
-    - script: pytest -v azext_iot/tests/test_iot_central_unit.py --junitxml=junit/test-central-unit-results.xml
-      displayName: 'Execute IoT Central unit tests'
-    - script: pytest -v azext_iot/tests/test_iot_pnp_unit.py --junitxml=junit/test-pnp-unit-results.xml
-      displayName: 'Execute IoT PnP unit tests'
-    - script: pytest -v azext_iot/tests/test_iot_digitaltwin_unit.py --junitxml=junit/test-dt-unit-results.xml
-      displayName: 'Execute IoT DigitalTwin unit tests'
-    - script: pytest -v azext_iot/tests/iothub/configurations/test_iot_config_unit.py --junitxml=junit/test-config-unit-results.xml
-      displayName: 'Execute IoT Configuration unit tests'
-    - script: pytest -v azext_iot/tests/iothub/jobs/test_iothub_jobs_unit.py --junitxml=junit/test-jobs-unit-results.xml
-      displayName: 'Execute IoT Hub job unit tests'
+    - template: set-testenv-sentinel.yml
+    - script: pytest -vv azext_iot/tests/ -k "_unit" --junitxml=junit/test-iotext-unit-results.xml
+      displayName: 'Execute IoT extension unit tests'
 
   - task: PublishTestResults@2
     condition: succeededOrFailed()

.azure-devops/templates/set-testenv-sentinel.yml

@@ -0,0 +1,28 @@
+steps:
+  - task: PythonScript@0
+    displayName : 'Set test envs with sentinel values'
+    name: 'setTestEnvSentinel'
+    inputs:
+      scriptSource: 'inline'
+      script: |
+        # This task is in place to get around DevOps pipelines env var naming rules which would require
+        # application code changes.
+        sentinel_value = "sentinel"
+        envvars = [
+            "azext_iot_testhub",
+            "azext_iot_testhub_cs",
+            "azext_iot_testrg",
+            "azext_iot_testdps",
+            "azext_pnp_endpoint",
+            "azext_pnp_repository",
+            "azext_pnp_cs",
+            "azext_iot_central_app_id",
+            "azext_iot_central_device_template_path",
+        ]
+        f = open("./pytest.ini", "w+")
+        f.write("[pytest]\n")
+        f.write("junit_family = xunit1\n")
+        f.write("env = \n")
+        envvars_sentinel = ["  {}={}\n".format(envvar, sentinel_value) for envvar in envvars]
+        f.writelines(envvars_sentinel)
+        f.close()

azext_iot/common/embedded_cli.py

@@ -21,10 +21,16 @@ def __init__(self):
         self.error_code = 0
         self.az_cli = get_default_cli()
 
-    def invoke(self, command):
+    def invoke(self, command: str, subscription: str = None):
         output_file = StringIO()
-        command = self._ensure_json_output(command)
-        self.error_code = self.az_cli.invoke(shlex.split(command), out_file=output_file) or 0
+        command = self._ensure_json_output(command=command)
+        if subscription:
+            command = self._ensure_subscription(
+                command=command, subscription=subscription
+            )
+        self.error_code = (
+            self.az_cli.invoke(shlex.split(command), out_file=output_file) or 0
+        )
         self.output = output_file.getvalue()
         logger.debug(
             "Embedded CLI received error code: %s, output: '%s'",
@@ -49,5 +55,8 @@ def success(self):
         logger.debug("Operation error code: %s", self.error_code)
         return self.error_code == 0
 
-    def _ensure_json_output(self, command):
+    def _ensure_json_output(self, command: str):
         return "{} -o json".format(command)
+
+    def _ensure_subscription(self, command: str, subscription: str):
+        return "{} --subscription '{}'".format(command, subscription)

azext_iot/constants.py

@@ -7,7 +7,7 @@
 
 import os
 
-VERSION = "0.9.4"
+VERSION = "0.9.5"
 EXTENSION_NAME = "azure-iot"
 EXTENSION_ROOT = os.path.dirname(os.path.abspath(__file__))
 EXTENSION_CONFIG_ROOT_KEY = "iotext"

azext_iot/digitaltwins/commands_resource.py

@@ -62,6 +62,7 @@ def add_endpoint_eventgrid(
     eventgrid_topic_name,
     eventgrid_resource_group,
     resource_group_name=None,
+    endpoint_subscription=None,
     tags=None,
 ):
     return _add_endpoint_eventgrid(
@@ -71,6 +72,7 @@ def add_endpoint_eventgrid(
         eventgrid_resource_group=eventgrid_resource_group,
         eventgrid_topic_name=eventgrid_topic_name,
         resource_group_name=resource_group_name,
+        endpoint_subscription=endpoint_subscription,
         tags=tags,
     )
 
@@ -81,8 +83,8 @@ def _add_endpoint_eventgrid(
     endpoint_name,
     eventgrid_topic_name,
     eventgrid_resource_group,
-    timeout=15,
     resource_group_name=None,
+    endpoint_subscription=None,
     tags=None,
 ):
     rp = ResourceProvider(cmd)
@@ -93,8 +95,8 @@ def _add_endpoint_eventgrid(
         endpoint_resource_type=ADTEndpointType.eventgridtopic,
         endpoint_resource_name=eventgrid_topic_name,
         endpoint_resource_group=eventgrid_resource_group,
+        endpoint_subscription=endpoint_subscription,
         tags=tags,
-        timeout=timeout,
     )
 
 
@@ -107,6 +109,7 @@ def add_endpoint_servicebus(
     servicebus_policy,
     servicebus_namespace,
     resource_group_name=None,
+    endpoint_subscription=None,
     tags=None,
 ):
     return _add_endpoint_servicebus(
@@ -118,6 +121,7 @@ def add_endpoint_servicebus(
         servicebus_policy=servicebus_policy,
         servicebus_namespace=servicebus_namespace,
         resource_group_name=resource_group_name,
+        endpoint_subscription=endpoint_subscription,
         tags=tags,
     )
 
@@ -130,8 +134,8 @@ def _add_endpoint_servicebus(
     servicebus_resource_group,
     servicebus_policy,
     servicebus_namespace,
-    timeout=15,
     resource_group_name=None,
+    endpoint_subscription=None,
     tags=None,
 ):
     rp = ResourceProvider(cmd)
@@ -144,8 +148,8 @@ def _add_endpoint_servicebus(
         endpoint_resource_group=servicebus_resource_group,
         endpoint_resource_namespace=servicebus_namespace,
         endpoint_resource_policy=servicebus_policy,
+        endpoint_subscription=endpoint_subscription,
         tags=tags,
-        timeout=timeout,
     )
 
 
@@ -158,6 +162,7 @@ def add_endpoint_eventhub(
     eventhub_policy,
     eventhub_namespace,
     resource_group_name=None,
+    endpoint_subscription=None,
     tags=None,
 ):
     return _add_endpoint_eventhub(
@@ -169,6 +174,7 @@ def add_endpoint_eventhub(
         eventhub_policy=eventhub_policy,
         eventhub_namespace=eventhub_namespace,
         resource_group_name=resource_group_name,
+        endpoint_subscription=endpoint_subscription,
         tags=tags,
     )
 
@@ -181,8 +187,8 @@ def _add_endpoint_eventhub(
     eventhub_resource_group,
     eventhub_policy,
     eventhub_namespace,
-    timeout=15,
     resource_group_name=None,
+    endpoint_subscription=None,
     tags=None,
 ):
     rp = ResourceProvider(cmd)
@@ -195,6 +201,6 @@ def _add_endpoint_eventhub(
         endpoint_resource_group=eventhub_resource_group,
         endpoint_resource_namespace=eventhub_namespace,
         endpoint_resource_policy=eventhub_policy,
+        endpoint_subscription=endpoint_subscription,
         tags=tags,
-        timeout=timeout,
     )

azext_iot/digitaltwins/params.py

@@ -113,6 +113,13 @@ def load_digitaltwins_arguments(self, _):
             help="Name of EventGrid Topic resource group.",
             arg_group="Event Grid Topic",
         )
+        context.argument(
+            "endpoint_subscription",
+            options_list=["--eventgrid-subscription", "--egs"],
+            help="Name or ID of subscription where the endpoint resource exists. "
+            "If no subscription is provided the default subscription is used.",
+            arg_group="Event Grid Topic",
+        )
 
     with self.argument_context("dt endpoint create eventhub") as context:
         context.argument(
@@ -139,6 +146,13 @@ def load_digitaltwins_arguments(self, _):
             help="Name of EventHub resource group.",
             arg_group="Event Hub",
         )
+        context.argument(
+            "endpoint_subscription",
+            options_list=["--eventhub-subscription", "--ehs"],
+            help="Name or ID of subscription where the endpoint resource exists. "
+            "If no subscription is provided the default subscription is used.",
+            arg_group="Event Hub",
+        )
 
     with self.argument_context("dt endpoint create servicebus") as context:
         context.argument(
@@ -165,6 +179,13 @@ def load_digitaltwins_arguments(self, _):
             help="Name of ServiceBus resource group.",
             arg_group="Service Bus Topic",
         )
+        context.argument(
+            "endpoint_subscription",
+            options_list=["--servicebus-subscription", "--sbs"],
+            help="Name or ID of subscription where the endpoint resource exists. "
+            "If no subscription is provided the default subscription is used.",
+            arg_group="Service Bus Topic",
+        )
 
     with self.argument_context("dt twin") as context:
         context.argument(
@@ -230,7 +251,7 @@ def load_digitaltwins_arguments(self, _):
     with self.argument_context("dt twin relationship") as context:
         context.argument(
             "twin_id",
-            options_list=["--twin-id", "-t"],
+            options_list=["--twin-id", "-t", "--source"],
             help="The source twin Id for a relationship.",
         )
         context.argument(

azext_iot/digitaltwins/providers/resource.py

@@ -20,7 +20,7 @@ def __init__(self, cmd):
         self.mgmt_sdk = self.get_mgmt_sdk()
         self.rbac = RbacProvider()
 
-    def create(self, name, resource_group_name, location, tags=None, timeout=15):
+    def create(self, name, resource_group_name, location, tags=None, timeout=20):
         if tags:
             tags = validate_key_value_pairs(tags)
 
@@ -207,6 +207,7 @@ def delete_endpoint(self, name, endpoint_name, resource_group_name=None):
         except ErrorResponseException as e:
             raise CLIError(unpack_msrest_error(e))
 
+    # TODO: Breakout and refactor
     def add_endpoint(
         self,
         name,
@@ -216,6 +217,7 @@ def add_endpoint(
         endpoint_resource_group,
         endpoint_resource_policy=None,
         endpoint_resource_namespace=None,
+        endpoint_subscription=None,
         tags=None,
         resource_group_name=None,
         timeout=20,
@@ -255,7 +257,8 @@ def add_endpoint(
             eg_topic_keys_op = cli.invoke(
                 "eventgrid topic key list -n {} -g {}".format(
                     endpoint_resource_name, endpoint_resource_group
-                )
+                ),
+                subscription=endpoint_subscription
             )
             if not eg_topic_keys_op.success():
                 raise CLIError("{} Event Grid topic keys.".format(error_prefix))
@@ -264,7 +267,8 @@ def add_endpoint(
             eg_topic_endpoint_op = cli.invoke(
                 "eventgrid topic show -n {} -g {}".format(
                     endpoint_resource_name, endpoint_resource_group
-                )
+                ),
+                subscription=endpoint_subscription
             )
             if not eg_topic_endpoint_op.success():
                 raise CLIError("{} Event Grid topic endpoint.".format(error_prefix))
@@ -283,7 +287,8 @@ def add_endpoint(
                     endpoint_resource_namespace,
                     endpoint_resource_group,
                     endpoint_resource_name,
-                )
+                ),
+                subscription=endpoint_subscription
             )
             if not sb_topic_keys_op.success():
                 raise CLIError("{} Service Bus topic keys.".format(error_prefix))
@@ -305,7 +310,8 @@ def add_endpoint(
                     endpoint_resource_namespace,
                     endpoint_resource_group,
                     endpoint_resource_name,
-                )
+                ),
+                subscription=endpoint_subscription
             )
             if not eventhub_topic_keys_op.success():
                 raise CLIError("{} Event Hub keys.".format(error_prefix))

azext_iot/tests/test_iot_utility_unit.py

@@ -20,6 +20,7 @@
 from azext_iot.common.deps import ensure_uamqp
 from azext_iot.constants import EVENT_LIB, EXTENSION_NAME
 from azext_iot._validators import mode2_iot_login_handler
+from azext_iot.common.embedded_cli import EmbeddedCLI
 
 
 class TestMinPython(object):
@@ -294,3 +295,49 @@ class TestVersionComparison:
     )
     def test_ensure_min_version(self, current, minimum, expected):
         assert ensure_min_version(current, minimum) == expected
+
+
+class TestEmbeddedCli:
+    @pytest.fixture(params=[0, 1])
+    def mocked_azclient(self, mocker, request):
+        def mock_invoke(args, out_file):
+            out_file.write(json.dumps({"generickey": "genericvalue"}))
+            return request.param
+
+        azclient = mocker.patch("azext_iot.common.embedded_cli.get_default_cli")
+        azclient.return_value.invoke.side_effect = mock_invoke
+        azclient.test_meta.error_code = request.param
+        return azclient
+
+    @pytest.mark.parametrize(
+        "command, subscription",
+        [
+            ("iot hub device-identity create -n abcd -d dcba", None),
+            ("iot hub device-twin show -n 'abcd' -d 'dcba'", "20a300e5-a444-4130-bb5a-1abd08ad930a"),
+        ],
+    )
+    def test_embedded_cli(self, mocked_azclient, command, subscription):
+        import shlex
+
+        cli = EmbeddedCLI()
+        cli.invoke(command=command, subscription=subscription)
+
+        # Due to forced json output
+        command += " -o json"
+
+        if subscription:
+            command += " --subscription '{}'".format(subscription)
+
+        expected_args = shlex.split(command)
+        call = mocked_azclient().invoke.call_args_list[0]
+        actual_args, _ = call
+        assert expected_args == actual_args[0]
+        success = cli.success()
+
+        if mocked_azclient.test_meta.error_code == 1:
+            assert not success
+        elif mocked_azclient.test_meta.error_code == 0:
+            assert success
+
+        assert cli.output
+        assert cli.as_json()