feat: rework grpc server testing

Signed-off-by: Alexander Trost <[email protected]>

gen/go/proto/services/auth/auth_test.go

@@ -5,24 +5,19 @@ import (
 	"fmt"
 	"os"
 	"testing"
-	"time"
 
 	permsauth "github.com/galexrt/fivenet/gen/go/proto/services/auth/perms"
+	"github.com/galexrt/fivenet/internal/modules"
 	"github.com/galexrt/fivenet/internal/tests/proto"
 	"github.com/galexrt/fivenet/internal/tests/servers"
-	"github.com/galexrt/fivenet/pkg/config"
-	"github.com/galexrt/fivenet/pkg/config/appconfig"
-	"github.com/galexrt/fivenet/pkg/grpc/auth"
-	"github.com/galexrt/fivenet/pkg/grpc/auth/userinfo"
-	"github.com/galexrt/fivenet/pkg/mstlystcdata"
+	grpcserver "github.com/galexrt/fivenet/pkg/grpc"
 	"github.com/galexrt/fivenet/pkg/perms"
-	"github.com/galexrt/fivenet/pkg/server/audit"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	tracesdk "go.opentelemetry.io/otel/sdk/trace"
+	"go.uber.org/fx"
 	"go.uber.org/fx/fxtest"
-	"go.uber.org/zap"
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
 )
 
@@ -47,71 +42,30 @@ func TestMain(m *testing.M) {
 func TestFullAuthFlow(t *testing.T) {
 	defer servers.TestDBServer.Reset()
 
-	db, err := servers.TestDBServer.DB()
-	require.NoError(t, err)
-
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	logger := zap.NewNop()
-	tp := tracesdk.NewTracerProvider()
-	ui := userinfo.NewMockUserInfoRetriever(map[int32]*userinfo.UserInfo{})
-	tm := auth.NewTokenMgr("")
-
-	cfg, err := config.LoadBaseConfigTest()
-	require.NoError(t, err)
-	require.NotNil(t, cfg)
 
-	cfg.NATS.URL = servers.TestNATSServer.GetURL()
-	cfg.Cache.RefreshTime = 1 * time.Hour
-
-	appCfg, err := appconfig.NewTest(ctx)
-	require.NoError(t, err)
-	require.NotNil(t, cfg)
-
-	js, err := servers.TestNATSServer.GetJS()
+	clientConn, grpcSrvModule, err := modules.TestGRPCServer(ctx)
 	require.NoError(t, err)
 
-	fxLC := fxtest.NewLifecycle(t)
-
-	p, err := perms.New(perms.Params{
-		LC:        fxLC,
-		Logger:    logger,
-		DB:        db,
-		TP:        tp,
-		JS:        js,
-		AppConfig: appCfg,
-	})
-	assert.NoError(t, err)
+	var srv *Server
+	app := fxtest.New(t,
+		modules.GetFxTestOpts(
+			fx.Provide(grpcSrvModule),
+			fx.Provide(grpcserver.AsService(func(p Params) *Server {
+				srv = NewServer(p)
+				return srv
+			})),
 
-	aud := &audit.Noop{}
-
-	c, err := mstlystcdata.NewCache(mstlystcdata.Params{
-		LC:     fxLC,
-		Logger: logger,
-		TP:     tp,
-		DB:     db,
-		Config: cfg,
-	})
-	assert.NoError(t, err)
-	enricher := mstlystcdata.NewEnricher(c, appCfg)
-
-	srv := NewServer(Params{
-		Logger:   logger,
-		DB:       db,
-		Auth:     auth.NewGRPCAuth(ui, tm),
-		TM:       tm,
-		Perms:    p,
-		Enricher: enricher,
-		Aud:      aud,
-		UI:       ui,
-		Config:   cfg,
-	})
+			fx.Invoke(func(*grpc.Server) {}),
+		)...,
+	)
+	assert.NotNil(t, app)
 
-	fxLC.RequireStart()
-	defer fxLC.RequireStop()
+	app.RequireStart()
+	assert.NotNil(t, srv)
 
-	client, _, cancel := NewTestAuthServiceClient(srv)
-	defer cancel()
+	client := NewAuthServiceClient(clientConn)
 
 	// First login without credentials
 	loginReq := &LoginRequest{}
@@ -184,19 +138,19 @@ func TestFullAuthFlow(t *testing.T) {
 	assert.Nil(t, chooseCharRes)
 	proto.CompareGRPCError(t, ErrUnableToChooseChar, err)
 
-	role, err := p.GetRoleByJobAndGrade(ctx, "ambulance", 1)
+	role, err := srv.ps.GetRoleByJobAndGrade(ctx, "ambulance", 1)
 	assert.NoError(t, err)
 	assert.NotNil(t, role)
 
-	perm, err := p.GetPermission(ctx, permsauth.AuthServicePerm, permsauth.AuthServiceChooseCharacterPerm)
+	perm, err := srv.ps.GetPermission(ctx, permsauth.AuthServicePerm, permsauth.AuthServiceChooseCharacterPerm)
 	assert.NoError(t, err)
 	assert.NotNil(t, perm)
 
 	// user-1: Choose valid character but we don't have permissions on the role
-	err = p.RemovePermissionsFromRole(ctx, role.ID, perm.Id)
+	err = srv.ps.RemovePermissionsFromRole(ctx, role.ID, perm.Id)
 	assert.NoError(t, err)
 	// Disable choose char perm from ambulance rank 1 role, `user-1` is a medic rank 1+
-	err = p.UpdateRolePermissions(ctx, role.ID, perms.AddPerm{
+	err = srv.ps.UpdateRolePermissions(ctx, role.ID, perms.AddPerm{
 		Id:  perm.Id,
 		Val: false,
 	})
@@ -208,7 +162,7 @@ func TestFullAuthFlow(t *testing.T) {
 	proto.CompareGRPCError(t, ErrUnableToChooseChar, err)
 
 	// user-1: Choose valid character, now we add a permssion
-	err = p.UpdateRolePermissions(ctx, role.ID, perms.AddPerm{
+	err = srv.ps.UpdateRolePermissions(ctx, role.ID, perms.AddPerm{
 		Id:  perm.Id,
 		Val: true,
 	})
@@ -217,4 +171,6 @@ func TestFullAuthFlow(t *testing.T) {
 	chooseCharRes, err = client.ChooseCharacter(ctx, chooseCharReq)
 	assert.NoError(t, err)
 	assert.NotNil(t, chooseCharRes)
+
+	app.RequireStop()
 }

gen/go/proto/services/centrum/centrum.go

@@ -20,7 +20,7 @@ import (
 	"github.com/galexrt/fivenet/pkg/perms"
 	"github.com/galexrt/fivenet/pkg/server/audit"
 	"github.com/galexrt/fivenet/pkg/tracker"
-	"github.com/galexrt/fivenet/pkg/utils"
+	"github.com/galexrt/fivenet/pkg/utils/broker"
 	"github.com/galexrt/fivenet/query/fivenet/model"
 	"github.com/nats-io/nats.go/jetstream"
 	tracesdk "go.opentelemetry.io/otel/sdk/trace"
@@ -48,7 +48,7 @@ type Server struct {
 	appCfg  appconfig.IConfig
 
 	brokersMutex sync.RWMutex
-	brokers      map[string]*utils.Broker[*StreamResponse]
+	brokers      map[string]*broker.Broker[*StreamResponse]
 
 	state *manager.Manager
 }
@@ -64,17 +64,17 @@ type Params struct {
 	Perms     perms.Permissions
 	Audit     audit.IAuditer
 	JS        jetstream.JetStream
-	Tracker   tracker.ITracker
-	Postals   postals.Postals
 	Config    *config.Config
 	AppConfig appconfig.IConfig
+	Tracker   tracker.ITracker
+	Postals   postals.Postals
 	Manager   *manager.Manager
 }
 
 func NewServer(p Params) (*Server, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 
-	brokers := map[string]*utils.Broker[*StreamResponse]{}
+	brokers := map[string]*broker.Broker[*StreamResponse]{}
 
 	s := &Server{
 		logger: p.Logger.Named("centrum"),
@@ -166,7 +166,7 @@ func (s *Server) handleAppConfigUpdate(ctx context.Context, cfg *appconfig.Cfg)
 
 	for _, job := range cfg.UserTracker.LivemapJobs {
 		if _, ok := s.brokers[job]; !ok {
-			s.brokers[job] = utils.NewBroker[*StreamResponse]()
+			s.brokers[job] = broker.New[*StreamResponse]()
 			go s.brokers[job].Start(ctx)
 		}
 	}
@@ -399,7 +399,7 @@ func (s *Server) Stream(req *StreamRequest, srv CentrumService_StreamServer) err
 	}
 }
 
-func (s *Server) getJobBroker(job string) (*utils.Broker[*StreamResponse], bool) {
+func (s *Server) getJobBroker(job string) (*broker.Broker[*StreamResponse], bool) {
 	s.brokersMutex.RLock()
 	defer s.brokersMutex.RUnlock()
 

gen/go/proto/services/centrum/centrum_test.go

@@ -5,22 +5,18 @@ import (
 	"fmt"
 	"os"
 	"testing"
-	"time"
 
 	"github.com/galexrt/fivenet/gen/go/proto/services/centrum/manager"
 	"github.com/galexrt/fivenet/gen/go/proto/services/centrum/state"
+	"github.com/galexrt/fivenet/internal/modules"
 	"github.com/galexrt/fivenet/internal/tests/servers"
-	"github.com/galexrt/fivenet/pkg/config"
-	"github.com/galexrt/fivenet/pkg/config/appconfig"
-	"github.com/galexrt/fivenet/pkg/coords/postals"
-	"github.com/galexrt/fivenet/pkg/mstlystcdata"
-	"github.com/galexrt/fivenet/pkg/perms"
-	"github.com/galexrt/fivenet/pkg/server/audit"
+	grpcserver "github.com/galexrt/fivenet/pkg/grpc"
 	"github.com/galexrt/fivenet/pkg/tracker"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	tracesdk "go.opentelemetry.io/otel/sdk/trace"
+	"go.uber.org/fx"
 	"go.uber.org/fx/fxtest"
-	"go.uber.org/zap"
+	"google.golang.org/grpc"
 )
 
 func TestMain(m *testing.M) {
@@ -44,97 +40,38 @@ func TestMain(m *testing.M) {
 func TestBasicCentrumFlow(t *testing.T) {
 	defer servers.TestDBServer.Reset()
 
-	db, err := servers.TestDBServer.DB()
-	require.NoError(t, err)
-
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	logger := zap.NewNop()
-	tp := tracesdk.NewTracerProvider()
-
-	cfg, err := config.LoadBaseConfigTest()
-	require.NoError(t, err)
-	require.NotNil(t, cfg)
-
-	cfg.NATS.URL = servers.TestNATSServer.GetURL()
-	cfg.Cache.RefreshTime = 1 * time.Hour
-
-	appCfg, err := appconfig.NewTest(ctx)
-	require.NoError(t, err)
-	require.NotNil(t, cfg)
-
-	js, err := servers.TestNATSServer.GetJS()
-	require.NoError(t, err)
 
-	fxLC := fxtest.NewLifecycle(t)
-
-	p, err := perms.New(perms.Params{
-		LC:        fxLC,
-		Logger:    logger,
-		DB:        db,
-		TP:        tp,
-		JS:        js,
-		AppConfig: appCfg,
-	})
+	clientConn, grpcSrvModule, err := modules.TestGRPCServer(ctx)
 	require.NoError(t, err)
 
-	aud := &audit.Noop{}
-
-	c, err := mstlystcdata.NewCache(mstlystcdata.Params{
-		LC:     fxLC,
-		Logger: logger,
-		TP:     tp,
-		DB:     db,
-		Config: cfg,
-	})
-	require.NoError(t, err)
-	enricher := mstlystcdata.NewEnricher(c, appCfg)
-
-	state, err := state.New(state.Params{
-		LC:        fxLC,
-		Logger:    logger,
-		JS:        js,
-		AppConfig: appCfg,
-	})
-	require.NoError(t, err)
-
-	mgr := manager.New(manager.Params{
-		LC:       fxLC,
-		Logger:   logger,
-		DB:       db,
-		TP:       tp,
-		JS:       js,
-		Enricher: enricher,
-		State:    state,
-	})
-
-	tracker := tracker.NewForTests(ctx)
-
-	postals, err := postals.NewForTests()
-	require.NoError(t, err)
-
-	srv, err := NewServer(Params{
-		LC:      fxLC,
-		Logger:  logger,
-		TP:      tp,
-		DB:      db,
-		Perms:   p,
-		Audit:   aud,
-		JS:      js,
-		Config:  cfg,
-		Manager: mgr,
-		Tracker: tracker,
-		Postals: postals,
-	})
-	require.NoError(t, err)
-
-	fxLC.RequireStart()
-	defer fxLC.RequireStop()
-
-	client, _, cancel := NewTestCentrumServiceClient(srv)
+	var srv *Server
+	app := fxtest.New(t,
+		modules.GetFxTestOpts(
+			fx.Provide(tracker.NewForTests),
+			state.StateModule,
+			manager.Module,
+			fx.Provide(grpcSrvModule),
+			fx.Provide(grpcserver.AsService(func(p Params) (*Server, error) {
+				srv, err = NewServer(p)
+				return srv, err
+			})),
+
+			fx.Invoke(func(*grpc.Server) {}),
+		)...,
+	)
+	assert.NotNil(t, app)
+
+	app.RequireStart()
+	assert.NotNil(t, srv)
+
+	client := NewCentrumServiceClient(clientConn)
 	defer cancel()
 	_ = ctx
 	_ = client
 
 	// TODO add tests
+
+	app.RequireStop()
 }

gen/go/proto/services/centrum/events/events.go

@@ -56,7 +56,7 @@ func RegisterStream(ctx context.Context, js jetstream.JetStream) (jetstream.Stre
 		Storage:     jetstream.MemoryStorage,
 		Duplicates:  20 * time.Second,
 	}
-	if _, err := natsutils.CreateOrUpdateStream(ctx, js, cfg); err != nil {
+	if _, err := js.CreateOrUpdateStream(ctx, cfg); err != nil {
 		return cfg, err
 	}
 

gen/go/proto/services/livemapper/events.go

@@ -28,7 +28,7 @@ func (s *Server) registerEvents(ctx context.Context, c context.Context) error {
 		Discard:     jetstream.DiscardOld,
 		MaxAge:      2 * time.Minute,
 	}
-	if _, err := natsutils.CreateOrUpdateStream(ctx, s.js, cfg); err != nil {
+	if _, err := s.js.CreateOrUpdateStream(ctx, cfg); err != nil {
 		return err
 	}