123 lines
3.0 KiB
Go
123 lines
3.0 KiB
Go
package socket_listener
|
|
|
|
import (
|
|
"net"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/influxdata/telegraf/testutil"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
func TestSocketListener_tcp(t *testing.T) {
|
|
sl := newSocketListener()
|
|
sl.ServiceAddress = "tcp://127.0.0.1:0"
|
|
|
|
acc := &testutil.Accumulator{}
|
|
err := sl.Start(acc)
|
|
require.NoError(t, err)
|
|
|
|
client, err := net.Dial("tcp", sl.Closer.(net.Listener).Addr().String())
|
|
require.NoError(t, err)
|
|
|
|
testSocketListener(t, sl, client)
|
|
}
|
|
|
|
func TestSocketListener_udp(t *testing.T) {
|
|
sl := newSocketListener()
|
|
sl.ServiceAddress = "udp://127.0.0.1:0"
|
|
|
|
acc := &testutil.Accumulator{}
|
|
err := sl.Start(acc)
|
|
require.NoError(t, err)
|
|
|
|
client, err := net.Dial("udp", sl.Closer.(net.PacketConn).LocalAddr().String())
|
|
require.NoError(t, err)
|
|
|
|
testSocketListener(t, sl, client)
|
|
}
|
|
|
|
func TestSocketListener_unix(t *testing.T) {
|
|
defer os.Remove("/tmp/telegraf_test.sock")
|
|
sl := newSocketListener()
|
|
sl.ServiceAddress = "unix:///tmp/telegraf_test.sock"
|
|
|
|
acc := &testutil.Accumulator{}
|
|
err := sl.Start(acc)
|
|
require.NoError(t, err)
|
|
|
|
client, err := net.Dial("unix", "/tmp/telegraf_test.sock")
|
|
require.NoError(t, err)
|
|
|
|
testSocketListener(t, sl, client)
|
|
}
|
|
|
|
func TestSocketListener_unixgram(t *testing.T) {
|
|
defer os.Remove("/tmp/telegraf_test.sock")
|
|
sl := newSocketListener()
|
|
sl.ServiceAddress = "unixgram:///tmp/telegraf_test.sock"
|
|
|
|
acc := &testutil.Accumulator{}
|
|
err := sl.Start(acc)
|
|
require.NoError(t, err)
|
|
|
|
client, err := net.Dial("unixgram", "/tmp/telegraf_test.sock")
|
|
require.NoError(t, err)
|
|
|
|
testSocketListener(t, sl, client)
|
|
}
|
|
|
|
func testSocketListener(t *testing.T, sl *SocketListener, client net.Conn) {
|
|
mstr12 := "test,foo=bar v=1i 123456789\ntest,foo=baz v=2i 123456790\n"
|
|
mstr3 := "test,foo=zab v=3i 123456791"
|
|
client.Write([]byte(mstr12))
|
|
client.Write([]byte(mstr3))
|
|
if _, ok := client.(net.Conn); ok {
|
|
// stream connection. needs trailing newline to terminate mstr3
|
|
client.Write([]byte{'\n'})
|
|
}
|
|
|
|
acc := sl.Accumulator.(*testutil.Accumulator)
|
|
|
|
acc.Lock()
|
|
if len(acc.Metrics) < 1 {
|
|
acc.Wait()
|
|
}
|
|
require.True(t, len(acc.Metrics) >= 1)
|
|
m := acc.Metrics[0]
|
|
acc.Unlock()
|
|
|
|
assert.Equal(t, "test", m.Measurement)
|
|
assert.Equal(t, map[string]string{"foo": "bar"}, m.Tags)
|
|
assert.Equal(t, map[string]interface{}{"v": int64(1)}, m.Fields)
|
|
assert.True(t, time.Unix(0, 123456789).Equal(m.Time))
|
|
|
|
acc.Lock()
|
|
if len(acc.Metrics) < 2 {
|
|
acc.Wait()
|
|
}
|
|
require.True(t, len(acc.Metrics) >= 2)
|
|
m = acc.Metrics[1]
|
|
acc.Unlock()
|
|
|
|
assert.Equal(t, "test", m.Measurement)
|
|
assert.Equal(t, map[string]string{"foo": "baz"}, m.Tags)
|
|
assert.Equal(t, map[string]interface{}{"v": int64(2)}, m.Fields)
|
|
assert.True(t, time.Unix(0, 123456790).Equal(m.Time))
|
|
|
|
acc.Lock()
|
|
if len(acc.Metrics) < 3 {
|
|
acc.Wait()
|
|
}
|
|
require.True(t, len(acc.Metrics) >= 3)
|
|
m = acc.Metrics[2]
|
|
acc.Unlock()
|
|
|
|
assert.Equal(t, "test", m.Measurement)
|
|
assert.Equal(t, map[string]string{"foo": "zab"}, m.Tags)
|
|
assert.Equal(t, map[string]interface{}{"v": int64(3)}, m.Fields)
|
|
assert.True(t, time.Unix(0, 123456791).Equal(m.Time))
|
|
}
|