Testing MQTT clients in Go works best with plain go test: a small helper that connects a client, a buffered channel that receives messages, and a select with a timeout that turns "the message never arrived" into a clean test failure. This guide shows that pattern for paho.mqtt.golang (MQTT 3.1.1) and autopaho (MQTT 5), plus table-driven tests and fakes for code that should not need a broker at all.
Two kinds of tests
| Test type | Talks to a broker? | Purpose |
|---|---|---|
| Unit test | No — a fake publisher | Your handler logic, payload encoding, topic building |
| Integration test | Yes — real broker | Connect options, TLS, auth, QoS, retained messages, ACLs |
Keep most logic testable without MQTT by depending on a small interface instead of the concrete client, and reserve the broker for what only a broker can prove. The rest of this guide is about integration tests; fakes come at the end.
A paho.mqtt.golang test helper
If you have not used the library before, MQTT in Go with Paho covers the basics. For tests, wrap connection setup in a helper that calls t.Helper() and registers cleanup, so every test is three lines long.
//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
}
}Three details prevent the most common hangs:
WaitTimeoutinstead ofWait.Wait()blocks forever if the broker never answers; a hung test gets killed bygo testonly after the default 10-minute timeout.- A buffered channel. paho delivers messages in order by default, so a handler that blocks on a full channel stalls the whole client — including the acknowledgements your publish is waiting for.
- Waiting for SUBACK.
Subscribereturns a token; wait for it before publishing, or the first message can race past the subscription.
A table-driven round-trip test
With the helpers in place, a test that checks delivery at every QoS level is short and readable:
//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)
}
})
}
}Each subtest gets a fresh client and topic, so t.Parallel() is safe. Note that the delivered QoS is the minimum of publish and subscription QoS — a mismatch here usually means the broker downgraded the subscription. More on that in testing QoS delivery.
Run the suite with the race detector and without test caching, since cached results say nothing about today's broker:
export MQTT_URL=tcp://broker.connectmq.com:1883
go test -tags integration -race -count=1 -timeout 2m ./...Asserting silence: ACL and filter tests
Some tests must prove that a message does not arrive — a wildcard that should not match, or a topic the client may not read. Add a second 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):
}
}Choose wait from measured latency: if a positive round trip normally takes 50 ms, two seconds of silence is strong evidence. Check your filter logic first with the topic matcher so you know which result to expect.
The same pattern with autopaho (MQTT 5)
autopaho from github.com/eclipse/paho.golang handles reconnection and speaks MQTT 5. Messages arrive through OnPublishReceived callbacks; forward them into a channel exactly as before.
//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")
}
}Asserting on a user property is a cheap way to check that MQTT 5 is negotiated end to end — a broker or proxy that silently falls back to 3.1.1 would drop it. autopaho's API has changed between minor versions, so pin the module version in go.mod and check the release notes when upgrading.
Unit tests without a broker
Code that only publishes rarely needs a real broker in unit tests. Depend on an interface and swap in a 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 already satisfies this interface, so production code passes the real client and tests pass the fake. mqtt.DummyToken is a ready-made token that completes immediately with no error. For behaviour that depends on broker semantics — retained messages, wildcard matching, session state — an embedded broker or a mock broker is the better tool; see mock brokers for unit tests.
Where the integration tests should point
Locally, a Mosquitto container or the shared broker.connectmq.com endpoint is fine for quick runs. In CI you want a broker that matches production: the same TLS port (ssl://...:8883 for paho, mqtts:// for autopaho), real credentials and the ACLs your devices are subject to. MQTT testing in GitHub Actions shows how to inject those as secrets.
When a test fails and you can't tell why, run it with -run and -v, and watch testmqtt/go/# in the online MQTT client at the same time. If the message shows up there but not in your test, the problem is your subscription or handler; if it never shows up, look at the publish side — the Go Paho error guide lists the usual causes.
Watch go test traffic in the browser
Go makes MQTT tests pleasant once the asynchrony is contained: helpers that fail fast with t.Fatalf, buffered channels, a select with a timeout for every expectation, and a unique client ID and topic per subtest. Keep broker-free logic behind an interface, run the real round trips behind a build tag, and the same suite will cover your laptop, CI and a post-deploy smoke test.
Frequently asked questions
Why does my Go MQTT test hang forever?
Usually a token is waited on with Wait() and the broker never answers, or a message handler blocks on an unbuffered channel and stalls the client's router. Use WaitTimeout, give message channels a buffer, and always select on a timeout when you read from them.
Should Go MQTT tests use t.Parallel?
Yes, as long as every test uses its own client ID and topic prefix. Integration tests spend most of their time waiting on the network, so running them in parallel shortens the suite considerably.
How do I skip MQTT integration tests in unit test runs?
Put them behind a build tag such as //go:build integration and run them with go test -tags integration, or call t.Skip when testing.Short() is true or when the MQTT_URL environment variable is not set.
