93 lines
3.0 KiB
Go
93 lines
3.0 KiB
Go
package mqtt
|
|
|
|
import (
|
|
"io"
|
|
"log/slog"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/eclipse/paho.golang/paho"
|
|
"robot-market/test-mqtt-5/internal/config"
|
|
"robot-market/test-mqtt-5/internal/events"
|
|
)
|
|
|
|
func TestCopyAndOverflow(t *testing.T) {
|
|
cfg := config.Defaults()
|
|
cfg.Limits.InboundQueue = 1
|
|
c := New(cfg, events.NewStore(4, slog.New(slog.NewTextHandler(io.Discard, nil))))
|
|
p := &paho.Publish{Topic: Topic(cfg, "acks"), Payload: []byte("payload"), Properties: &paho.PublishProperties{CorrelationData: []byte("id"), MessageExpiry: paho.Uint32(5)}}
|
|
c.receive(paho.PublishReceived{Packet: p})
|
|
p.Payload[0] = 'X'
|
|
p.Properties.CorrelationData[0] = 'X'
|
|
*p.Properties.MessageExpiry = 99
|
|
c.receive(paho.PublishReceived{Packet: p})
|
|
if c.dropped.Load() != 1 {
|
|
t.Fatal("overflow not counted")
|
|
}
|
|
got := <-c.Incoming
|
|
if string(got.Payload) != "payload" || string(got.Properties.CorrelationData) != "id" || *got.Properties.MessageExpiry != 5 {
|
|
t.Fatal("callback input was aliased")
|
|
}
|
|
p.Topic = "production/topic"
|
|
c.receive(paho.PublishReceived{Packet: p})
|
|
if c.invalid.Load() != 1 {
|
|
t.Fatal("foreign topic not rejected")
|
|
}
|
|
}
|
|
func TestBoundedBackoff(t *testing.T) {
|
|
f := Backoff(time.Second, 30*time.Second)
|
|
for attempt := 0; attempt < 100; attempt++ {
|
|
for n := 0; n < 10; n++ {
|
|
d := f(attempt)
|
|
if d < time.Second || d > 30*time.Second {
|
|
t.Fatal("backoff outside bounds")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
func TestTLSNeverSkipsVerification(t *testing.T) {
|
|
cfg := config.Defaults()
|
|
tc, err := TLSConfig(cfg)
|
|
if err != nil || tc.InsecureSkipVerify || tc.MinVersion == 0 {
|
|
t.Fatal("unsafe TLS")
|
|
}
|
|
}
|
|
|
|
func TestObservedDeviceTopicIdentity(t *testing.T) {
|
|
cfg := config.Defaults()
|
|
cfg.Topics.ObserveDevices = "sim-001, esp32-lab-001,esp32-lab-001"
|
|
if len(ObservedDevices(cfg)) != 2 {
|
|
t.Fatal("duplicate subscriptions")
|
|
}
|
|
device, kind := Identity(cfg, "robot-market/test/v1/esp32-lab-001/telemetry")
|
|
if device != "esp32-lab-001" || kind != "telemetry" {
|
|
t.Fatal("device topic missing")
|
|
}
|
|
for _, topic := range []string{"robot-market/test/v1/other/telemetry", "production/esp32-lab-001/telemetry", "robot-market/test/v1/esp32-lab-001/unknown"} {
|
|
if _, kind := Identity(cfg, topic); kind != "" {
|
|
t.Fatal("unconfigured topic admitted")
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestRawObservationAdmitsNamespaceWithoutDeviceOrCategoryRestriction(t *testing.T) {
|
|
cfg := config.Defaults()
|
|
cfg.Topics.ObserveAll = true
|
|
c := New(cfg, events.NewStore(8, slog.New(slog.NewTextHandler(io.Discard, nil))))
|
|
for _, topic := range []string{cfg.Topics.Namespace, cfg.Topics.Namespace + "/custom", cfg.Topics.Namespace + "/new-device/custom-topic"} {
|
|
c.receive(paho.PublishReceived{Packet: &paho.Publish{Topic: topic, Payload: []byte("not JSON")}})
|
|
select {
|
|
case got := <-c.Incoming:
|
|
if got.Topic != topic {
|
|
t.Fatal("topic changed")
|
|
}
|
|
default:
|
|
t.Fatal("raw topic filtered")
|
|
}
|
|
}
|
|
c.receive(paho.PublishReceived{Packet: &paho.Publish{Topic: cfg.Topics.Namespace + "-other/device/custom"}})
|
|
if c.invalid.Load() != 1 {
|
|
t.Fatal("namespace boundary lost")
|
|
}
|
|
}
|