package gateway import ( "context" "log/slog" "testing" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" ) func TestMQTTClientHandleMessagePublishesRawAndUnified(t *testing.T) { sink := &recordingSink{} client, err := NewMQTTClient(MQTTClientConfig{ EndpointName: "endpoint-a", Broker: "tcp://127.0.0.1:1883", ClientID: "test-client", Topics: []string{"/ytforward/shln/+"}, Sink: sink, Logger: slog.New(slog.NewTextHandler(testWriter{t: t}, nil)), }) if err != nil { t.Fatalf("NewMQTTClient() error = %v", err) } client.handleMessage(context.Background(), "/ytforward/shln/dev1", []byte(`{ "device":"LTEST000000000001", "time":"20260413100000", "data":{"METER_SPEED":52.3,"TOTAL_MILEAGE":123456.7} }`)) if len(sink.raw) != 1 || len(sink.unified) != 1 { t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified)) } if sink.raw[0].Protocol != envelope.ProtocolYutongMQTT || sink.raw[0].VIN != "LTEST000000000001" { t.Fatalf("unexpected raw envelope: %#v", sink.raw[0]) } } func TestMQTTClientHandleBadPayloadPublishesOnlyRaw(t *testing.T) { sink := &recordingSink{} client, err := NewMQTTClient(MQTTClientConfig{ EndpointName: "endpoint-a", Broker: "tcp://127.0.0.1:1883", ClientID: "test-client", Topics: []string{"/ytforward/shln/+"}, Sink: sink, Logger: slog.New(slog.NewTextHandler(testWriter{t: t}, nil)), }) if err != nil { t.Fatalf("NewMQTTClient() error = %v", err) } client.handleMessage(context.Background(), "/ytforward/shln/bad", []byte("{bad-json")) if len(sink.raw) != 1 || len(sink.unified) != 0 { t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified)) } if sink.raw[0].ParseStatus != envelope.ParseBadFrame { t.Fatalf("parse status = %q", sink.raw[0].ParseStatus) } }