Probar clientes MQTT en Go funciona mejor con go test a secas: un pequeño helper que conecta un cliente, un canal con buffer que recibe los mensajes y un select con timeout que convierte "el mensaje nunca llegó" en un fallo limpio. Esta guía muestra ese patrón para paho.mqtt.golang (MQTT 3.1.1) y autopaho (MQTT 5), además de tests por tabla y fakes para el código que no debería necesitar un broker.
Dos tipos de tests
| Tipo de test | ¿Habla con un broker? | Propósito |
|---|---|---|
| Test unitario | No — un publisher falso | La lógica de tus handlers, la codificación del payload, la construcción de topics |
| Test de integración | Sí — broker real | Opciones de conexión, TLS, autenticación, QoS, mensajes retenidos, ACLs |
Mantén la mayor parte de la lógica testeable sin MQTT dependiendo de una interfaz pequeña en lugar del cliente concreto, y reserva el broker para lo que solo un broker puede demostrar. El resto de la guía trata de tests de integración; los fakes llegan al final.
Un helper de test para paho.mqtt.golang
Si no has usado la librería antes, MQTT en Go con Paho cubre lo básico. Para los tests, envuelve la conexión en un helper que llame a t.Helper() y registre la limpieza, de modo que cada test ocupe tres líneas.
//go:build integration
package telemetry
import (
"crypto/rand"
"encoding/hex"
"os"
"testing"
"time"
mqtt "github.com/eclipse/paho.mqtt.golang"
)
func brokerURL() string {
if u := os.Getenv("MQTT_URL"); u != "" {
return u
}
return "tcp://broker.connectmq.com:1883"
}
func randomID(t *testing.T) string {
t.Helper()
b := make([]byte, 4)
if _, err := rand.Read(b); err != nil {
t.Fatal(err)
}
return hex.EncodeToString(b)
}
// newClient connects a client and disconnects it when the test ends.
func newClient(t *testing.T) mqtt.Client {
t.Helper()
opts := mqtt.NewClientOptions().
AddBroker(brokerURL()).
SetClientID("gotest-" + randomID(t)).
SetUsername(os.Getenv("MQTT_USER")).
SetPassword(os.Getenv("MQTT_PASS")).
SetCleanSession(true).
SetConnectTimeout(10 * time.Second)
c := mqtt.NewClient(opts)
tok := c.Connect()
if !tok.WaitTimeout(15 * time.Second) {
t.Fatal("connect: timed out")
}
if err := tok.Error(); err != nil {
t.Fatalf("connect: %v", err)
}
t.Cleanup(func() { c.Disconnect(250) })
return c
}
// subscribe returns a buffered channel that receives every message on filter.
func subscribe(t *testing.T, c mqtt.Client, filter string, qos byte) <-chan mqtt.Message {
t.Helper()
ch := make(chan mqtt.Message, 100)
tok := c.Subscribe(filter, qos, func(_ mqtt.Client, m mqtt.Message) { ch <- m })
if !tok.WaitTimeout(5*time.Second) || tok.Error() != nil {
t.Fatalf("subscribe %s: %v", filter, tok.Error())
}
return ch
}
// expectMessage fails the test unless a message with payload want arrives in time.
func expectMessage(t *testing.T, ch <-chan mqtt.Message, want string, timeout time.Duration) mqtt.Message {
t.Helper()
select {
case m := <-ch:
if got := string(m.Payload()); got != want {
t.Fatalf("topic %s: got payload %q, want %q", m.Topic(), got, want)
}
return m
case <-time.After(timeout):
t.Fatalf("no message within %s", timeout)
return nil
}
}Tres detalles evitan los cuelgues más habituales:
WaitTimeouten lugar deWait.Wait()bloquea para siempre si el broker no responde;go testsolo mata un test colgado tras el timeout por defecto de 10 minutos.- Un canal con buffer. paho entrega los mensajes en orden por defecto, así que un handler bloqueado en un canal lleno detiene todo el cliente, incluidas las confirmaciones que espera tu publish.
- Esperar el SUBACK.
Subscribedevuelve un token; espéralo antes de publicar, o el primer mensaje puede adelantarse a la suscripción.
Un test de ida y vuelta guiado por tabla
Con los helpers listos, un test que comprueba la entrega en cada nivel de QoS es corto y legible:
//go:build integration
package telemetry
import (
"testing"
"time"
)
func TestRoundTrip(t *testing.T) {
tests := []struct {
name string
qos byte
payload string
}{
{"qos0 small", 0, "21.5"},
{"qos1 json", 1, `{"temp":21.5,"unit":"C"}`},
{"qos2 text", 2, "door=open"},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
c := newClient(t)
topic := "testmqtt/go/" + randomID(t) + "/temperature"
msgs := subscribe(t, c, topic, tc.qos)
tok := c.Publish(topic, tc.qos, false, tc.payload)
if !tok.WaitTimeout(5*time.Second) || tok.Error() != nil {
t.Fatalf("publish: %v", tok.Error())
}
m := expectMessage(t, msgs, tc.payload, 5*time.Second)
if m.Qos() != tc.qos {
t.Errorf("delivered at QoS %d, want %d", m.Qos(), tc.qos)
}
})
}
}Cada subtest recibe un cliente y un topic nuevos, así que t.Parallel() es seguro. Ten en cuenta que el QoS entregado es el mínimo entre el QoS de publicación y el de la suscripción: una discrepancia aquí suele significar que el broker rebajó la suscripción. Más detalles en probar la entrega por QoS.
Ejecuta la suite con el detector de carreras y sin caché de tests, porque un resultado cacheado no dice nada del broker de hoy:
export MQTT_URL=tcp://broker.connectmq.com:1883
go test -tags integration -race -count=1 -timeout 2m ./...Verificar silencio: tests de ACL y de filtros
Algunos tests deben demostrar que un mensaje no llega: un wildcard que no debería coincidir o un topic que el cliente no puede leer. Añade un segundo helper:
func expectNoMessage(t *testing.T, ch <-chan mqtt.Message, wait time.Duration) {
t.Helper()
select {
case m := <-ch:
t.Fatalf("unexpected message on %s: %q", m.Topic(), m.Payload())
case <-time.After(wait):
}
}Elige wait a partir de la latencia medida: si un round trip positivo tarda normalmente 50 ms, dos segundos de silencio son una prueba sólida. Comprueba antes la lógica de tu filtro con el topic matcher para saber qué resultado esperar.
El mismo patrón con autopaho (MQTT 5)
autopaho, de github.com/eclipse/paho.golang, gestiona la reconexión y habla MQTT 5. Los mensajes llegan a través de los callbacks OnPublishReceived; reenvíalos a un canal igual que antes.
//go:build integration
package telemetry
import (
"context"
"net/url"
"testing"
"time"
"github.com/eclipse/paho.golang/autopaho"
"github.com/eclipse/paho.golang/paho"
)
func TestAutopahoRoundTrip(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
u, _ := url.Parse("mqtt://broker.connectmq.com:1883")
msgs := make(chan *paho.Publish, 10)
topic := "testmqtt/go/" + randomID(t) + "/v5"
cm, err := autopaho.NewConnection(ctx, autopaho.ClientConfig{
ServerUrls: []*url.URL{u},
KeepAlive: 20,
CleanStartOnInitialConnection: true,
ClientConfig: paho.ClientConfig{
ClientID: "gotest-v5-" + randomID(t),
OnPublishReceived: []func(paho.PublishReceived) (bool, error){
func(pr paho.PublishReceived) (bool, error) {
msgs <- pr.Packet
return true, nil
},
},
},
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = cm.Disconnect(context.Background()) })
if err := cm.AwaitConnection(ctx); err != nil {
t.Fatalf("connect: %v", err)
}
if _, err := cm.Subscribe(ctx, &paho.Subscribe{
Subscriptions: []paho.SubscribeOptions{{Topic: topic, QoS: 1}},
}); err != nil {
t.Fatalf("subscribe: %v", err)
}
if _, err := cm.Publish(ctx, &paho.Publish{
Topic: topic, QoS: 1, Payload: []byte("hello v5"),
Properties: &paho.PublishProperties{
User: paho.UserProperties{{Key: "test", Value: t.Name()}},
},
}); err != nil {
t.Fatalf("publish: %v", err)
}
select {
case p := <-msgs:
if string(p.Payload) != "hello v5" {
t.Fatalf("got %q", p.Payload)
}
if p.Properties.User.Get("test") != t.Name() {
t.Errorf("user property lost")
}
case <-ctx.Done():
t.Fatal("no message before deadline")
}
}Verificar una user property es una forma barata de comprobar que MQTT 5 se negocia de extremo a extremo: un broker o proxy que vuelva en silencio a 3.1.1 la descartaría. La API de autopaho ha cambiado entre versiones menores, así que fija la versión del módulo en go.mod y revisa las notas de versión al actualizar.
Tests unitarios sin broker
El código que solo publica rara vez necesita un broker real en los tests unitarios. Depende de una interfaz y sustitúyela por un fake:
type Publisher interface {
Publish(topic string, qos byte, retained bool, payload interface{}) mqtt.Token
}
// In the test file:
type sentMsg struct {
topic string
payload interface{}
}
type fakePublisher struct{ sent []sentMsg }
func (f *fakePublisher) Publish(topic string, _ byte, _ bool, p interface{}) mqtt.Token {
f.sent = append(f.sent, sentMsg{topic, p})
return &mqtt.DummyToken{}
}mqtt.Client ya cumple esta interfaz, así que el código de producción pasa el cliente real y los tests pasan el fake. mqtt.DummyToken es un token listo para usar que se completa al instante sin error. Para comportamientos que dependen de la semántica del broker (mensajes retenidos, coincidencia de wildcards, estado de sesión), un broker embebido o un broker mock es mejor herramienta; consulta brokers mock para tests unitarios.
Hacia dónde deben apuntar los tests de integración
En local, un contenedor de Mosquitto o el endpoint compartido broker.connectmq.com bastan para ejecuciones rápidas. En CI quieres un broker que se parezca a producción: el mismo puerto TLS (ssl://...:8883 para paho, mqtts:// para autopaho), credenciales reales y las ACLs a las que están sujetos tus dispositivos. Pruebas MQTT en GitHub Actions muestra cómo inyectarlas como secrets.
Cuando un test falla y no sabes por qué, ejecútalo con -run y -v y observa al mismo tiempo testmqtt/go/# en el cliente MQTT online. Si el mensaje aparece ahí pero no en tu test, el problema está en tu suscripción o tu handler; si nunca aparece, mira el lado del publish: la guía de errores de Paho en Go recoge las causas habituales.
Mira el tráfico de go test en el navegador
Go hace agradables los tests MQTT una vez contenida la asincronía: helpers que fallan rápido con t.Fatalf, canales con buffer, un select con timeout para cada expectativa y un client ID y un topic únicos por subtest. Mantén la lógica sin broker detrás de una interfaz, ejecuta los round trips reales detrás de un build tag y la misma suite cubrirá tu portátil, la CI y un smoke test tras el despliegue.
Preguntas frecuentes
¿Por qué mi test MQTT en Go se queda colgado para siempre?
Normalmente se espera un token con Wait() y el broker nunca responde, o un handler de mensajes se bloquea en un canal sin buffer y detiene el router del cliente. Usa WaitTimeout, dale buffer a los canales de mensajes y haz siempre select con un timeout al leerlos.
¿Deben los tests MQTT en Go usar t.Parallel?
Sí, siempre que cada test use su propio client ID y prefijo de topic. Los tests de integración pasan la mayor parte del tiempo esperando la red, así que ejecutarlos en paralelo acorta mucho la suite.
¿Cómo omito los tests de integración MQTT en las ejecuciones unitarias?
Ponlos detrás de un build tag como //go:build integration y ejecútalos con go test -tags integration, o llama a t.Skip cuando testing.Short() sea true o cuando la variable de entorno MQTT_URL no esté definida.
