Merge pull request #1 from badgerodon/ci-test

fix tests, add coveralls

.github/workflows/golangci-lint.yml → .github/workflows/lint.yml

@@ -1,4 +1,4 @@
-name: golangci-lint
+name: Lint
 on:
   push:
     tags:
@@ -8,11 +8,13 @@ on:
   pull_request:
 jobs:
   golangci:
-    name: lint
+    name: Go
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
-      - name: golangci-lint
+      - name: Checkout
+        uses: actions/checkout@v2
+
+      - name: Lint
         uses: golangci/golangci-lint-action@v2
         with:
-          version: latest
+          version: latest

.github/workflows/test.yml

@@ -0,0 +1,57 @@
+name: Test
+on:
+  push:
+    tags:
+      - v*
+    branches:
+      - main
+  pull_request:
+jobs:
+  test:
+    name: Go
+    runs-on: ubuntu-latest
+    steps:
+      - name: Install go
+        uses: actions/setup-go@v2
+        with:
+          go-version: '1.16.6'
+
+      - id: go-cache-paths
+        run: |
+          echo "::set-output name=go-build::$(go env GOCACHE)"
+          echo "::set-output name=go-mod::$(go env GOMODCACHE)"
+
+      - name: Checkout
+        uses: actions/checkout@v2
+
+      - name: Cache go build
+        uses: actions/cache@v2
+        with:
+          path: ${{ steps.go-cache-paths.outputs.go-build }}
+          key: ${{ runner.os }}-go-build
+
+      - name: Cache go modules
+        uses: actions/cache@v2
+        with:
+          path: ${{ steps.go-cache-paths.outputs.go-mod }}
+          key: ${{ runner.os }}-go-mod-${{ hashFiles('**/go.sum') }}
+          restore-keys: |
+            ${{ runner.os }}-go-mod-
+
+      - name: Pull test docker images
+        run: |
+          docker pull fixl/google-cloud-sdk-emulators:349.0.0-pubsub
+
+      - name: Download go modules
+        run: |
+          go mod download
+
+      - name: Run tests
+        run: |
+          go test -race -covermode atomic -coverprofile=covprofile ./...
+
+      - name: Send coverage
+        env:
+          COVERALLS_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          go run github.com/mattn/goveralls -coverprofile=covprofile -service=github

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/docker/go-connections v0.4.0 // indirect
 	github.com/gotestyourself/gotestyourself v2.2.0+incompatible // indirect
 	github.com/lib/pq v1.10.2 // indirect
+	github.com/mattn/goveralls v0.0.9
 	github.com/oklog/ulid/v2 v2.0.2
 	github.com/opencontainers/image-spec v1.0.1 // indirect
 	github.com/opencontainers/runc v1.0.1 // indirect

go.sum

@@ -218,6 +218,8 @@ github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/lib/pq v1.10.2 h1:AqzbZs4ZoCBp+GtejcpCpcxM3zlSMx29dXbUSeVtJb8=
 github.com/lib/pq v1.10.2/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
+github.com/mattn/goveralls v0.0.9 h1:XmIwwrO9a9pqSW6IpI89BSCShzQxx0j/oKnnvELQNME=
+github.com/mattn/goveralls v0.0.9/go.mod h1:FRbM1PS8oVsOe9JtdzAAXM+DsvDMMHcM1C7drGJD8HY=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=

providers/google/queue.go

@@ -5,6 +5,7 @@ package google
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	googlepubsub "cloud.google.com/go/pubsub"
@@ -127,7 +128,7 @@ func (s *subscription) Receive(ctx context.Context, handler pubsub.SubscribeHand
 	err := s.subscription.Receive(ctx, func(ctx context.Context, msg *googlepubsub.Message) {
 		handler(ctx, subscriberMessage{msg})
 	})
-	if err != nil {
+	if failed(err) {
 		return fmt.Errorf("google: error receiving messages (project=%s subscription=%s): %w",
 			s.t.q.cfg.projectID, s.subscription.ID(), err)
 	}
@@ -175,3 +176,7 @@ func getSubscriptionConfigToUpdate(opts ...pubsub.SubscribeOption) googlepubsub.
 		},
 	}
 }
+
+func failed(err error) bool {
+	return err != nil && !errors.Is(err, context.Canceled) && status.Code(err) != codes.Canceled
+}

providers/google/queue_test.go

@@ -20,6 +20,6 @@ func TestQueue(t *testing.T) {
 		}
 		defer func() { _ = q.Close() }()
 
-		testutil.Run(t, q)
+		testutil.RunTestSuite(t, q)
 	}))
 }

providers/memory/queue_test.go

@@ -12,5 +12,5 @@ func TestQueue(t *testing.T) {
 	ctx, clearTimeout := context.WithTimeout(context.Background(), time.Second*10)
 	defer clearTimeout()
 
-	testutil.Run(t, New(ctx))
+	testutil.RunTestSuite(t, New(ctx))
 }

testutil/googlepubsub.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/ory/dockertest"
+	"github.com/ory/dockertest/docker"
 )
 
 // WithGooglePubSubEmulator runs a gcloud pubsub emulator in a docker container,
@@ -19,9 +20,10 @@ func WithGooglePubSubEmulator(ctx context.Context, callback func()) error {
 	}
 
 	resource, err := dockerPool.RunWithOptions(&dockertest.RunOptions{
-		Repository: "bigtruedata/gcloud-pubsub-emulator",
-		Tag:        "latest",
-		Cmd:        []string{"start", "--host-port", "0.0.0.0:8538"},
+		Repository:   "fixl/google-cloud-sdk-emulators",
+		Tag:          "349.0.0-pubsub",
+		Cmd:          []string{"start", "--host-port=0.0.0.0:8085"},
+		ExposedPorts: []string{"8085"},
 	})
 	if err != nil {
 		return fmt.Errorf("error running pubsub emulator: %w", err)
@@ -31,8 +33,9 @@ func WithGooglePubSubEmulator(ctx context.Context, callback func()) error {
 	} else {
 		_ = resource.Expire(60 * 10)
 	}
+	go TailLogs(ctx, dockerPool, resource)
 
-	addr := resource.GetHostPort("8538/tcp")
+	addr := resource.GetHostPort("8085/tcp")
 
 	if err = dockerPool.Retry(func() error {
 		conn, err := net.Dial("tcp", addr)
@@ -56,3 +59,14 @@ func WithGooglePubSubEmulator(ctx context.Context, callback func()) error {
 
 	return nil
 }
+
+func TailLogs(ctx context.Context, pool *dockertest.Pool, resource *dockertest.Resource) {
+	_ = pool.Client.Logs(docker.LogsOptions{
+		Context:      ctx,
+		Stderr:       true,
+		Stdout:       true,
+		Follow:       true,
+		Container:    resource.Container.ID,
+		OutputStream: os.Stderr,
+	})
+}

testutil/testsuite.go

@@ -2,13 +2,14 @@ package testutil
 
 import (
 	"context"
+	"sync/atomic"
 	"testing"
 
 	"github.com/badgerodon/pubsub"
 	"github.com/stretchr/testify/assert"
 )
 
-func Run(t *testing.T, q pubsub.Queue) {
+func RunTestSuite(t *testing.T, q pubsub.Queue) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -17,16 +18,20 @@ func Run(t *testing.T, q pubsub.Queue) {
 		return
 	}
 
-
 	subscription, err := topic.Subscribe(ctx, "SUBSCRIPTION-1")
 	if !assert.NoError(t, err) {
 		return
 	}
 
 	recv := make(chan pubsub.Message, 4)
 	go func() {
-		err := subscription.Receive(ctx, func(ctx context.Context, msg pubsub.SubscriberMessage) {
-			recv <-msg
+		receiveContext, receiveCancel := context.WithCancel(ctx)
+		var cnt int64
+		err := subscription.Receive(receiveContext, func(ctx context.Context, msg pubsub.SubscriberMessage) {
+			if atomic.AddInt64(&cnt, 1) >= 4 {
+				receiveCancel()
+			}
+			recv <- msg
 			msg.Ack()
 		})
 		assert.NoError(t, err)
@@ -48,9 +53,9 @@ func Run(t *testing.T, q pubsub.Queue) {
 	seen := map[string]struct{}{}
 	for i := 0; i < len(msgs); i++ {
 		msg := <-recv
-		seen[msg.ID()] = struct{}{}
+		seen[string(msg.Data())] = struct{}{}
 	}
 	assert.Equal(t, map[string]struct{}{
-		"1":{}, "2":{}, "3":{}, "4":{},
+		"1": {}, "2": {}, "3": {}, "4": {},
 	}, seen)
 }

tools.go

@@ -0,0 +1,7 @@
+//+build tools
+
+package pubsub
+
+import (
+	_ "github.com/mattn/goveralls" // used for code coverage
+)