Pruebas end-to-end IoT sobre MQTT: del dispositivo a la UI

Los tests unitarios demuestran que cada servicio funciona. A los clientes les importa si la lectura que envía un sensor aparece en su dashboard y si el botón que pulsan llega al dispositivo. Las pruebas end-to-end para sistemas IoT sobre MQTT comprueban exactamente eso: un dispositivo simulado publica a través del broker real y el test verifica cada salto, desde el servicio de ingesta y la base de datos hasta la API y la UI; después envía un comando de vuelta y comprueba que el dispositivo lo recibió.

Esta guía muestra cómo estructurar ese test, con código ejecutable de pytest y Playwright, y cómo mantenerlo rápido y fiable para ejecutarlo en cada merge.

El camino que recorre un evento

Un sistema IoT típico basado en MQTT tiene dos direcciones, y una suite end-to-end debe cubrir ambas:

Dirección	Saltos	Qué puede romperse
Uplink (telemetría)	Dispositivo → broker → servicio de ingesta → base de datos → API → UI	Cambios de topic o payload, ACL, parseo, zonas horarias, caché
Downlink (comandos)	UI → API → backend → broker → dispositivo → ack	Topic equivocado, permisos que faltan, QoS, dispositivos offline
Presencia	Desconexión del dispositivo → Last Will → backend → estado en la UI	Ajustes de keep alive, topic del LWT, estado obsoleto

Cada salto ya tiene sus propios tests (consulta pruebas de integración MQTT para la frontera con el broker). La suite end-to-end es pequeña, un puñado de escenarios, pero cada uno cruza todas las fronteras a la vez.

Construye un simulador que se comporte como el firmware

El simulador es la pieza más importante. Debe usar el mismo esquema de client ID, los mismos topics, QoS, formato de payload y Last Will que el firmware real; si no, estás probando otro sistema. Una versión mínima en Python con 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 recibido: listo 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):
        """Corta la conexión TCP sin DISCONNECT para que el broker envíe el Last Will."""
        self.crashed = True
        self.client.loop_stop()
        self.client.socket().close()

    def stop(self):
        if not self.crashed:
            # borra el estado retenido para dejar el broker limpio
            self.client.publish(f"{self.base}/status", "", qos=1, retain=True).wait_for_publish()
            self.client.disconnect()
        self.client.loop_stop()

Si ya tienes un simulador de flota, reutilízalo; consulta cómo construir un simulador de dispositivos MQTT para escalar la misma idea a muchos dispositivos. Para evitar que el payload diverja entre firmware y simulador, un JSON Schema compartido y las pruebas de contrato mantienen a ambos alineados.

Espera condiciones, nunca uses sleep

Los pipelines asíncronos tienen latencia variable. Un time.sleep(2) fijo es demasiado corto en un runner de CI cargado o desperdicia tiempo en el resto de casos. Consulta una condición hasta que se cumpla o venza un plazo:

eventually.py

import time

def eventually(check, timeout=15.0, interval=0.25):
    """Llama a check() hasta que deje de lanzar AssertionError, o relanza tras el timeout."""
    deadline = time.monotonic() + timeout
    while True:
        try:
            return check()
        except AssertionError:
            if time.monotonic() > deadline:
                raise
            time.sleep(interval)

El test de uplink: del dispositivo a la base de datos y a la API

Cada ejecución recibe un ID de dispositivo único, de modo que las ejecuciones en paralelo y los restos de ejecuciones anteriores nunca interfieren:

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)

La comprobación en la base de datos te dice que la ingesta funciona; la de la API, que las consultas, la serialización y la caché funcionan por encima. Cuando el test falla, la primera aserción que falla nombra el salto roto, y ese es el principal valor de verificar cada capa en lugar de solo la UI. La guía MQTT a PostgreSQL muestra un servicio de ingesta equivalente.

El test de UI y downlink con Playwright

El test de navegador comprueba lo que ve el usuario y recorre el camino de comandos de vuelta al dispositivo. Con 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)        # el simulador lo recibió por MQTT
    assert cmd["action"] == "reboot"
    expect(page.get_by_test_id("last-command")).to_contain_text("acknowledged", timeout=15_000)

Esto reutiliza la fixture device (muévela a conftest.py). El expect de Playwright ya reintenta hasta su timeout, así que no hace falta más polling en el navegador. Usa atributos data-testid en lugar de clases CSS; sobreviven a los rediseños.

Presencia: prueba el camino del Last Will

La detección de dispositivos offline es una función en la que los usuarios confían y que rara vez se prueba. Simula un fallo, no una desconexión limpia, porque un DISCONNECT normal suprime el 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()

    # un socket cerrado dispara el Will al instante; un enlace muerto en silencio tarda hasta 1,5 x keepalive
    expect(page.get_by_test_id("device-status")).to_have_text("offline", timeout=60_000)

Un socket TCP cerrado dispara el Will de inmediato, pero un dispositivo que se queda sin energía o sin cobertura no envía nada, y el broker solo lo detecta tras unas 1,5 veces el intervalo de keep alive, así que mantén el timeout por encima de eso. Consulta cómo probar el Last Will para las reglas de tiempos. Observar testmqtt/e2e/# en el cliente MQTT online durante una ejecución local muestra en orden los mensajes de estado, telemetría, comando y ack, y es la forma más rápida de depurar un escenario que falla.

Observa el tráfico del test end-to-end

Entorno y CI

Ejecuta el backend, la base de datos y la UI con Docker Compose y apúntalos, igual que el simulador, al mismo broker:

Ejecutar la suite en local

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

En CI, los mismos comandos funcionan dentro de un job; consulta pruebas MQTT en GitHub Actions. Mantén la configuración del broker (TLS, ACL, versión de MQTT) idéntica a la de producción, porque los errores de permisos y de TLS son justo lo que deben detectar las pruebas end-to-end.

Mantener la suite fiable

IDs únicos por ejecución. El ID de dispositivo, el prefijo de topic y las filas de la base de datos derivan de un mismo sufijo aleatorio.
Espera el SUBACK. start() del simulador se bloquea hasta confirmar la suscripción de comandos; sin eso, el primer comando puede llegar antes de que nadie escuche.
Limpia los mensajes retenidos. Publicar un payload retenido vacío en el topic de estado en stop() mantiene el broker ordenado; consulta mensajes retenidos.
Pocos escenarios, profundidad completa. Cinco escenarios end-to-end que cruzan todos los saltos valen más que cincuenta que duplican tests unitarios.
Falla con el nombre del salto. Escribe mensajes de aserción como "reading not stored yet", para que el informe diga dónde se rompió la cadena.

Una suite end-to-end para un sistema MQTT es un simulador, un helper de polling y unos pocos tests que siguen un evento del dispositivo a la pantalla y de vuelta. Se ejecuta en minutos, detecta los errores de integración que los tests unitarios no pueden ver y te da una respuesta concreta a «¿funciona?» antes de cada versión.

Preguntas frecuentes

¿Qué diferencia hay entre pruebas de integración y end-to-end en IoT?

Una prueba de integración comprueba una frontera, por ejemplo que tu servicio de ingesta recibe un mensaje del broker. Una prueba end-to-end sigue un único evento por todos los componentes: dispositivo, broker, backend, base de datos, API y UI, además del camino de comandos de vuelta al dispositivo.

¿Necesito hardware real para las pruebas end-to-end de IoT?

No para la mayoría. Un simulador de dispositivo por software que use los mismos topics y payloads MQTT que el firmware cubre la parte cloud de forma fiable y se ejecuta en CI. Mantén una pequeña suite hardware-in-the-loop para radio, energía y comportamiento específico del firmware.

¿Cómo evito que las pruebas end-to-end con MQTT sean inestables?

Da a cada ejecución un ID de dispositivo y un prefijo de topic únicos, espera condiciones con un timeout en lugar de sleeps fijos, espera el SUBACK antes de publicar y limpia las filas y los mensajes retenidos que creó el test.

Pruebas end-to-end para sistemas IoT sobre MQTT