IoT

Testing AWS IoT and Azure IoT Hub Integrations

HelpMeTest

HelpMeTest

— 5 min read
Testing AWS IoT and Azure IoT Hub Integrations

AWS IoT Core and Azure IoT Hub are the two dominant managed IoT platforms. Both abstract away the infrastructure complexity, but they introduce new testing challenges: device provisioning flows, message routing rules, device shadows/twins, and serverless processing functions.

This guide shows you how to test these integrations without sending real hardware traffic to production cloud services.

The Core Testing Challenge

IoT cloud platforms sit in the middle of your architecture — between your physical devices and your application layer. Testing the full stack requires:

  1. Simulated devices — software that emulates real hardware sending telemetry
  2. Message routing — verifying that the right messages trigger the right actions
  3. Device state — testing device shadow/twin updates and delta notifications
  4. Provisioning — verifying the certificate and credential workflow

Testing AWS IoT Core

Local Testing with LocalStack

LocalStack supports a subset of AWS IoT Core. Use it for integration tests that would otherwise require a real AWS account:

import boto3
import pytest
import json
import time

@pytest.fixture
def iot_client():
    return boto3.client(
        'iot',
        endpoint_url='http://localhost:4566',
        region_name='us-east-1',
        aws_access_key_id='test',
        aws_secret_access_key='test'
    )

@pytest.fixture
def iot_data_client():
    return boto3.client(
        'iot-data',
        endpoint_url='http://localhost:4566',
        region_name='us-east-1',
        aws_access_key_id='test',
        aws_secret_access_key='test'
    )

def test_thing_creation(iot_client):
    response = iot_client.create_thing(thingName='test-sensor-001')
    
    assert response['ResponseMetadata']['HTTPStatusCode'] == 200
    assert response['thingName'] == 'test-sensor-001'
    assert 'thingArn' in response
    assert 'thingId' in response

def test_device_shadow_update(iot_data_client):
    thing_name = 'test-sensor-001'
    
    # Update shadow state (simulates device reporting state)
    payload = json.dumps({
        "state": {
            "reported": {
                "temperature": 22.5,
                "humidity": 45.0,
                "firmware_version": "2.1.0"
            }
        }
    })
    
    response = iot_data_client.update_thing_shadow(
        thingName=thing_name,
        payload=payload.encode()
    )
    
    assert response['ResponseMetadata']['HTTPStatusCode'] == 200
    
    # Read shadow back
    shadow = iot_data_client.get_thing_shadow(thingName=thing_name)
    shadow_state = json.loads(shadow['payload'].read())
    
    assert shadow_state['state']['reported']['temperature'] == 22.5
    assert shadow_state['state']['reported']['firmware_version'] == '2.1.0'

Testing Device Shadow Delta

Device shadows are most useful for desired vs. reported state. Test the delta notification flow:

def test_shadow_delta_generation(iot_data_client):
    thing_name = 'test-thermostat-001'
    
    # Cloud sets desired state (e.g., from a user app)
    iot_data_client.update_thing_shadow(
        thingName=thing_name,
        payload=json.dumps({
            "state": {
                "desired": {"target_temperature": 20.0}
            }
        }).encode()
    )
    
    # Device reports current state (different from desired)
    iot_data_client.update_thing_shadow(
        thingName=thing_name,
        payload=json.dumps({
            "state": {
                "reported": {"target_temperature": 18.5}
            }
        }).encode()
    )
    
    # Delta should exist — device needs to act
    shadow = iot_data_client.get_thing_shadow(thingName=thing_name)
    state = json.loads(shadow['payload'].read())
    
    assert 'delta' in state['state']
    assert state['state']['delta']['target_temperature'] == 20.0

Testing IoT Rules and Message Routing

AWS IoT Rules Engine routes messages from devices to services like Lambda, DynamoDB, and Kinesis. Test that rules trigger correctly:

def test_iot_rule_routes_to_lambda(iot_client, lambda_client):
    # Create a rule that routes temperature alerts to Lambda
    rule_payload = {
        "sql": "SELECT * FROM 'devices/+/alerts' WHERE temperature > 80",
        "actions": [{
            "lambda": {
                "functionArn": "arn:aws:lambda:us-east-1:000000000000:function:temperature-alert-handler"
            }
        }]
    }
    
    iot_client.create_topic_rule(
        ruleName='TemperatureAlertRule',
        topicRulePayload=rule_payload
    )
    
    # Publish a message that should match the rule
    iot_data_client = boto3.client('iot-data', endpoint_url='http://localhost:4566',
                                    region_name='us-east-1',
                                    aws_access_key_id='test', aws_secret_access_key='test')
    
    iot_data_client.publish(
        topic='devices/sensor-001/alerts',
        qos=1,
        payload=json.dumps({"temperature": 95.0, "device_id": "sensor-001"}).encode()
    )
    
    # Check Lambda was invoked
    time.sleep(1)
    invocations = lambda_client.get_function_invocations('temperature-alert-handler')
    assert len(invocations) > 0
    assert invocations[-1]['payload']['temperature'] == 95.0

Simulating Devices with AWS IoT Device SDK

For higher-fidelity integration tests, use the AWS IoT Device SDK to simulate real device connections:

from awsiot import mqtt_connection_builder
import awscrt.io as io

def create_test_device(endpoint, cert_path, key_path, ca_path, client_id):
    event_loop_group = io.EventLoopGroup(1)
    host_resolver = io.DefaultHostResolver(event_loop_group)
    client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)
    
    return mqtt_connection_builder.mtls_from_path(
        endpoint=endpoint,
        cert_filepath=cert_path,
        pri_key_filepath=key_path,
        client_bootstrap=client_bootstrap,
        ca_filepath=ca_path,
        client_id=client_id,
        clean_session=False,
        keep_alive_secs=30
    )

def test_device_connectivity_and_telemetry(test_certificates):
    connection = create_test_device(
        endpoint="your-iot-endpoint.amazonaws.com",
        cert_path=test_certificates.cert,
        key_path=test_certificates.key,
        ca_path=test_certificates.ca,
        client_id="test-device-integration-001"
    )
    
    connect_future = connection.connect()
    connect_future.result(timeout=10)  # 10 second timeout
    
    # Publish telemetry
    payload = json.dumps({"temp": 22.0, "ts": int(time.time())})
    publish_future, _ = connection.publish(
        topic="devices/test-device-integration-001/telemetry",
        payload=payload,
        qos=mqtt.QoS.AT_LEAST_ONCE
    )
    publish_future.result(timeout=5)
    
    # Disconnect
    disconnect_future = connection.disconnect()
    disconnect_future.result(timeout=5)

Testing Azure IoT Hub

Local Testing with the Azure IoT Hub Emulator

Microsoft provides an official IoT Hub emulator for local development:

# Pull and run the emulator
docker pull mcr.microsoft.com/azure-iot-hub-emulator:latest

docker run -d \
  -p 8883:8883 \
  -p 443:443 \
  -e connectionString=<span class="hljs-string">"HostName=localhost;SharedAccessKeyName=iothubowner;SharedAccessKey=dummykey" \
  mcr.microsoft.com/azure-iot-hub-emulator:latest
from azure.iot.hub import IoTHubRegistryManager
from azure.iot.device import IoTHubDeviceClient, Message
import pytest

EMULATOR_CONNECTION_STRING = (
    "HostName=localhost;SharedAccessKeyName=iothubowner;SharedAccessKey=dummykey"
)

@pytest.fixture
def registry_manager():
    return IoTHubRegistryManager.from_connection_string(EMULATOR_CONNECTION_STRING)

def test_device_registration(registry_manager):
    device_id = "test-device-001"
    
    device = registry_manager.create_device_with_sas(
        device_id=device_id,
        primary_key="primaryKeyBase64==",
        secondary_key="secondaryKeyBase64==",
        status="enabled"
    )
    
    assert device.device_id == device_id
    assert device.status == "enabled"
    
    # Cleanup
    registry_manager.delete_device(device_id)

Testing Device Twin

Azure's equivalent of AWS Device Shadow is the Device Twin. Test desired and reported properties:

def test_device_twin_desired_properties(registry_manager):
    device_id = "test-twin-device-001"
    
    # Update desired properties (from cloud/application side)
    twin_patch = {
        "properties": {
            "desired": {
                "telemetryInterval": 30,
                "fanMode": "auto"
            }
        }
    }
    
    registry_manager.update_twin(device_id, twin_patch, "*")
    
    # Read the twin back
    twin = registry_manager.get_twin(device_id)
    desired = twin.properties.desired
    
    assert desired.get("telemetryInterval") == 30
    assert desired.get("fanMode") == "auto"

def test_device_reports_twin_properties():
    device_connection_string = (
        "HostName=localhost;DeviceId=test-device-001;SharedAccessKey=primaryKeyBase64=="
    )
    
    client = IoTHubDeviceClient.create_from_connection_string(device_connection_string)
    client.connect()
    
    # Device reports its current state
    reported_properties = {
        "firmware_version": "1.2.3",
        "battery_level": 87,
        "sensor_calibration_date": "2026-01-15"
    }
    
    client.patch_twin_reported_properties(reported_properties)
    client.disconnect()
    
    # Verify via registry manager
    registry = IoTHubRegistryManager.from_connection_string(EMULATOR_CONNECTION_STRING)
    twin = registry.get_twin("test-device-001")
    
    reported = twin.properties.reported
    assert reported.get("firmware_version") == "1.2.3"
    assert reported.get("battery_level") == 87

Testing Direct Methods

Both platforms support invoking methods on devices from the cloud:

def test_direct_method_invocation(registry_manager):
    device_id = "test-device-001"
    
    # In production, this would go to the device. In tests, use a simulated device.
    method_name = "reboot"
    payload = {"reason": "scheduled_maintenance"}
    
    response = registry_manager.invoke_device_method(
        device_id=device_id,
        direct_method_request={
            "methodName": method_name,
            "payload": json.dumps(payload),
            "responseTimeoutInSeconds": 30
        }
    )
    
    assert response.status == 200
    result = json.loads(response.payload)
    assert result.get("acknowledged") is True

Test Infrastructure Patterns

Use Test Certificates (Don't Share Production Certs)

import tempfile
import os
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa

@pytest.fixture(scope="session")
def test_certificates(tmp_path_factory):
    """Generate self-signed certificates for IoT testing."""
    cert_dir = tmp_path_factory.mktemp("certs")
    
    # Generate key
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    
    # Generate self-signed cert
    cert = (
        x509.CertificateBuilder()
        .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "test-device")]))
        .issuer_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "test-ca")]))
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(datetime.utcnow())
        .not_valid_after(datetime.utcnow() + timedelta(days=1))
        .sign(key, hashes.SHA256())
    )
    
    # Write files
    key_path = cert_dir / "device.key"
    cert_path = cert_dir / "device.crt"
    
    key_path.write_bytes(key.private_bytes(serialization.Encoding.PEM,
                                            serialization.PrivateFormat.TraditionalOpenSSL,
                                            serialization.NoEncryption()))
    cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
    
    return type('Certs', (), {'key': str(key_path), 'cert': str(cert_path)})()

Comparison: AWS IoT vs Azure IoT Testing

Aspect AWS IoT Core Azure IoT Hub
Local emulator LocalStack (partial) Official emulator
Device simulation IoT Device SDK + Python Azure IoT Device SDK
State management Device Shadow Device Twin
Message routing Rules Engine (SQL) Message Routing (query)
Direct device calls Lambda via rule Direct Methods
Best test approach LocalStack for unit, staging for E2E Official emulator for most, staging for cert flows

Testing IoT cloud integrations requires accepting that you can't mock everything. Use emulators and LocalStack for fast integration tests, but maintain a staging IoT Hub/Core with real device certificates for pre-production validation of provisioning flows and certificate rotation.

Read more