mqtt_consumer_test.go

package mqtt_consumer

import (
	"testing"

	"github.com/influxdata/telegraf/plugins/parsers"
	"github.com/influxdata/telegraf/testutil"

	"github.com/stretchr/testify/assert"

	"github.com/eclipse/paho.mqtt.golang"
)

const (
	testMsg         = "cpu_load_short,host=server01 value=23422.0 1422568543702900257\n"
	testMsgNeg      = "cpu_load_short,host=server01 value=-23422.0 1422568543702900257\n"
	testMsgGraphite = "cpu.load.short.graphite 23422 1454780029"
	testMsgJSON     = "{\"a\": 5, \"b\": {\"c\": 6}}\n"
	invalidMsg      = "cpu_load_short,host=server01 1422568543702900257\n"
)

func newTestMQTTConsumer() (*MQTTConsumer, chan mqtt.Message) {
	in := make(chan mqtt.Message, 100)
	n := &MQTTConsumer{
		Topics:  []string{"telegraf"},
		Servers: []string{"localhost:1883"},
		in:      in,
		done:    make(chan struct{}),
	}
	return n, in
}

// Test that default client has random ID
func TestRandomClientID(t *testing.T) {
	m1 := &MQTTConsumer{
		Servers: []string{"localhost:1883"}}
	opts, err := m1.createOpts()
	assert.NoError(t, err)

	m2 := &MQTTConsumer{
		Servers: []string{"localhost:1883"}}
	opts2, err2 := m2.createOpts()
	assert.NoError(t, err2)

	assert.NotEqual(t, opts.ClientID, opts2.ClientID)
}

// Test that default client has random ID
func TestClientID(t *testing.T) {
	m1 := &MQTTConsumer{
		Servers:  []string{"localhost:1883"},
		ClientID: "telegraf-test",
	}
	opts, err := m1.createOpts()
	assert.NoError(t, err)

	m2 := &MQTTConsumer{
		Servers:  []string{"localhost:1883"},
		ClientID: "telegraf-test",
	}
	opts2, err2 := m2.createOpts()
	assert.NoError(t, err2)

	assert.Equal(t, "telegraf-test", opts2.ClientID)
	assert.Equal(t, "telegraf-test", opts.ClientID)
}

// Test that Start() fails if client ID is not set but persistent is
func TestPersistentClientIDFail(t *testing.T) {
	m1 := &MQTTConsumer{
		Servers:           []string{"localhost:1883"},
		PersistentSession: true,
	}
	acc := testutil.Accumulator{}
	err := m1.Start(&acc)
	assert.Error(t, err)
}

func TestRunParser(t *testing.T) {
	n, in := newTestMQTTConsumer()
	acc := testutil.Accumulator{}
	n.acc = &acc
	defer close(n.done)

	n.parser, _ = parsers.NewInfluxParser()
	go n.receiver()
	in <- mqttMsg(testMsgNeg)
	acc.Wait(1)

	if a := acc.NFields(); a != 1 {
		t.Errorf("got %v, expected %v", a, 1)
	}
}

func TestRunParserNegativeNumber(t *testing.T) {
	n, in := newTestMQTTConsumer()
	acc := testutil.Accumulator{}
	n.acc = &acc
	defer close(n.done)

	n.parser, _ = parsers.NewInfluxParser()
	go n.receiver()
	in <- mqttMsg(testMsg)
	acc.Wait(1)

	if a := acc.NFields(); a != 1 {
		t.Errorf("got %v, expected %v", a, 1)
	}
}

// Test that the parser ignores invalid messages
func TestRunParserInvalidMsg(t *testing.T) {
	n, in := newTestMQTTConsumer()
	acc := testutil.Accumulator{}
	n.acc = &acc
	defer close(n.done)

	n.parser, _ = parsers.NewInfluxParser()
	go n.receiver()
	in <- mqttMsg(invalidMsg)
	acc.WaitError(1)

	if a := acc.NFields(); a != 0 {
		t.Errorf("got %v, expected %v", a, 0)
	}
	assert.Contains(t, acc.Errors[0].Error(), "MQTT Parse Error")
}

// Test that the parser parses line format messages into metrics
func TestRunParserAndGather(t *testing.T) {
	n, in := newTestMQTTConsumer()
	acc := testutil.Accumulator{}
	n.acc = &acc
	defer close(n.done)

	n.parser, _ = parsers.NewInfluxParser()
	go n.receiver()
	in <- mqttMsg(testMsg)
	acc.Wait(1)

	n.Gather(&acc)

	acc.AssertContainsFields(t, "cpu_load_short",
		map[string]interface{}{"value": float64(23422)})
}

// Test that the parser parses graphite format messages into metrics
func TestRunParserAndGatherGraphite(t *testing.T) {
	n, in := newTestMQTTConsumer()
	acc := testutil.Accumulator{}
	n.acc = &acc
	defer close(n.done)

	n.parser, _ = parsers.NewGraphiteParser("_", []string{}, nil)
	go n.receiver()
	in <- mqttMsg(testMsgGraphite)

	n.Gather(&acc)
	acc.Wait(1)

	acc.AssertContainsFields(t, "cpu_load_short_graphite",
		map[string]interface{}{"value": float64(23422)})
}

// Test that the parser parses json format messages into metrics
func TestRunParserAndGatherJSON(t *testing.T) {
	n, in := newTestMQTTConsumer()
	acc := testutil.Accumulator{}
	n.acc = &acc
	defer close(n.done)

	n.parser, _ = parsers.NewJSONParser("nats_json_test", []string{}, nil)
	go n.receiver()
	in <- mqttMsg(testMsgJSON)

	n.Gather(&acc)

	acc.Wait(1)

	acc.AssertContainsFields(t, "nats_json_test",
		map[string]interface{}{
			"a":   float64(5),
			"b_c": float64(6),
		})
}

func mqttMsg(val string) mqtt.Message {
	return &message{
		topic:   "telegraf/unit_test",
		payload: []byte(val),
	}
}

// Take the message struct from the paho mqtt client library for returning
// a test message interface.
type message struct {
	duplicate bool
	qos       byte
	retained  bool
	topic     string
	messageID uint16
	payload   []byte
}

func (m *message) Duplicate() bool {
	return m.duplicate
}

func (m *message) Qos() byte {
	return m.qos
}

func (m *message) Retained() bool {
	return m.retained
}

func (m *message) Topic() string {
	return m.topic
}

func (m *message) MessageID() uint16 {
	return m.messageID
}

func (m *message) Payload() []byte {
	return m.payload
}