package global import ( "context" "io" "log/slog" "sync" "testing" "github.com/redis/go-redis/v9" "go.mongodb.org/mongo-driver/mongo" "gorm.io/gorm" platformmq "kra/pkg/mq" platformws "kra/pkg/websocket" ) type storageStub struct{} func (*storageStub) Open(context.Context, string) (io.ReadCloser, error) { return nil, nil } func (*storageStub) Delete(context.Context, string) error { return nil } func (*storageStub) DeletePrefix(context.Context, string) error { return nil } type schedulerStub struct{} func (*schedulerStub) Start(context.Context) error { return nil } func (*schedulerStub) Stop(context.Context) error { return nil } type mqStub struct{} func (*mqStub) Register(platformmq.SubscriptionSet) error { return nil } func (*mqStub) Unregister(string) error { return nil } func (*mqStub) Client(string) platformmq.Client { return nil } func (*mqStub) PublishTo(context.Context, string, string, []byte, byte, bool) error { return nil } func (*mqStub) SubscribeTo(context.Context, string, string, byte, platformmq.Handler) error { return nil } func (*mqStub) UnsubscribeFrom(context.Context, string, ...string) error { return nil } func (*mqStub) ConnectedTo(string) bool { return true } type webSocketStub struct{} func (*webSocketStub) Enabled() bool { return true } func (*webSocketStub) Broadcast([]byte) error { return nil } func (*webSocketStub) BroadcastBinary([]byte) error { return nil } func (*webSocketStub) Sessions() ([]*platformws.Session, error) { return nil, nil } func (*webSocketStub) Len() int { return 0 } func (*webSocketStub) Send(*platformws.Session, []byte) error { return nil } func (*webSocketStub) SendBinary(*platformws.Session, []byte) error { return nil } func (*webSocketStub) OnMessage(func(*platformws.Session, []byte)) {} func (*webSocketStub) OnBinaryMessage(func(*platformws.Session, []byte)) {} func (*webSocketStub) OnConnect(func(*platformws.Session)) {} func (*webSocketStub) OnDisconnect(func(*platformws.Session)) {} func TestResourceRegistryDefaults(t *testing.T) { registry := NewResourceRegistry() resources := registry.Snapshot() if resources.Logger == nil { t.Fatal("default logger is nil") } if resources.DB != nil || len(resources.NamedDBs) != 0 || resources.Redis != nil || len(resources.NamedRedis) != 0 || resources.Mongo != nil || resources.Storage != nil || resources.MQ != nil || resources.WebSocket != nil || resources.Scheduler != nil { t.Fatalf("unexpected initialized resource: %#v", resources) } } func TestResourceRegistryReplaceAndSnapshot(t *testing.T) { logger := slog.Default().With("test", true) db := &gorm.DB{} redisClient := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"}) t.Cleanup(func() { _ = redisClient.Close() }) mongoClient := &mongo.Client{} storage := &storageStub{} mq := &mqStub{} webSocket := &webSocketStub{} scheduler := &schedulerStub{} registry := NewResourceRegistry() inputDBs := map[string]*gorm.DB{"reporting": db} inputRedis := map[string]redis.UniversalClient{"cache": redisClient} registry.Replace(Resources{ Logger: logger, DB: db, NamedDBs: inputDBs, Redis: redisClient, NamedRedis: inputRedis, Mongo: mongoClient, Storage: storage, MQ: mq, WebSocket: webSocket, Scheduler: scheduler, }) resources := registry.Snapshot() if resources.Logger != logger || resources.DB != db || resources.Redis != redisClient || resources.Mongo != mongoClient || resources.Storage != storage || resources.MQ != mq || resources.WebSocket != webSocket || resources.Scheduler != scheduler { t.Fatalf("replacement snapshot does not match: %#v", resources) } if resources.NamedDBs["reporting"] != db || resources.NamedRedis["cache"] != redisClient { t.Fatal("named resource lookup returned the wrong resource") } inputDBs["mutated"] = &gorm.DB{} inputRedis["mutated"] = redisClient if _, exists := registry.Snapshot().NamedDBs["mutated"]; exists { t.Fatal("named database map was not copied on replacement") } if _, exists := registry.Snapshot().NamedRedis["mutated"]; exists { t.Fatal("named redis map was not copied on replacement") } resources.NamedDBs["mutated"] = db if _, exists := registry.Snapshot().NamedDBs["mutated"]; exists { t.Fatal("snapshot shares its named database map with the registry") } } func TestResourceRegistryNormalizesTypedNilInterfaces(t *testing.T) { registry := NewResourceRegistry() var redisClient *redis.Client var storage *storageStub var mq *mqStub var webSocket *webSocketStub var scheduler *schedulerStub registry.Replace(Resources{Redis: redisClient, Storage: storage, MQ: mq, WebSocket: webSocket, Scheduler: scheduler}) resources := registry.Snapshot() if resources.Redis != nil || resources.Storage != nil || resources.MQ != nil || resources.WebSocket != nil || resources.Scheduler != nil { t.Fatalf("typed nil resource was retained: %#v", resources) } if resources.Logger == nil { t.Fatal("nil logger was not replaced with the default logger") } } func TestResourceRegistryConcurrentAccess(t *testing.T) { registry := NewResourceRegistry() const iterations = 1000 var wait sync.WaitGroup for index := 0; index < 8; index++ { wait.Add(1) go func() { defer wait.Done() for step := 0; step < iterations; step++ { registry.Replace(Resources{DB: &gorm.DB{}}) if registry.Snapshot().Logger == nil { t.Error("concurrent snapshot lost the default logger") return } } }() } wait.Wait() if registry.Snapshot().DB == nil { t.Fatal("concurrent replacement lost the database") } }