Repository: lf-edge/ekuiper
Pull request #2914 (2024-06-11 05:20:33)

test: revise mqtt test with embedded broker

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2914?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `82.60870%` with `4 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@6e0a346`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#section-missing-base-commit) about missing BASE report.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2914?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/testx/testUtil.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2914?src=pr&el=tree&filepath=internal%2Ftestx%2FtestUtil.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdGVzdHgvdGVzdFV0aWwuZ28=) | 82.61% | [2 Missing and 2 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2914?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff              @@
##             master-v2    #2914   +/-   ##
============================================
  Coverage             ?   68.65%           
============================================
  Files                ?      297           
  Lines                ?    28654           
  Branches             ?        0           
============================================
  Hits                 ?    19672           
  Misses               ?     7252           
  Partials             ?     1730           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2914?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/go.mod b/go.mod
index d0c045a6..b3e6db32 100644
--- a/go.mod
+++ b/go.mod
@@ -26,6 +26,7 @@ require (
 	github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible
 	github.com/lf-edge/ekuiper/contract/v2 v2.0.0-alpha.2
 	github.com/mitchellh/mapstructure v1.5.0
+	github.com/mochi-mqtt/server/v2 v2.6.4
 	github.com/modern-go/reflect2 v1.0.2
 	github.com/montanaflynn/stats v0.7.1
 	github.com/msgpack-rpc/msgpack-rpc-go v0.0.0-20131026060856-c76397e1782b
@@ -104,6 +105,7 @@ require (
 	github.com/prometheus/procfs v0.13.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/rogpeppe/go-internal v1.12.0 // indirect
+	github.com/rs/xid v1.4.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/shopspring/decimal v1.3.1 // indirect
 	github.com/spf13/cast v1.6.0 // indirect
@@ -114,7 +116,7 @@ require (
 	go.uber.org/goleak v1.3.0 // indirect
 	golang.org/x/crypto v0.21.0 // indirect
 	golang.org/x/mod v0.16.0 // indirect
-	golang.org/x/net v0.22.0 // indirect
+	golang.org/x/net v0.23.0 // indirect
 	golang.org/x/sync v0.6.0 // indirect
 	golang.org/x/sys v0.18.0 // indirect
 	golang.org/x/tools v0.19.0 // indirect
diff --git a/go.sum b/go.sum
index 32961b5e..06413f79 100644
--- a/go.sum
+++ b/go.sum
@@ -32,6 +32,8 @@ github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
 github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/bufbuild/protocompile v0.9.0 h1:DI8qLG5PEO0Mu1Oj51YFPqtx6I3qYXUAhJVJ/IzAVl0=
 github.com/bufbuild/protocompile v0.9.0/go.mod h1:s89m1O8CqSYpyE/YaSGtg1r1YFMF5nLTwh4vlj6O444=
+github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
+github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
 github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
@@ -151,6 +153,8 @@ github.com/imdario/mergo v0.3.16 h1:wwQJbIsHYGMUyLSPrEq1CT16AhnhNJQ51+4fdHUnCl4=
 github.com/imdario/mergo v0.3.16/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY=
 github.com/jhump/protoreflect v1.15.6 h1:WMYJbw2Wo+KOWwZFvgY0jMoVHM6i4XIvRs2RcBj5VmI=
 github.com/jhump/protoreflect v1.15.6/go.mod h1:jCHoyYQIJnaabEYnbGwyo9hUqfyUMTbJw/tAut5t97E=
+github.com/jinzhu/copier v0.3.5 h1:GlvfUwHk62RokgqVNvYsku0TATCF7bAHVwEXoBh3iJg=
+github.com/jinzhu/copier v0.3.5/go.mod h1:DfbEm0FYsaqBcKcFuvmOZb218JkPGtvSHsKg8S8hyyg=
 github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
 github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
 github.com/jonboulle/clockwork v0.2.2 h1:UOGuzwb1PwsrDAObMuhUnj0p5ULPj8V/xJ7Kx9qUBdQ=
@@ -190,6 +194,8 @@ github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RR
 github.com/mitchellh/reflectwalk v1.0.0/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
 github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zxSIeXaQ=
 github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
+github.com/mochi-mqtt/server/v2 v2.6.4 h1:zuKokG/YzmefLecpodu1VSOSXJf1GP9mk2LdVcp1Jp4=
+github.com/mochi-mqtt/server/v2 v2.6.4/go.mod h1:TqztjKGO0/ArOjJt9x9idk0kqPT3CVN8Pb+l+PS5Gdo=
 github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/montanaflynn/stats v0.7.1 h1:etflOAAHORrCC44V+aR6Ftzort912ZU+YLiSTuV8eaE=
@@ -250,6 +256,8 @@ github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzG
 github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
 github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
+github.com/rs/xid v1.4.0 h1:qd7wPTDkN6KQx2VmMBLrpHkiyQwgFXRnkOLacUiaSNY=
+github.com/rs/xid v1.4.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
@@ -330,8 +338,8 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.2.0/go.mod h1:KqCZLdyyvdV855qA2rE3GC2aiw5xGR5TEjj8smXukLY=
-golang.org/x/net v0.22.0 h1:9sGLhx7iRIHEiX0oAJ3MRZMUCElJgy7Br1nO+AMN3Tc=
-golang.org/x/net v0.22.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
+golang.org/x/net v0.23.0 h1:7EYJ93RZ9vYSZAIb2x3lnuvqO5zneoD6IvWjuhfxjTs=
+golang.org/x/net v0.23.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=


Test patch:
diff --git a/.github/workflows/run_test_case.yaml b/.github/workflows/run_test_case.yaml
index 3ca6488a..0dc38ffa 100644
--- a/.github/workflows/run_test_case.yaml
+++ b/.github/workflows/run_test_case.yaml
@@ -20,11 +20,6 @@ on:
 jobs:
   run_unit_tests:
     runs-on: ubuntu-latest
-    services:
-      emqx:
-        image: emqx/emqx:4.3.10
-        ports:
-          - 1883:1883
     steps:
     - uses: actions/checkout@v3
     - uses: actions/setup-go@v4
diff --git a/internal/io/mqtt/client/client_test.go b/internal/io/mqtt/client/client_test.go
index b4e180bf..b5881a04 100644
--- a/internal/io/mqtt/client/client_test.go
+++ b/internal/io/mqtt/client/client_test.go
@@ -18,12 +18,20 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
 )
 
-// NOTICE!!! Need to run a MQTT broker in localhost:1883 for this test or change the url to your broker
 const url = "tcp://127.0.0.1:1883"
 
 func TestValidate(t *testing.T) {
+	cancel, err := testx.InitBroker()
+	require.NoError(t, err)
+	defer func() {
+		cancel()
+	}()
+
 	tests := []struct {
 		name  string
 		props map[string]any
diff --git a/internal/io/mqtt/source_sink_test.go b/internal/io/mqtt/source_sink_test.go
index f7544079..c93c9037 100644
--- a/internal/io/mqtt/source_sink_test.go
+++ b/internal/io/mqtt/source_sink_test.go
@@ -24,12 +24,18 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mockclock"
 	"github.com/lf-edge/ekuiper/v2/pkg/mock"
 	"github.com/lf-edge/ekuiper/v2/pkg/model"
 )
 
 func TestSourceSink(t *testing.T) {
+	cancel, err := testx.InitBroker()
+	require.NoError(t, err)
+	defer func() {
+		cancel()
+	}()
 	dataDir, err := conf.GetDataLoc()
 	require.NoError(t, err)
 	require.NoError(t, store.SetupDefault(dataDir))
diff --git a/internal/io/mqtt/source_test.go b/internal/io/mqtt/source_test.go
index 0fefb880..51e2f5e5 100644
--- a/internal/io/mqtt/source_test.go
+++ b/internal/io/mqtt/source_test.go
@@ -30,7 +30,6 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
-// NOTICE!!! Need to run a MQTT broker in localhost:1883 for this test or change the url to your broker
 const url = "tcp://127.0.0.1:1883"
 
 func init() {
@@ -39,6 +38,11 @@ func init() {
 }
 
 func TestProvision(t *testing.T) {
+	cancel, err := testx.InitBroker()
+	require.NoError(t, err)
+	defer func() {
+		cancel()
+	}()
 	dataDir, err := conf.GetDataLoc()
 	require.NoError(t, err)
 	require.NoError(t, store.SetupDefault(dataDir))
diff --git a/internal/testx/testUtil.go b/internal/testx/testUtil.go
index eb500e3d..680a89e2 100644
--- a/internal/testx/testUtil.go
+++ b/internal/testx/testUtil.go
@@ -15,6 +15,14 @@
 package testx
 
 import (
+	"context"
+	"log"
+	"sync"
+
+	mqtt "github.com/mochi-mqtt/server/v2"
+	"github.com/mochi-mqtt/server/v2/hooks/auth"
+	"github.com/mochi-mqtt/server/v2/listeners"
+
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 )
@@ -39,3 +47,38 @@ func InitEnv(id string) {
 		conf.Log.Fatal(err)
 	}
 }
+
+func InitBroker() (func(), error) {
+	// Create the new MQTT Server.
+	server := mqtt.New(nil)
+	// Allow all connections.
+	_ = server.AddHook(new(auth.AllowHook), nil)
+
+	// Create a TCP listener on a standard port.
+	tcp := listeners.NewTCP(listeners.Config{ID: "t1", Address: ":1883"})
+	err := server.AddListener(tcp)
+	if err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		select {
+		case <-ctx.Done():
+			server.Close()
+			wg.Done()
+		}
+	}()
+	go func() {
+		err := server.Serve()
+		if err != nil {
+			log.Fatal(err)
+		}
+	}()
+	return func() {
+		cancel()
+		// wait server close
+		wg.Wait()
+	}, nil
+}