package backend import ( "context" "io" "log/slog" "testing" "time" "github.com/eclipse/paho.golang/paho" "robot-market/test-mqtt-5/internal/config" "robot-market/test-mqtt-5/internal/contracts" "robot-market/test-mqtt-5/internal/events" "robot-market/test-mqtt-5/internal/mqtt" ) func testBackend() *Backend { c := config.Defaults() return New(mqtt.New(c, events.NewStore(32, slog.New(slog.NewTextHandler(io.Discard, nil))))) } func TestNativeCorrelationAndResult(t *testing.T) { for _, bad := range []string{"topic", "device", "correlation", "retained", "response", "result", "expired"} { t.Run(bad, func(t *testing.T) { b := testBackend() cmd := NewCommand("sim-001", "ping", ID(), 15*time.Second) p, err := b.register(cmd) if err != nil { t.Fatal(err) } yes := true ack := contracts.Ack{Envelope: cmd.Envelope, RequestID: cmd.RequestID, Timestamp: Timestamp(time.Now()), Stage: "completed", OK: &yes, Result: &contracts.Result{Pong: &yes}} if bad == "device" { ack.DeviceID = "other" } if bad == "result" { ack.Result = &contracts.Result{Version: "1.0.0"} } packet := mqtt.Packet(mqtt.Topic(b.Client.Config, "acks"), 1, false, ack) packet.Properties.CorrelationData = []byte(cmd.RequestID) switch bad { case "topic": packet.Topic = "robot-market/test/v1/other/acks" case "correlation": packet.Properties.CorrelationData = []byte("wrong") case "retained": packet.Retain = true case "response": packet.Properties.ResponseTopic = "unexpected" case "expired": p.expires = time.Now().Add(-time.Second) } b.Handle(packet) select { case <-p.result: t.Fatal("invalid ACK completed request") default: } }) } b := testBackend() cmd := NewCommand("sim-001", "ping", ID(), 15*time.Second) p, _ := b.register(cmd) yes := true ack := contracts.Ack{Envelope: cmd.Envelope, RequestID: cmd.RequestID, Timestamp: Timestamp(time.Now()), Stage: "completed", OK: &yes, Result: &contracts.Result{Pong: &yes}} packet := mqtt.Packet(mqtt.Topic(b.Client.Config, "acks"), 1, false, ack) packet.Properties.CorrelationData = []byte(cmd.RequestID) // Completion may precede accepted. A subsequent accepted/duplicate cannot regress it. b.Handle(packet) ack.Stage = "accepted" ack.Result = nil accepted := mqtt.Packet(packet.Topic, 1, false, ack) accepted.Properties.CorrelationData = []byte(cmd.RequestID) b.Handle(accepted) b.Handle(packet) select { case result := <-p.result: if result.Err != nil || result.Ack.Stage != "completed" { t.Fatal("wrong completion") } default: t.Fatal("valid completion ignored") } select { case <-p.result: t.Fatal("completion delivered twice") default: } } func TestPendingLimitsAndDisconnect(t *testing.T) { b := testBackend() b.Client.Config.Limits.PendingCommands = 1 cmd := NewCommand("sim-001", "ping", ID(), 15*time.Second) p, _ := b.register(cmd) if _, err := b.register(NewCommand("sim-001", "ping", ID(), 15*time.Second)); err == nil { t.Fatal("capacity ignored") } b.Disconnect() b.Disconnect() result := <-p.result if result.Err == nil { t.Fatal("disconnect claimed success") } if _, err := b.Command(context.Background(), cmd); err == nil { t.Fatal("disconnected command admitted") } } func TestObserveDeviceAndInvalidPayloadDiagnostics(t *testing.T) { b := testBackend() b.Client.Config.Topics.ObserveDevices = "sim-001,esp32-lab-001" b.Client.Config.Logging.RawPayloads = true payload := []byte(`{"schema_version":1,"device_id":"esp32-lab-001","message_id":"real-device","boot_id":"boot-1","sequence":1,"timestamp":"2026-10-07T00:00:00Z","readings":{"uptime_s":12}}`) b.Handle(&paho.Publish{Topic: "robot-market/test/v1/esp32-lab-001/telemetry", Payload: payload}) b.Handle(&paho.Publish{Topic: "robot-market/test/v1/esp32-lab-001/telemetry", Payload: []byte("hello from esp32")}) history, counts := b.Client.Events.Snapshot() if counts[events.MessageReceived] != 1 || counts[events.ValidationFailed] != 1 { t.Fatal("device receipt and contract failure must be distinct") } if history[0].DeviceID != "esp32-lab-001" || history[0].Payload.Text != string(payload) { t.Fatal("ESP32 identity/payload was lost") } if history[1].Payload.Text != "hello from esp32" || history[1].Outcome == "" { t.Fatal("invalid external payload needs diagnostic visibility") } b.Client.Config.Logging.RawPayloads = false b.Handle(&paho.Publish{Topic: "robot-market/test/v1/esp32-lab-001/telemetry", Payload: []byte("private invalid input")}) history, _ = b.Client.Events.Snapshot() if !history[2].Payload.Omitted || history[2].Payload.Text != "" { t.Fatal("raw payload display must require explicit opt-in") } } func TestOtherObservedDeviceCannotCompletePendingCommand(t *testing.T) { b := testBackend() b.Client.Config.Topics.ObserveDevices = "esp32-lab-001" cmd := NewCommand("sim-001", "ping", ID(), 15*time.Second) pending, _ := b.register(cmd) yes := true ack := contracts.Ack{Envelope: contracts.Envelope{SchemaVersion: 1, DeviceID: "esp32-lab-001"}, RequestID: cmd.RequestID, Timestamp: Timestamp(time.Now()), Stage: "completed", OK: &yes, Result: &contracts.Result{Pong: &yes}} packet := mqtt.Packet("robot-market/test/v1/esp32-lab-001/acks", 1, false, ack) packet.Properties.CorrelationData = []byte(cmd.RequestID) b.Handle(packet) select { case <-pending.result: t.Fatal("another device completed an own command") default: } } func TestRawObservationAcceptsArbitraryPayloadsAndTopics(t *testing.T) { b := testBackend() b.Client.Config.Topics.ObserveAll = true for _, payload := range []string{"hello", `{"device_id":"different","timestamp":"{{$isoTimestamp}}"}`, "", "[1,2,3]"} { b.Handle(&paho.Publish{Topic: "robot-market/test/v1/arbitrary/custom-topic", Payload: []byte(payload)}) } history, counts := b.Client.Events.Snapshot() if counts[events.MessageReceived] != 4 || counts[events.ValidationFailed] != 0 { t.Fatal("raw observation incorrectly validated a payload") } if history[1].Payload.Text != `{"device_id":"different","timestamp":"{{$isoTimestamp}}"}` || history[1].DeviceID != "arbitrary" { t.Fatal("raw observation lost payload or topic metadata") } }