test received MQTT messages

test/e2e/data.go

@@ -7,32 +7,20 @@ const (
 	TIC_MODE_STANDARD
 )
 
-type MQTTResult struct {
-	// TODO
-}
-
-type TestStep struct {
-	Sent     []string
-	Expected []MQTTResult
-}
-
 type TestCase struct {
 	Name     string
 	Mode     TicMode
-	Steps []TestStep
+	Sent     [][]string
+	Expected map[string][]string
 }
 
 var testCases []TestCase = []TestCase{
 	{
 		Name: "historique_simple",
 		Mode: TIC_MODE_HISTORIQUE,
-		Steps: []TestStep{
+		Sent: [][]string{
+			{}, // LibTeleinfo explicitely discards the first frame
 			{
-				// LibTeleinfo explicitely discards the first frame
-				Sent: []string{},
-			},
-			{
-				Sent: []string{
 				"MOTDETAT 000000 B",
 				"PPOT 00 #",
 				"OPTARIF HC.. <",
@@ -50,9 +38,7 @@ var testCases []TestCase = []TestCase{
 				"PAPP 00540 *",
 				"HHPHC A ,",
 			},
-			},
 			{
-				Sent: []string{
 				"MOTDETAT 000000 B",
 				"PPOT 00 #",
 				"OPTARIF HC.. <",
@@ -70,9 +56,7 @@ var testCases []TestCase = []TestCase{
 				"PAPP 02420 )",
 				"HHPHC A ,",
 			},
-			},
 			{
-				Sent: []string{
 				"MOTDETAT 000000 B",
 				"PPOT 00 #",
 				"OPTARIF HC.. <",
@@ -91,18 +75,32 @@ var testCases []TestCase = []TestCase{
 				"HHPHC A ,",
 			},
 		},
+		Expected: map[string][]string{
+			"test/esp-tic/status/tic/MOTDETAT": {"000000"},
+			"test/esp-tic/status/tic/PPOT":     {"00"},
+			"test/esp-tic/status/tic/OPTARIF":  {"HC.."},
+			"test/esp-tic/status/tic/ISOUSC":   {"25"},
+			"test/esp-tic/status/tic/HCHC":     {"015558379"},
+			"test/esp-tic/status/tic/HCHP":     {"011651340", "011651341", "011651343"},
+			"test/esp-tic/status/tic/PTEC":     {"HP.."},
+			"test/esp-tic/status/tic/IINST1":   {"001"},
+			"test/esp-tic/status/tic/IINST2":   {"001", "009", "006"},
+			"test/esp-tic/status/tic/IINST3":   {"000"},
+			"test/esp-tic/status/tic/IMAX1":    {"060"},
+			"test/esp-tic/status/tic/IMAX2":    {"060"},
+			"test/esp-tic/status/tic/IMAX3":    {"060"},
+			"test/esp-tic/status/tic/PMAX":     {"08611"},
+			"test/esp-tic/status/tic/PAPP":     {"00540", "02420", "01690"},
+			"test/esp-tic/status/tic/HHPHC":    {"A"},
 		},
 	},
 	{
 		Name: "historique_adps_tri",
 		Mode: TIC_MODE_HISTORIQUE,
-		Steps: []TestStep{
+		Sent: [][]string{
+			{}, // LibTeleinfo explicitely discards the first frame
 			{
-				// LibTeleinfo explicitely discards the first frame
+				"ADCO 123456789012 G",
-				Sent: []string{},
-			},
-			{
-				Sent: []string{
 				"MOTDETAT 000000 B",
 				"PPOT 00 #",
 				"OPTARIF HC.. <",
@@ -120,36 +118,29 @@ var testCases []TestCase = []TestCase{
 				"PAPP 00540 *",
 				"HHPHC A ,",
 			},
-			},
 			{
-				Sent: []string{
 				"ADCO 123456789012 G",
 				"ADIR1 001 \"",
 				"IINST1 001 I",
 				"IINST2 009 R",
 				"IINST3 000 J",
 			},
-			},
 			{
-				Sent: []string{
 				"ADCO 123456789012 G",
 				"ADIR1 001 \"",
 				"IINST1 001 I",
 				"IINST2 009 R",
 				"IINST3 000 J",
 			},
-			},
 			{
-				Sent: []string{
 				"ADCO 123456789012 G",
 				"ADIR1 001 \"",
 				"IINST1 001 I",
 				"IINST2 009 R",
 				"IINST3 000 J",
 			},
-			},
 			{
-				Sent: []string{
+				"ADCO 123456789012 G",
 				"MOTDETAT 000000 B",
 				"PPOT 00 #",
 				"OPTARIF HC.. <",
@@ -168,6 +159,25 @@ var testCases []TestCase = []TestCase{
 				"HHPHC A ,",
 			},
 		},
+		Expected: map[string][]string{
+			"test/esp-tic/status/tic/ADIR1":    {"001", "001", "001"},
+			"test/esp-tic/status/tic/ADCO":     {"123456789012"},
+			"test/esp-tic/status/tic/MOTDETAT": {"000000"},
+			"test/esp-tic/status/tic/PPOT":     {"00"},
+			"test/esp-tic/status/tic/OPTARIF":  {"HC.."},
+			"test/esp-tic/status/tic/ISOUSC":   {"25"},
+			"test/esp-tic/status/tic/HCHC":     {"015558379"},
+			"test/esp-tic/status/tic/HCHP":     {"011651340", "011651343"},
+			"test/esp-tic/status/tic/PTEC":     {"HP.."},
+			"test/esp-tic/status/tic/IINST1":   {"001"},
+			"test/esp-tic/status/tic/IINST2":   {"001", "009", "006"},
+			"test/esp-tic/status/tic/IINST3":   {"000"},
+			"test/esp-tic/status/tic/IMAX1":    {"060"},
+			"test/esp-tic/status/tic/IMAX2":    {"060"},
+			"test/esp-tic/status/tic/IMAX3":    {"060"},
+			"test/esp-tic/status/tic/PMAX":     {"08611"},
+			"test/esp-tic/status/tic/PAPP":     {"00540", "01690"},
+			"test/esp-tic/status/tic/HHPHC":    {"A"},
 		},
 	},
 }

test/e2e/go.mod

@@ -2,4 +2,9 @@ module github.com/nmasse-itix/tic-to-mqtt/test/e2e
 
 go 1.16
 
-require go.bug.st/serial v1.3.4 // indirect
+require (
+	github.com/eclipse/paho.mqtt.golang v1.3.5 // indirect
+	github.com/r3labs/diff/v2 v2.15.0 // indirect
+	go.bug.st/serial v1.3.4 // indirect
+	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+)

test/e2e/go.sum

@@ -1,12 +1,37 @@
 github.com/creack/goselect v0.1.2 h1:2DNy14+JPjRBgPzAd1thbQp4BSIihxcBf0IXhQXDRa0=
 github.com/creack/goselect v0.1.2/go.mod h1:a/NhLweNvqIYMuxcMOuWY516Cimucms3DglDzQP3hKY=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/eclipse/paho.mqtt.golang v1.3.5 h1:sWtmgNxYM9P2sP+xEItMozsR3w0cqZFlqnNN1bdl41Y=
+github.com/eclipse/paho.mqtt.golang v1.3.5/go.mod h1:eTzb4gxwwyWpqBUHGQZ4ABAV7+Jgm1PklsYT/eo8Hcc=
+github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg=
+github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
+github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/r3labs/diff/v2 v2.15.0 h1:3TEoJ6dBqESl1YgL+7curys5PvuEnwrtjkFNskgUvfg=
+github.com/r3labs/diff/v2 v2.15.0/go.mod h1:I8noH9Fc2fjSaMxqF3G2lhDdC0b+JXCfyx85tWFM9kc=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/vmihailenco/msgpack v4.0.4+incompatible h1:dSLoQfGFAo3F6OoNhwUmLwVgaUXK79GlxNBwueZn0xI=
+github.com/vmihailenco/msgpack v4.0.4+incompatible/go.mod h1:fy3FlTQTDXWkZ7Bh6AcGMlsjHatGryHQYUTf1ShIgkk=
 go.bug.st/serial v1.3.4 h1:fMpfNEOsPQjYGZ3VHcs/xxsxoaPgbcjrm4YnMkcir3Y=
 go.bug.st/serial v1.3.4/go.mod h1:z8CesKorE90Qr/oRSJiEuvzYRKol9r/anJZEb5kt304=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
+golang.org/x/net v0.0.0-20200425230154-ff2c4b7c35a0 h1:Jcxah/M+oLZ/R4/z5RzfPzGbPXnVDPkEDtf2JnuxN+U=
+golang.org/x/net v0.0.0-20200425230154-ff2c4b7c35a0/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf h1:2ucpDCmfkl8Bd/FsLtiD653Wf96cW37s+iGx93zsu4k=
 golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+google.golang.org/appengine v1.6.6 h1:lMO5rYAqUxkmaj76jAkRUvt5JZgFymx/+Q5Mzfivuhc=
+google.golang.org/appengine v1.6.6/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

test/e2e/main.go

@@ -2,24 +2,58 @@ package main
 
 import (
 	"bytes"
+	"encoding/json"
 	"flag"
 	"fmt"
 	"os"
-	"sort"
+	"reflect"
+	"strings"
+	"sync"
 	"time"
 
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	diff "github.com/r3labs/diff/v2"
 	"go.bug.st/serial"
+	yaml "gopkg.in/yaml.v3"
 )
 
+const (
+	MQTT_QOS_0 = 0
+	MQTT_QOS_1 = 1
+	MQTT_QOS_2 = 2
+)
+
+type TicData struct {
+	Value     string `json:"val"`
+	Timestamp int64  `json:"ts"`
+}
+
 func main() {
+	//mqtt.ERROR = log.New(os.Stdout, "[ERROR] ", 0)
+	//mqtt.CRITICAL = log.New(os.Stdout, "[CRIT] ", 0)
+	//mqtt.WARN = log.New(os.Stdout, "[WARN]  ", 0)
+	//mqtt.DEBUG = log.New(os.Stdout, "[DEBUG]  ", 0)
+
 	var portName string
+	var mqttUri string
+	var mqttUsername string
+	var mqttPassword string
+	var mqttClientId string
+	var mqttClient mqtt.Client
 
 	flag.StringVar(&portName, "p", "/dev/ttyUSB1", "Serial port to use")
+	flag.StringVar(&mqttUri, "s", "", "MQTT Server URI (tcp://server:port or ssl://server:port)")
+	flag.StringVar(&mqttUsername, "u", "", "MQTT Username")
+	flag.StringVar(&mqttPassword, "w", "", "MQTT Password")
+	flag.StringVar(&mqttClientId, "c", "esptic-e2e-test", "MQTT Client ID")
 	flag.Parse()
 
-	wanted := flag.Args()
+	cases := flag.Args()
-	if len(wanted) == 0 {
+	if len(cases) == 0 || len(cases) > 1 {
-		fmt.Println("Usage: e2e [-t /dev/ttyUSBX ] test_case1 test_case2 ...")
+		fmt.Println("Usage: e2e [options] test_case")
+		fmt.Println()
+		fmt.Println("Options:")
+		flag.PrintDefaults()
 		fmt.Println()
 		fmt.Println("Available test cases:")
 		for _, testCase := range testCases {
@@ -27,10 +61,38 @@ func main() {
 		}
 		os.Exit(1)
 	}
-	sort.Strings(wanted)
+	testcaseName := cases[0]
+
+	if mqttUri != "" {
+		opts := mqtt.NewClientOptions()
+		opts.AddBroker(mqttUri)
+		opts.SetAutoReconnect(false)
+		opts.SetConnectRetry(false)
+		opts.SetOrderMatters(false)
+		opts.SetCleanSession(true)
+		opts.SetClientID(mqttClientId)
+		if mqttUsername != "" {
+			opts.SetUsername(mqttUsername)
+		}
+		if mqttPassword != "" {
+			opts.SetPassword(mqttPassword)
+		}
+		mqttClient = mqtt.NewClient(opts)
+		fmt.Printf("Connecting to MQTT server %s...\n", mqttUri)
+		ct := mqttClient.Connect()
+		if !ct.WaitTimeout(30 * time.Second) {
+			fmt.Println("Timeout waiting for MQTT server!")
+			os.Exit(1)
+		}
+		if ct.Error() != nil {
+			fmt.Println(ct.Error())
+			os.Exit(1)
+		}
+	}
+
 	for _, testCase := range testCases {
-		i := sort.SearchStrings(wanted, testCase.Name)
+		var mut sync.Mutex
-		if i >= len(wanted) || wanted[i] != testCase.Name {
+		if testCase.Name != testcaseName {
 			continue
 		}
 
@@ -66,11 +128,40 @@ func main() {
 		}
 		defer port.Close()
 
-		for i, step := range testCase.Steps {
+		var mqttObservedValues map[string][]string = make(map[string][]string, len(testCase.Expected))
+		if mqttUri != "" {
+			var topics map[string]byte = make(map[string]byte, len(testCase.Expected))
+			for topic := range testCase.Expected {
+				topics[topic] = MQTT_QOS_1
+			}
+			st := mqttClient.SubscribeMultiple(topics, func(c mqtt.Client, m mqtt.Message) {
+				if m.Retained() {
+					return
+				}
+				var data TicData
+				err := json.Unmarshal(m.Payload(), &data)
+				if err != nil {
+					fmt.Printf("Cannot unmarshal JSON payload '%s'\n", string(m.Payload()))
+					return
+				}
+
+				mut.Lock()
+				defer mut.Unlock()
+				values, ok := mqttObservedValues[m.Topic()]
+				if ok {
+					mqttObservedValues[m.Topic()] = append(values, data.Value)
+				} else {
+					mqttObservedValues[m.Topic()] = []string{data.Value}
+				}
+			})
+			st.Wait() // Happy path: there won't be a timeout...
+		}
+
+		for i, frame := range testCase.Sent {
 			fmt.Printf("Sending trame %d...\n", i)
 			var b bytes.Buffer
 			b.WriteByte(0x02)
-			for _, info := range step.Sent {
+			for _, info := range frame {
 				b.WriteString(fmt.Sprintf("\n%s\r", info))
 			}
 			b.WriteByte(0x03)
@@ -85,7 +176,75 @@ func main() {
 			// Can be any value between 16.7 and 33.4 ms
 			time.Sleep(33 * time.Millisecond)
 		}
+
+		fmt.Println("Waiting for MQTT messages...")
+		success := false
+		for i := 0; i < 15; i++ {
+			mut.Lock()
+			ok := reflect.DeepEqual(mqttObservedValues, testCase.Expected)
+			mut.Unlock()
+			if ok {
+				success = true
+				break
+			}
+
+			// there may be ongoing MQTT messages waiting to be sent/delivered
+			// wait one second and try again
+			time.Sleep(time.Second)
+		}
+
+		if !success {
+			changelog, err := diff.Diff(testCase.Expected, mqttObservedValues)
+			if err != nil {
+				fmt.Println(err)
+				os.Exit(1)
+			}
+			fmt.Println("TEST FAILED!")
+			fmt.Println()
+			fmt.Println("Changelog:")
+			fmt.Println()
+			for _, change := range changelog {
+				fmt.Printf("%s: %s: %s -> %s\n", change.Type, strings.Join(change.Path, "."), str(change.From), str(change.To))
+			}
+			fmt.Println()
+			fmt.Println("Expected:")
+			b, _ := yaml.Marshal(testCase.Expected)
+			fmt.Println(string(b))
+			fmt.Println()
+			fmt.Println("Got:")
+			b, _ = yaml.Marshal(mqttObservedValues)
+			fmt.Println(string(b))
+			fmt.Println()
+			os.Exit(1)
+		}
+
+		if mqttUri != "" {
+			var topics []string = make([]string, len(testCase.Expected))
+			i := 0
+			for topic := range testCase.Expected {
+				topics[i] = topic
+				i++
+			}
+			ut := mqttClient.Unsubscribe(topics...)
+			ut.Wait() // Happy path: there won't be a timeout...
+		}
+
+		fmt.Println("TEST SUCCEEDED!")
+
+		break
+	}
+
+	if mqttClient.IsConnected() {
+		mqttClient.Disconnect(0)
 	}
 
 	fmt.Println("Done.")
 }
+
+func str(s interface{}) string {
+	if s == nil {
+		return "nil"
+	}
+
+	return fmt.Sprintf("%s", s)
+}