Update jobs stats tests

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

5 files changed, 30 insertions, 27 deletions

diff --git a/tests/plugins/jobs/helpers.go b/tests/plugins/jobs/helpers.go
index 9c314494..5ec81cec 100644
--- a/tests/plugins/jobs/helpers.go
+++ b/tests/plugins/jobs/helpers.go
@@ -210,22 +210,24 @@ func deleteProxy(name string, t *testing.T) {
 	}
 }
 
-func stats(t *testing.T, state *jobState.State) {
-	conn, err := net.Dial("tcp", "127.0.0.1:6001")
-	assert.NoError(t, err)
-	client := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn))
+func stats(state *jobState.State) func(t *testing.T) {
+	return func(t *testing.T) {
+		conn, err := net.Dial("tcp", "127.0.0.1:6001")
+		assert.NoError(t, err)
+		client := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn))
 
-	st := &jobsv1beta.Stats{}
-	er := &jobsv1beta.Empty{}
+		st := &jobsv1beta.Stats{}
+		er := &jobsv1beta.Empty{}
 
-	err = client.Call(stat, er, st)
-	require.NoError(t, err)
-	require.NotNil(t, st)
-
-	state.Queue = st.Stats[0].Queue
-	state.Pipeline = st.Stats[0].Pipeline
-	state.Driver = st.Stats[0].Driver
-	state.Active = st.Stats[0].Active
-	state.Delayed = st.Stats[0].Delayed
-	state.Reserved = st.Stats[0].Reserved
+		err = client.Call(stat, er, st)
+		require.NoError(t, err)
+		require.NotNil(t, st)
+
+		state.Queue = st.Stats[0].Queue
+		state.Pipeline = st.Stats[0].Pipeline
+		state.Driver = st.Stats[0].Driver
+		state.Active = st.Stats[0].Active
+		state.Delayed = st.Stats[0].Delayed
+		state.Reserved = st.Stats[0].Reserved
+	}
 }
diff --git a/tests/plugins/jobs/jobs_amqp_test.go b/tests/plugins/jobs/jobs_amqp_test.go
index 4c1f600a..c35244c5 100644
--- a/tests/plugins/jobs/jobs_amqp_test.go
+++ b/tests/plugins/jobs/jobs_amqp_test.go
@@ -428,7 +428,8 @@ func TestAMQPStats(t *testing.T) {
 	t.Run("PushPipelineDelayed", pushToPipeDelayed("test-3", 5))
 
 	out := &jobState.State{}
-	stats(t, out)
+	t.Run("Stats", stats(out))
+
 	assert.Equal(t, out.Pipeline, "test-3")
 	assert.Equal(t, out.Driver, "amqp")
 	assert.Equal(t, out.Queue, "default")
@@ -442,7 +443,7 @@ func TestAMQPStats(t *testing.T) {
 	time.Sleep(time.Second * 7)
 
 	out = &jobState.State{}
-	stats(t, out)
+	t.Run("Stats", stats(out))
 
 	assert.Equal(t, out.Pipeline, "test-3")
 	assert.Equal(t, out.Driver, "amqp")
diff --git a/tests/plugins/jobs/jobs_beanstalk_test.go b/tests/plugins/jobs/jobs_beanstalk_test.go
index a6eb3d4e..ba798ee0 100644
--- a/tests/plugins/jobs/jobs_beanstalk_test.go
+++ b/tests/plugins/jobs/jobs_beanstalk_test.go
@@ -408,22 +408,22 @@ func TestBeanstalkStats(t *testing.T) {
 	time.Sleep(time.Second * 2)
 
 	out := &jobState.State{}
-	stats(t, out)
+	t.Run("Stats", stats(out))
 
 	assert.Equal(t, out.Pipeline, "test-3")
 	assert.Equal(t, out.Driver, "beanstalk")
 	assert.Equal(t, out.Queue, "default")
 
-	assert.Equal(t, int64(0), out.Active)
+	assert.Equal(t, int64(1), out.Active)
 	assert.Equal(t, int64(1), out.Delayed)
-	assert.Equal(t, int64(1), out.Reserved)
+	assert.Equal(t, int64(0), out.Reserved)
 
 	time.Sleep(time.Second)
 	t.Run("ResumePipeline", resumePipes("test-3"))
 	time.Sleep(time.Second * 7)
 
 	out = &jobState.State{}
-	stats(t, out)
+	t.Run("Stats", stats(out))
 
 	assert.Equal(t, out.Pipeline, "test-3")
 	assert.Equal(t, out.Driver, "beanstalk")
@@ -436,7 +436,7 @@ func TestBeanstalkStats(t *testing.T) {
 	time.Sleep(time.Second)
 	t.Run("DestroyPipeline", destroyPipelines("test-3"))
 
-	time.Sleep(time.Second * 5)
+	time.Sleep(time.Second)
 	stopCh <- struct{}{}
 	wg.Wait()
 }
diff --git a/tests/plugins/jobs/jobs_ephemeral_test.go b/tests/plugins/jobs/jobs_ephemeral_test.go
index 3f296c83..f793da6d 100644
--- a/tests/plugins/jobs/jobs_ephemeral_test.go
+++ b/tests/plugins/jobs/jobs_ephemeral_test.go
@@ -493,7 +493,7 @@ func TestEphemeralStats(t *testing.T) {
 
 	time.Sleep(time.Second)
 	out := &jobState.State{}
-	stats(t, out)
+	t.Run("Stats", stats(out))
 
 	assert.Equal(t, out.Pipeline, "test-3")
 	assert.Equal(t, out.Driver, "ephemeral")
@@ -508,7 +508,7 @@ func TestEphemeralStats(t *testing.T) {
 	time.Sleep(time.Second * 7)
 
 	out = &jobState.State{}
-	stats(t, out)
+	t.Run("Stats", stats(out))
 
 	assert.Equal(t, out.Pipeline, "test-3")
 	assert.Equal(t, out.Driver, "ephemeral")
diff --git a/tests/plugins/jobs/jobs_sqs_test.go b/tests/plugins/jobs/jobs_sqs_test.go
index e5a92916..191c8212 100644
--- a/tests/plugins/jobs/jobs_sqs_test.go
+++ b/tests/plugins/jobs/jobs_sqs_test.go
@@ -438,7 +438,7 @@ func TestSQSStat(t *testing.T) {
 	time.Sleep(time.Second)
 
 	out := &jobState.State{}
-	stats(t, out)
+	t.Run("Stats", stats(out))
 
 	assert.Equal(t, out.Pipeline, "test-3")
 	assert.Equal(t, out.Driver, "sqs")
@@ -453,7 +453,7 @@ func TestSQSStat(t *testing.T) {
 	time.Sleep(time.Second * 7)
 
 	out = &jobState.State{}
-	stats(t, out)
+	t.Run("Stats", stats(out))
 
 	assert.Equal(t, out.Pipeline, "test-3")
 	assert.Equal(t, out.Driver, "sqs")