chore: improved testing (#96)

* publisher: log: add tests

* worker: test for ack function

* worker: tests: remove unused mocks

* worker: add tests for error cases

* worker: test: use constructor for tests

* app: worker: fix flush always timing out

* app: move kafka specific log messages to kafka publisher package

* wip: worker metrics test

* worker: add tests against instrumented metrics

* worker: rename timeSource to Clock

* chore: remove client example

* worker: use idiomatic null device for tests

* config: add tests for validation errors

* config: add test for valid configuration

* config: add tests for internal prepare()

* config: add tests for Walk()

* config: kafka: add tests for ToConfigMap()

* config: kafka: add more tests for config map serializer

* config: add tests for internal helper functions

* refactor: extract Clock to a stand alone package

* collector: add tests for TimePushed mutation

* serialization: add tests for SerialiseProto

* deserialization: add tests for DeserializeFunc

* worker: add test for flush time out

* config: refactor ack value names

* tests: integration e2e tests results in coverage data

* ci: test: fix coverage merge failing

* ci: prospective fix for coverdata #1

* ci: prospective fix for coverdata #2

* ci: prospective fix for coverdata #3

* ci: upload test coverage as artifact

* ci: refactor Docker build for release and testing

* services: grpc: add tests for synchronous event handling

* services: grpc: add tests to cover metric instrumentation

* services: rest: remove deadcode

For an incoming request, req.Body is never nil
https://pkg.go.dev/net/http#Request

* services: rest: add tests for error reading request body

* services: rest: add tests for malformed request body

* services: rest: refactor ack handling

* services: rest: add tests for different error paths

* ci: cache image pulls to speed up tests

* ci: fix docker image caching

* ci: ignore proto package during tests

* ci: remove docker image caching

* Revert "ci: ignore proto package during tests"

This reverts commit 75d0807.

* Revert "Revert "ci: ignore proto package during tests""

This reverts commit 1038f46.

* Revert "Revert "Revert "ci: ignore proto package during tests"""

This reverts commit a39cad6.

* publisher: log: move test event to root proto package

* ci: blacklist proto package during tests

* hack: publisher: log: use single spaced fields

* Revert "publisher: log: move test event to root proto package"

This reverts commit 7f837bf.

* Revert "ci: blacklist proto package during tests"

This reverts commit 0856073.

* Revert "hack: publisher: log: use single spaced fields"

This reverts commit 5ffb846.

* Revert "Revert "hack: publisher: log: use single spaced fields""

This reverts commit 67797d9.

* ci: ignore proto folders when computing coverage

* hack: publisher: log: use double spaced fields

* ci: fix coverage data merge

* ci: fix cover merge resulting in empty coverage file

* publish: kinesis: add tests for stream's existence check

* misc: improve test case description

* publisher: kinesis: add tests for hitting rate limits and quotas

* publisher: kinesis: refactor tests

.github/workflows/test.yaml

@@ -18,21 +18,32 @@ jobs:
           go-version: "1.22.4"
       - name: Checkout code
         uses: actions/checkout@v2
+      - name: Create coverage data directory
+        run: mkdir raccoon-coverage
       - name: Start raccoon
         run: make docker-run
       - name: Run tests
-        run: go test ./... -v -coverprofile=coverage.out
+        run: go test ./... -v -cover -test.gocoverdir=$PWD/raccoon-coverage/
         env:
           INTEGTEST_BOOTSTRAP_SERVER: 'localhost:9094'
           INTEGTEST_HOST: 'localhost:8080'
           INTEGTEST_TOPIC_FORMAT: 'clickstream-%s-log'
           GRPC_SERVER_ADDR: 'localhost:8081'
           PUBSUB_EMULATOR_HOST: 'localhost:8085'
           LOCALSTACK_HOST: 'http://localhost:4566'
-      - name: Upload coverage data 
+      - name: Stop raccoon
+        run: docker compose down
+      - name: Merge coverage data
+        run: go tool covdata textfmt -i=raccoon-coverage -pkg "$(go list ./... | grep -v proto | paste -sd ',')" -o coverage.out
+      - name: Upload coverage data to coveralls
         uses: shogo82148/actions-goveralls@v1
         with:
           path-to-profile: coverage.out
+      - name: Upload coverage data as workflow artifact
+        uses: actions/upload-artifact@v4
+        with:
+          name: coverage
+          path: coverage.out
   smoke-test:
     runs-on: ubuntu-latest
     steps:

.goreleaser.yaml

@@ -66,7 +66,6 @@ archives:
 
 dockers:
   - id: dockerhub
-    dockerfile: Dockerfile.release
     image_templates:
       - "docker.io/raystack/{{.ProjectName}}:latest"
       - "docker.io/raystack/{{.ProjectName}}:{{ .Version }}"

Dockerfile

@@ -1,17 +1,4 @@
-FROM golang:1.22.4
-
-WORKDIR /app
-RUN apt-get update && apt-get install unzip  --no-install-recommends --assume-yes
-RUN PROTOC_ZIP=protoc-3.17.3-linux-x86_64.zip && \
-    curl -OL https://github.com/protocolbuffers/protobuf/releases/download/v3.17.3/$PROTOC_ZIP && \
-    unzip -o $PROTOC_ZIP -d /usr/local bin/protoc && \
-    unzip -o $PROTOC_ZIP -d /usr/local 'include/*' && \
-    rm -f $PROTOC_ZIP
-COPY . .
-RUN make build
-
-
 FROM debian:bookworm-slim
 WORKDIR /app
-COPY --from=0 /app/raccoon ./raccoon
-ENTRYPOINT [ "/app/raccoon" ] 
+COPY raccoon .
+ENTRYPOINT [ "/app/raccoon" ]

Dockerfile.test

@@ -0,0 +1,17 @@
+FROM golang:1.22.4
+
+WORKDIR /app
+RUN apt-get update && apt-get install unzip  --no-install-recommends --assume-yes
+RUN PROTOC_ZIP=protoc-3.17.3-linux-x86_64.zip && \
+    curl -OL https://github.com/protocolbuffers/protobuf/releases/download/v3.17.3/$PROTOC_ZIP && \
+    unzip -o $PROTOC_ZIP -d /usr/local bin/protoc && \
+    unzip -o $PROTOC_ZIP -d /usr/local 'include/*' && \
+    rm -f $PROTOC_ZIP
+COPY . .
+RUN make build-cov
+
+
+FROM debian:bookworm-slim
+WORKDIR /app
+COPY --from=0 /app/raccoon ./raccoon
+ENTRYPOINT [ "/app/raccoon" ] 

Makefile

@@ -46,6 +46,11 @@ build: ## Build the raccoon binary
 	go build ${BUILD_FLAGS} ${LD_FLAGS} ${NAME}
 	@echo "Build complete"
 
+build-cov: ## Build raccoon with coverage instrumentation
+	@echo "Building raccoon version ${VERSION}..."
+	go build -cover ${BUILD_FLAGS} ${LD_FLAGS} ${NAME}
+	@echo "Build complete"
+
 install:
 	@echo "Installing Raccoon to ${GOBIN}..."
 	go install ${BUILD_FLAGS} ${LD_FLAGS} ${NAME}

app/server.go

@@ -64,16 +64,16 @@ func shutDownServer(ctx context.Context, cancel context.CancelFunc, httpServices
 		sig := <-signalChan
 		switch sig {
 		case syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT:
-			logger.Info(fmt.Sprintf("[App.Server] Received a signal %s", sig))
+			logger.Infof("[App.Server] Received a signal %s", sig)
 			httpServices.Shutdown(ctx)
 			logger.Info("Server shutdown all the listeners")
+			close(bufferChannel)
+			logger.Info("Flushing workers")
 			timedOut := workerPool.FlushWithTimeOut(workerFlushTimeout)
 			if timedOut {
-				logger.Info(fmt.Sprintf("WorkerPool flush timedout %t", timedOut))
+				logger.Infof("WorkerPool flush timed out")
 			}
-			flushInterval := config.Publisher.Kafka.FlushInterval
 			logger.Infof("Closing %q producer\n", pub.Name())
-			logger.Info(fmt.Sprintf("Wait %d ms for all messages to be delivered", flushInterval))
 			eventsInProducer := 0
 
 			err := pub.Close()
@@ -91,6 +91,8 @@ func shutDownServer(ctx context.Context, cancel context.CancelFunc, httpServices
 			Until then we fall back to approximation */
 			eventsInChannel := len(bufferChannel) * 7
 			logger.Info(fmt.Sprintf("Outstanding unprocessed events in the channel, data lost ~ (No batches %d * 5 events) = ~%d", len(bufferChannel), eventsInChannel))
+
+			// NOTE(turtledev): aren't these metrics misleading?
 			metrics.Count(
 				fmt.Sprintf("%s_messages_delivered_total", pub.Name()),
 				int64(eventsInChannel+eventsInProducer),
@@ -101,9 +103,13 @@ func shutDownServer(ctx context.Context, cancel context.CancelFunc, httpServices
 				},
 			)
 			logger.Info("Exiting server")
+
+			// NOTE(turtledev): what's the purpose of this?
+			// everything has exited at this point. The context
+			// is not saved by any instances.
 			cancel()
 		default:
-			logger.Info(fmt.Sprintf("[App.Server] Received a unexpected signal %s", sig))
+			logger.Infof("[App.Server] Received a unexpected signal %s", sig)
 		}
 	}
 }

clock/clock.go

@@ -0,0 +1,21 @@
+package clock
+
+import (
+	"time"
+)
+
+// Clock represents a time source
+// It can be used as a way to abstract away the implicit
+// dependency on time.Now() from code.
+type Clock interface {
+	Now() time.Time
+}
+
+type defaultClock struct{}
+
+func (defaultClock) Now() time.Time {
+	return time.Now()
+}
+
+// Default clock. Uses time.Now() internally.
+var Default = defaultClock{}

clock/mock.go

@@ -0,0 +1,16 @@
+package clock
+
+import (
+	"time"
+
+	"github.com/stretchr/testify/mock"
+)
+
+// Mock clock for testing
+type Mock struct {
+	mock.Mock
+}
+
+func (m *Mock) Now() time.Time {
+	return m.Called().Get(0).(time.Time)
+}

collector/service.go

@@ -2,21 +2,24 @@ package collector
 
 import (
 	"context"
-	"time"
+
+	"github.com/raystack/raccoon/clock"
 )
 
 type ChannelCollector struct {
-	ch chan CollectRequest
+	ch    chan CollectRequest
+	clock clock.Clock
 }
 
 func NewChannelCollector(c chan CollectRequest) Collector {
 	return &ChannelCollector{
-		ch: c,
+		ch:    c,
+		clock: clock.Default,
 	}
 }
 
 func (c *ChannelCollector) Collect(ctx context.Context, req *CollectRequest) error {
-	req.TimePushed = time.Now()
+	req.TimePushed = c.clock.Now()
 	c.ch <- *req
 	return nil
 }

collector/service_test.go

@@ -1,8 +1,13 @@
 package collector
 
 import (
+	"context"
 	"reflect"
 	"testing"
+	"time"
+
+	"github.com/raystack/raccoon/clock"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestNewChannelCollector(t *testing.T) {
@@ -22,7 +27,8 @@ func TestNewChannelCollector(t *testing.T) {
 				c: c,
 			},
 			want: &ChannelCollector{
-				ch: c,
+				ch:    c,
+				clock: clock.Default,
 			},
 		},
 	}
@@ -34,3 +40,32 @@ func TestNewChannelCollector(t *testing.T) {
 		})
 	}
 }
+
+func TestCollect(t *testing.T) {
+	t.Run("It should mutate TimePushed to the time the collect request is acknowledged", func(t *testing.T) {
+		now := time.Now()
+		clk := &clock.Mock{}
+		clk.On("Now").Return(now).Once()
+		defer clk.AssertExpectations(t)
+
+		ch := make(chan CollectRequest)
+		defer close(ch)
+
+		collector := &ChannelCollector{
+			ch:    ch,
+			clock: clk,
+		}
+
+		consumer := func(requests chan CollectRequest) {
+			for range requests {
+			}
+		}
+		go consumer(collector.ch)
+
+		req := &CollectRequest{}
+		assert.Nil(
+			t, collector.Collect(context.Background(), req),
+		)
+		assert.Equal(t, req.TimePushed, now)
+	})
+}

config/config.go

@@ -33,6 +33,9 @@ func prepare() {
 
 	// add default CORS headers
 	corsHeaders := []string{"Content-Type"}
+
+	// TODO(turtledev): evalute the need for this block.
+	// It may be a candiate for removal
 	provisionalHeaders := []string{
 		"SERVER_WEBSOCKET_CONN_GROUP_HEADER",
 		"SERVER_WEBSOCKET_CONN_ID_HEADER",