IoT System Testing: Unit, Integration, Hardware-in-the-Loop, and End-to-End

IoT System Testing: Unit, Integration, Hardware-in-the-Loop, and End-to-End

IoT system testing strategies differ from web application testing in ways that catch new teams off guard. You can't mock a temperature sensor reading that's coming from a faulty ADC. You can't replay a network partition at will when the failure mode is a physical cable. And a firmware bug that makes it to production requires OTA updates to every deployed device — if you're lucky — or field visits if you're not.

The consequence: IoT testing must be more rigorous, more automated, and planned earlier than most teams expect. This guide covers the full pyramid from firmware unit tests to chaos testing in production.

The IoT Testing Pyramid

                    ┌─────────────┐
                    │  Chaos/     │  (production, controlled)
                    │  Resilience │
                  ┌─┴─────────────┴─┐
                  │   End-to-End    │  (full stack, real devices)
                ┌─┴─────────────────┴─┐
                │ Hardware-in-the-    │  (real hardware, simulated cloud)
                │ Loop (HIL)          │
              ┌─┴─────────────────────┴─┐
              │   Integration           │  (MQTT mock, backend + DB)
            ┌─┴─────────────────────────┴─┐
            │   Unit (firmware + backend)   │  (fast, no hardware)
            └───────────────────────────────┘

Run everything below HIL in CI on every commit. HIL on every pull request merge. End-to-end nightly. Chaos testing weekly.

Layer 1: Firmware Unit Testing with Unity

Unity is the industry-standard C unit testing framework for embedded firmware. It runs on the target hardware or on a Linux host (using hardware abstraction layers to mock peripherals).

Structure your firmware so business logic is separated from hardware drivers — this is the essential prerequisite for unit testing:

// sensor_processor.h — pure business logic, no hardware calls
typedef struct {
  float temperature;
  float humidity;
  uint32_t timestamp;
} SensorReading;
typedef struct {
  float tempMin;
  float tempMax;
  float humidityMin;
  float humidityMax;
} AlertThresholds;
// Returns true if reading is within expected bounds
bool sensor_reading_is_valid(const SensorReading* reading);
// Returns alert bitmask (bit 0 = temp high, bit 1 = temp low, etc.)
uint8_t sensor_check_alerts(const SensorReading* reading,
                             const AlertThresholds* thresholds);
// Pack reading into MQTT payload — returns bytes written
int sensor_reading_to_json(const SensorReading* reading,
                            char* buf, size_t buf_size);

// test_sensor_processor.c — Unity tests, no hardware
#include "unity.h"
#include "sensor_processor.h"
void setUp(void) {}
void tearDown(void) {}
void test_reading_validation_rejects_impossible_humidity(void) {
  SensorReading r = {.temperature = 25.0f, .humidity = 150.0f, .timestamp = 1000};
  TEST_ASSERT_FALSE(sensor_reading_is_valid(&r));
}
void test_alert_fires_on_high_temperature(void) {
  SensorReading r = {.temperature = 85.0f, .humidity = 50.0f, .timestamp = 1000};
  AlertThresholds t = {.tempMin = 0.0f, .tempMax = 80.0f,
                       .humidityMin = 10.0f, .humidityMax = 90.0f};
  uint8_t alerts = sensor_check_alerts(&r, &t);
  TEST_ASSERT_BIT_HIGH(0, alerts); // bit 0 = temp high
}
void test_json_serialization_produces_valid_output(void) {
  SensorReading r = {.temperature = 22.5f, .humidity = 48.3f, .timestamp = 12345};
  char buf[256];
  int len = sensor_reading_to_json(&r, buf, sizeof(buf));
  TEST_ASSERT_GREATER_THAN(0, len);
  TEST_ASSERT_NOT_NULL(strstr(buf, ""t":22.50"));
  TEST_ASSERT_NOT_NULL(strstr(buf, ""h":48.30"));
}
int main(void) {
  UNITY_BEGIN();
  RUN_TEST(test_reading_validation_rejects_impossible_humidity);
  RUN_TEST(test_alert_fires_on_high_temperature);
  RUN_TEST(test_json_serialization_produces_valid_output);
  return UNITY_END();
}

Run Unity tests in CI using a Linux build target — no hardware needed. We use CMake + GitHub Actions for this:

# .github/workflows/firmware-test.yml
name: Firmware Unit Tests
on: [push, pull_request]
jobs:
  firmware-unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Install build tools
        run: sudo apt-get install -y cmake gcc
      - name: Build and run Unity tests
        run: |
          cmake -B build/test -DTARGET=host -DBUILD_TESTS=ON
          cmake --build build/test
          cd build/test && ctest --output-on-failure

Layer 2: Integration Testing with MQTT Mock Broker

Test the full message flow — firmware logic → MQTT → backend processing — without real hardware or a live cloud broker.

Mosquitto with a test config or the mqtt-mock library works well. We use Aedes (JavaScript MQTT broker) for integration tests because it runs in the same Node.js process as the backend:

// integration/test-helpers/mqtt-broker.ts
import Aedes from 'aedes'
import { createServer } from 'net'
export async function startTestBroker(port = 1883): Promise<{
  broker: Aedes
  stop: () => Promise<void>
}> {
  const broker = new Aedes()
  const server = createServer(broker.handle)
  await new Promise<void>((resolve) => server.listen(port, resolve))
  return {
    broker,
    stop: () => new Promise((resolve) => {
      broker.close(() => server.close(() => resolve()))
    }),
  }
}
// integration/telemetry-pipeline.test.ts
import { startTestBroker } from './test-helpers/mqtt-broker'
import { startTelemetryProcessor } from '../src/telemetry-processor'
import mqtt from 'mqtt'
describe('Telemetry pipeline integration', () => {
  let broker: Awaited<ReturnType<typeof startTestBroker>>
  let processor: Awaited<ReturnType<typeof startTelemetryProcessor>>
  beforeAll(async () => {
    broker  = await startTestBroker(1883)
    processor = await startTelemetryProcessor({ brokerUrl: 'mqtt://localhost:1883' })
  })
  afterAll(async () => {
    await processor.stop()
    await broker.stop()
  })
  it('processes batched telemetry and stores all readings', async () => {
    const client = mqtt.connect('mqtt://localhost:1883')
    await new Promise((r) => client.on('connect', r))
    const batch = {
      readings: [
        { t: 22.5, h: 48.0, ts: Date.now() - 60000 },
        { t: 22.8, h: 47.5, ts: Date.now() },
      ],
    }
    client.publish('devices/test-device-01/telemetry/batch', JSON.stringify(batch))
    // Wait for async processing
    await new Promise((r) => setTimeout(r, 500))
    const stored = await getStoredReadings('test-device-01')
    expect(stored).toHaveLength(2)
    expect(stored[0].temperature).toBeCloseTo(22.5)
    client.end()
  })
})

Layer 3: Hardware-in-the-Loop (HIL)

HIL testing uses real hardware but simulated or instrumented signals. A test jig applies known inputs (simulated sensor voltages, injected CAN messages, scripted GPIO events) and verifies the device's MQTT output.

Key components of a HIL rig:

# hil/test_temperature_alert.py — runs on test controller (Raspberry Pi)
import paho.mqtt.client as mqtt
import RPi.GPIO as GPIO
import time
import pytest
BROKER_HOST = "192.168.1.100"
DEVICE_MQTT_ID = "device-under-test-001"
DAC_CHANNEL = 0  # MCP4725 DAC for injecting voltage
class HILTestSession:
    def __init__(self):
        self.received_alerts = []
        self.client = mqtt.Client()
        self.client.on_message = self._on_message
        self.client.connect(BROKER_HOST, 1883)
        self.client.subscribe(f"devices/{DEVICE_MQTT_ID}/#")
        self.client.loop_start()
    def _on_message(self, client, userdata, msg):
        self.received_alerts.append({
            'topic': msg.topic,
            'payload': msg.payload.decode(),
            'timestamp': time.time()
        })
    def inject_temperature_voltage(self, celsius: float):
        # Convert celsius to DAC voltage for thermistor simulation
        voltage = temp_to_voltage(celsius)
        set_dac_voltage(DAC_CHANNEL, voltage)
def test_high_temperature_alert_fires_within_30_seconds():
    session = HILTestSession()
    # Inject normal temperature first
    session.inject_temperature_voltage(22.0)
    time.sleep(15)
    # Inject critically high temperature
    session.inject_temperature_voltage(92.0)
    start = time.time()
    # Wait for alert topic to appear
    timeout = 30
    while time.time() - start < timeout:
        alerts = [m for m in session.received_alerts
                  if 'alert' in m['topic']]
        if alerts:
            latency = time.time() - start
            assert latency < 30, f"Alert fired too late: {latency:.1f}s"
            return
        time.sleep(1)
    pytest.fail("No alert received within 30 seconds of temperature injection")

HIL tests are slower to write but catch hardware-software integration bugs that no amount of unit testing finds: wrong ADC gain settings, off-by-one in ADC bit depth conversion, timer drift in sampling intervals.

Layer 4: End-to-End and Chaos Testing

End-to-end tests use production firmware, a staging cloud environment, and real devices. Run nightly with a test device fleet (we keep 5–10 devices per project in a permanent staging rack).

Chaos testing verifies resilience. Deliberately induce failures and verify recovery:

// chaos/network-partition-test.ts
// Verify device reconnects and delivers queued messages after network outage
async function testNetworkPartitionRecovery() {
  const deviceId = 'chaos-test-device-01'
  const messagesBefore: number[] = []
  // Collect baseline message count over 2 minutes
  await collectMessages(deviceId, 120_000, (msg) => messagesBefore.push(msg.ts))
  // Cut network to device for 5 minutes (via managed switch API or iptables on gateway)
  await setNetworkPartition(deviceId, 'isolated')
  console.log('Network partitioned — waiting 5 minutes')
  await sleep(5 * 60 * 1000)
  await setNetworkPartition(deviceId, 'connected')
  console.log('Network restored — waiting for recovery')
  // Device should reconnect within 60 seconds and deliver queued messages
  const messagesAfter: number[] = []
  await collectMessages(deviceId, 120_000, (msg) => messagesAfter.push(msg.ts))
  // Verify: no gap in timestamps larger than the partition duration
  const sortedTs = [...messagesBefore, ...messagesAfter].sort()
  const maxGap = Math.max(...sortedTs.slice(1).map((ts, i) => ts - sortedTs[i]))
  console.assert(maxGap < 8 * 60 * 1000, `Message gap too large: ${maxGap / 1000}s`)
  console.log(`Test passed. Max message gap: ${maxGap / 1000}s`)
}

Other chaos scenarios worth automating:

CI/CD Pipeline for IoT

# Simplified CI pipeline
stages:
  - firmware-unit-test     # 2 min — runs on every commit
  - backend-unit-test      # 2 min — runs on every commit
  - backend-integration    # 5 min — runs on every commit
  - firmware-build         # 8 min — cross-compile for ESP32
  - hil-tests              # 20 min — PR merge only, requires jig
  - staging-deploy         # 5 min — deploy to staging environment
  - e2e-tests              # 30 min — nightly, on real device fleet
  - chaos-tests            # 90 min — weekly

The HIL and E2E stages require physical lab infrastructure — self-hosted GitHub Actions runners connected to the test rig. Budget this into your project plan from the start; retrofitting a test rig after you've shipped firmware to 5,000 devices in the field is painful.

IoT system testing strategies require more investment than web testing, but the asymmetry of failure costs makes it non-negotiable. A firmware bug caught in HIL takes 20 minutes to fix. The same bug in production takes an OTA release cycle, field team coordination, and customer communication.

Need help? [Contact Code Caracal](/contact) — we've shipped these systems for clients across 15+ countries.