DriverVault/API Server/internal/plugins/builtin/ankersolix/cloudmqtt_test.go

package ankersolix

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"strings"
	"testing"
	"time"

	"drivervault/apiserver/internal/mqtt"
)

func TestMqttCredentialsAddressAndAppName(t *testing.T) {
	c := mqttCredentials{EndpointAddr: "aiot-mqtt-eu.anker.com"}
	if got := c.address(); got != "aiot-mqtt-eu.anker.com:8883" {
		t.Errorf("address = %q, want the TLS port appended", got)
	}
	// A response that already carries a port must not have another appended.
	c.EndpointAddr = "aiot-mqtt-eu.anker.com:8884"
	if got := c.address(); got != "aiot-mqtt-eu.anker.com:8884" {
		t.Errorf("address = %q, want the endpoint's own port kept", got)
	}
	if got := c.appName(); got != "anker_power" {
		t.Errorf("appName = %q, want the default when the cloud sent none", got)
	}
	c.AppName = "anker_charging"
	if got := c.appName(); got != "anker_charging" {
		t.Errorf("appName = %q, want the cloud's own value", got)
	}
}

func TestMqttCredentialsNeedCertificateAndKey(t *testing.T) {
	full := mqttCredentials{EndpointAddr: "host", CertificatePE: "cert", PrivateKey: "key"}
	if !full.valid() {
		t.Error("complete credentials were rejected")
	}
	for _, c := range []mqttCredentials{
		{CertificatePE: "cert", PrivateKey: "key"},
		{EndpointAddr: "host", PrivateKey: "key"},
		{EndpointAddr: "host", CertificatePE: "cert"},
	} {
		if c.valid() {
			t.Errorf("credentials missing a field were accepted: %+v", c)
		}
	}
}

func TestTopicsAddressOneCharger(t *testing.T) {
	c := mqttCredentials{AppName: "anker_power"}
	if got := commandTopic(c, "A5191", "SN123"); got != "cmd/anker_power/A5191/SN123/req" {
		t.Errorf("commandTopic = %q", got)
	}
	if got := dataTopic(c, "A5191", "SN123"); got != "dt/anker_power/A5191/SN123/#" {
		t.Errorf("dataTopic = %q", got)
	}
}

// envelope builds a message shaped like the ones the cloud delivers: JSON, whose
// payload is itself a JSON string, whose data field is a base64 device frame.
func envelope(t *testing.T, sn string, frame []byte) mqtt.Message {
	t.Helper()
	inner, err := json.Marshal(map[string]any{
		"device_sn": sn,
		"data":      base64.StdEncoding.EncodeToString(frame),
	})
	if err != nil {
		t.Fatal(err)
	}
	outer, err := json.Marshal(map[string]any{
		"head":    map[string]any{"device_sn": sn, "timestamp": 1756813256},
		"payload": string(inner),
	})
	if err != nil {
		t.Fatal(err)
	}
	return mqtt.Message{Topic: "dt/anker_power/A5191/" + sn + "/param_info", Payload: outer}
}

func TestParseEnvelopeUnwrapsTheDeviceFrame(t *testing.T) {
	frame := buildInbound(t, msgEVTelemetry, field(0xbb, typeUint8, 0x02))
	sn, data, ok := parseEnvelope(envelope(t, "SN123", frame))
	if !ok {
		t.Fatal("a well-formed envelope was rejected")
	}
	if sn != "SN123" {
		t.Errorf("serial = %q, want SN123", sn)
	}
	if string(data) != string(frame) {
		t.Errorf("frame came back changed")
	}
}

// The serial is not always repeated in the payload; the topic carries it too,
// and a message we cannot attribute to a charger must be dropped rather than
// folded into some other charger's state.
func TestParseEnvelopeFallsBackToTheTopic(t *testing.T) {
	inner, _ := json.Marshal(map[string]any{"data": base64.StdEncoding.EncodeToString([]byte{1, 2, 3})})
	outer, _ := json.Marshal(map[string]any{"payload": string(inner)})
	sn, _, ok := parseEnvelope(mqtt.Message{Topic: "dt/anker_power/A5191/SN999/param_info", Payload: outer})
	if !ok || sn != "SN999" {
		t.Errorf("serial = %q (ok=%v), want SN999 from the topic", sn, ok)
	}

	for _, bad := range []mqtt.Message{
		{Topic: "dt/a/b/SN/x", Payload: []byte("not json")},
		{Topic: "dt/a/b/SN/x", Payload: []byte(`{"payload":"not json"}`)},
		{Topic: "dt/a/b/SN/x", Payload: []byte(`{"payload":"{\"device_sn\":\"SN\"}"}`)},   // no data
		{Topic: "dt/a/b/SN/x", Payload: []byte(`{"payload":"{\"data\":\"!!not b64\"}"}`)}, // undecodable
		{Topic: "short", Payload: []byte(`{"payload":"{\"data\":\"AQID\"}"}`)},            // no serial anywhere
	} {
		if _, _, ok := parseEnvelope(bad); ok {
			t.Errorf("a malformed envelope was accepted: %s", bad.Payload)
		}
	}
}

// A charger's state is assembled from two message types that arrive at different
// times: telemetry must not erase the settings that came with the last command,
// and vice versa.
func TestIngestMergesTelemetryAndSettings(t *testing.T) {
	c := &mqttConn{subs: map[string]bool{}, devices: map[string]*deviceState{}, shutdown: make(chan struct{})}

	c.ingest(envelope(t, "SN1", buildInbound(t, msgEVParams, field(0xa8, typeInt16LE, 0x40, 0x01))))
	c.ingest(envelope(t, "SN1", buildInbound(t, msgEVTelemetry, field(0xbb, typeUint8, 0x02))))

	values, telemetryAt, settingsAt, _ := c.snapshotOf("SN1")
	if v, _ := values["maxCurrentSetA"].(float64); v != 32 {
		t.Errorf("the settings value was lost when telemetry arrived: %v", values["maxCurrentSetA"])
	}
	if v, _ := values["status"].(float64); v != 2 {
		t.Errorf("status = %v, want 2", values["status"])
	}
	if telemetryAt.IsZero() || settingsAt.IsZero() {
		t.Errorf("both halves should be timestamped: telemetry %v, settings %v", telemetryAt, settingsAt)
	}
	if !telemetryAt.After(settingsAt) && !telemetryAt.Equal(settingsAt) {
		t.Errorf("telemetry arrived second but is stamped earlier")
	}

	// A message from a charger we have no map for leaves the state untouched.
	c.ingest(mqtt.Message{Topic: "dt/a/b/SN1/x", Payload: []byte("rubbish")})
	after, _, _, _ := c.snapshotOf("SN1")
	if len(after) != len(values) {
		t.Errorf("an unreadable message changed the charger's state")
	}
}

// A message type we have no map for still carries its fields into the state,
// but it must not pass for a settings report: settingsAt is what a control
// command waits on to say the charger acknowledged it, and a frame we cannot
// read is not an acknowledgement.
func TestIngestOfAnUnmappedMessageKeepsFieldsButNotTheStamp(t *testing.T) {
	c := &mqttConn{subs: map[string]bool{}, devices: map[string]*deviceState{}, shutdown: make(chan struct{})}

	c.ingest(envelope(t, "SN1", buildInbound(t, "0400", field(0xa2, typeUint8, 0x01))))

	values, telemetryAt, settingsAt, _ := c.snapshotOf("SN1")
	if v, _ := values["0400.a2"].(float64); v != 1 {
		t.Errorf("values = %v, want the unnamed field kept as 0400.a2", values)
	}
	if !settingsAt.IsZero() || !telemetryAt.IsZero() {
		t.Errorf("an unmapped message stamped the state: telemetry %v, settings %v", telemetryAt, settingsAt)
	}
}

func TestWaitForReturnsWhenTheStateArrives(t *testing.T) {
	c := &mqttConn{subs: map[string]bool{}, devices: map[string]*deviceState{}, shutdown: make(chan struct{})}
	go func() {
		time.Sleep(20 * time.Millisecond)
		c.ingest(envelope(t, "SN1", buildInbound(t, msgEVTelemetry, field(0xbb, typeUint8, 0x02))))
	}()
	ok, err := c.waitFor(context.Background(), "SN1", func(st *deviceState) bool {
		return !st.telemetryAt.IsZero()
	}, 2*time.Second)
	if err != nil || !ok {
		t.Fatalf("waitFor = %v, %v; want it to see the message", ok, err)
	}
}

func TestWaitForGivesUpAndReportsADeadConnection(t *testing.T) {
	c := &mqttConn{subs: map[string]bool{}, devices: map[string]*deviceState{}, shutdown: make(chan struct{})}
	ok, err := c.waitFor(context.Background(), "SN1", func(*deviceState) bool { return false }, 30*time.Millisecond)
	if err != nil || ok {
		t.Errorf("waitFor = %v, %v; want a quiet timeout", ok, err)
	}

	// A connection that dies while a caller is waiting must wake it with the
	// reason rather than making it sit out the whole timeout.
	c2 := &mqttConn{subs: map[string]bool{}, devices: map[string]*deviceState{}, shutdown: make(chan struct{})}
	go func() {
		time.Sleep(20 * time.Millisecond)
		c2.fail(mqtt.ErrClosed)
	}()
	start := time.Now()
	if _, err := c2.waitFor(context.Background(), "SN1", func(*deviceState) bool { return false }, 5*time.Second); err == nil {
		t.Error("a dropped connection ended the wait without an error")
	}
	if time.Since(start) > time.Second {
		t.Error("the waiter was not woken when the connection dropped")
	}
}

// A command that cannot be sent should be refused before it costs a sign-in, a
// certificate fetch and a broker connection — so the check runs on a plugin with
// no session at all.
func TestMqttCommandValidatesBeforeReachingTheCloud(t *testing.T) {
	p := &Plugin{}
	for _, tc := range []struct {
		command string
		amps    float64
		want    string
	}{
		{"reboot", 0, "not a cloud command"},
		{"limit", 3, "below the charger's 6 A floor"},
		{"limit", 40, "outside the charger's range"},
	} {
		_, err := p.mqttCommand(context.Background(), "SN1", tc.command, tc.amps)
		if err == nil {
			t.Errorf("%s(%v) was accepted", tc.command, tc.amps)
			continue
		}
		if !strings.Contains(err.Error(), tc.want) {
			t.Errorf("%s(%v) failed with %q, want it to mention %q", tc.command, tc.amps, err, tc.want)
		}
	}

	// A valid command gets past validation and fails on the missing session
	// instead, which is what proves the order.
	if _, err := p.mqttCommand(context.Background(), "SN1", "start", 0); err == nil ||
		!strings.Contains(err.Error(), "not initialised") {
		t.Errorf("start failed with %v, want it to reach the session lookup", err)
	}
}

func TestMqttModeValuesMatchTheChargerEnum(t *testing.T) {
	want := map[string]uint8{
		modeStartCharge: 1, modeStopCharge: 2, modeSkipDelay: 3, modeBoostCharge: 4,
	}
	for mode, v := range want {
		if mqttModeValues[mode] != v {
			t.Errorf("%s = %d, want %d", mode, mqttModeValues[mode], v)
		}
	}
	// Every accepted command name must resolve to one of those modes, or the
	// endpoint can offer a name the transport cannot send.
	for name, mode := range mqttCommands {
		if _, ok := mqttModeValues[mode]; !ok {
			t.Errorf("command %q maps to %q, which is not a charger mode", name, mode)
		}
	}
}

func TestClientIDDoesNotCollideWithTheApp(t *testing.T) {
	creds := mqttCredentials{ThingName: "abc-anker_power"}
	a, b := clientIDFor(creds), clientIDFor(creds)
	if !strings.HasPrefix(a, "abc-anker_power_") {
		t.Errorf("client id %q does not carry the account's thing name", a)
	}
	if a == b {
		t.Error("two connections were given the same client id, which would evict each other")
	}
	// With no thing name the user id stands in, so the id is still account-scoped.
	if got := clientIDFor(mqttCredentials{UserID: "u1"}); !strings.HasPrefix(got, "u1_") {
		t.Errorf("client id %q does not fall back to the user id", got)
	}
}

// A charger that goes quiet long enough to be written off is not written off for
// good. The miss counter decides whether a status read waits for the settings
// frame at all, so a counter that only ever rose meant one early patch of
// silence cost every later read its settings — which is the half the settings
// card is drawn from.
func TestAnsweringClearsTheStatusRequestMisses(t *testing.T) {
	c := &mqttConn{subs: map[string]bool{}, devices: map[string]*deviceState{}, shutdown: make(chan struct{})}

	// Three unanswered requests: the reads stop paying the wait.
	c.ingest(envelope(t, "SN1", buildInbound(t, msgEVTelemetry, field(0xbb, typeUint8, 0x02))))
	for i := 0; i < statusReqTries; i++ {
		c.noteStatusMiss("SN1")
	}
	if c.statusReqAnswered("SN1") {
		t.Fatalf("after %d misses the charger should not be waited for", statusReqTries)
	}

	// Then it answers one.
	c.ingest(envelope(t, "SN1", buildInbound(t, msgEVParams, field(0xa8, typeInt16LE, 0x40, 0x01))))
	if !c.statusReqAnswered("SN1") {
		t.Error("a charger that reported its settings is still being written off")
	}
}