104 lines
3.0 KiB
Go
104 lines
3.0 KiB
Go
package simulator
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"log/slog"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/eclipse/paho.golang/paho"
|
|
"robot-market/test-mqtt-5/internal/backend"
|
|
"robot-market/test-mqtt-5/internal/config"
|
|
"robot-market/test-mqtt-5/internal/events"
|
|
"robot-market/test-mqtt-5/internal/mqtt"
|
|
)
|
|
|
|
func testSimulator() *Simulator {
|
|
cfg := config.Defaults()
|
|
return New(mqtt.New(cfg, events.NewStore(64, slog.New(slog.NewTextHandler(io.Discard, nil)))))
|
|
}
|
|
func commandPacket(s *Simulator, cmdName, id string) *paho.Publish {
|
|
cmd := backend.NewCommand("sim-001", cmdName, id, 15*time.Second)
|
|
p := mqtt.Packet(mqtt.Topic(s.Client.Config, "commands"), 1, false, cmd)
|
|
p.Properties.ResponseTopic = mqtt.Topic(s.Client.Config, "acks")
|
|
p.Properties.CorrelationData = []byte(id)
|
|
p.Properties.MessageExpiry = paho.Uint32(15)
|
|
return p
|
|
}
|
|
func TestDuplicateAndConflict(t *testing.T) {
|
|
s := testSimulator()
|
|
ctx := context.Background()
|
|
id := backend.ID()
|
|
p := commandPacket(s, "ping", id)
|
|
s.Handle(ctx, p)
|
|
s.Handle(ctx, p)
|
|
if len(s.cache) != 1 {
|
|
t.Fatal("duplicate not bounded")
|
|
}
|
|
s.CompleteDue(ctx, time.Now())
|
|
s.Handle(ctx, p)
|
|
s.CompleteDue(ctx, time.Now())
|
|
if s.Executions != 1 || !s.cache[id].complete {
|
|
t.Fatal("duplicate executed again")
|
|
}
|
|
conflict := commandPacket(s, "get_version", id)
|
|
s.Handle(ctx, conflict)
|
|
s.CompleteDue(ctx, time.Now())
|
|
if s.Executions != 1 || s.cache[id].command.Command != "ping" {
|
|
t.Fatal("conflicting reuse executed")
|
|
}
|
|
}
|
|
func TestInvalidMetadataAndCapacity(t *testing.T) {
|
|
for _, bad := range []string{"retained", "topic", "correlation", "expiry", "qos"} {
|
|
t.Run(bad, func(t *testing.T) {
|
|
s := testSimulator()
|
|
p := commandPacket(s, "ping", backend.ID())
|
|
switch bad {
|
|
case "retained":
|
|
p.Retain = true
|
|
case "topic":
|
|
p.Properties.ResponseTopic = "other/acks"
|
|
case "correlation":
|
|
p.Properties.CorrelationData = nil
|
|
case "expiry":
|
|
p.Properties.MessageExpiry = nil
|
|
case "qos":
|
|
p.QoS = 0
|
|
}
|
|
s.Handle(context.Background(), p)
|
|
if len(s.cache) != 0 {
|
|
t.Fatal("invalid command admitted")
|
|
}
|
|
})
|
|
}
|
|
s := testSimulator()
|
|
s.Client.Config.Limits.DedupEntries = 1
|
|
s.Client.Config.Simulator.AckDelay = time.Second
|
|
s.Handle(context.Background(), commandPacket(s, "ping", backend.ID()))
|
|
s.Handle(context.Background(), commandPacket(s, "ping", backend.ID()))
|
|
if len(s.cache) != 1 {
|
|
t.Fatal("cache capacity ignored")
|
|
}
|
|
s.CompleteDue(context.Background(), time.Now().Add(time.Minute))
|
|
if s.Executions != 0 {
|
|
t.Fatal("expired command executed")
|
|
}
|
|
s.purge(time.Now().Add(time.Hour))
|
|
if len(s.cache) != 0 {
|
|
t.Fatal("completed cache not purged")
|
|
}
|
|
}
|
|
|
|
func TestCancellationStopsAdmissionAndExecution(t *testing.T) {
|
|
s := testSimulator()
|
|
s.Handle(context.Background(), commandPacket(s, "ping", backend.ID()))
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
s.Handle(ctx, commandPacket(s, "ping", backend.ID()))
|
|
s.CompleteDue(ctx, time.Now().Add(time.Second))
|
|
if len(s.cache) != 1 || s.Executions != 0 {
|
|
t.Fatal("work admitted/executed after cancellation")
|
|
}
|
|
}
|