Testes end-to-end IoT sobre MQTT: do dispositivo à UI

Testes unitários provam que cada serviço funciona. O cliente quer saber se a leitura que um sensor envia aparece no dashboard e se o botão que ele aperta chega ao dispositivo. Os testes end-to-end para sistemas IoT sobre MQTT verificam exatamente isso: um dispositivo simulado publica pelo broker real e o teste verifica cada salto, do serviço de ingestão e do banco de dados até a API e a UI; depois envia um comando de volta e confere se o dispositivo o recebeu.

Este guia mostra como estruturar esse teste, com código executável de pytest e Playwright, e como mantê-lo rápido e confiável o bastante para rodar a cada merge.

O caminho que um evento percorre

Um sistema IoT típico baseado em MQTT tem duas direções, e uma suíte end-to-end deve cobrir as duas:

Direção	Saltos	O que pode quebrar
Uplink (telemetria)	Dispositivo → broker → serviço de ingestão → banco de dados → API → UI	Mudanças de topic ou payload, ACLs, parsing, fusos horários, cache
Downlink (comandos)	UI → API → backend → broker → dispositivo → ack	Topic errado, permissões faltando, QoS, dispositivos offline
Presença	Desconexão do dispositivo → Last Will → backend → status na UI	Configuração de keep alive, topic do LWT, estado desatualizado

Cada salto já tem seus próprios testes (veja testes de integração MQTT para a fronteira com o broker). A suíte end-to-end é pequena, um punhado de cenários, mas cada um atravessa todas as fronteiras de uma vez.

Crie um simulador que se comporte como o firmware

O simulador é a peça mais importante. Ele precisa usar o mesmo esquema de client ID, os mesmos topics, QoS, formato de payload e Last Will do firmware real; caso contrário, você está testando outro sistema. Uma versão mínima em Python com paho-mqtt 2.x:

sim_device.py

import json, queue, time
from datetime import datetime, timezone
import paho.mqtt.client as mqtt

class SimDevice:
    def __init__(self, device_id, host, user, password, prefix="testmqtt/e2e"):
        self.id = device_id
        self.base = f"{prefix}/{device_id}"
        self.commands = queue.Queue()
        self.subscribed = queue.Queue()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=device_id,
                                  protocol=mqtt.MQTTv5)
        self.client.username_pw_set(user, password)
        self.client.tls_set()
        self.client.will_set(f"{self.base}/status", "offline", qos=1, retain=True)
        self.client.on_connect = self._on_connect
        self.client.on_subscribe = lambda c, u, mid, rc, p: self.subscribed.put(mid)
        self.client.on_message = self._on_message
        self.host = host
        self.crashed = False

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        client.publish(f"{self.base}/status", "online", qos=1, retain=True)
        client.subscribe(f"{self.base}/cmd", qos=1)

    def _on_message(self, client, userdata, msg):
        cmd = json.loads(msg.payload)
        self.commands.put(cmd)
        client.publish(f"{self.base}/cmd/ack", json.dumps({"id": cmd["id"], "ok": True}), qos=1)

    def start(self):
        self.client.connect(self.host, 8883, keepalive=30)
        self.client.loop_start()
        self.subscribed.get(timeout=10)        # SUBACK recebido: pronto para comandos

    def send_reading(self, temp_c):
        payload = {"schema": "telemetry/v1", "device": self.id, "temp_c": temp_c,
                   "ts": datetime.now(timezone.utc).isoformat()}
        self.client.publish(f"{self.base}/telemetry", json.dumps(payload), qos=1).wait_for_publish()

    def crash(self):
        """Derruba a conexão TCP sem DISCONNECT para que o broker envie o Last Will."""
        self.crashed = True
        self.client.loop_stop()
        self.client.socket().close()

    def stop(self):
        if not self.crashed:
            # limpa o status retido para manter o broker organizado
            self.client.publish(f"{self.base}/status", "", qos=1, retain=True).wait_for_publish()
            self.client.disconnect()
        self.client.loop_stop()

Se você já tem um simulador de frota, reutilize-o; veja como construir um simulador de dispositivos MQTT para escalar a mesma ideia para muitos dispositivos. Para evitar que o payload divirja entre firmware e simulador, um JSON Schema compartilhado e testes de contrato mantêm os dois alinhados.

Espere por condições, nunca use sleep

Pipelines assíncronos têm latência variável. Um time.sleep(2) fixo é curto demais em um runner de CI sobrecarregado ou desperdiça tempo em todos os outros casos. Verifique uma condição até que ela seja verdadeira ou um prazo expire:

eventually.py

import time

def eventually(check, timeout=15.0, interval=0.25):
    """Chama check() até parar de lançar AssertionError, ou relança após o timeout."""
    deadline = time.monotonic() + timeout
    while True:
        try:
            return check()
        except AssertionError:
            if time.monotonic() > deadline:
                raise
            time.sleep(interval)

O teste de uplink: do dispositivo ao banco e à API

Cada execução recebe um ID de dispositivo único, para que execuções paralelas e restos de execuções anteriores nunca interfiram:

test_e2e_uplink.py

import os, uuid
import httpx, psycopg, pytest
from eventually import eventually
from sim_device import SimDevice

HOST = "broker.connectmq.com"
API = os.environ.get("API_URL", "http://localhost:8000")
DSN = os.environ.get("DATABASE_URL", "postgresql://app:app@localhost:5432/app")

@pytest.fixture
def device():
    dev = SimDevice(f"e2e-{uuid.uuid4().hex[:8]}", HOST,
                    os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    dev.start()
    yield dev
    dev.stop()
    with psycopg.connect(DSN) as db:
        db.execute("DELETE FROM readings WHERE device_id = %s", (dev.id,))

def test_reading_reaches_database_and_api(device):
    device.send_reading(21.5)

    def stored():
        with psycopg.connect(DSN) as db:
            row = db.execute(
                "SELECT temp_c FROM readings WHERE device_id = %s ORDER BY ts DESC LIMIT 1",
                (device.id,),
            ).fetchone()
        assert row is not None, "reading not stored yet"
        assert row[0] == pytest.approx(21.5)
    eventually(stored)

    def served():
        r = httpx.get(f"{API}/devices/{device.id}/latest", timeout=5)
        assert r.status_code == 200
        assert r.json()["temp_c"] == pytest.approx(21.5)
    eventually(served)

A verificação no banco diz que a ingestão funciona; a da API diz que consultas, serialização e cache funcionam por cima dela. Quando o teste falha, a primeira asserção que falha nomeia o salto quebrado, e esse é o principal valor de verificar cada camada em vez de só a UI. O guia MQTT para PostgreSQL mostra um serviço de ingestão correspondente.

O teste de UI e downlink com Playwright

O teste no navegador verifica o que o usuário vê e percorre o caminho de comandos de volta ao dispositivo. Com Playwright para Python:

test_e2e_ui.py

import os
from playwright.sync_api import Page, expect
from eventually import eventually

APP = os.environ.get("APP_URL", "http://localhost:3000")

def test_dashboard_shows_reading_and_sends_command(page: Page, device):
    device.send_reading(23.0)

    page.goto(f"{APP}/devices/{device.id}")
    expect(page.get_by_test_id("device-status")).to_have_text("online", timeout=15_000)
    expect(page.get_by_test_id("temp-value")).to_have_text("23.0 °C", timeout=15_000)

    page.get_by_role("button", name="Reboot").click()

    cmd = device.commands.get(timeout=15)        # o simulador recebeu via MQTT
    assert cmd["action"] == "reboot"
    expect(page.get_by_test_id("last-command")).to_contain_text("acknowledged", timeout=15_000)

Isso reutiliza a fixture device (mova-a para conftest.py). O expect do Playwright já tenta de novo até o timeout, então não é preciso polling extra no navegador. Use atributos data-testid em vez de classes CSS; eles sobrevivem a redesenhos.

Presença: teste o caminho do Last Will

A detecção de dispositivos offline é um recurso do qual os usuários dependem e que raramente é testado. Simule uma queda, não uma desconexão limpa, porque um DISCONNECT normal suprime o Last Will:

test_e2e_presence.py

from playwright.sync_api import Page, expect

def test_crashed_device_shows_offline(page: Page, device):
    page.goto(f"http://localhost:3000/devices/{device.id}")
    expect(page.get_by_test_id("device-status")).to_have_text("online", timeout=15_000)

    device.crash()

    # um socket fechado dispara o Will na hora; um link morto em silêncio leva até 1,5 x keepalive
    expect(page.get_by_test_id("device-status")).to_have_text("offline", timeout=60_000)

Um socket TCP fechado dispara o Will imediatamente, mas um dispositivo que perde energia ou cobertura não envia nada, e o broker só percebe depois de cerca de 1,5 vez o intervalo de keep alive, então mantenha o timeout acima disso. Veja como testar o Last Will para as regras de tempo. Acompanhar testmqtt/e2e/# no cliente MQTT online durante uma execução local mostra em ordem as mensagens de status, telemetria, comando e ack, e é o jeito mais rápido de depurar um cenário com falha.

Acompanhe o tráfego do teste end-to-end

Ambiente e CI

Rode o backend, o banco de dados e a UI com Docker Compose e aponte-os, assim como o simulador, para o mesmo broker:

Rodar a suíte localmente

export MQTT_USER=username MQTT_PASS=password
docker compose up -d --wait          # api, ingest, postgres, web
pip install pytest pytest-playwright paho-mqtt httpx "psycopg[binary]"
playwright install chromium
pytest -q tests/e2e --maxfail=1

No CI, os mesmos comandos funcionam dentro de um job; veja testes MQTT no GitHub Actions. Mantenha a configuração do broker (TLS, ACLs, versão do MQTT) idêntica à de produção, porque erros de permissão e de TLS são exatamente o que testes end-to-end devem pegar.

Mantendo a suíte confiável

IDs únicos por execução. ID do dispositivo, prefixo de topic e linhas do banco derivam de um único sufixo aleatório.
Espere o SUBACK. O start() do simulador bloqueia até a assinatura de comandos ser confirmada; sem isso, o primeiro comando pode chegar antes de alguém estar ouvindo.
Limpe as mensagens retidas. Publicar um payload retido vazio no topic de status em stop() mantém o broker organizado; veja mensagens retidas.
Poucos cenários, profundidade total. Cinco cenários end-to-end que atravessam todos os saltos valem mais que cinquenta que duplicam testes unitários.
Falhe com o nome do salto. Escreva mensagens de asserção como "reading not stored yet", para que o relatório diga onde a cadeia quebrou.

Uma suíte end-to-end para um sistema MQTT é um simulador, um helper de polling e alguns testes que acompanham um evento do dispositivo até a tela e de volta. Ela roda em minutos, pega os erros de integração que testes unitários não enxergam e dá uma resposta concreta para "funciona?" antes de cada release.

Perguntas frequentes

Qual a diferença entre testes de integração e end-to-end em IoT?

Um teste de integração verifica uma fronteira, por exemplo se o seu serviço de ingestão recebe uma mensagem do broker. Um teste end-to-end acompanha um único evento por todos os componentes: dispositivo, broker, backend, banco de dados, API e UI, além do caminho de comandos de volta ao dispositivo.

Preciso de hardware real para testes end-to-end de IoT?

Não para a maioria. Um simulador de dispositivo em software que usa os mesmos topics e payloads MQTT do firmware cobre o lado cloud de forma confiável e roda no CI. Mantenha uma pequena suíte hardware-in-the-loop para rádio, energia e comportamentos específicos do firmware.

Como evitar que testes end-to-end com MQTT fiquem instáveis?

Dê a cada execução um ID de dispositivo e um prefixo de topic únicos, espere condições com timeout em vez de sleeps fixos, espere o SUBACK antes de publicar e limpe as linhas e as mensagens retidas que o teste criou.

Testes end-to-end para sistemas IoT sobre MQTT