Added kafka testing

pkg/repository/kafka_test.go

@@ -1,31 +1,31 @@
 package repository_test
 
 import (
-	// "strconv"
+	"strconv"
 
 	. "github.com/onsi/ginkgo/v2"
-	// . "github.com/onsi/gomega"
-	//
-	// "github.com/ShatteredRealms/go-backend/pkg/config"
-	// "github.com/ShatteredRealms/go-backend/pkg/repository"
-	// testdb "github.com/ShatteredRealms/go-backend/test/db"
+	. "github.com/onsi/gomega"
+
+	"github.com/ShatteredRealms/go-backend/pkg/config"
+	"github.com/ShatteredRealms/go-backend/pkg/repository"
+	testdb "github.com/ShatteredRealms/go-backend/test/db"
 )
 
 var _ = Describe("Kafka repository", func() {
 	Describe("ConnectKafka", func() {
 		Context("valid input", func() {
 			It("should work", func() {
-				// cleanupFunc, _, kResource := testdb.SetupKafkaWithDocker()
-				// sPort := kResource.GetPort("29092/tcp")
-				// port, err := strconv.ParseUint(sPort, 10, 64)
-				// Expect(err).NotTo(HaveOccurred())
-				// kafka, err := repository.ConnectKafka(config.ServerAddress{
-				// 	Port: uint(port),
-				// 	Host: "127.0.0.1",
-				// })
-				// Expect(err).NotTo(HaveOccurred())
-				// Expect(kafka).NotTo(BeNil())
-				// cleanupFunc()
+				cleanupFunc, _, kResource := testdb.SetupKafkaWithDocker()
+				defer cleanupFunc()
+				sPort := kResource.GetPort("29093/tcp")
+				port, err := strconv.ParseUint(sPort, 10, 64)
+				Expect(err).NotTo(HaveOccurred())
+				kafka, err := repository.ConnectKafka(config.ServerAddress{
+					Port: uint(port),
+					Host: "127.0.0.1",
+				})
+				Expect(err).NotTo(HaveOccurred())
+				Expect(kafka).NotTo(BeNil())
 			})
 		})
 	})

pkg/service/chat_s_test.go

@@ -0,0 +1,129 @@
+package service_test
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/ShatteredRealms/go-backend/pkg/config"
+	"github.com/ShatteredRealms/go-backend/pkg/mocks"
+	"github.com/ShatteredRealms/go-backend/pkg/model"
+	"github.com/ShatteredRealms/go-backend/pkg/service"
+	testdb "github.com/ShatteredRealms/go-backend/test/db"
+	"github.com/bxcodec/faker/v4"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/ory/dockertest/v3"
+	"github.com/sirupsen/logrus/hooks/test"
+	"go.uber.org/mock/gomock"
+	"gorm.io/gorm"
+)
+
+var _ = Describe("Chat service", Ordered, func() {
+
+	var (
+		hook *test.Hook
+
+		mockController *gomock.Controller
+		mockRepository *mocks.MockChatRepository
+		cleanupFunc    func()
+		kResource      *dockertest.Resource
+
+		chatService service.ChatService
+
+		kafkaPort uint
+
+		err       error
+		fakeError = fmt.Errorf("error")
+		channels  = []*model.ChatChannel{
+			{
+				Model: gorm.Model{
+					ID:        1,
+					CreatedAt: time.Now(),
+					UpdatedAt: time.Now(),
+				},
+				Name:      faker.Username(),
+				Dimension: faker.Username(),
+			},
+			{
+				Model: gorm.Model{
+					ID:        2,
+					CreatedAt: time.Now().Add(time.Minute),
+					UpdatedAt: time.Now().Add(time.Minute),
+				},
+				Name:      faker.Username(),
+				Dimension: faker.Username(),
+			},
+		}
+	)
+
+	BeforeAll(func() {
+		_, hook = test.NewNullLogger()
+		mockController = gomock.NewController(GinkgoT())
+		mockRepository = mocks.NewMockChatRepository(mockController)
+
+		cleanupFunc, _, kResource = testdb.SetupKafkaWithDocker()
+		sPort := kResource.GetPort("29093/tcp")
+		port, err := strconv.ParseUint(sPort, 10, 64)
+		kafkaPort = uint(port)
+		Expect(err).NotTo(HaveOccurred())
+		hook.Reset()
+	})
+
+	Describe("NewChatService", Ordered, func() {
+		When("given invalid input", func() {
+			It("should fail if migration fails", func() {
+				mockRepository.EXPECT().Migrate(gomock.Any()).Return(fakeError)
+				chatService, err = service.NewChatService(context.Background(), mockRepository, config.ServerAddress{
+					Port: kafkaPort,
+					Host: "localhost",
+				})
+
+				Expect(err).To(MatchError(fakeError))
+				Expect(chatService).To(BeNil())
+			})
+
+			It("should fail if kafka connect fails", func() {
+				mockRepository.EXPECT().Migrate(gomock.Any()).Return(nil)
+				chatService, err = service.NewChatService(context.Background(), mockRepository, config.ServerAddress{
+					Port: 0,
+					Host: "nowhere",
+				})
+
+				Expect(err).To(HaveOccurred())
+				Expect(chatService).To(BeNil())
+			})
+
+			It("should fail if kafka all channels", func() {
+				mockRepository.EXPECT().Migrate(gomock.Any()).Return(nil)
+				mockRepository.EXPECT().AllChannels(gomock.Any()).Return(model.ChatChannels{}, fakeError)
+				chatService, err = service.NewChatService(context.Background(), mockRepository, config.ServerAddress{
+					Port: kafkaPort,
+					Host: "localhost",
+				})
+
+				Expect(err).To(HaveOccurred())
+				Expect(chatService).To(BeNil())
+			})
+		})
+
+		When("valid input is given", func() {
+			It("should succeed", func() {
+				mockRepository.EXPECT().Migrate(gomock.Any()).Return(nil)
+				mockRepository.EXPECT().AllChannels(gomock.Any()).Return(channels, nil)
+				chatService, err = service.NewChatService(context.Background(), mockRepository, config.ServerAddress{
+					Port: kafkaPort,
+					Host: "localhost",
+				})
+
+				Expect(err).To(BeNil())
+				Expect(chatService).NotTo(BeNil())
+			})
+		})
+	})
+
+	AfterAll(func() {
+		cleanupFunc()
+	})
+})

test/db/docker.go

@@ -53,16 +53,16 @@ func SetupKafkaWithDocker() (func(), *dockertest.Resource, *dockertest.Resource)
 		Env: []string{
 			"KAFKA_BROKER_ID=1",
 			"KAFKA_ZOOKEEPER_CONNECT=gozookeeper:2181",
-			"KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://gokafka:9092,PLAINTEXT_HOST://gokafka:29092",
+			"KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9093,PLAINTEXT_HOST://localhost:29093",
 			"KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT",
 			"KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT",
 			// "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1",
 		},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"39092/tcp": {{HostIP: "gokafka", HostPort: "29092/tcp"}},
-			"49092/tcp": {{HostIP: "gokafka", HostPort: "9092/tcp"}},
+			"29093/tcp": {{HostIP: "localhost", HostPort: "29093/tcp"}},
+			"9093/tcp":  {{HostIP: "localhost", HostPort: "9093/tcp"}},
 		},
-		ExposedPorts: []string{"29092/tcp"},
+		ExposedPorts: []string{"29093/tcp"},
 		Networks:     []*dockertest.Network{net},
 	}
 
@@ -85,9 +85,12 @@ func SetupKafkaWithDocker() (func(), *dockertest.Resource, *dockertest.Resource)
 	}))
 
 	fnCleanup := func() {
-		chk(kafkaResource.Close())
-		chk(zookeeperResource.Close())
-		chk(net.Close())
+		err1 := kafkaResource.Close()
+		err2 := zookeeperResource.Close()
+		err3 := net.Close()
+		chk(err1)
+		chk(err2)
+		chk(err3)
 	}
 
 	return fnCleanup, zookeeperResource, kafkaResource