mqttplug/mqttplug_test.go
Rama main · réplica de lectura del repositorio original.
package mqttplug
import (
"context"
"strings"
"testing"
mqtt "github.com/eclipse/paho.mqtt.golang"
"github.com/ccmpbll/printspy/models"
)
func TestParseTopic(t *testing.T) {
cases := []struct {
in string
topic, kind, suffix string
ok bool
}{
{"stat/testplug/POWER", "testplug", "stat", "POWER", true},
{"stat/testplug/POWER2", "testplug", "stat", "POWER2", true},
{"tele/testplug/SENSOR", "testplug", "tele", "SENSOR", true},
{"garbage", "", "", "", false},
}
for _, c := range cases {
topic, kind, suffix, ok := parseTopic(c.in)
if ok != c.ok || topic != c.topic || kind != c.kind || suffix != c.suffix {
t.Errorf("parseTopic(%q) = (%q,%q,%q,%v), want (%q,%q,%q,%v)",
c.in, topic, kind, suffix, ok, c.topic, c.kind, c.suffix, c.ok)
}
}
}
func TestCommandTopic(t *testing.T) {
if got := commandTopic("testplug", "1"); got != "cmnd/testplug/Power" {
t.Errorf("commandTopic idx=1 = %q, want cmnd/testplug/Power", got)
}
if got := commandTopic("testplug", "2"); got != "cmnd/testplug/Power2" {
t.Errorf("commandTopic idx=2 = %q, want cmnd/testplug/Power2", got)
}
if got := commandTopic("testplug", ""); got != "cmnd/testplug/Power" {
t.Errorf("commandTopic idx=\"\" = %q, want cmnd/testplug/Power", got)
}
}
func TestApplyPowerAndEnergy(t *testing.T) {
c := New()
c.subs = map[string]topicSubs{
"testplug": {relays: map[string]relayMeta{"1": {Label: "Printer"}}},
}
c.applyPower("testplug", "POWER", true)
ps, ok := c.GetState("testplug", "1")
if !ok || !ps.On || ps.Label != "Printer" || ps.ID != "mqtt:testplug:1" {
t.Fatalf("applyPower: got %+v, ok=%v", ps, ok)
}
c.applyEnergy("testplug", []byte(`{"ENERGY":{"Power":12.3,"Voltage":120,"Current":0.1,"Total":0.5}}`))
ps, ok = c.GetState("testplug", "1")
if !ok || ps.Watts != 12.3 || ps.Voltage != 120 || ps.Current != 0.1 || ps.TotalKWh != 0.5 {
t.Fatalf("applyEnergy: got %+v", ps)
}
// On stays true - energy telemetry shouldn't clobber the last known
// power state, they arrive on separate topics.
if !ps.On {
t.Errorf("On = false after applyEnergy, want true (unaffected)")
}
}
func TestApplyPowerBareSuffixIsRelay1(t *testing.T) {
c := New()
c.applyPower("testplug", "POWER", true)
if _, ok := c.GetState("testplug", "1"); !ok {
t.Error("bare POWER suffix should key to idx \"1\"")
}
}
func TestSyncDedupesByTopic(t *testing.T) {
c := New()
plugs := []models.SmartPlug{
{MQTTTopic: "shared", Idx: "1", Label: "A"},
{MQTTTopic: "shared", Idx: "2", Label: "B"},
{MQTTTopic: "other", Idx: "1", Label: "C"},
{MQTTTopic: "", IP: "1.2.3.4", Idx: "1", Label: "HTTP-mode, ignored"},
}
c.Sync(plugs)
if len(c.subs) != 2 {
t.Fatalf("subs = %d topics, want 2", len(c.subs))
}
if len(c.subs["shared"].relays) != 2 {
t.Fatalf("shared topic relays = %d, want 2", len(c.subs["shared"].relays))
}
}
func TestSetCachedOnWinsOverStaleRead(t *testing.T) {
// Regression: a poll cycle already in flight when SetState is called
// reads whatever's cached - if that's the pre-command device echo
// (device hasn't caught up yet), it must not win over the command we
// just successfully published.
c := New()
c.subs = map[string]topicSubs{"testplug": {relays: map[string]relayMeta{"1": {Label: "Light"}}}}
c.applyPower("testplug", "POWER", true) // device was last seen ON
c.setCachedOn("testplug", "1", false) // we just commanded OFF
ps, ok := c.GetState("testplug", "1")
if !ok || ps.On {
t.Fatalf("GetState after setCachedOn = %+v, ok=%v, want On=false", ps, ok)
}
}
func TestGetStateUnknown(t *testing.T) {
c := New()
if _, ok := c.GetState("nope", "1"); ok {
t.Error("GetState on unknown topic:idx should return ok=false")
}
}
// fakePublishClient embeds the mqtt.Client interface (nil for every method
// but Publish) so it satisfies the interface without stubbing methods
// handleMessage never calls.
type fakePublishClient struct {
mqtt.Client
published []string
retained []bool
}
func (f *fakePublishClient) Publish(topic string, qos byte, retained bool, payload interface{}) mqtt.Token {
f.published = append(f.published, topic)
f.retained = append(f.retained, retained)
return &fakeDoneToken{}
}
func (f *fakePublishClient) IsConnectionOpen() bool { return true }
// fakeDoneToken is an already-completed mqtt.Token - Publish (mqttplug.go)
// selects on Done()/ctx.Done(), so a nil token (fine for handleMessage's
// fire-and-forget query publishes) would panic here. Embeds mqtt.Token (nil)
// like fakePublishClient embeds mqtt.Client, overriding only what's called.
type fakeDoneToken struct {
mqtt.Token
}
func (t *fakeDoneToken) Done() <-chan struct{} { ch := make(chan struct{}); close(ch); return ch }
func (t *fakeDoneToken) Error() error { return nil }
type fakeMessage struct {
topic string
payload []byte
}
func (m *fakeMessage) Duplicate() bool { return false }
func (m *fakeMessage) Qos() byte { return 0 }
func (m *fakeMessage) Retained() bool { return false }
func (m *fakeMessage) Topic() string { return m.topic }
func (m *fakeMessage) MessageID() uint16 { return 0 }
func (m *fakeMessage) Payload() []byte { return m.payload }
func (m *fakeMessage) Ack() {}
func TestLWTOnlineTriggersRequery(t *testing.T) {
// Regression: a device that reconnects to the broker (reboot, power
// blip, wifi drop) may come back in a different state than we last
// knew, with nothing else telling us - LWT flipping to Online is the
// signal to ask again instead of trusting the stale cache.
c := New()
c.subs = map[string]topicSubs{"testplug": {relays: map[string]relayMeta{"1": {}}}}
fc := &fakePublishClient{}
c.handleMessage(fc, &fakeMessage{topic: "tele/testplug/LWT", payload: []byte("Online")})
if len(fc.published) != 1 || fc.published[0] != "cmnd/testplug/Power" {
t.Fatalf("expected one Power query publish, got %v", fc.published)
}
}
func TestLWTOfflineDoesNotRequery(t *testing.T) {
c := New()
c.subs = map[string]topicSubs{"testplug": {relays: map[string]relayMeta{"1": {}}}}
fc := &fakePublishClient{}
c.handleMessage(fc, &fakeMessage{topic: "tele/testplug/LWT", payload: []byte("Offline")})
if len(fc.published) != 0 {
t.Fatalf("expected no publish for an Offline LWT, got %v", fc.published)
}
}
func TestLWTOfflineMarksCacheOff(t *testing.T) {
// Regression: a device that drops off the broker (most commonly lost
// power, being an inline relay) must not leave the dashboard showing
// its last confirmed reading forever - the cache has to reflect "off,
// unconfirmed" instead of silently staying stuck on a stale "On".
c := New()
c.subs = map[string]topicSubs{"testplug": {relays: map[string]relayMeta{"1": {Label: "Printer"}}}}
c.applyPower("testplug", "POWER", true) // device was last seen ON
fc := &fakePublishClient{}
c.handleMessage(fc, &fakeMessage{topic: "tele/testplug/LWT", payload: []byte("Offline")})
ps, ok := c.GetState("testplug", "1")
if !ok || ps.On || ps.Source != "mqtt-offline" {
t.Fatalf("GetState after Offline LWT = %+v, ok=%v, want On=false, Source=mqtt-offline", ps, ok)
}
}
func TestPublishNotConfigured(t *testing.T) {
c := New()
if err := c.Publish(context.Background(), "printspy/printer/1/state", []byte("{}")); err == nil {
t.Fatal("Publish with no configured client should error")
}
}
func TestOnConnectPublishesAvailability(t *testing.T) {
// Regression: onConnect must announce "online" itself - the broker's LWT
// only ever fires "offline" (on connection loss), nothing publishes the
// online side automatically.
c := New()
fc := &fakePublishClient{}
c.onConnect(fc)
if len(fc.published) != 1 || fc.published[0] != availabilityTopic {
t.Fatalf("expected one publish to %s, got %v", availabilityTopic, fc.published)
}
if len(fc.retained) != 1 || !fc.retained[0] {
t.Fatalf("expected availability publish to be retained, got %v", fc.retained)
}
}
func TestPublish(t *testing.T) {
c := New()
fc := &fakePublishClient{}
c.cli = fc
payload := []byte(`{"id":1,"state":"printing"}`)
if err := c.Publish(context.Background(), "printspy/printer/1/state", payload); err != nil {
t.Fatalf("Publish returned error: %v", err)
}
if len(fc.published) != 1 || fc.published[0] != "printspy/printer/1/state" {
t.Fatalf("expected one publish to printspy/printer/1/state, got %v", fc.published)
}
if len(fc.retained) != 1 || !fc.retained[0] {
t.Fatalf("expected publish to be retained, got %v", fc.retained)
}
}
func TestValidateBrokerURL(t *testing.T) {
for in, ok := range map[string]bool{
"": true,
"tcp://10.0.0.5:1883": true,
"mqtts://broker.lan:8883": true,
"wss://broker.lan/mqtt": true,
"unix:///var/run/x.sock": false,
"file:///etc/passwd": false,
"10.0.0.5:1883": false,
"tcp://": false,
} {
if err := ValidateBrokerURL(in); (err == nil) != ok {
t.Errorf("ValidateBrokerURL(%q) = %v, want ok=%v", in, err, ok)
}
}
}
func TestRedactedBrokerHidesPassword(t *testing.T) {
got := redactedBroker("tcp://user:s3cret@10.0.0.5:1883")
if strings.Contains(got, "s3cret") {
t.Errorf("redactedBroker leaked password: %q", got)
}
}
// A bad new config must be rejected by the probe without touching the
// working client - previously Configure tore the old one down first.
func TestConfigureFailureKeepsExistingClient(t *testing.T) {
c := New()
old := &fakePublishClient{}
c.cli = old
if err := c.Configure("tcp://127.0.0.1:1", "", ""); err == nil {
t.Fatal("want error for unreachable broker")
}
if c.cli != mqtt.Client(old) {
t.Error("working client was replaced by a failed Configure")
}
if len(old.published) != 0 {
t.Errorf("working client was told to go offline: %v", old.published)
}
}