test: add TLS 1.3 integration tests

Author: yuelinMS

azext_iot/tests/dps/core/test_dps_linked_hub_int.py

@@ -0,0 +1,174 @@
+# coding=utf-8
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+"""
+Integration tests for TLS 1.3 DPS linked hub features.
+
+Tests cover:
+- Linked hub create with hostname types (auto, classic, device)
+- Hostname resolution for GWv2 hubs
+- Linked hub list after creation
+"""
+
+import pytest
+from azext_iot.common.embedded_cli import EmbeddedCLI
+
+cli = EmbeddedCLI()
+
+
+def _find_gwv2_hub(rg):
+    """Find a GWv2 hub in the RG"""
+    hubs = cli.invoke(f"iot hub list -g {rg}").as_json()
+    for hub in hubs:
+        if hub.get("properties", {}).get("deviceHostName"):
+            return hub
+    return None
+
+
+def _cleanup_linked_hub(dps_name, rg, linked_hub_name):
+    cli.invoke(
+        f"iot dps linked-hub delete --dps-name {dps_name} -g {rg} "
+        f"--linked-hub {linked_hub_name}"
+    )
+
+
+def test_linked_hub_create_auto_hostname(provisioned_iot_dps_no_hub_module):
+    """On a GWv2 hub, auto (default) should resolve to the device hostname."""
+    dps_name = provisioned_iot_dps_no_hub_module["name"]
+    dps_rg = provisioned_iot_dps_no_hub_module["resourceGroup"]
+
+    gwv2_hub = _find_gwv2_hub(dps_rg)
+    if not gwv2_hub:
+        pytest.skip("No GWv2 hub available in resource group for TLS 1.3 testing")
+
+    hub_name = gwv2_hub["name"]
+    device_hostname = gwv2_hub["properties"]["deviceHostName"]
+
+    try:
+        result = cli.invoke(
+            f"iot dps linked-hub create --dps-name {dps_name} -g {dps_rg} "
+            f"--hub-name {hub_name}"
+        ).as_json()
+
+        assert result, "Linked hub create should return a result"
+        linked_hubs = result if isinstance(result, list) else [result]
+        matching = [h for h in linked_hubs if h["name"] == device_hostname]
+        assert len(matching) == 1, \
+            f"Expected linked hub with name '{device_hostname}'. Got: {[h['name'] for h in linked_hubs]}"
+        assert device_hostname in matching[0]["connectionString"], \
+            "Connection string should use device hostname"
+    finally:
+        _cleanup_linked_hub(dps_name, dps_rg, device_hostname)
+
+
+def test_linked_hub_create_classic_hostname(provisioned_iot_dps_no_hub_module):
+    """Create linked hub with explicit classic hostname type."""
+    dps_name = provisioned_iot_dps_no_hub_module["name"]
+    dps_rg = provisioned_iot_dps_no_hub_module["resourceGroup"]
+
+    gwv2_hub = _find_gwv2_hub(dps_rg)
+    if not gwv2_hub:
+        pytest.skip("No GWv2 hub available in resource group")
+
+    hub_name = gwv2_hub["name"]
+    classic_hostname = gwv2_hub["properties"]["hostName"]
+
+    try:
+        result = cli.invoke(
+            f"iot dps linked-hub create --dps-name {dps_name} -g {dps_rg} "
+            f"--hub-name {hub_name} --hostname-type classic"
+        ).as_json()
+
+        assert result, "Linked hub create should return a result"
+        linked_hubs = result if isinstance(result, list) else [result]
+        matching = [h for h in linked_hubs if h["name"] == classic_hostname]
+        assert len(matching) == 1, \
+            f"Expected linked hub with name '{classic_hostname}'. Got: {[h['name'] for h in linked_hubs]}"
+        assert ".device." not in matching[0]["name"], \
+            "Classic hostname should not contain .device. segment"
+    finally:
+        _cleanup_linked_hub(dps_name, dps_rg, classic_hostname)
+
+
+def test_linked_hub_create_device_hostname(provisioned_iot_dps_no_hub_module):
+    """Create linked hub with explicit device hostname type on a GWv2 hub."""
+    dps_name = provisioned_iot_dps_no_hub_module["name"]
+    dps_rg = provisioned_iot_dps_no_hub_module["resourceGroup"]
+
+    gwv2_hub = _find_gwv2_hub(dps_rg)
+    if not gwv2_hub:
+        pytest.skip("No GWv2 hub available in resource group")
+
+    hub_name = gwv2_hub["name"]
+    device_hostname = gwv2_hub["properties"]["deviceHostName"]
+
+    try:
+        result = cli.invoke(
+            f"iot dps linked-hub create --dps-name {dps_name} -g {dps_rg} "
+            f"--hub-name {hub_name} --hostname-type device"
+        ).as_json()
+
+        assert result, "Linked hub create should return a result"
+        linked_hubs = result if isinstance(result, list) else [result]
+        matching = [h for h in linked_hubs if h["name"] == device_hostname]
+        assert len(matching) == 1, \
+            f"Expected linked hub with name '{device_hostname}'. Got: {[h['name'] for h in linked_hubs]}"
+        assert ".device." in matching[0]["name"], \
+            "Device hostname should contain .device. segment"
+    finally:
+        _cleanup_linked_hub(dps_name, dps_rg, device_hostname)
+
+
+def test_hub_show_returns_tls13_hostnames(provisioned_iot_dps_no_hub_module):
+    """Verify hub show returns TLS 1.3 hostname properties for GWv2 hubs."""
+    dps_rg = provisioned_iot_dps_no_hub_module["resourceGroup"]
+
+    gwv2_hub = _find_gwv2_hub(dps_rg)
+    if not gwv2_hub:
+        pytest.skip("No GWv2 hub available in resource group")
+
+    hub_name = gwv2_hub["name"]
+    result = cli.invoke(f"iot hub show -n {hub_name}").as_json()
+    props = result["properties"]
+
+    assert props.get("hostName"), "hostName (classic) should be present"
+    assert props.get("deviceHostName"), "deviceHostName should be present for GWv2 hub"
+    assert props.get("serviceHostName"), "serviceHostName should be present for GWv2 hub"
+
+    assert ".device." in props["deviceHostName"]
+    assert ".service." in props["serviceHostName"]
+    assert hub_name in props["hostName"]
+    assert hub_name in props["deviceHostName"]
+    assert hub_name in props["serviceHostName"]
+
+
+def test_linked_hub_list_shows_hostname(provisioned_iot_dps_no_hub_module):
+    """Verify linked hub list returns the correct hostname after linking."""
+    dps_name = provisioned_iot_dps_no_hub_module["name"]
+    dps_rg = provisioned_iot_dps_no_hub_module["resourceGroup"]
+
+    gwv2_hub = _find_gwv2_hub(dps_rg)
+    if not gwv2_hub:
+        pytest.skip("No GWv2 hub available in resource group")
+
+    hub_name = gwv2_hub["name"]
+    device_hostname = gwv2_hub["properties"]["deviceHostName"]
+
+    try:
+        cli.invoke(
+            f"iot dps linked-hub create --dps-name {dps_name} -g {dps_rg} "
+            f"--hub-name {hub_name}"
+        )
+
+        linked_hubs = cli.invoke(
+            f"iot dps linked-hub list --dps-name {dps_name} -g {dps_rg}"
+        ).as_json()
+
+        matching = [h for h in linked_hubs if h["name"] == device_hostname]
+        assert len(matching) == 1, \
+            f"Should find linked hub with name '{device_hostname}'. Found: {[h['name'] for h in linked_hubs]}"
+    finally:
+        _cleanup_linked_hub(dps_name, dps_rg, device_hostname)

azext_iot/tests/iothub/tls13/test_tls13_int.py

@@ -0,0 +1,127 @@
+# coding=utf-8
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+"""
+Integration tests for IoT Hub discovery and hostname resolution.
+
+Tests cover:
+- Hub discovery returns TLS 1.3 hostname properties
+- Service hostname used for data plane target on GWv2 hubs
+- Connection string uses correct hostname based on GWv2 status
+"""
+
+import pytest
+from azure.cli.testsdk.reverse_dependency import get_dummy_cli
+from azext_iot.common.embedded_cli import EmbeddedCLI
+from azext_iot.iothub.providers.discovery import IotHubDiscovery
+from azext_iot.tests.settings import (
+    DynamoSettings,
+    Setting,
+    ENV_SET_TEST_IOTHUB_REQUIRED,
+    ENV_SET_TEST_IOTHUB_OPTIONAL,
+)
+
+cli = EmbeddedCLI()
+settings = DynamoSettings(
+    req_env_set=ENV_SET_TEST_IOTHUB_REQUIRED,
+    opt_env_set=ENV_SET_TEST_IOTHUB_OPTIONAL,
+)
+ENTITY_RG = settings.env.azext_iot_testrg
+ENTITY_NAME = settings.env.azext_iot_testhub
+
+
+@pytest.fixture(scope="module")
+def discovery():
+    cmd_shell = Setting()
+    setattr(cmd_shell, "cli_ctx", get_dummy_cli())
+    return IotHubDiscovery(cmd_shell)
+
+
+def test_find_resource_returns_hostname_properties(discovery):
+    """Verify find_resource returns TLS 1.3 hostname properties."""
+    if not ENTITY_NAME:
+        pytest.skip("azext_iot_testhub not set")
+
+    resource = discovery.find_resource(resource_name=ENTITY_NAME)
+    props = resource.get("properties", {})
+
+    assert props.get("hostName"), "hostName (classic) should be present"
+
+    device_hostname = props.get("deviceHostName")
+    service_hostname = props.get("serviceHostName")
+    if device_hostname:
+        assert ENTITY_NAME in device_hostname
+        assert ".device." in device_hostname
+    if service_hostname:
+        assert ENTITY_NAME in service_hostname
+        assert ".service." in service_hostname
+
+
+def test_build_target_includes_hostname_fields(discovery):
+    """Verify _build_target populates deviceHostName and serviceHostName."""
+    if not ENTITY_NAME:
+        pytest.skip("azext_iot_testhub not set")
+
+    resource = discovery.find_resource(resource_name=ENTITY_NAME)
+    policy = discovery.find_policy(resource_name=ENTITY_NAME, rg=ENTITY_RG)
+    target = discovery._build_target(resource=resource, policy=policy)
+
+    assert target.get("entity"), "entity (hostname) should be set"
+    assert target.get("cs"), "connection string should be set"
+    assert "deviceHostName" in target, "target should include deviceHostName key"
+    assert "serviceHostName" in target, "target should include serviceHostName key"
+
+
+def test_gwv2_target_uses_service_hostname(discovery):
+    """For GWv2 hubs, _build_target should use service hostname for entity."""
+    if not ENTITY_NAME:
+        pytest.skip("azext_iot_testhub not set")
+
+    resource = discovery.find_resource(resource_name=ENTITY_NAME)
+    props = resource.get("properties", {})
+    gw_version = props.get("iotHubDetails", {}).get("gatewayVersion")
+    service_hostname = props.get("serviceHostName")
+
+    if gw_version != "V2" or not service_hostname:
+        pytest.skip("Hub is not GWv2 — skipping service hostname test")
+
+    policy = discovery.find_policy(resource_name=ENTITY_NAME, rg=ENTITY_RG)
+    target = discovery._build_target(resource=resource, policy=policy)
+
+    assert target["entity"] == service_hostname, \
+        f"GWv2 hub entity should be service hostname. Got: {target['entity']}"
+    assert service_hostname in target["cs"], \
+        "Connection string should use service hostname for GWv2 hub"
+
+
+def test_connection_string_uses_gwv2_hostname():
+    """Verify connection-string show uses the device hostname for GWv2 hubs."""
+    if not ENTITY_NAME:
+        pytest.skip("azext_iot_testhub not set")
+
+    hub = cli.invoke(f"iot hub show -n {ENTITY_NAME}").as_json()
+    props = hub.get("properties", {})
+    device_hostname = props.get("deviceHostName")
+
+    result = cli.invoke(f"iot hub connection-string show -n {ENTITY_NAME}").as_json()
+    cs = result["connectionString"]
+    assert "HostName=" in cs
+
+    cs_hostname = None
+    for part in cs.split(";"):
+        if part.startswith("HostName="):
+            cs_hostname = part.split("=", 1)[1]
+            break
+
+    assert cs_hostname, "Should extract hostname from connection string"
+
+    if device_hostname:
+        assert cs_hostname == device_hostname, \
+            f"GWv2 connection string should use device hostname '{device_hostname}'. Got: '{cs_hostname}'"
+    else:
+        classic_hostname = props.get("hostName")
+        assert cs_hostname == classic_hostname, \
+            f"V1 connection string should use classic hostname '{classic_hostname}'. Got: '{cs_hostname}'"