MQTT-Clients in Go testen

//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
	}
}

//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)
			}
		})
	}
}

export MQTT_URL=tcp://broker.connectmq.com:1883
go test -tags integration -race -count=1 -timeout 2m ./...

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):
	}
}

//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")
	}
}

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{}
}