FranklinWH-Toolbox/service/repositories/mqtt/fwh_mqtt/fwh_mqtt_client_test.go

package fwh_mqtt

import (
	"context"
	"encoding/json"
	"fwh-toolbox-server/config"
	"sync"
	"testing"
	"time"
)

// fakeTransport 内存实现，记录订阅和发送，并可模拟 broker 推送
type fakeTransport struct {
	mu          sync.Mutex
	connected   bool
	subs        map[string]func([]byte)
	published   []string
	subCalls    int
	unsubCalls  int
	disconnects int
}

func newFake() *fakeTransport { return &fakeTransport{subs: map[string]func([]byte){}} }

func (f *fakeTransport) Connect() error { f.mu.Lock(); f.connected = true; f.mu.Unlock(); return nil }
func (f *fakeTransport) IsConnected() bool {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.connected
}
func (f *fakeTransport) Disconnect() {
	f.mu.Lock()
	f.connected = false
	f.disconnects++
	f.mu.Unlock()
}
func (f *fakeTransport) SubscribeMessage(topic string, cb func([]byte)) error {
	f.mu.Lock()
	f.subs[topic] = cb
	f.subCalls++
	f.mu.Unlock()
	return nil
}
func (f *fakeTransport) UnSubscribeMessage(topic string) error {
	f.mu.Lock()
	delete(f.subs, topic)
	f.unsubCalls++
	f.mu.Unlock()
	return nil
}
func (f *fakeTransport) SendMessageOnce(topic string, msg []byte) error {
	f.mu.Lock()
	f.published = append(f.published, topic)
	f.mu.Unlock()
	return nil
}
func (f *fakeTransport) GetClientStatus() string { return "fake" }

// push 模拟 broker 推送
func (f *fakeTransport) push(topic string, cmdType uint) bool {
	f.mu.Lock()
	cb := f.subs[topic]
	f.mu.Unlock()
	if cb == nil {
		return false
	}
	b, _ := json.Marshal(FwhMqttMsg{CmdType: cmdType, EquipNo: "SN", DataArea: map[string]int{"cmd": 2}})
	cb(b)
	return true
}

func (f *fakeTransport) counts() (sub, unsub int) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.subCalls, f.unsubCalls
}

func waitFor(t *testing.T, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for !cond() {
		if time.Now().After(deadline) {
			t.Fatal("condition not met in time")
		}
		time.Sleep(5 * time.Millisecond)
	}
}

func TestParseEnv(t *testing.T) {
	cases := []struct {
		in      string
		want    Env
		wantErr bool
	}{
		{"", EnvProd, false},
		{"prod", EnvProd, false},
		{"test", EnvTest, false},
		{"TEST", "", true},
		{"dev", "", true},
	}
	for _, c := range cases {
		got, err := ParseEnv(c.in)
		if (err != nil) != c.wantErr || got != c.want {
			t.Errorf("ParseEnv(%q) = %q, %v; want %q, err=%v", c.in, got, err, c.want, c.wantErr)
		}
	}
}

func TestInitWithoutBrokerFails(t *testing.T) {
	if err := InitFwhMqttClient(EnvTest, &config.FwhMqttConfig{}); err == nil {
		t.Fatal("expected error when broker not configured")
	}
	if _, err := GetClient(EnvTest); err == nil {
		t.Fatal("client should not be registered after failed init")
	}
}

func TestSubscribeFanOutAndRefCount(t *testing.T) {
	f := newFake()
	c := newFwhMqttClient(EnvTest, f)
	ctx := context.Background()

	var mu sync.Mutex
	got := map[string]int{}
	h := func(name string) Handler {
		return func(m Message) {
			mu.Lock()
			got[name]++
			mu.Unlock()
		}
	}

	cancelA, err := c.Subscribe(ctx, "t", h("a"))
	if err != nil {
		t.Fatal(err)
	}
	cancelB, err := c.Subscribe(ctx, "t", h("b"))
	if err != nil {
		t.Fatal(err)
	}
	if sub, _ := f.counts(); sub != 1 {
		t.Fatalf("broker subscribe calls = %d, want 1", sub)
	}

	f.push("t", 379)
	waitFor(t, func() bool { mu.Lock(); defer mu.Unlock(); return got["a"] == 1 && got["b"] == 1 })

	cancelA()
	cancelA() // 重复取消无副作用
	f.push("t", 379)
	waitFor(t, func() bool { mu.Lock(); defer mu.Unlock(); return got["b"] == 2 })
	mu.Lock()
	if got["a"] != 1 {
		t.Errorf("cancelled subscriber received message")
	}
	mu.Unlock()
	if _, unsub := f.counts(); unsub != 0 {
		t.Errorf("should not unsubscribe while subscribers remain")
	}

	cancelB()
	waitFor(t, func() bool { _, unsub := f.counts(); return unsub == 1 })
	if c.hasTopics() {
		t.Error("topics should be empty")
	}
}

func TestRequestMatchesResponse(t *testing.T) {
	f := newFake()
	c := newFwhMqttClient(EnvTest, f)

	go func() {
		waitFor(t, func() bool {
			f.mu.Lock()
			defer f.mu.Unlock()
			return len(f.published) == 1
		})
		f.push(C2STopic("mng/SN"), 379) // 不匹配，应被忽略
		f.push(C2STopic("mng/SN"), 378)
	}()

	res, err := c.Request(context.Background(), RequestOpt{
		Topic:    S2CTopic("mng/SN"),
		Msg:      FwhMqttMsg{CmdType: 377},
		ResTopic: C2STopic("mng/SN"),
		Match:    func(m Message) bool { return m.Envelope.CmdType == 378 },
		Timeout:  2 * time.Second,
	})
	if err != nil {
		t.Fatal(err)
	}
	if res.Envelope.CmdType != 378 || len(res.Raw) == 0 {
		t.Errorf("unexpected response: %+v", res.Envelope)
	}
	waitFor(t, func() bool { _, unsub := f.counts(); return unsub == 1 })
}

func TestRequestTimeout(t *testing.T) {
	c := newFwhMqttClient(EnvTest, newFake())
	_, err := c.Request(context.Background(), RequestOpt{
		Topic: "a", ResTopic: "b", Timeout: 50 * time.Millisecond,
	})
	if err == nil {
		t.Fatal("expected timeout")
	}
}

func TestNewEnvelope(t *testing.T) {
	m, err := NewEnvelope(377, "SN", map[string]any{"cmd": 4, "data": map[string]any{}})
	if err != nil {
		t.Fatal(err)
	}
	// {"cmd":4,"data":{}} 共 19 字节
	if m.Len != 19 || len(m.Crc) != 8 || m.EquipNo != "SN" || m.CmdType != 377 {
		t.Errorf("envelope: %+v", m)
	}
}