diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..9f8d579 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,33 @@ +name: Tests + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + test: + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, windows-latest, macos-latest] + python-version: ["3.9", "3.10"] + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -e . + pip install pytest pytest-mock pytest-testdox pytest-timeout + + - name: Run tests + run: python -m pytest tests/ -v diff --git a/README.md b/README.md index c1106b9..57a2e84 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# Azure IoTHub Service SDK +# Azure IoTHub Service SDK The Azure IoTHub Service SDK for Python provides functionality for communicating with the Azure IoT Hub. @@ -8,21 +8,90 @@ The SDK provides the following clients: * ### IoT Hub Registry Manager - * Provides CRUD operations for device on IoTHub - * Get statistics about the IoTHub service and devices + * CRUD operations for devices and modules on IoT Hub + * Get service and device registry statistics + * Query device twins using a SQL-like language + * Retrieve and update device twins and module twins + * Invoke direct methods on devices and modules + * **Cloud-to-Device (C2D) messaging** over AMQP (requires `uamqp` — see [Installation](#installation) below) + * Bulk create, update, or delete device identities + +* ### IoT Hub Configuration Manager + + * CRUD operations for IoT Hub configurations + +* ### IoT Hub Job Manager + + * Schedule and manage jobs for twin updates and direct method invocations + * Import and export device identities in bulk + +* ### IoT Hub HTTP Runtime Manager + + * Receive and complete/reject/abandon Cloud-to-Device messages from the device feedback queue + +* ### Digital Twin Client + + * Get and update digital twins + * Invoke commands on digital twin components ## Installation -```python +### Standard installation + +```bash pip install azure-iot-hub ``` +This installs all dependencies needed for every feature **except** Cloud-to-Device (C2D) AMQP messaging on ARM macOS / Apple Silicon (see below). + +### Cloud-to-Device (C2D) messaging and `uamqp` + +The `send_c2d_message` method on `IoTHubRegistryManager` uses the [uamqp](https://pypi.org/project/uamqp/) library, which is a C extension that must be compiled from source. + +| Platform | C2D messaging support | Notes | +|---|---|---| +| Windows | Included automatically | `uamqp` is installed as part of `pip install azure-iot-hub` | +| Linux | Included automatically | `uamqp` is installed as part of `pip install azure-iot-hub` | +| macOS (Intel) | Included automatically | `uamqp` is installed as part of `pip install azure-iot-hub` | +| macOS (Apple Silicon / ARM) | **Not installed automatically** | Recent versions of Xcode/clang enforce stricter C type checking that breaks the `uamqp` build. See below. | + +#### Apple Silicon macOS: opting in to C2D messaging + +If your Xcode and clang version are compatible, you can install `uamqp` explicitly using the `uamqp` extra: + +```bash +pip install azure-iot-hub[uamqp] +``` + +If `uamqp` cannot be built in your environment, all other SDK features (device/module CRUD, twin operations, direct methods, digital twins, jobs, etc.) work without it. Calling `send_c2d_message` without `uamqp` installed raises an `ImportError` with instructions. + +#### AMQP transport options + +When `uamqp` is available, C2D messaging supports two transport modes, controlled by the `transport_type` parameter: + +```python +from uamqp import TransportType + +# Default: AMQP over TCP (port 5671) +registry_manager = IoTHubRegistryManager.from_connection_string(connection_string) + +# AMQP over WebSocket (port 443) — useful when port 5671 is blocked by a firewall +registry_manager = IoTHubRegistryManager.from_connection_string( + connection_string, + transport_type=TransportType.AmqpOverWebsocket +) +``` + ## IoTHub Samples -Check out the [samples repository](https://github.com/Azure/azure-iot-hub-python/tree/main/samples) for more detailed samples +Check out the [samples repository](https://github.com/Azure/azure-iot-hub-python/tree/main/samples) for more detailed samples. -## Getting help and finding API docs +Notable C2D samples: + +- [`iothub_registry_manager_c2d_sample.py`](https://github.com/Azure/azure-iot-hub-python/tree/main/samples/iothub_registry_manager_c2d_sample.py) — basic C2D messaging +- [`iothub_registry_manager_c2d_amqp_over_websocket_sample.py`](https://github.com/Azure/azure-iot-hub-python/tree/main/samples/iothub_registry_manager_c2d_amqp_over_websocket_sample.py) — C2D over WebSocket +## Getting help and finding API docs API documentation for this package is available via [Microsoft Docs](https://docs.microsoft.com/python/api/azure-iot-hub/azure.iot.hub?view=azure-python) diff --git a/setup.py b/setup.py index cd0559c..35bf911 100644 --- a/setup.py +++ b/setup.py @@ -76,10 +76,20 @@ ], install_requires=[ "msrest>=0.6.21,<1.0.0", - # NOTE: Python 2.7, 3.5 support dropped >= 1.4.0 - "uamqp>=1.2.14,<2.0.0", "azure-core>=1.10.0,<2.0.0", + "six>=1.12.0", + # uamqp is a C extension that fails to build on ARM macOS (Apple Silicon) with recent + # clang versions due to stricter type checking. It is excluded from automatic installation + # on that platform only; all other platforms (Windows, Linux, Intel Mac) get it by default. + # ARM Mac users who need C2D messaging can opt in: pip install azure-iot-hub[uamqp] + # NOTE: Python 2.7, 3.5 support dropped in uamqp >= 1.4.0 + "uamqp>=1.2.14,<2.0.0; platform_machine != 'arm64' or sys_platform != 'darwin'", ], + extras_require={ + # Opt-in for ARM macOS users whose environment can build uamqp. + # Also usable on any platform to pin/re-add the dependency explicitly. + "uamqp": ["uamqp>=1.2.14,<2.0.0"], + }, python_requires=">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, <4", packages=find_packages( where="src", diff --git a/src/azure/iot/hub/constant.py b/src/azure/iot/hub/constant.py index 00ce90b..140058f 100644 --- a/src/azure/iot/hub/constant.py +++ b/src/azure/iot/hub/constant.py @@ -6,4 +6,4 @@ """This module defines constants for use across the azure-iot-hub package """ -VERSION = "2.7.0" +VERSION = "2.8.0" diff --git a/src/azure/iot/hub/iothub_amqp_client.py b/src/azure/iot/hub/iothub_amqp_client.py index ab995d9..dd0a1fc 100644 --- a/src/azure/iot/hub/iothub_amqp_client.py +++ b/src/azure/iot/hub/iothub_amqp_client.py @@ -11,8 +11,19 @@ from uuid import uuid4 import six.moves.urllib as urllib from azure.core.credentials import AccessToken -import uamqp +try: + import uamqp + + HAS_UAMQP = True +except ImportError: + HAS_UAMQP = False + +_UAMQP_MISSING_ERROR = ( + "uamqp is required for AMQP-based C2D messaging but is not installed. " + "On ARM macOS (Apple Silicon) it is not installed automatically due to build compatibility " + "issues with recent clang versions. Install it separately with: pip install azure-iot-hub[uamqp]" +) default_sas_expiry = 3600 @@ -35,6 +46,8 @@ def send_message_to_device(self, device_id, message, app_props): :raises: Exception if the Send command is not able to send the message """ + if not HAS_UAMQP: + raise ImportError(_UAMQP_MISSING_ERROR) msg_content = message msg_props = uamqp.message.MessageProperties() msg_props.message_id = str(uuid4()) @@ -68,7 +81,12 @@ def send_message_to_device(self, device_id, message, app_props): class IoTHubAmqpClientSharedAccessKeyAuth(IoTHubAmqpClientBase): - def __init__(self, hostname, shared_access_key_name, shared_access_key, transport_type=uamqp.TransportType.Amqp): + def __init__(self, hostname, shared_access_key_name, shared_access_key, transport_type=None): + if not HAS_UAMQP: + raise ImportError(_UAMQP_MISSING_ERROR) + if transport_type is None: + transport_type = uamqp.TransportType.Amqp + def get_token(): expiry = int(time.time() + default_sas_expiry) sas = base64.b64decode(shared_access_key) @@ -100,8 +118,13 @@ def get_token(): class IoTHubAmqpClientTokenAuth(IoTHubAmqpClientBase): def __init__( - self, hostname, token_credential, token_scope="https://iothubs.azure.net/.default", transport_type=uamqp.TransportType.Amqp + self, hostname, token_credential, token_scope="https://iothubs.azure.net/.default", transport_type=None ): + if not HAS_UAMQP: + raise ImportError(_UAMQP_MISSING_ERROR) + if transport_type is None: + transport_type = uamqp.TransportType.Amqp + def get_token(): result = token_credential.get_token(token_scope) return AccessToken("Bearer " + result.token, result.expires_on) diff --git a/src/azure/iot/hub/iothub_registry_manager.py b/src/azure/iot/hub/iothub_registry_manager.py index 6bb59b5..40e898b 100644 --- a/src/azure/iot/hub/iothub_registry_manager.py +++ b/src/azure/iot/hub/iothub_registry_manager.py @@ -14,7 +14,22 @@ AuthenticationMechanism, DeviceCapabilities, ) -from uamqp import TransportType + +try: + from uamqp import TransportType +except ImportError: + from enum import IntEnum + + class TransportType(IntEnum): + """Transport type for AMQP connections. + + Mirrors uamqp.TransportType. Only used when uamqp is not installed; + C2D messaging (send_c2d_message) requires uamqp and will raise an + ImportError if called without it. + """ + + Amqp = 1 + AmqpOverWebsocket = 3 def _ensure_quoted(etag): @@ -68,7 +83,7 @@ def __init__(self, connection_string=None, host=None, token_credential=None, tra Default value: None :param transport_type: The underlying transport protocol type: Amqp: AMQP over the default TCP transport protocol, it uses port 5671. AmqpOverWebsocket: Amqp over the Web Sockets transport protocol, it uses port 443. Default value: Amqp - :type transport_type: :class:`uamqp.TransportType` + :type transport_type: :class:`TransportType` :returns: Instance of the IoTHubRegistryManager object. :rtype: :class:`azure.iot.hub.IoTHubRegistryManager` @@ -79,19 +94,21 @@ def __init__(self, connection_string=None, host=None, token_credential=None, tra self.protocol = protocol_client( conn_string_auth, "https://" + conn_string_auth["HostName"] ) - self.amqp_svc_client = iothub_amqp_client.IoTHubAmqpClientSharedAccessKeyAuth( - conn_string_auth["HostName"], - conn_string_auth["SharedAccessKeyName"], - conn_string_auth["SharedAccessKey"], - transport_type, - ) + if iothub_amqp_client.HAS_UAMQP: + self.amqp_svc_client = iothub_amqp_client.IoTHubAmqpClientSharedAccessKeyAuth( + conn_string_auth["HostName"], + conn_string_auth["SharedAccessKeyName"], + conn_string_auth["SharedAccessKey"], + transport_type, + ) else: self.protocol = protocol_client( AzureIdentityCredentialAdapter(token_credential), "https://" + host ) - self.amqp_svc_client = iothub_amqp_client.IoTHubAmqpClientTokenAuth( - host, token_credential, transport_type=transport_type - ) + if iothub_amqp_client.HAS_UAMQP: + self.amqp_svc_client = iothub_amqp_client.IoTHubAmqpClientTokenAuth( + host, token_credential, transport_type=transport_type + ) @classmethod def from_connection_string(cls, connection_string, transport_type=TransportType.Amqp): @@ -105,7 +122,7 @@ def from_connection_string(cls, connection_string, transport_type=TransportType. with IoTHub. :param transport_type: The underlying transport protocol type: Amqp: AMQP over the default TCP transport protocol, it uses port 5671. AmqpOverWebsocket: Amqp over the Web Sockets transport protocol, it uses port 443. Default value: Amqp - :type transport_type: :class:`uamqp.TransportType` + :type transport_type: :class:`TransportType` :rtype: :class:`azure.iot.hub.IoTHubRegistryManager` """ @@ -124,7 +141,7 @@ def from_token_credential(cls, url, token_credential, transport_type=TransportTy :type token_credential: :class:`azure.core.TokenCredential` :param transport_type: The underlying transport protocol type: Amqp: AMQP over the default TCP transport protocol, it uses port 5671. AmqpOverWebsocket: Amqp over the Web Sockets transport protocol, it uses port 443. Default value: Amqp - :type transport_type: :class:`uamqp.TransportType` + :type transport_type: :class:`TransportType` :rtype: :class:`azure.iot.hub.IoTHubRegistryManager` """ @@ -926,7 +943,7 @@ def invoke_device_module_method(self, device_id, module_id, direct_method_reques return self.protocol.modules.invoke_method(device_id, module_id, direct_method_request) - def send_c2d_message(self, device_id, message, properties={}): + def send_c2d_message(self, device_id, message, properties=None): """Send a C2D message to a IoTHub Device. :param str device_id: The name (Id) of the device. @@ -936,4 +953,8 @@ def send_c2d_message(self, device_id, message, properties={}): :raises: Exception if the Send command is not able to send the message """ + if self.amqp_svc_client is None: + raise ImportError(iothub_amqp_client._UAMQP_MISSING_ERROR) + if properties is None: + properties = {} self.amqp_svc_client.send_message_to_device(device_id, message, properties) diff --git a/tests/test_iothub_amqp_client.py b/tests/test_iothub_amqp_client.py index 4df9938..434d901 100644 --- a/tests/test_iothub_amqp_client.py +++ b/tests/test_iothub_amqp_client.py @@ -7,7 +7,9 @@ import base64 import hashlib import pytest -import uamqp + +uamqp = pytest.importorskip("uamqp") + import hmac from azure.core.credentials import AccessToken from azure.iot.hub.iothub_amqp_client import ( diff --git a/tests/test_iothub_registry_manager.py b/tests/test_iothub_registry_manager.py index 3d91cb1..1d24c0c 100644 --- a/tests/test_iothub_registry_manager.py +++ b/tests/test_iothub_registry_manager.py @@ -6,10 +6,13 @@ import pytest from azure.iot.hub.protocol.models import AuthenticationMechanism, DeviceCapabilities -from azure.iot.hub.iothub_registry_manager import IoTHubRegistryManager +from azure.iot.hub.iothub_registry_manager import IoTHubRegistryManager, TransportType from azure.iot.hub import iothub_amqp_client from azure.iot.hub.protocol.iot_hub_gateway_service_ap_is import IotHubGatewayServiceAPIs -from uamqp import TransportType + +requires_uamqp = pytest.mark.skipif( + not iothub_amqp_client.HAS_UAMQP, reason="uamqp not installed" +) """---Constants---""" @@ -161,6 +164,7 @@ class TestFromConnectionString: ), ], ) + @requires_uamqp @pytest.mark.it( "Creates an instance of IotHubGatewayServiceAPIs and IoTHubAmqpClientSharedAccessKeyAuth with the correct arguments" ) @@ -205,6 +209,7 @@ def test_connection_string_auth(self, mocker, connection_string): ), ], ) + @requires_uamqp @pytest.mark.it( "Creates an instance of IotHubGatewayServiceAPIs and IoTHubAmqpClientSharedAccessKeyAuth with the correct arguments and using AMQP over Websocket" ) @@ -275,6 +280,7 @@ def test_instantiates_with_connection_string_no_shared_access_key(self): @pytest.mark.describe("IoTHubRegistryManager - .from_token_credential()") class TestFromTokenCredential: + @requires_uamqp @pytest.mark.it( "Creates an instance of IotHubGatewayServiceAPIs and IoTHubAmqpClientTokenAuth with the correct arguments" ) @@ -294,6 +300,7 @@ def test_token_credential_auth(self, mocker): assert amqp_client_init_mock.call_args == mocker.call( fake_hostname, mock_azure_identity_TokenCredential, transport_type=TransportType.Amqp ) + @requires_uamqp def test_token_credential_auth_with_amqp_over_websocket(self, mocker): mock_azure_identity_TokenCredential = mocker.MagicMock() amqp_client_init_mock = mocker.patch.object(iothub_amqp_client, "IoTHubAmqpClientTokenAuth") @@ -1430,6 +1437,7 @@ def test_invoke_device_module_method_payload_none( @pytest.mark.describe("IoTHubRegistryManager - .send_c2d_message()") +@requires_uamqp class TestSendC2dMessage(object): @pytest.mark.it("Test send c2d message") def test_send_c2d_message( @@ -1445,6 +1453,7 @@ def test_send_c2d_message( @pytest.mark.describe("IoTHubRegistryManager - .send_c2d_message() with properties") +@requires_uamqp class TestSendC2dMessageWithProperties(object): @pytest.mark.it("Test send c2d message with properties") def test_send_c2d_message_with_properties( diff --git a/tests/test_no_uamqp.py b/tests/test_no_uamqp.py new file mode 100644 index 0000000..fba422a --- /dev/null +++ b/tests/test_no_uamqp.py @@ -0,0 +1,239 @@ +# -------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- + +"""Tests that validate SDK behaviour when uamqp is NOT installed. + +The fix for https://github.com/ansible-collections/azure/issues/1511 makes +uamqp optional on ARM macOS. These tests simulate its absence by patching +HAS_UAMQP to False and making the AMQP client constructors raise ImportError, +then verify that: + + 1. IoTHubRegistryManager can still be constructed (REST operations work). + 2. REST-based operations (get_device, etc.) are unaffected. + 3. send_c2d_message raises a clear ImportError. + 4. AMQP client classes raise ImportError at construction time. +""" + +import pytest + +from azure.iot.hub.iothub_registry_manager import IoTHubRegistryManager +from azure.iot.hub import iothub_amqp_client + +# ---Constants--- +fake_hostname = "beauxbatons.academy-net" +fake_device_id = "MyPensieve" +fake_shared_access_key_name = "alohomora" +fake_shared_access_key = "Zm9vYmFy" +fake_connection_string = ( + "HostName={hostname};DeviceId={device_id};" + "SharedAccessKeyName={skn};SharedAccessKey={sk}" +).format( + hostname=fake_hostname, + device_id=fake_device_id, + skn=fake_shared_access_key_name, + sk=fake_shared_access_key, +) + + +# ────────────────────────────────────────────────────────────────────── +# Fixtures +# ────────────────────────────────────────────────────────────────────── + + +@pytest.fixture(autouse=True) +def mock_protocol_client(mocker): + """Prevent real HTTP calls; return a mock protocol client.""" + return mocker.patch( + "azure.iot.hub.iothub_registry_manager.protocol_client" + ) + + +@pytest.fixture() +def simulate_no_uamqp(mocker): + """Patch HAS_UAMQP to False so AMQP client constructors raise ImportError.""" + mocker.patch.object(iothub_amqp_client, "HAS_UAMQP", False) + + +# ────────────────────────────────────────────────────────────────────── +# IoTHubRegistryManager — construction without uamqp +# ────────────────────────────────────────────────────────────────────── + + +@pytest.mark.describe("IoTHubRegistryManager — construction without uamqp") +class TestRegistryManagerConstructionWithoutUamqp: + + @pytest.mark.it( + "Can be constructed from a connection string when uamqp is absent" + ) + def test_from_connection_string_succeeds(self, simulate_no_uamqp): + manager = IoTHubRegistryManager.from_connection_string( + fake_connection_string + ) + assert manager is not None + assert manager.amqp_svc_client is None + + @pytest.mark.it( + "Can be constructed from token credential when uamqp is absent" + ) + def test_from_token_credential_succeeds(self, simulate_no_uamqp, mocker): + mock_credential = mocker.MagicMock() + manager = IoTHubRegistryManager.from_token_credential( + fake_hostname, mock_credential + ) + assert manager is not None + assert manager.amqp_svc_client is None + + +# ────────────────────────────────────────────────────────────────────── +# IoTHubRegistryManager — REST operations work without uamqp +# ────────────────────────────────────────────────────────────────────── + + +@pytest.mark.describe("IoTHubRegistryManager — REST operations without uamqp") +class TestRegistryManagerRestOperationsWithoutUamqp: + + @pytest.fixture + def manager(self, simulate_no_uamqp): + return IoTHubRegistryManager.from_connection_string( + fake_connection_string + ) + + @pytest.mark.it("get_device works without uamqp") + def test_get_device(self, manager): + manager.get_device(fake_device_id) + manager.protocol.devices.get_identity.assert_called_once_with( + fake_device_id + ) + + @pytest.mark.it("get_module works without uamqp") + def test_get_module(self, manager): + manager.get_module(fake_device_id, "module1") + manager.protocol.modules.get_identity.assert_called_once_with( + fake_device_id, "module1" + ) + + @pytest.mark.it("get_service_statistics works without uamqp") + def test_get_service_statistics(self, manager): + manager.get_service_statistics() + manager.protocol.statistics.get_service_statistics.assert_called_once() + + @pytest.mark.it("delete_device works without uamqp") + def test_delete_device(self, manager): + manager.delete_device(fake_device_id) + manager.protocol.devices.delete_identity.assert_called_once_with( + fake_device_id, '"*"' + ) + + @pytest.mark.it("get_twin works without uamqp") + def test_get_twin(self, manager): + manager.get_twin(fake_device_id) + manager.protocol.devices.get_twin.assert_called_once_with( + fake_device_id + ) + + +# ────────────────────────────────────────────────────────────────────── +# IoTHubRegistryManager — send_c2d_message fails clearly without uamqp +# ────────────────────────────────────────────────────────────────────── + + +@pytest.mark.describe("IoTHubRegistryManager — send_c2d_message without uamqp") +class TestSendC2dMessageWithoutUamqp: + + @pytest.fixture + def manager(self, simulate_no_uamqp): + return IoTHubRegistryManager.from_connection_string( + fake_connection_string + ) + + @pytest.mark.it("Raises ImportError with install instructions") + def test_send_c2d_message_raises_import_error(self, manager): + with pytest.raises(ImportError, match="pip install azure-iot-hub"): + manager.send_c2d_message(fake_device_id, "hello") + + @pytest.mark.it("Error message mentions ARM macOS") + def test_error_message_mentions_platform(self, manager): + with pytest.raises(ImportError, match="ARM macOS"): + manager.send_c2d_message(fake_device_id, "hello") + + +# ────────────────────────────────────────────────────────────────────── +# AMQP client classes — raise ImportError without uamqp +# ────────────────────────────────────────────────────────────────────── + + +@pytest.mark.describe("IoTHubAmqpClient classes — instantiation without uamqp") +class TestAmqpClientWithoutUamqp: + + @pytest.mark.it( + "IoTHubAmqpClientSharedAccessKeyAuth raises ImportError" + ) + def test_shared_access_key_auth_raises(self, simulate_no_uamqp): + with pytest.raises(ImportError, match="pip install azure-iot-hub"): + iothub_amqp_client.IoTHubAmqpClientSharedAccessKeyAuth( + fake_hostname, + fake_shared_access_key_name, + fake_shared_access_key, + ) + + @pytest.mark.it("IoTHubAmqpClientTokenAuth raises ImportError") + def test_token_auth_raises(self, simulate_no_uamqp, mocker): + mock_credential = mocker.MagicMock() + with pytest.raises(ImportError, match="pip install azure-iot-hub"): + iothub_amqp_client.IoTHubAmqpClientTokenAuth( + fake_hostname, mock_credential + ) + + +# ────────────────────────────────────────────────────────────────────── +# TransportType fallback enum +# ────────────────────────────────────────────────────────────────────── + + +@pytest.mark.describe("TransportType fallback enum") +class TestTransportTypeFallback: + + @pytest.mark.it("Has Amqp and AmqpOverWebsocket members") + def test_fallback_enum_members(self): + from azure.iot.hub.iothub_registry_manager import TransportType + + assert hasattr(TransportType, "Amqp") + assert hasattr(TransportType, "AmqpOverWebsocket") + + @pytest.mark.it("Fallback IntEnum values match uamqp convention") + def test_fallback_enum_int_values(self, monkeypatch): + """Force uamqp ImportError and reload the module to exercise the fallback.""" + from enum import IntEnum + import builtins + import importlib + import azure.iot.hub.iothub_registry_manager as registry_module + + real_import = builtins.__import__ + + def _import_without_uamqp(name, *args, **kwargs): + if name == "uamqp": + raise ImportError("No module named 'uamqp'") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", _import_without_uamqp) + reloaded_module = importlib.reload(registry_module) + FallbackTransportType = reloaded_module.TransportType + + assert issubclass(FallbackTransportType, IntEnum) + assert FallbackTransportType.Amqp == 1 + assert FallbackTransportType.AmqpOverWebsocket == 3 + + # Restore original module state: undo monkeypatch first, then reload + monkeypatch.setattr(builtins, "__import__", real_import) + importlib.reload(registry_module) + + @pytest.mark.it("Default transport_type in from_connection_string is Amqp") + def test_default_transport_type(self, simulate_no_uamqp): + manager = IoTHubRegistryManager.from_connection_string( + fake_connection_string + ) + # Construction succeeded — the default TransportType.Amqp resolved fine + assert manager is not None