lingniu-vehicle-ingest/go/vehicle-gateway/internal/gateway/mqtt_client_test.go

package gateway

import (
	"context"
	"crypto/rand"
	"crypto/rsa"
	"crypto/x509"
	"crypto/x509/pkix"
	"encoding/pem"
	"log/slog"
	"math/big"
	"os"
	"path/filepath"
	"testing"
	"time"

	"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)
	}
}

func TestMQTTClientBuildOptionsLoadsTLSCertificates(t *testing.T) {
	dir := t.TempDir()
	caPath, certPath, keyPath := writeTestTLSMaterial(t, dir)
	client, err := NewMQTTClient(MQTTClientConfig{
		EndpointName:            "endpoint-a",
		Broker:                  "ssl://mqtt.example.test:8883",
		ClientID:                "test-client",
		Topics:                  []string{"/ytforward/shln/+"},
		QoS:                     1,
		Sink:                    &recordingSink{},
		Logger:                  slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
		TLSCACertPath:           caPath,
		TLSClientCertPath:       certPath,
		TLSClientKeyPath:        keyPath,
		TLSHostnameVerification: false,
		CleanSession:            true,
		KeepAlive:               20 * time.Second,
		ConnectTimeout:          10 * time.Second,
	})
	if err != nil {
		t.Fatalf("NewMQTTClient() error = %v", err)
	}

	opts, err := client.buildOptions(context.Background())
	if err != nil {
		t.Fatalf("buildOptions() error = %v", err)
	}
	if opts.TLSConfig == nil {
		t.Fatal("TLSConfig is nil")
	}
	if opts.TLSConfig.RootCAs == nil {
		t.Fatal("RootCAs is nil")
	}
	if len(opts.TLSConfig.Certificates) != 1 {
		t.Fatalf("client certificates = %d, want 1", len(opts.TLSConfig.Certificates))
	}
	if !opts.TLSConfig.InsecureSkipVerify {
		t.Fatal("InsecureSkipVerify should be true when hostname verification is disabled")
	}
	if !opts.CleanSession {
		t.Fatal("CleanSession should be true")
	}
	if opts.Order {
		t.Fatal("OrderMatters should be false so MQTT network handling is not blocked by Kafka/DB work")
	}
	if !opts.ResumeSubs {
		t.Fatal("ResumeSubs should be true to avoid subscribe failures during reconnect churn")
	}
	if got := opts.KeepAlive; got != 20 {
		t.Fatalf("KeepAlive = %d, want 20", got)
	}
	if got := opts.ConnectTimeout; got != 10*time.Second {
		t.Fatalf("ConnectTimeout = %v, want 10s", got)
	}
}

func writeTestTLSMaterial(t *testing.T, dir string) (string, string, string) {
	t.Helper()
	caKey, err := rsa.GenerateKey(rand.Reader, 2048)
	if err != nil {
		t.Fatalf("generate ca key: %v", err)
	}
	caTemplate := &x509.Certificate{
		SerialNumber:          big.NewInt(1),
		Subject:               pkix.Name{CommonName: "test-ca"},
		NotBefore:             time.Now().Add(-time.Hour),
		NotAfter:              time.Now().Add(time.Hour),
		KeyUsage:              x509.KeyUsageCertSign | x509.KeyUsageCRLSign,
		BasicConstraintsValid: true,
		IsCA:                  true,
	}
	caDER, err := x509.CreateCertificate(rand.Reader, caTemplate, caTemplate, &caKey.PublicKey, caKey)
	if err != nil {
		t.Fatalf("create ca cert: %v", err)
	}
	clientKey, err := rsa.GenerateKey(rand.Reader, 2048)
	if err != nil {
		t.Fatalf("generate client key: %v", err)
	}
	clientTemplate := &x509.Certificate{
		SerialNumber: big.NewInt(2),
		Subject:      pkix.Name{CommonName: "test-client"},
		NotBefore:    time.Now().Add(-time.Hour),
		NotAfter:     time.Now().Add(time.Hour),
		KeyUsage:     x509.KeyUsageDigitalSignature,
		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
	}
	clientDER, err := x509.CreateCertificate(rand.Reader, clientTemplate, caTemplate, &clientKey.PublicKey, caKey)
	if err != nil {
		t.Fatalf("create client cert: %v", err)
	}
	caPath := filepath.Join(dir, "ca.pem")
	certPath := filepath.Join(dir, "client.pem")
	keyPath := filepath.Join(dir, "client-key.pem")
	writePEM(t, caPath, "CERTIFICATE", caDER)
	writePEM(t, certPath, "CERTIFICATE", clientDER)
	keyDER := x509.MarshalPKCS1PrivateKey(clientKey)
	writePEM(t, keyPath, "RSA PRIVATE KEY", keyDER)
	return caPath, certPath, keyPath
}

func writePEM(t *testing.T, path, typ string, der []byte) {
	t.Helper()
	file, err := os.Create(path)
	if err != nil {
		t.Fatalf("create %s: %v", path, err)
	}
	defer file.Close()
	if err := pem.Encode(file, &pem.Block{Type: typ, Bytes: der}); err != nil {
		t.Fatalf("write pem %s: %v", path, err)
	}
}