109 lines
3.9 KiB
Go
109 lines
3.9 KiB
Go
package backend
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/eclipse/paho.golang/paho"
|
|
"robot-market/test-mqtt-5/internal/events"
|
|
)
|
|
|
|
func TestReceiptRawPayloadRoutingAndCorrelation(t *testing.T) {
|
|
b := testBackend()
|
|
b.Client.Config.Topics.ObserveAll = true
|
|
for _, payload := range []string{"hello", "", "[1,2]", `{"device_id":"different","timestamp":"{{$isoTimestamp}}","message_id":"msg-1"}`} {
|
|
reply := b.receipt(&paho.Publish{Topic: "robot-market/test/v1/real-device/telemetry", Payload: []byte(payload)})
|
|
if reply == nil || reply.Topic != "robot-market/test/v1/real-device/acks" || reply.Retain || reply.QoS != 1 {
|
|
t.Fatal("missing/routed/retained reply")
|
|
}
|
|
var ack ReceiptAck
|
|
if json.Unmarshal(reply.Payload, &ack) != nil || !ack.OK || ack.Type != "receipt_ack" || ack.PayloadBytes != len(payload) || len(ack.PayloadSHA256) != 64 {
|
|
t.Fatal("invalid receipt")
|
|
}
|
|
}
|
|
p := &paho.Publish{Topic: "robot-market/test/v1/device/telemetry", Payload: []byte("x"), Properties: &paho.PublishProperties{ResponseTopic: "robot-market/test/v1/replies/custom", CorrelationData: []byte{0, 255, 1}}}
|
|
reply := b.receipt(p)
|
|
p.Properties.CorrelationData[0] = 99
|
|
if reply.Topic != "robot-market/test/v1/replies/custom" || reply.Properties.CorrelationData[0] != 0 || reply.Properties.ResponseTopic != "" {
|
|
t.Fatal("native response/correlation lost or aliased")
|
|
}
|
|
p.Properties.ResponseTopic = "outside/test"
|
|
if b.receipt(p).Topic != "robot-market/test/v1/device/acks" {
|
|
t.Fatal("unusable response topic should use default receipt topic")
|
|
}
|
|
}
|
|
|
|
func TestReceiptLoopAndRetainedReplayPrevention(t *testing.T) {
|
|
b := testBackend()
|
|
b.Client.Config.Topics.ObserveAll = true
|
|
for _, topic := range []string{"acks", "commands", "receipts"} {
|
|
if b.receipt(&paho.Publish{Topic: "robot-market/test/v1/device/" + topic, Payload: []byte("hello")}) != nil {
|
|
t.Fatal("ACK loop possible")
|
|
}
|
|
}
|
|
for _, p := range []*paho.Publish{
|
|
{Topic: "robot-market/test/v1/device/status", Retain: true, Payload: []byte("online")},
|
|
{Topic: "robot-market/test/v1/custom/topic", Payload: []byte(`{"type":"receipt_ack","ok":true}`)},
|
|
{Topic: "outside/device/topic", Payload: []byte("hello")},
|
|
} {
|
|
if b.receipt(p) != nil {
|
|
t.Fatal("reply created for replay, receipt or external topic")
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestReceiptDuplicateDeliveryRepliesWithoutCommandExecution(t *testing.T) {
|
|
b := testBackend()
|
|
b.Client.Config.Topics.ObserveAll = true
|
|
b.Client.Config.Receipts.Enabled = true
|
|
p := &paho.Publish{Topic: "robot-market/test/v1/device/telemetry", Payload: []byte("same message")}
|
|
b.Handle(p)
|
|
b.Handle(p)
|
|
if len(b.receipts) != 2 || len(b.pending) != 0 {
|
|
t.Fatal("duplicate receipt handling must not execute commands")
|
|
}
|
|
}
|
|
|
|
func TestReceiptQueueFailureAndShutdown(t *testing.T) {
|
|
b := testBackend()
|
|
b.Client.Config.Topics.ObserveAll = true
|
|
b.Client.Config.Receipts.Enabled = true
|
|
b.receipts = make(chan *paho.Publish, 1)
|
|
p := &paho.Publish{Topic: "robot-market/test/v1/device/telemetry", Payload: []byte("hello")}
|
|
b.Handle(p)
|
|
b.Handle(p)
|
|
_, counts := b.Client.Events.Snapshot()
|
|
if counts[events.Kind("receipt_ack_dropped")] != 1 {
|
|
t.Fatal("queue overflow invisible")
|
|
}
|
|
b.PublishReceipt = func(context.Context, *paho.Publish) error { return errors.New("test disconnect") }
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() { b.receiptWorker(ctx); close(done) }()
|
|
deadline := time.Now().Add(time.Second)
|
|
for {
|
|
_, counts = b.Client.Events.Snapshot()
|
|
if counts[events.Kind("receipt_ack_failed")] == 1 {
|
|
break
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("failure not recorded")
|
|
}
|
|
time.Sleep(time.Millisecond)
|
|
}
|
|
cancel()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("worker shutdown blocked")
|
|
}
|
|
history, _ := b.Client.Events.Snapshot()
|
|
if !strings.Contains(history[len(history)-1].Outcome, "not confirmed") {
|
|
t.Fatal("failure claims success")
|
|
}
|
|
}
|