DriverVault/API Server/internal/plugins/builtin/greencell/greencell_test.go

package greencell

import (
	"bufio"
	"context"
	"encoding/binary"
	"encoding/json"
	"io"
	"net"
	"strings"
	"sync"
	"testing"
	"time"

	"drivervault/apiserver/internal/plugins"
)

// ---- descriptor & registration -----------------------------------------------

func TestDescriptor(t *testing.T) {
	d := (&Plugin{}).Descriptor()
	if d.Name != "greencell" {
		t.Fatalf("name = %q, want greencell", d.Name)
	}
	if d.Kind != plugins.KindBuiltin {
		t.Fatalf("kind = %q, want builtin", d.Kind)
	}
	if len(d.Capabilities) == 0 {
		t.Fatal("expected capabilities")
	}

	fields := map[string]plugins.ConfigField{}
	for _, f := range d.ConfigFields {
		fields[f.Key] = f
	}
	for _, k := range []string{"host", "port", "tls", "username", "password", "serial", "commandTopic", "timeout"} {
		if _, ok := fields[k]; !ok {
			t.Errorf("config field %q should be present", k)
		}
	}
	// Nothing is required at the global layer — the per-user cascade may supply
	// the broker instead, exactly as it does for the other connectors.
	for k, f := range fields {
		if f.Required {
			t.Errorf("config field %q must not be required at the global layer", k)
		}
	}
	if !fields["password"].Secret {
		t.Error("the broker password must be marked secret")
	}
	if fields["username"].Secret {
		t.Error("the broker username is not a secret")
	}
	if fields["tls"].Type != "select" {
		t.Errorf("tls type = %q, want select", fields["tls"].Type)
	}
	if fields["port"].Default != defaultPort {
		t.Errorf("port default = %q, want %q", fields["port"].Default, defaultPort)
	}
}

func TestRegistered(t *testing.T) {
	// init() must have registered the factory; registering twice panics, so a
	// duplicate name would surface at startup rather than here.
	defer func() {
		if r := recover(); r == nil {
			t.Fatal("expected a panic on duplicate registration, meaning the plugin registered itself")
		}
	}()
	plugins.Register("greencell", func() plugins.Plugin { return &Plugin{} })
}

// ---- config ------------------------------------------------------------------

func TestInitBuildsBrokerAddress(t *testing.T) {
	cases := []struct {
		name    string
		config  map[string]string
		address string
		tls     bool
	}{
		{"explicit port", map[string]string{"host": "10.2.1.10", "port": "1884"}, "10.2.1.10:1884", false},
		{"default plain port", map[string]string{"host": "mqtt.local"}, "mqtt.local:1883", false},
		{"default tls port", map[string]string{"host": "mqtt.local", "tls": "on"}, "mqtt.local:8883", true},
		{"tls with explicit port", map[string]string{"host": "mqtt.local", "tls": "true", "port": "9001"}, "mqtt.local:9001", true},
		{"ipv6 host", map[string]string{"host": "fd00::1", "port": "1883"}, "[fd00::1]:1883", false},
		{"no host", map[string]string{}, "", false},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			var p Plugin
			if err := p.Init(context.Background(), tc.config); err != nil {
				t.Fatalf("Init: %v", err)
			}
			if p.address != tc.address {
				t.Errorf("address = %q, want %q", p.address, tc.address)
			}
			if p.useTLS != tc.tls {
				t.Errorf("useTLS = %v, want %v", p.useTLS, tc.tls)
			}
		})
	}
}

func TestSnapshotWithoutBrokerIsAnError(t *testing.T) {
	var p Plugin
	_ = p.Init(context.Background(), map[string]string{})
	if _, _, err := p.snapshot(); err == nil {
		t.Fatal("expected an error when no broker is configured")
	}
}

func TestParseTimeout(t *testing.T) {
	cases := map[string]time.Duration{
		"":       defaultTimeout,
		"abc":    defaultTimeout,
		"0":      defaultTimeout,
		"-5":     defaultTimeout,
		" 20 ":   20 * time.Second,
		"1":      minTimeout,
		"600":    maxTimeout, // clamped
		"999999": maxTimeout,
	}
	for in, want := range cases {
		if got := parseTimeout(in); got != want {
			t.Errorf("parseTimeout(%q) = %v, want %v", in, got, want)
		}
	}
}

func TestTruthy(t *testing.T) {
	for _, v := range []string{"on", "ON", "true", "yes", "1", " tls "} {
		if !truthy(v) {
			t.Errorf("truthy(%q) = false", v)
		}
	}
	for _, v := range []string{"", "off", "false", "no", "0", "maybe"} {
		if truthy(v) {
			t.Errorf("truthy(%q) = true", v)
		}
	}
}

// ---- serials & announcements -------------------------------------------------

func TestDeviceModel(t *testing.T) {
	// The HabuDen pattern comes from greencell_client's GreencellUtils.
	if got := deviceModel("EVGC021B22752405ZM0018"); got != nameHabuDen {
		t.Errorf("HabuDen serial named %q, want %q", got, nameHabuDen)
	}
	for _, sn := range []string{"EVGC031B22752405ZM0018", "EVGC022B22752405ZM0018", "HABU_DEN", ""} {
		if got := deviceModel(sn); got != nameGeneric {
			t.Errorf("deviceModel(%q) = %q, want %q", sn, got, nameGeneric)
		}
	}
}

func TestParseAnnouncement(t *testing.T) {
	dev, ok := parseAnnouncement([]byte(`{"id":"EVGC021B22752405ZM0018","fw":"1.2.3"}`))
	if !ok {
		t.Fatal("a payload with an id should parse")
	}
	if dev.SN != "EVGC021B22752405ZM0018" {
		t.Errorf("sn = %q", dev.SN)
	}
	if dev.Model != nameHabuDen || !strings.HasPrefix(dev.Name, nameHabuDen+" ") {
		t.Errorf("name/model = %q / %q", dev.Name, dev.Model)
	}
	// Fields this connector has no schema for must survive.
	if !strings.Contains(string(dev.Announcement), `"fw":"1.2.3"`) {
		t.Errorf("announcement should be passed through verbatim, got %s", dev.Announcement)
	}

	for _, bad := range []string{`{"id":""}`, `{"id":"   "}`, `{"name":"BROADCAST"}`, `not json`, ``} {
		if _, ok := parseAnnouncement([]byte(bad)); ok {
			t.Errorf("parseAnnouncement(%q) should have been rejected", bad)
		}
	}
}

func TestTopicSuffix(t *testing.T) {
	sn := "SN1"
	for _, want := range telemetryTopics {
		got, ok := topicSuffix(evsePrefix+sn+"/"+want, sn)
		if !ok || got != want {
			t.Errorf("topicSuffix for %q = %q/%v", want, got, ok)
		}
	}
	for _, topic := range []string{
		evsePrefix + "OTHER/current", // another charger
		evsePrefix + sn + "/unknown", // a topic we do not model
		discoveryTopic,
		evsePrefix + sn, // no suffix at all
	} {
		if _, ok := topicSuffix(topic, sn); ok {
			t.Errorf("topicSuffix(%q) should not match", topic)
		}
	}
}

// ---- payload parsing ---------------------------------------------------------

func TestAsNumber(t *testing.T) {
	f := asNumber(json.RawMessage(`230.5`))
	if f == nil || *f != 230.5 {
		t.Errorf("number = %v", f)
	}
	// Some firmware quotes its numbers.
	f = asNumber(json.RawMessage(`"16000"`))
	if f == nil || *f != 16000 {
		t.Errorf("quoted number = %v", f)
	}
	for _, bad := range []string{`null`, `"abc"`, `{}`, ``} {
		if got := asNumber(json.RawMessage(bad)); got != nil {
			t.Errorf("asNumber(%q) = %v, want nil", bad, *got)
		}
	}
}

func TestParsePhasesLeavesMissingPhasesUnset(t *testing.T) {
	// A single-phase charge reports only l1; l2/l3 must stay nil rather than
	// reading as a real zero.
	p := parsePhases(json.RawMessage(`{"l1":16000}`))
	if p.L1 == nil || *p.L1 != 16000 {
		t.Errorf("l1 = %v", p.L1)
	}
	if p.L2 != nil || p.L3 != nil {
		t.Errorf("missing phases should be nil, got %v %v", p.L2, p.L3)
	}
	if got := parsePhases(json.RawMessage(`not json`)); got.L1 != nil {
		t.Error("a malformed payload should yield no phases")
	}
	if got := parsePhases(nil); got.L1 != nil {
		t.Error("a missing payload should yield no phases")
	}
}

func TestNormalizeState(t *testing.T) {
	if got := normalizeState("CHARGING"); got != stateCharging {
		t.Errorf("CHARGING -> %q", got)
	}
	if got := normalizeState(" waiting_for_car "); got != stateWaitingForCar {
		t.Errorf("waiting_for_car -> %q", got)
	}
	for _, in := range []string{"", "SOMETHING_NEW", "UNKNOWN"} {
		if got := normalizeState(in); got != stateUnknown {
			t.Errorf("normalizeState(%q) = %q, want unknown", in, got)
		}
	}
}

func TestNormalizeAccess(t *testing.T) {
	cases := map[string]string{
		"EXECUTE": accessExecute, "read": accessRead, "DISABLED": accessDisabled,
		"UNAVAILABLE": accessUnavailable,
		// OFFLINE is greencell_client's deprecated spelling of UNAVAILABLE.
		"OFFLINE": accessUnavailable,
		"":        accessUnknown, "nonsense": accessUnknown,
	}
	for in, want := range cases {
		if got := normalizeAccess(in); got != want {
			t.Errorf("normalizeAccess(%q) = %q, want %q", in, got, want)
		}
	}
}

func TestBuildStateThreePhaseCharge(t *testing.T) {
	raw := map[string]json.RawMessage{
		topicCurrent:     json.RawMessage(`{"l1":16000,"l2":15980,"l3":16010}`),
		topicVoltage:     json.RawMessage(`{"l1":230.1,"l2":229.7,"l3":231.0}`),
		topicPower:       json.RawMessage(`{"momentary":11040}`),
		topicStatus:      json.RawMessage(`{"state":"CHARGING"}`),
		topicDeviceState: json.RawMessage(`{"level":"EXECUTE"}`),
	}
	st := buildState("EVGC021B22752405ZM0018", raw)

	if !st.Complete {
		t.Error("all five topics reported, so the reading is complete")
	}
	// Current is published in milliamps.
	if st.CurrentA.L1 == nil || *st.CurrentA.L1 != 16 {
		t.Errorf("l1 current = %v A, want 16", st.CurrentA.L1)
	}
	if st.VoltageV.L1 == nil || *st.VoltageV.L1 != 230.1 {
		t.Errorf("l1 voltage = %v", st.VoltageV.L1)
	}
	if st.PowerW == nil || *st.PowerW != 11040 {
		t.Errorf("power = %v", st.PowerW)
	}
	if st.Status != stateCharging || !st.Charging || !st.Plugged || st.Fault {
		t.Errorf("status flags: %q charging=%v plugged=%v fault=%v", st.Status, st.Charging, st.Plugged, st.Fault)
	}
	if st.AccessLevel != accessExecute || !st.CanExecute || !st.Available {
		t.Errorf("access: %q canExecute=%v available=%v", st.AccessLevel, st.CanExecute, st.Available)
	}
	if st.LivePhases != 3 {
		t.Errorf("livePhases = %d, want 3", st.LivePhases)
	}
	if st.Model != nameHabuDen {
		t.Errorf("model = %q", st.Model)
	}
	if st.ObservedAt.IsZero() {
		t.Error("observedAt should be stamped")
	}
}

func TestBuildStateSinglePhaseAndPartial(t *testing.T) {
	// Only two topics reported: the rest must read as unset, not as zero.
	raw := map[string]json.RawMessage{
		topicCurrent: json.RawMessage(`{"l1":16000,"l2":30,"l3":0}`),
		topicStatus:  json.RawMessage(`{"state":"CHARGING"}`),
	}
	st := buildState("SN1", raw)

	if st.Complete {
		t.Error("a partial reading must not claim to be complete")
	}
	if st.Received[topicCurrent] != true || st.Received[topicPower] != false {
		t.Errorf("received map = %v", st.Received)
	}
	if st.PowerW != nil {
		t.Errorf("power = %v, want nil when the topic was silent", *st.PowerW)
	}
	// 30 mA is 0.03 A — noise on an idle phase, not a live one.
	if st.LivePhases != 1 {
		t.Errorf("livePhases = %d, want 1", st.LivePhases)
	}
	if st.AccessLevel != accessUnknown {
		t.Errorf("access level = %q, want unknown when device_state was silent", st.AccessLevel)
	}
	if st.Model != nameGeneric {
		t.Errorf("model = %q, want the generic name for a non-HabuDen serial", st.Model)
	}
}

func TestBuildStateUnavailable(t *testing.T) {
	// The device signals unavailability through the status topic itself, which
	// upstream detects by substring because the payload is not a state document.
	for _, payload := range []string{`{"state":"UNAVAILABLE"}`, `"OFFLINE"`, `{"status":"UNAVAILABLE"}`} {
		st := buildState("SN1", map[string]json.RawMessage{
			topicStatus:      json.RawMessage(payload),
			topicDeviceState: json.RawMessage(`{"level":"READ"}`),
		})
		if st.Status != stateUnavailable {
			t.Errorf("payload %s -> status %q, want unavailable", payload, st.Status)
		}
		if st.Available {
			t.Errorf("payload %s should not read as available", payload)
		}
	}
}

func TestBuildStateErrorStates(t *testing.T) {
	for _, tc := range []struct {
		state   string
		fault   bool
		plugged bool
	}{
		{"ERROR_CAR", true, true},
		{"ERROR_EVSE", true, false},
		{"IDLE", false, false},
		{"FINISHED", false, true},
	} {
		st := buildState("SN1", map[string]json.RawMessage{
			topicStatus: json.RawMessage(`{"state":"` + tc.state + `"}`),
		})
		if st.Fault != tc.fault || st.Plugged != tc.plugged {
			t.Errorf("%s: fault=%v plugged=%v, want %v/%v", tc.state, st.Fault, st.Plugged, tc.fault, tc.plugged)
		}
	}
}

func TestBuildStateDisabledIsNotAvailable(t *testing.T) {
	st := buildState("SN1", map[string]json.RawMessage{
		topicStatus:      json.RawMessage(`{"state":"IDLE"}`),
		topicDeviceState: json.RawMessage(`{"level":"DISABLED"}`),
	})
	if st.Available || st.CanExecute {
		t.Errorf("a DISABLED device is neither available nor executable: %+v", st)
	}
}

// ---- Invoke dispatch ---------------------------------------------------------

func TestInvokeUnknownAction(t *testing.T) {
	var p Plugin
	_ = p.Init(context.Background(), map[string]string{"host": "127.0.0.1"})
	if _, err := p.Invoke(context.Background(), "nope", nil); err == nil {
		t.Fatal("expected an error for an unknown action")
	}
}

func TestInvokeChargerStateRequiresSerial(t *testing.T) {
	var p Plugin
	_ = p.Init(context.Background(), map[string]string{"host": "127.0.0.1"})
	_, err := p.Invoke(context.Background(), "charger-state", nil)
	if err == nil || !strings.Contains(err.Error(), "serial") {
		t.Fatalf("error = %v, want a missing-serial error", err)
	}
}

func TestInvokeRejectsMalformedParams(t *testing.T) {
	var p Plugin
	_ = p.Init(context.Background(), map[string]string{"host": "127.0.0.1"})
	if _, err := p.Invoke(context.Background(), "chargers", json.RawMessage(`[`)); err == nil {
		t.Fatal("expected an error for malformed params")
	}
}

// ---- end to end against a broker ---------------------------------------------

func TestDiscoverCollectsEveryDeviceThatAnswers(t *testing.T) {
	addr := startBroker(t, []brokerMsg{
		{discoveryTopic, `{"id":"EVGC021B22752405ZM0018"}`},
		{discoveryTopic, `{"id":"SOMETHINGELSE"}`},
		{discoveryTopic, `{"id":"EVGC021B22752405ZM0018"}`}, // a repeat is one device
		{discoveryTopic, `{"name":"BROADCAST"}`},            // our own echo, not a device
	})

	p := pluginAt(t, addr, map[string]string{"timeout": "5"})
	devices, err := p.discover(context.Background())
	if err != nil {
		t.Fatalf("discover: %v", err)
	}
	if len(devices) != 2 {
		t.Fatalf("found %d devices, want 2: %+v", len(devices), devices)
	}
	if devices[0].SN != "EVGC021B22752405ZM0018" || devices[0].Model != nameHabuDen {
		t.Errorf("first device = %+v", devices[0])
	}
	if devices[1].Model != nameGeneric {
		t.Errorf("second device model = %q, want the generic name", devices[1].Model)
	}
}

func TestDiscoverReturnsEmptyOnSilence(t *testing.T) {
	addr := startBroker(t, nil)
	p := pluginAt(t, addr, map[string]string{"timeout": "1"})
	devices, err := p.discover(context.Background())
	if err != nil {
		t.Fatalf("discover: %v", err)
	}
	if len(devices) != 0 {
		t.Fatalf("want no devices, got %+v", devices)
	}
}

func TestChargerStateReadsTelemetry(t *testing.T) {
	sn := "EVGC021B22752405ZM0018"
	addr := startBroker(t, []brokerMsg{
		{evsePrefix + sn + "/current", `{"l1":16000,"l2":0,"l3":0}`},
		{evsePrefix + sn + "/voltage", `{"l1":230,"l2":231,"l3":229}`},
		{evsePrefix + sn + "/power", `{"momentary":3680}`},
		{evsePrefix + sn + "/status", `{"state":"CHARGING"}`},
		{evsePrefix + sn + "/device_state", `{"level":"READ"}`},
		// Another charger on the same broker must not bleed into this reading.
		{evsePrefix + "OTHER/power", `{"momentary":99999}`},
	})

	p := pluginAt(t, addr, map[string]string{"timeout": "5"})
	st, err := p.chargerState(context.Background(), sn)
	if err != nil {
		t.Fatalf("chargerState: %v", err)
	}
	if !st.Complete {
		t.Errorf("expected a complete reading, got %v", st.Received)
	}
	if st.PowerW == nil || *st.PowerW != 3680 {
		t.Errorf("power = %v, want 3680 (the other charger's reading must not leak in)", st.PowerW)
	}
	if st.CurrentA.L1 == nil || *st.CurrentA.L1 != 16 {
		t.Errorf("l1 = %v A, want 16", st.CurrentA.L1)
	}
	if st.LivePhases != 1 {
		t.Errorf("livePhases = %d, want 1", st.LivePhases)
	}
	if st.AccessLevel != accessRead || st.CanExecute {
		t.Errorf("READ mode should not report as executable: %+v", st)
	}
}

func TestChargerStateReturnsPartialAtTheDeadline(t *testing.T) {
	sn := "SN1"
	addr := startBroker(t, []brokerMsg{
		{evsePrefix + sn + "/status", `{"state":"IDLE"}`},
	})

	p := pluginAt(t, addr, map[string]string{"timeout": "1"})
	st, err := p.chargerState(context.Background(), sn)
	if err != nil {
		t.Fatalf("a partial reading should not be an error: %v", err)
	}
	if st.Complete || st.Status != stateIdle {
		t.Errorf("state = %+v", st)
	}
	if st.Received[topicVoltage] {
		t.Error("voltage was never published, so it must read as not received")
	}
}

func TestChargerStateErrorsOnSilence(t *testing.T) {
	addr := startBroker(t, nil)
	p := pluginAt(t, addr, map[string]string{"timeout": "1"})
	_, err := p.chargerState(context.Background(), "SN1")
	if err == nil || !strings.Contains(err.Error(), "no data from charger") {
		t.Fatalf("error = %v, want a no-data error", err)
	}
}

func TestChargerStateFailsWithoutABroker(t *testing.T) {
	// Nothing is listening, so the session cannot be opened at all.
	p := pluginAt(t, "127.0.0.1:1", map[string]string{"timeout": "1"})
	if _, err := p.chargerState(context.Background(), "SN1"); err == nil {
		t.Fatal("expected a connection error")
	}
}

func TestHealthCheckClassifiesTheThreeOutcomes(t *testing.T) {
	t.Run("ok", func(t *testing.T) {
		addr := startBroker(t, []brokerMsg{{discoveryTopic, `{"id":"EVGC021B22752405ZM0018"}`}})
		p := pluginAt(t, addr, map[string]string{"timeout": "5"})
		h := p.HealthCheck(context.Background())
		if h.Status != plugins.StatusOK {
			t.Fatalf("status = %q (%s), want ok", h.Status, h.Detail)
		}
		if !strings.Contains(h.Detail, "1 Greencell device") {
			t.Errorf("detail = %q", h.Detail)
		}
	})

	t.Run("degraded when the broker answers but no charger does", func(t *testing.T) {
		addr := startBroker(t, nil)
		p := pluginAt(t, addr, map[string]string{"timeout": "1"})
		h := p.HealthCheck(context.Background())
		if h.Status != plugins.StatusDegraded {
			t.Fatalf("status = %q (%s), want degraded", h.Status, h.Detail)
		}
	})

	t.Run("down when the broker is unreachable", func(t *testing.T) {
		p := pluginAt(t, "127.0.0.1:1", map[string]string{"timeout": "1"})
		h := p.HealthCheck(context.Background())
		if h.Status != plugins.StatusDown {
			t.Fatalf("status = %q (%s), want down", h.Status, h.Detail)
		}
	})
}

func TestInvokeChargersShape(t *testing.T) {
	addr := startBroker(t, []brokerMsg{{discoveryTopic, `{"id":"EVGC021B22752405ZM0018"}`}})
	p := pluginAt(t, addr, map[string]string{"timeout": "5"})

	raw, err := p.Invoke(context.Background(), "chargers", nil)
	if err != nil {
		t.Fatalf("Invoke: %v", err)
	}
	var doc struct {
		Chargers []Device `json:"chargers"`
	}
	if err := json.Unmarshal(raw, &doc); err != nil {
		t.Fatalf("unmarshal: %v", err)
	}
	if len(doc.Chargers) != 1 || doc.Chargers[0].SN != "EVGC021B22752405ZM0018" {
		t.Fatalf("chargers = %+v", doc.Chargers)
	}
}

func TestInvokeChargerStateFallsBackToTheConfiguredSerial(t *testing.T) {
	sn := "EVGC021B22752405ZM0018"
	addr := startBroker(t, []brokerMsg{{evsePrefix + sn + "/status", `{"state":"FINISHED"}`}})
	p := pluginAt(t, addr, map[string]string{"timeout": "1", "serial": sn})

	raw, err := p.Invoke(context.Background(), "charger-state", nil)
	if err != nil {
		t.Fatalf("Invoke: %v", err)
	}
	var st State
	if err := json.Unmarshal(raw, &st); err != nil {
		t.Fatalf("unmarshal: %v", err)
	}
	if st.SN != sn || st.Status != stateFinished {
		t.Fatalf("state = %+v", st)
	}
}

// ---- test broker -------------------------------------------------------------

// brokerMsg is one publication the test broker pushes once the plugin has
// subscribed and sent its discovery broadcast.
type brokerMsg struct {
	topic   string
	payload string
}

// pluginAt returns a plugin configured to talk to addr.
func pluginAt(t *testing.T, addr string, extra map[string]string) *Plugin {
	t.Helper()
	host, port, err := net.SplitHostPort(addr)
	if err != nil {
		t.Fatalf("bad address %q: %v", addr, err)
	}
	config := map[string]string{"host": host, "port": port}
	for k, v := range extra {
		config[k] = v
	}
	p := &Plugin{}
	if err := p.Init(context.Background(), config); err != nil {
		t.Fatalf("Init: %v", err)
	}
	return p
}

// startBroker runs an MQTT 3.1.1 broker that does just enough for these tests:
// accept a connection, acknowledge the subscription, and — once the client sends
// its discovery broadcast — push the scripted messages back. It speaks the wire
// format directly rather than reusing internal/mqtt, so a bug in the client
// cannot hide behind a matching bug in the fixture.
func startBroker(t *testing.T, script []brokerMsg) string {
	t.Helper()
	return startBrokerRecording(t, nil, script)
}

// startBrokerRecording is startBroker with a channel that receives everything the
// client publishes, so a test can assert on what went out as well as what came
// back. A nil channel records nothing.
func startBrokerRecording(t *testing.T, seen chan<- brokerMsg, script []brokerMsg) string {
	t.Helper()
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("listen: %v", err)
	}
	var wg sync.WaitGroup
	// Close the listener before waiting: the accept loop only ends when the
	// listener does, and cleanups run last-registered first.
	t.Cleanup(func() { _ = ln.Close(); wg.Wait() })
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			conn, err := ln.Accept()
			if err != nil {
				return
			}
			wg.Add(1)
			go func() {
				defer wg.Done()
				defer conn.Close()
				serveBroker(conn, seen, script)
			}()
		}
	}()
	return ln.Addr().String()
}

func serveBroker(conn net.Conn, seen chan<- brokerMsg, script []brokerMsg) {
	br := bufio.NewReader(conn)
	for {
		typ, _, body, err := readRaw(br)
		if err != nil {
			return
		}
		switch typ {
		case 1: // CONNECT
			if _, err := conn.Write(frame(2, 0, []byte{0, 0})); err != nil {
				return
			}
		case 8: // SUBSCRIBE — acknowledge every requested filter with QoS 0
			if len(body) < 2 {
				return
			}
			ack := append([]byte{body[0], body[1]}, make([]byte, countFilters(body[2:]))...)
			if _, err := conn.Write(frame(9, 0, ack)); err != nil {
				return
			}
		case 3: // PUBLISH — from the client; record it, then answer with the script
			if seen != nil {
				if topic, rest, err := splitRawString(body); err == nil {
					select {
					case seen <- brokerMsg{topic, string(rest)}:
					default:
					}
				}
			}
			for _, m := range script {
				pub := append(encRawString(m.topic), m.payload...)
				if _, err := conn.Write(frame(3, 0, pub)); err != nil {
					return
				}
			}
		case 12: // PINGREQ
			if _, err := conn.Write(frame(13, 0, nil)); err != nil {
				return
			}
		case 14: // DISCONNECT
			return
		}
	}
}

// countFilters counts the topic filters in a SUBSCRIBE payload so the SUBACK can
// carry one return code per filter.
func countFilters(b []byte) int {
	n := 0
	for len(b) >= 3 {
		l := int(binary.BigEndian.Uint16(b))
		if len(b) < 2+l+1 {
			return n
		}
		b = b[2+l+1:]
		n++
	}
	return n
}

// splitRawString peels an MQTT UTF-8 string off the front of b.
func splitRawString(b []byte) (string, []byte, error) {
	if len(b) < 2 {
		return "", nil, io.ErrUnexpectedEOF
	}
	n := int(binary.BigEndian.Uint16(b))
	if len(b) < 2+n {
		return "", nil, io.ErrUnexpectedEOF
	}
	return string(b[2 : 2+n]), b[2+n:], nil
}

func encRawString(s string) []byte {
	out := binary.BigEndian.AppendUint16(nil, uint16(len(s)))
	return append(out, s...)
}

func frame(typ, flags byte, body []byte) []byte {
	out := []byte{typ<<4 | flags}
	n := len(body)
	for {
		digit := byte(n % 128)
		n /= 128
		if n > 0 {
			digit |= 0x80
		}
		out = append(out, digit)
		if n == 0 {
			break
		}
	}
	return append(out, body...)
}

func readRaw(br *bufio.Reader) (typ, flags byte, body []byte, err error) {
	head, err := br.ReadByte()
	if err != nil {
		return 0, 0, nil, err
	}
	var (
		length     int
		multiplier = 1
	)
	for i := 0; i < 4; i++ {
		digit, err := br.ReadByte()
		if err != nil {
			return 0, 0, nil, err
		}
		length += int(digit&0x7F) * multiplier
		if digit&0x80 == 0 {
			break
		}
		multiplier *= 128
	}
	body = make([]byte, length)
	if _, err := io.ReadFull(br, body); err != nil {
		return 0, 0, nil, err
	}
	return head >> 4, head & 0x0F, body, nil
}

// ---- the operator-supplied QUERY topic ---------------------------------------

func TestCommandTopicFor(t *testing.T) {
	var p Plugin
	_ = p.Init(context.Background(), map[string]string{
		"host": "h", "commandTopic": "/greencell/evse/" + snPlaceholder + "/command",
	})
	if got := p.commandTopicFor("SN1"); got != "/greencell/evse/SN1/command" {
		t.Errorf("commandTopicFor = %q", got)
	}

	// A topic without the placeholder is used as given.
	_ = p.Init(context.Background(), map[string]string{"host": "h", "commandTopic": "/fixed/topic"})
	if got := p.commandTopicFor("SN1"); got != "/fixed/topic" {
		t.Errorf("commandTopicFor = %q, want the literal topic", got)
	}

	// Unset means listen-only; nothing is ever published.
	_ = p.Init(context.Background(), map[string]string{"host": "h"})
	if got := p.commandTopicFor("SN1"); got != "" {
		t.Errorf("commandTopicFor = %q, want empty when unconfigured", got)
	}
}

func TestChargerStateSendsQueryWhenATopicIsConfigured(t *testing.T) {
	sn := "SN1"
	seen := make(chan brokerMsg, 4)
	addr := startBrokerRecording(t, seen, []brokerMsg{
		{evsePrefix + sn + "/status", `{"state":"IDLE"}`},
	})

	p := pluginAt(t, addr, map[string]string{
		"timeout": "1", "commandTopic": "/greencell/evse/" + snPlaceholder + "/command",
	})
	if _, err := p.chargerState(context.Background(), sn); err != nil {
		t.Fatalf("chargerState: %v", err)
	}

	var got []brokerMsg
	close(seen)
	for m := range seen {
		got = append(got, m)
	}
	// The discovery broadcast, then the QUERY on the configured topic.
	var query *brokerMsg
	for i := range got {
		if got[i].topic == "/greencell/evse/SN1/command" {
			query = &got[i]
		}
	}
	if query == nil {
		t.Fatalf("no QUERY was published; broker saw %+v", got)
	}
	if query.payload != queryCommand {
		t.Errorf("QUERY payload = %q, want %q", query.payload, queryCommand)
	}
}

func TestChargerStateSendsNothingWithoutACommandTopic(t *testing.T) {
	sn := "SN1"
	seen := make(chan brokerMsg, 4)
	addr := startBrokerRecording(t, seen, []brokerMsg{
		{evsePrefix + sn + "/status", `{"state":"IDLE"}`},
	})

	p := pluginAt(t, addr, map[string]string{"timeout": "1"})
	if _, err := p.chargerState(context.Background(), sn); err != nil {
		t.Fatalf("chargerState: %v", err)
	}

	close(seen)
	for m := range seen {
		if m.topic != broadcastTopic {
			t.Errorf("unconfigured, the plugin must publish only the discovery broadcast; it also sent %+v", m)
		}
	}
}