104 lines
3.1 KiB
Go
104 lines
3.1 KiB
Go
package eventbus
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"reflect"
|
|
"testing"
|
|
|
|
"github.com/segmentio/kafka-go"
|
|
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
|
)
|
|
|
|
func TestKafkaSinkRoutesRawTopics(t *testing.T) {
|
|
writer := &recordingWriter{}
|
|
sink := newKafkaSinkWithWriter(writer, KafkaConfig{})
|
|
|
|
env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"}
|
|
if err := sink.PublishRaw(context.Background(), env); err != nil {
|
|
t.Fatalf("PublishRaw() error = %v", err)
|
|
}
|
|
if len(writer.messages) != 1 {
|
|
t.Fatalf("messages = %d", len(writer.messages))
|
|
}
|
|
msg := writer.messages[0]
|
|
if msg.Topic != "vehicle.raw.jt808.v1" {
|
|
t.Fatalf("topic = %q", msg.Topic)
|
|
}
|
|
if string(msg.Key) != "JT808:13307795425" {
|
|
t.Fatalf("key = %q", string(msg.Key))
|
|
}
|
|
var decoded envelope.FrameEnvelope
|
|
if err := json.Unmarshal(msg.Value, &decoded); err != nil {
|
|
t.Fatalf("json.Unmarshal() error = %v", err)
|
|
}
|
|
if decoded.EventID == "" || decoded.ParseStatus != envelope.ParseOK {
|
|
t.Fatalf("unexpected payload defaults: %#v", decoded)
|
|
}
|
|
}
|
|
|
|
func TestKafkaSinkRoutesUnifiedTopic(t *testing.T) {
|
|
writer := &recordingWriter{}
|
|
sink := newKafkaSinkWithWriter(writer, KafkaConfig{UnifiedTopic: "custom.unified"})
|
|
|
|
env := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "LNBVIN00000000001", MessageID: "0x02"}
|
|
if err := sink.PublishUnified(context.Background(), env); err != nil {
|
|
t.Fatalf("PublishUnified() error = %v", err)
|
|
}
|
|
if writer.messages[0].Topic != "custom.unified" {
|
|
t.Fatalf("topic = %q", writer.messages[0].Topic)
|
|
}
|
|
if string(writer.messages[0].Key) != "LNBVIN00000000001" {
|
|
t.Fatalf("key = %q", string(writer.messages[0].Key))
|
|
}
|
|
}
|
|
|
|
func TestKafkaSinkRejectsUnknownProtocol(t *testing.T) {
|
|
writer := &recordingWriter{}
|
|
sink := newKafkaSinkWithWriter(writer, KafkaConfig{})
|
|
|
|
err := sink.PublishRaw(context.Background(), envelope.FrameEnvelope{Protocol: envelope.Protocol("UNKNOWN")})
|
|
if err == nil {
|
|
t.Fatal("expected unknown protocol error")
|
|
}
|
|
}
|
|
|
|
func TestNewKafkaSinkUsesProductionDeliveryGuarantees(t *testing.T) {
|
|
sink, err := NewKafkaSink(KafkaConfig{Brokers: []string{"127.0.0.1:9092"}})
|
|
if err != nil {
|
|
t.Fatalf("NewKafkaSink() error = %v", err)
|
|
}
|
|
defer sink.Close()
|
|
|
|
writer, ok := sink.writer.(*kafka.Writer)
|
|
if !ok {
|
|
t.Fatalf("writer type = %s", reflect.TypeOf(sink.writer))
|
|
}
|
|
if writer.RequiredAcks != kafka.RequireAll {
|
|
t.Fatalf("RequiredAcks = %v, want %v", writer.RequiredAcks, kafka.RequireAll)
|
|
}
|
|
if writer.Async {
|
|
t.Fatal("writer must publish synchronously so Kafka errors propagate")
|
|
}
|
|
if writer.AllowAutoTopicCreation {
|
|
t.Fatal("writer must not auto-create production topics")
|
|
}
|
|
if _, ok := writer.Balancer.(*kafka.Hash); !ok {
|
|
t.Fatalf("balancer type = %s, want *kafka.Hash", reflect.TypeOf(writer.Balancer))
|
|
}
|
|
}
|
|
|
|
type recordingWriter struct {
|
|
messages []kafka.Message
|
|
}
|
|
|
|
func (w *recordingWriter) WriteMessages(_ context.Context, messages ...kafka.Message) error {
|
|
w.messages = append(w.messages, messages...)
|
|
return nil
|
|
}
|
|
|
|
func (w *recordingWriter) Close() error {
|
|
return nil
|
|
}
|