Allow specifying state notifier in tests (#414)

Author: biglittlebigben

test/integration.go

@@ -34,6 +34,7 @@ import (
 
 	"github.com/livekit/ingress/pkg/config"
 	"github.com/livekit/ingress/pkg/params"
+	"github.com/livekit/ingress/pkg/utils"
 )
 
 type TestConfig struct {
@@ -132,7 +133,7 @@ func getConfig(t *testing.T) *TestConfig {
 	return tc
 }
 
-func RunTestSuite(t *testing.T, conf *TestConfig, bus psrpc.MessageBus, newCmd func(ctx context.Context, p *params.Params) (*exec.Cmd, error)) {
+func RunTestSuite(t *testing.T, conf *TestConfig, bus psrpc.MessageBus, getStateNotifier func(psrpcClient rpc.IOInfoClient) utils.StateNotifier, newCmd func(ctx context.Context, p *params.Params) (*exec.Cmd, error)) {
 	psrpcClient, err := rpc.NewIOInfoClient(bus)
 	require.NoError(t, err)
 
@@ -142,19 +143,21 @@ func RunTestSuite(t *testing.T, conf *TestConfig, bus psrpc.MessageBus, newCmd f
 	commandPsrpcClient, err := rpc.NewIngressHandlerClient(bus, psrpc.WithClientTimeout(5*time.Second))
 	require.NoError(t, err)
 
+	sn := getStateNotifier(psrpcClient)
+
 	if !conf.WhipOnly && !conf.URLOnly {
 		t.Run("RTMP", func(t *testing.T) {
-			RunRTMPTest(t, conf, bus, commandPsrpcClient, psrpcClient, newCmd)
+			RunRTMPTest(t, conf, bus, commandPsrpcClient, psrpcClient, sn, newCmd)
 		})
 	}
 	if !conf.RtmpOnly && !conf.URLOnly {
 		t.Run("WHIP", func(t *testing.T) {
-			RunWHIPTest(t, conf, bus, commandPsrpcClient, psrpcClient, newCmd)
+			RunWHIPTest(t, conf, bus, commandPsrpcClient, psrpcClient, sn, newCmd)
 		})
 	}
 	if !conf.RtmpOnly && !conf.WhipOnly {
 		t.Run("URL pul", func(t *testing.T) {
-			RunURLTest(t, conf, bus, commandPsrpcClient, psrpcClient, newCmd)
+			RunURLTest(t, conf, bus, commandPsrpcClient, psrpcClient, sn, newCmd)
 		})
 	}
 

test/integration_test.go

@@ -19,9 +19,12 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"github.com/livekit/ingress/pkg/service"
 	"github.com/livekit/protocol/redis"
+	"github.com/livekit/protocol/rpc"
 	"github.com/livekit/psrpc"
+
+	"github.com/livekit/ingress/pkg/service"
+	"github.com/livekit/ingress/pkg/utils"
 )
 
 func TestIngress(t *testing.T) {
@@ -33,5 +36,7 @@ func TestIngress(t *testing.T) {
 
 	bus := psrpc.NewRedisMessageBus(rc)
 
-	RunTestSuite(t, conf, bus, service.NewCmd)
+	RunTestSuite(t, conf, bus, func(psrpcClient rpc.IOInfoClient) utils.StateNotifier {
+		return utils.NewServiceStateNotifier(psrpcClient)
+	}, service.NewCmd)
 }

test/rtmp.go

@@ -37,11 +37,11 @@ import (
 	"github.com/livekit/ingress/pkg/utils"
 )
 
-func RunRTMPTest(t *testing.T, conf *TestConfig, bus psrpc.MessageBus, commandPsrpcClient rpc.IngressHandlerClient, psrpcClient rpc.IOInfoClient, newCmd func(ctx context.Context, p *params.Params) (*exec.Cmd, error)) {
+func RunRTMPTest(t *testing.T, conf *TestConfig, bus psrpc.MessageBus, commandPsrpcClient rpc.IngressHandlerClient, psrpcClient rpc.IOInfoClient, sn utils.StateNotifier, newCmd func(ctx context.Context, p *params.Params) (*exec.Cmd, error)) {
 	rtmpsrv := rtmp.NewRTMPServer()
 	relay := service.NewRelay(rtmpsrv, nil)
 
-	svc, err := service.NewService(conf.Config, psrpcClient, utils.NewNoopStateNotifier(), bus, rtmpsrv, nil, newCmd, "")
+	svc, err := service.NewService(conf.Config, psrpcClient, sn, bus, rtmpsrv, nil, newCmd, "")
 	require.NoError(t, err)
 	go func() {
 		err := svc.Run()

test/url.go

@@ -34,8 +34,8 @@ import (
 	"github.com/livekit/ingress/pkg/utils"
 )
 
-func RunURLTest(t *testing.T, conf *TestConfig, bus psrpc.MessageBus, commandPsrpcClient rpc.IngressHandlerClient, psrpcClient rpc.IOInfoClient, newCmd func(ctx context.Context, p *params.Params) (*exec.Cmd, error)) {
-	svc, err := service.NewService(conf.Config, psrpcClient, utils.NewNoopStateNotifier(), bus, nil, nil, newCmd, "")
+func RunURLTest(t *testing.T, conf *TestConfig, bus psrpc.MessageBus, commandPsrpcClient rpc.IngressHandlerClient, psrpcClient rpc.IOInfoClient, sn utils.StateNotifier, newCmd func(ctx context.Context, p *params.Params) (*exec.Cmd, error)) {
+	svc, err := service.NewService(conf.Config, psrpcClient, sn, bus, nil, nil, newCmd, "")
 	require.NoError(t, err)
 	svc.StartDebugHandlers()
 

test/whip.go

@@ -42,12 +42,12 @@ const (
 	whipClientPath = "livekit-whip-bot/cmd/whip-client/whip-client"
 )
 
-func RunWHIPTest(t *testing.T, conf *TestConfig, bus psrpc.MessageBus, commandPsrpcClient rpc.IngressHandlerClient, psrpcClient rpc.IOInfoClient, newCmd func(ctx context.Context, p *params.Params) (*exec.Cmd, error)) {
+func RunWHIPTest(t *testing.T, conf *TestConfig, bus psrpc.MessageBus, commandPsrpcClient rpc.IngressHandlerClient, psrpcClient rpc.IOInfoClient, sn utils.StateNotifier, newCmd func(ctx context.Context, p *params.Params) (*exec.Cmd, error)) {
 	whipsrv, err := whip.NewWHIPServer(bus)
 	require.NoError(t, err)
 	relay := service.NewRelay(nil, whipsrv)
 
-	svc, err := service.NewService(conf.Config, psrpcClient, utils.NewNoopStateNotifier(), bus, nil, whipsrv, newCmd, "")
+	svc, err := service.NewService(conf.Config, psrpcClient, sn, bus, nil, whipsrv, newCmd, "")
 	require.NoError(t, err)
 	go func() {
 		err := svc.Run()