207 lines
6.6 KiB
Go
207 lines
6.6 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"testing"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/segmentio/kafka-go"
|
|
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics"
|
|
)
|
|
|
|
func TestBridgeBatchWritesKafkaThenAcks(t *testing.T) {
|
|
env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"}
|
|
payload, err := json.Marshal(env)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
writer := &recordingBridgeWriter{}
|
|
acked := 0
|
|
|
|
err = bridgeBatch(context.Background(), nil, writer, []bridgeMessage{
|
|
{subject: "vehicle.raw.jt808.v1", data: payload, ack: func() error {
|
|
acked++
|
|
return nil
|
|
}},
|
|
}, map[string]string{"vehicle.raw.jt808.v1": "vehicle.raw.jt808.v1"})
|
|
if err != nil {
|
|
t.Fatalf("bridgeBatch() error = %v", err)
|
|
}
|
|
if len(writer.messages) != 1 {
|
|
t.Fatalf("kafka writes = %d, want 1", len(writer.messages))
|
|
}
|
|
if got, want := writer.messages[0].Topic, "vehicle.raw.jt808.v1"; got != want {
|
|
t.Fatalf("topic = %q, want %q", got, want)
|
|
}
|
|
if got, want := string(writer.messages[0].Key), "JT808:13307795425"; got != want {
|
|
t.Fatalf("key = %q, want %q", got, want)
|
|
}
|
|
if acked != 1 {
|
|
t.Fatalf("acks = %d, want 1", acked)
|
|
}
|
|
}
|
|
|
|
func TestBridgeBatchDoesNotAckWhenKafkaFails(t *testing.T) {
|
|
writer := &recordingBridgeWriter{err: errors.New("kafka unavailable")}
|
|
acked := 0
|
|
|
|
err := bridgeBatch(context.Background(), nil, writer, []bridgeMessage{
|
|
{subject: "vehicle.event.unified.v1", data: []byte(`{"protocol":"JT808"}`), ack: func() error {
|
|
acked++
|
|
return nil
|
|
}},
|
|
}, map[string]string{"vehicle.event.unified.v1": "vehicle.event.unified.v1"})
|
|
if err == nil {
|
|
t.Fatal("bridgeBatch() error is nil, want kafka error")
|
|
}
|
|
if acked != 0 {
|
|
t.Fatalf("acks = %d, want 0", acked)
|
|
}
|
|
}
|
|
|
|
func TestBridgeBatchRecordsMetrics(t *testing.T) {
|
|
env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"}
|
|
payload, err := json.Marshal(env)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
registry := metrics.NewRegistry()
|
|
writer := &recordingBridgeWriter{}
|
|
|
|
err = bridgeBatch(context.Background(), registry, writer, []bridgeMessage{
|
|
{subject: "vehicle.raw.jt808.v1", data: payload, ack: func() error { return nil }},
|
|
}, map[string]string{"vehicle.raw.jt808.v1": "vehicle.raw.jt808.v1"})
|
|
if err != nil {
|
|
t.Fatalf("bridgeBatch() error = %v", err)
|
|
}
|
|
text := registry.Render()
|
|
for _, want := range []string{
|
|
`vehicle_bridge_messages_total{status="received",subject="vehicle.raw.jt808.v1"} 1`,
|
|
`vehicle_bridge_kafka_writes_total{status="ok",topic="vehicle.raw.jt808.v1"} 1`,
|
|
`vehicle_bridge_nats_acks_total{status="ok",subject="vehicle.raw.jt808.v1"} 1`,
|
|
} {
|
|
if !strings.Contains(text, want) {
|
|
t.Fatalf("metrics missing %s:\n%s", want, text)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestBridgeBatchExposesPendingAndDurationMetrics(t *testing.T) {
|
|
env := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"}
|
|
payload, err := json.Marshal(env)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
registry := metrics.NewRegistry()
|
|
writer := &recordingBridgeWriter{
|
|
onWrite: func() {
|
|
text := registry.Render()
|
|
for _, want := range []string{
|
|
`vehicle_bridge_batch_pending_messages 2`,
|
|
`vehicle_bridge_batch_pending_kafka_messages 2`,
|
|
} {
|
|
if !strings.Contains(text, want) {
|
|
t.Fatalf("pending bridge metric missing %s during write:\n%s", want, text)
|
|
}
|
|
}
|
|
},
|
|
}
|
|
|
|
err = bridgeBatch(context.Background(), registry, writer, []bridgeMessage{
|
|
{subject: "vehicle.raw.go.gb32960.v1", data: payload, ack: func() error { return nil }},
|
|
{subject: "vehicle.raw.go.gb32960.v1", data: payload, ack: func() error { return nil }},
|
|
}, map[string]string{"vehicle.raw.go.gb32960.v1": "vehicle.raw.go.gb32960.v1"})
|
|
if err != nil {
|
|
t.Fatalf("bridgeBatch() error = %v", err)
|
|
}
|
|
|
|
text := registry.Render()
|
|
for _, want := range []string{
|
|
`vehicle_bridge_batch_pending_messages 0`,
|
|
`vehicle_bridge_batch_pending_kafka_messages 0`,
|
|
`vehicle_bridge_batch_duration_ms_histogram_bucket{le="+Inf",status="ok"} 1`,
|
|
`vehicle_bridge_batch_duration_ms_histogram_count{status="ok"} 1`,
|
|
`vehicle_bridge_batch_duration_ms_histogram_sum{status="ok"}`,
|
|
} {
|
|
if !strings.Contains(text, want) {
|
|
t.Fatalf("bridge batch metric missing %s:\n%s", want, text)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestRecordNATSConsumerInfoMetrics(t *testing.T) {
|
|
registry := metrics.NewRegistry()
|
|
cfg := config{NATSStream: "VEHICLE_INGEST", NATSDurable: "vehicle-kafka-bridge"}
|
|
|
|
recordNATSConsumerInfoMetrics(registry, cfg, &nats.ConsumerInfo{
|
|
NumPending: 17,
|
|
NumAckPending: 3,
|
|
NumWaiting: 2,
|
|
})
|
|
|
|
text := registry.Render()
|
|
for _, want := range []string{
|
|
`vehicle_bridge_nats_consumer_pending{consumer="vehicle-kafka-bridge",stream="VEHICLE_INGEST"} 17`,
|
|
`vehicle_bridge_nats_consumer_ack_pending{consumer="vehicle-kafka-bridge",stream="VEHICLE_INGEST"} 3`,
|
|
`vehicle_bridge_nats_consumer_waiting{consumer="vehicle-kafka-bridge",stream="VEHICLE_INGEST"} 2`,
|
|
} {
|
|
if !strings.Contains(text, want) {
|
|
t.Fatalf("metrics missing %s:\n%s", want, text)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestLoadConfigDefaultsToGoSubjectRoutes(t *testing.T) {
|
|
cfg := loadConfig()
|
|
|
|
want := map[string]string{
|
|
"vehicle.raw.go.gb32960.v1": "vehicle.raw.go.gb32960.v1",
|
|
"vehicle.raw.go.jt808.v1": "vehicle.raw.go.jt808.v1",
|
|
"vehicle.raw.go.yutong-mqtt.v1": "vehicle.raw.go.yutong-mqtt.v1",
|
|
"vehicle.fields.go.gb32960.v1": "vehicle.fields.go.gb32960.v1",
|
|
"vehicle.fields.go.jt808.v1": "vehicle.fields.go.jt808.v1",
|
|
"vehicle.fields.go.yutong-mqtt.v1": "vehicle.fields.go.yutong-mqtt.v1",
|
|
}
|
|
if len(cfg.Route) != len(want) {
|
|
t.Fatalf("route len = %d, want %d: %#v", len(cfg.Route), len(want), cfg.Route)
|
|
}
|
|
for subject, topic := range want {
|
|
if got := cfg.Route[subject]; got != topic {
|
|
t.Fatalf("route[%q] = %q, want %q; route=%#v", subject, got, topic, cfg.Route)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestLoadConfigIncludesUnifiedOnlyWhenExplicitlyConfigured(t *testing.T) {
|
|
t.Setenv("NATS_SUBJECT_UNIFIED", "vehicle.event.go.unified.v1")
|
|
t.Setenv("KAFKA_TOPIC_UNIFIED", "vehicle.event.go.unified.v1")
|
|
|
|
cfg := loadConfig()
|
|
|
|
if got := cfg.Route["vehicle.event.go.unified.v1"]; got != "vehicle.event.go.unified.v1" {
|
|
t.Fatalf("unified route = %q, route=%#v", got, cfg.Route)
|
|
}
|
|
}
|
|
|
|
type recordingBridgeWriter struct {
|
|
messages []kafka.Message
|
|
err error
|
|
onWrite func()
|
|
}
|
|
|
|
func (w *recordingBridgeWriter) WriteMessages(_ context.Context, messages ...kafka.Message) error {
|
|
if w.onWrite != nil {
|
|
w.onWrite()
|
|
}
|
|
if w.err != nil {
|
|
return w.err
|
|
}
|
|
w.messages = append(w.messages, messages...)
|
|
return nil
|
|
}
|