1574  Integration Testing for IoT Systems

1574.1 Learning Objectives

By the end of this chapter, you will be able to:

• Test Hardware-Software Interfaces: Validate GPIO, I2C, SPI, and ADC interactions
• Test Protocol Implementations: Verify MQTT, CoAP, HTTP, and BLE protocol behavior
• Test Cloud Integration: Validate end-to-end device-to-cloud data flows
• Design Integration Test Strategies: Create comprehensive test plans for IoT subsystems

1574.2 Prerequisites

Before diving into this chapter, you should be familiar with:

• Testing Fundamentals: Understanding the testing pyramid
• Unit Testing Firmware: Mocking and unit test concepts

NoteKey Takeaway

In one sentence: Integration tests validate that modules work together correctly, catching interface bugs that unit tests miss.

Remember this rule: If two modules communicate, test their interface. If timing matters, test real timing. If state persists, test state consistency.

---

1574.3 Hardware-Software Integration

Integration tests bridge unit tests (logic only) and end-to-end tests (full system).

Testing hardware-software interfaces requires real hardware or high-fidelity simulators:

Hardware-software integration testing flowchart

Figure 1574.1: Hardware-software integration tests validate the interface between firmware and physical hardware components

1574.3.1 Key Integration Test Categories

Category	What It Tests	Tools
GPIO	Pin state, timing, interrupts	Logic analyzer, oscilloscope
I2C/SPI	Sensor/peripheral communication	Protocol analyzer (Saleae, Bus Pirate)
ADC	Analog-to-digital conversion accuracy	Signal generator, calibrated reference
UART	Serial communication, parsing	Serial monitor, terminal emulator
Wi-Fi	Connection, reconnection, roaming	Network emulator, access point

1574.3.2 Example: Testing GPIO-based LED Control

// test_integration_gpio.c
#include "test_framework.h"
#include "gpio_driver.h"
#include "logic_analyzer.h"  // Interface to external tool

void test_led_toggle_timing(void) {
    // Arrange: Configure LED pin and start logic analyzer capture
    gpio_configure(LED_PIN, GPIO_OUTPUT);
    logic_analyzer_start_capture(LED_PIN);

    // Act: Toggle LED with 100ms period
    for (int i = 0; i < 10; i++) {
        gpio_write(LED_PIN, HIGH);
        delay_ms(50);
        gpio_write(LED_PIN, LOW);
        delay_ms(50);
    }

    // Assert: Verify timing via logic analyzer
    TimingResult result = logic_analyzer_analyze_period(LED_PIN);

    // Allow 10% tolerance for timing
    TEST_ASSERT_WITHIN(10, 100, result.period_ms);
    TEST_ASSERT_WITHIN(5, 50, result.high_time_ms);
    TEST_ASSERT_EQUAL(10, result.cycle_count);
}

1574.3.3 Example: Testing I2C Temperature Sensor

void test_i2c_sensor_read(void) {
    // Arrange: Known calibrated temperature (thermal chamber)
    float chamber_temp = 25.0;  // Set by test fixture
    thermal_chamber_set_temperature(chamber_temp);
    delay_ms(30000);  // Wait for stabilization

    // Act: Read sensor
    float sensor_temp = temperature_sensor_read();

    // Assert: Within sensor accuracy spec (±0.5°C)
    TEST_ASSERT_FLOAT_WITHIN(0.5, chamber_temp, sensor_temp);
}

void test_i2c_sensor_disconnect_detection(void) {
    // Arrange: Disconnect sensor from I2C bus
    i2c_bus_disconnect(SENSOR_ADDRESS);

    // Act: Attempt to read
    SensorStatus status = temperature_sensor_read_safe();

    // Assert: Proper error handling
    TEST_ASSERT_EQUAL(SENSOR_NOT_FOUND, status);
    TEST_ASSERT_TRUE(error_logged(ERR_I2C_NACK));
}

---

1574.4 Protocol Testing

Protocol tests validate that your firmware correctly implements communication protocols.

1574.4.1 MQTT Client Testing

# test_mqtt_integration.py
import pytest
import paho.mqtt.client as mqtt
import time
import json

@pytest.fixture
def mqtt_broker():
    """Start a test MQTT broker."""
    broker = start_mosquitto_broker(port=1883)
    yield broker
    broker.stop()

@pytest.fixture
def device(mqtt_broker):
    """Flash and boot device under test."""
    flash_firmware("firmware.bin")
    power_cycle_device()
    wait_for_boot(timeout=30)
    return DeviceInterface()

class TestMQTTPublish:

    def test_sensor_data_published_on_interval(self, device, mqtt_broker):
        """Device should publish sensor data every 60 seconds."""
        messages = []

        def on_message(client, userdata, msg):
            messages.append(json.loads(msg.payload))

        client = mqtt.Client()
        client.on_message = on_message
        client.connect("localhost", 1883)
        client.subscribe("sensors/temperature")
        client.loop_start()

        # Wait for 3 publish intervals
        time.sleep(180)
        client.loop_stop()

        # Assert: Should have ~3 messages
        assert len(messages) >= 2
        assert len(messages) <= 4

        # Verify message format
        for msg in messages:
            assert "temperature" in msg
            assert "timestamp" in msg
            assert isinstance(msg["temperature"], float)

    def test_publish_qos1_with_ack(self, device, mqtt_broker):
        """Device should retry QoS 1 messages until acknowledged."""
        # Capture MQTT packets
        pcap = start_mqtt_capture()

        # Temporarily block PUBACK from broker
        mqtt_broker.block_puback()

        # Trigger device publish
        device.trigger_sensor_read()
        time.sleep(2)

        # Verify retransmissions
        packets = pcap.get_packets()
        publish_count = sum(1 for p in packets if p.type == "PUBLISH")

        assert publish_count >= 2  # Original + at least 1 retry

        # Restore and verify eventual delivery
        mqtt_broker.allow_puback()
        time.sleep(5)

        assert mqtt_broker.received_message_count() >= 1

class TestMQTTSubscribe:

    def test_command_received_and_executed(self, device, mqtt_broker):
        """Device should execute commands received via MQTT."""
        client = mqtt.Client()
        client.connect("localhost", 1883)

        # Send command
        command = {"action": "set_led", "state": "on"}
        client.publish("device/commands", json.dumps(command))

        time.sleep(1)

        # Verify device executed command
        assert device.read_gpio_state(LED_PIN) == HIGH

    def test_reconnect_after_broker_restart(self, device, mqtt_broker):
        """Device should reconnect after broker becomes available."""
        # Verify initial connection
        assert device.mqtt_connected() == True

        # Restart broker
        mqtt_broker.stop()
        time.sleep(5)
        mqtt_broker.start()

        # Wait for reconnection
        for _ in range(30):
            if device.mqtt_connected():
                break
            time.sleep(1)

        assert device.mqtt_connected() == True

1574.4.2 Protocol Compliance Testing

Validate correct implementation of protocol specifications:

Protocol	Key Tests	Tools
MQTT	QoS handling, will messages, keepalive, clean session	Wireshark, mqtt-spy
CoAP	Confirmable messages, block transfer, observe	libcoap test suite
HTTP	Status codes, headers, chunked encoding	curl, Postman
BLE	GATT services, advertising, pairing	nRF Connect, hcitool

---

1574.5 Cloud Integration Testing

End-to-end tests validate the complete data path from sensor to cloud and back.

Cloud integration test architecture

Figure 1574.2: Cloud integration tests validate the complete device-to-cloud data flow including authentication, data formatting, and error handling

1574.5.1 Testing Cloud API Integration

# test_cloud_integration.py
import pytest
import requests
import time

CLOUD_API = "https://api.staging.example.com"
DEVICE_ID = "TEST-001"

@pytest.fixture
def device():
    """Configure device for staging environment."""
    flash_firmware_with_config({
        "cloud_url": CLOUD_API,
        "device_id": DEVICE_ID
    })
    power_cycle_device()
    wait_for_boot(timeout=30)
    return DeviceInterface()

class TestDeviceToCloud:

    def test_telemetry_received_by_cloud(self, device):
        """Sensor data should appear in cloud API within 60s."""
        # Trigger device to send data
        device.trigger_sensor_read()

        # Poll cloud API for data
        for _ in range(60):
            response = requests.get(
                f"{CLOUD_API}/devices/{DEVICE_ID}/telemetry/latest"
            )
            if response.status_code == 200:
                data = response.json()
                assert "temperature" in data
                return
            time.sleep(1)

        pytest.fail("Telemetry not received by cloud within 60s")

    def test_device_registration(self, device):
        """New device should auto-register with cloud."""
        # Check device appears in cloud registry
        response = requests.get(f"{CLOUD_API}/devices/{DEVICE_ID}")

        assert response.status_code == 200
        data = response.json()
        assert data["status"] == "online"
        assert data["firmware_version"] == device.firmware_version()

class TestCloudToDevice:

    def test_config_update_applied(self, device):
        """Config changes from cloud should be applied on device."""
        # Set new config via cloud API
        new_config = {"reporting_interval": 120}
        requests.post(
            f"{CLOUD_API}/devices/{DEVICE_ID}/config",
            json=new_config
        )

        # Wait for device to fetch config
        time.sleep(10)

        # Verify device applied config
        assert device.get_config("reporting_interval") == 120

    def test_firmware_ota_update(self, device):
        """Device should accept and install OTA firmware update."""
        original_version = device.firmware_version()

        # Trigger OTA via cloud
        requests.post(
            f"{CLOUD_API}/devices/{DEVICE_ID}/ota",
            json={"version": "2.0.0"}
        )

        # Wait for update (max 5 minutes)
        for _ in range(60):
            time.sleep(5)
            if device.firmware_version() != original_version:
                break

        assert device.firmware_version() == "2.0.0"
        assert device.is_operational()  # Device still works

---

1574.6 Test Environment Setup

Integration tests require controlled environments that simulate production conditions:

1574.6.1 Network Condition Simulation

# test_network_conditions.py
import netem  # Network emulator

class TestNetworkResilience:

    def test_high_latency_mqtt(self, device):
        """Device handles 500ms network latency."""
        # Add 500ms latency
        netem.add_delay("eth0", delay_ms=500)

        try:
            device.trigger_sensor_read()
            time.sleep(5)

            # Verify message eventually delivered
            assert mqtt_broker.received_message_count() >= 1
        finally:
            netem.reset("eth0")

    def test_packet_loss_recovery(self, device):
        """Device retries under 20% packet loss."""
        netem.add_packet_loss("eth0", loss_percent=20)

        try:
            device.trigger_sensor_read()
            time.sleep(30)  # Allow retries

            assert mqtt_broker.received_message_count() >= 1
        finally:
            netem.reset("eth0")

    def test_intermittent_connectivity(self, device):
        """Device buffers data during network outage."""
        # Disconnect for 60 seconds
        netem.block_all_traffic("eth0")
        time.sleep(60)

        # Reconnect
        netem.allow_all_traffic("eth0")
        time.sleep(30)

        # Verify buffered data was sent
        messages = mqtt_broker.get_all_messages()
        # Should have ~6 messages buffered (10s interval)
        assert len(messages) >= 5

---

1574.7 Knowledge Check

InlineKnowledgeCheck({
  questionId: "kc-testing-integration-1",
  question: "Your smart thermostat unit tests all pass (100% coverage), but the device fails in the field: it reads 35C when the room is 22C. Investigation reveals the I2C temperature sensor returns 16-bit values in big-endian format, but your firmware interprets them as little-endian. Which testing gap allowed this bug to escape?",
  options: [
    "Unit tests failed - they should have caught the endianness bug",
    "Missing integration test with real hardware - mocks returned correctly-formatted data",
    "Environmental testing gap - the bug only appears at certain temperatures",
    "Code coverage was insufficient - the parsing code wasn't tested"
  ],
  correctAnswer: 1,
  feedback: [
    "Incorrect. Unit tests verify the parsing function works as designed - but the design was based on incorrect assumptions about the sensor data format.",
    "Correct! Unit tests mocked the sensor, returning properly-formatted test data. Integration tests with real hardware would have revealed the byte-order mismatch immediately.",
    "Incorrect. The bug appears at all temperatures - the endianness error affects all readings equally (though wrong values might be more noticeable at certain ranges).",
    "Incorrect. The parsing code was covered (100% coverage). The bug was in the assumptions about input format, not in untested code."
  ],
  hint: "Think about what mocks assume about real hardware interfaces."
})

InlineKnowledgeCheck({
  questionId: "kc-testing-integration-2",
  question: "Your IoT gateway passes all integration tests on your CI server. In the field, MQTT connections drop every 2-3 hours. Logs show: 'TCP keepalive timeout'. Your test environment uses wired Ethernet; field deployments use Wi-Fi with occasional packet loss. What testing gap caused this?",
  options: [
    "Integration tests are fundamentally flawed - they can't predict network issues",
    "Missing network condition simulation - tests didn't include latency, jitter, or packet loss",
    "The bug is in the MQTT broker, not the device - nothing to test on device side",
    "Add more unit tests for the TCP keepalive code path"
  ],
  correctAnswer: 1,
  feedback: [
    "Incorrect. Integration tests can absolutely test network resilience - with proper network simulation.",
    "Correct! Testing on pristine wired networks misses real-world conditions. Use netem, tc, or commercial network emulators to inject latency, jitter, and packet loss.",
    "Incorrect. The device's MQTT client must handle network degradation gracefully regardless of broker behavior.",
    "Incorrect. Unit tests can't test real TCP behavior - network testing requires actual network stacks and delays."
  ],
  hint: "Consider what's different between your lab network and field deployments."
})

---

1574.8 Summary

Integration testing validates that IoT system components work together:

• Hardware-Software: Test GPIO, I2C, SPI with real hardware or high-fidelity simulators
• Protocol Testing: Validate MQTT, CoAP, HTTP, BLE compliance and edge cases
• Cloud Integration: End-to-end tests from sensor to cloud and back
• Network Simulation: Test resilience under latency, packet loss, and disconnection
• Environment Control: Use staging environments, not production, for integration tests

---

1574.9 What’s Next?

Continue your testing journey with these chapters:

• Hardware-in-the-Loop Testing: Automate firmware validation with simulated sensors
• Environmental Testing: Validate operation under temperature, humidity, EMC
• Testing Overview: Return to the complete testing guide