Update broadcast tests, add redis flusing. Initial impl of the job

drivers state.

Signed-off-by: Valery Piashchynski <[email protected]>

5 files changed, 31 insertions, 4 deletions

diff --git a/tests/env/docker-compose.yaml b/tests/env/docker-compose.yaml
index 4f58d543..4735070a 100644
--- a/tests/env/docker-compose.yaml
+++ b/tests/env/docker-compose.yaml
@@ -1,4 +1,4 @@
-version: '3.8'
+version: '3'
 
 services:
   memcached:
diff --git a/tests/plugins/broadcast/broadcast_plugin_test.go b/tests/plugins/broadcast/broadcast_plugin_test.go
index a78b17e1..c7041cc9 100644
--- a/tests/plugins/broadcast/broadcast_plugin_test.go
+++ b/tests/plugins/broadcast/broadcast_plugin_test.go
@@ -1,6 +1,7 @@
 package broadcast
 
 import (
+	"context"
 	"net"
 	"net/rpc"
 	"os"
@@ -10,6 +11,7 @@ import (
 	"testing"
 	"time"
 
+	goRedis "github.com/go-redis/redis/v8"
 	"github.com/golang/mock/gomock"
 	endure "github.com/spiral/endure/pkg/container"
 	goridgeRpc "github.com/spiral/goridge/v3/pkg/rpc"
@@ -176,6 +178,9 @@ func TestBroadcastNoConfig(t *testing.T) {
 }
 
 func TestBroadcastSameSubscriber(t *testing.T) {
+	t.Run("RedisFlush", redisFlushAll("127.0.0.1:6379"))
+	t.Run("RedisFlush", redisFlushAll("127.0.0.1:6378"))
+
 	cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.ErrorLevel), endure.GracefulShutdownTimeout(time.Second))
 	assert.NoError(t, err)
 
@@ -283,9 +288,15 @@ func TestBroadcastSameSubscriber(t *testing.T) {
 	stopCh <- struct{}{}
 
 	wg.Wait()
+
+	t.Run("RedisFlush", redisFlushAll("127.0.0.1:6379"))
+	t.Run("RedisFlush", redisFlushAll("127.0.0.1:6378"))
 }
 
 func TestBroadcastSameSubscriberGlobal(t *testing.T) {
+	t.Run("RedisFlush", redisFlushAll("127.0.0.1:6379"))
+	t.Run("RedisFlush", redisFlushAll("127.0.0.1:6378"))
+
 	cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.ErrorLevel), endure.GracefulShutdownTimeout(time.Second))
 	assert.NoError(t, err)
 
@@ -394,6 +405,9 @@ func TestBroadcastSameSubscriberGlobal(t *testing.T) {
 
 	wg.Wait()
 	time.Sleep(time.Second * 5)
+
+	t.Run("RedisFlush", redisFlushAll("127.0.0.1:6379"))
+	t.Run("RedisFlush", redisFlushAll("127.0.0.1:6378"))
 }
 
 func BroadcastPublishFooFoo2Foo3(port string) func(t *testing.T) {
@@ -446,6 +460,7 @@ func BroadcastPublishFoo3(port string) func(t *testing.T) {
 		}
 	}
 }
+
 func BroadcastPublishAsyncFooFoo2Foo3(port string) func(t *testing.T) {
 	return func(t *testing.T) {
 		conn, err := net.Dial("tcp", "127.0.0.1:"+port)
@@ -475,3 +490,15 @@ func makeMessage(payload []byte, topics ...string) *websocketsv1.Request {
 
 	return m
 }
+
+func redisFlushAll(addr string) func(t *testing.T) {
+	return func(t *testing.T) {
+		rdb := goRedis.NewClient(&goRedis.Options{
+			Addr:     addr,
+			Password: "", // no password set
+			DB:       0,  // use default DB
+		})
+
+		rdb.FlushAll(context.Background())
+	}
+}
diff --git a/tests/plugins/http/http_plugin_test.go b/tests/plugins/http/http_plugin_test.go
index bd804264..a48c8972 100644
--- a/tests/plugins/http/http_plugin_test.go
+++ b/tests/plugins/http/http_plugin_test.go
@@ -22,7 +22,7 @@ import (
 	"github.com/golang/mock/gomock"
 	endure "github.com/spiral/endure/pkg/container"
 	goridgeRpc "github.com/spiral/goridge/v3/pkg/rpc"
-	"github.com/spiral/roadrunner/v2/pkg/process"
+	"github.com/spiral/roadrunner/v2/pkg/state/process"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/gzip"
 	"github.com/spiral/roadrunner/v2/plugins/informer"
diff --git a/tests/plugins/informer/informer_test.go b/tests/plugins/informer/informer_test.go
index 61be85a1..c3b5c6a6 100644
--- a/tests/plugins/informer/informer_test.go
+++ b/tests/plugins/informer/informer_test.go
@@ -12,7 +12,7 @@ import (
 
 	endure "github.com/spiral/endure/pkg/container"
 	goridgeRpc "github.com/spiral/goridge/v3/pkg/rpc"
-	"github.com/spiral/roadrunner/v2/pkg/process"
+	"github.com/spiral/roadrunner/v2/pkg/state/process"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/informer"
 	"github.com/spiral/roadrunner/v2/plugins/logger"
diff --git a/tests/plugins/informer/test_plugin.go b/tests/plugins/informer/test_plugin.go
index 095140b8..21897f40 100644
--- a/tests/plugins/informer/test_plugin.go
+++ b/tests/plugins/informer/test_plugin.go
@@ -5,7 +5,7 @@ import (
 	"time"
 
 	"github.com/spiral/roadrunner/v2/pkg/pool"
-	"github.com/spiral/roadrunner/v2/pkg/process"
+	"github.com/spiral/roadrunner/v2/pkg/state/process"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/server"
 )