Run integration test with GCS mock (#55)

mariiatuzovska · amanjpro · juanli16 · web-flow · commit 45fd71376232 · 2024-12-11T14:10:20.000-05:00
Co-authored-by: amanjpro &lt;amanjpro@gmail.com&gt;
Co-authored-by: Justin Li &lt;justin.li@optable.co&gt;
diff --git a/.github/workflows/pr.yml b/.github/workflows/pr.yml
@@ -47,5 +47,25 @@ jobs:
         run: go get ./...
       - name: Build
         run: make build
+      - name: Start fake-gcs-server
+        run: |
+          docker run -d --name fake-gcs-server -p 4443:4443 fsouza/fake-gcs-server \
+            -scheme http \
+            -public-host 0.0.0.0:4443
+      - name: Wait for fake-gcs-server to start
+        run: |
+          for i in {1..10}; do
+            if curl -s http://0.0.0.0:4443/storage/v1/; then
+              echo "fake-gcs-server is ready"
+              break
+            fi
+            echo "Waiting for fake-gcs-server..."
+            sleep 2
+          done
       - name: Run go tests
+        env:
+          STORAGE_EMULATOR_HOST: http://0.0.0.0:4443
         run: go test ./...
+      - name: Cleanup
+        if: always()
+        run: docker stop fake-gcs-server && docker rm fake-gcs-server
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
@@ -50,8 +50,28 @@ jobs:
         run: go get ./...
       - name: build
         run: make release
+      - name: Start fake-gcs-server
+        run: |
+          docker run -d --name fake-gcs-server -p 4443:4443 fsouza/fake-gcs-server \
+            -scheme http \
+            -public-host 0.0.0.0:4443
+      - name: Wait for fake-gcs-server to start
+        run: |
+          for i in {1..10}; do
+            if curl -s http://0.0.0.0:4443/storage/v1/; then
+              echo "fake-gcs-server is ready"
+              break
+            fi
+            echo "Waiting for fake-gcs-server..."
+            sleep 2
+          done
       - name: Run go tests
+        env:
+          STORAGE_EMULATOR_HOST: http://0.0.0.0:4443
         run: go test ./...
+      - name: Cleanup
+        if: always()
+        run: docker stop fake-gcs-server && docker rm fake-gcs-server
       - name: Create release
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
diff --git a/Makefile b/Makefile
@@ -3,6 +3,14 @@ BUILD_VERSION := $(shell git describe --tags --always)
 GO=$(shell which go)
 RM=rm -f ./bin/*
 
+# fake-gcs-server variables
+FAKE_GCS_DOCKER_IMAGE = fsouza/fake-gcs-server
+FAKE_GCS_DOCKER_CONTAINER = fake-gcs-server
+FAKE_GCS_SCHEME = http
+FAKE_GCS_PORT = 4443
+FAKE_GCS_HOST = 0.0.0.0
+STORAGE_EMULATOR_HOST = $(FAKE_GCS_SCHEME)://$(FAKE_GCS_HOST):$(FAKE_GCS_PORT)
+
 # windows specific commands
 ifeq ($(OS), Windows_NT)
 	MV=move bin\opair bin\opair.exe
@@ -53,7 +61,7 @@ windows:
 	mkdir -p release && cp bin/opair release/opair-windows-amd64.exe
 
 .PHONY: clean
-clean: clean-bin clean-release
+clean: clean-bin clean-release clean-fake-gcs-server
 
 .PHONY: clean-bin
 clean-bin:
@@ -62,3 +70,32 @@ clean-bin:
 .PHONY: clean-release
 clean-release:
 	rm -f release/*
+
+.PHONY: start-fake-gcs-server
+start-fake-gcs-server:
+	@if [ $$(docker ps -q -f name=$(FAKE_GCS_DOCKER_CONTAINER)) ]; then \
+		echo "Container $(FAKE_GCS_DOCKER_CONTAINER) is already running."; \
+	elif [ $$(docker ps -aq -f name=$(FAKE_GCS_DOCKER_CONTAINER)) ]; then \
+		echo "Starting existing container $(FAKE_GCS_DOCKER_CONTAINER)..."; \
+		docker start $(FAKE_GCS_DOCKER_CONTAINER); \
+	else \
+		echo "Creating and starting container $(FAKE_GCS_DOCKER_CONTAINER)..."; \
+		docker run -d --name $(FAKE_GCS_DOCKER_CONTAINER) -p $(FAKE_GCS_PORT):$(FAKE_GCS_PORT) $(FAKE_GCS_DOCKER_IMAGE) \
+			-scheme $(FAKE_GCS_SCHEME) \
+			-public-host $(FAKE_GCS_HOST):$(FAKE_GCS_PORT); \
+	fi
+
+.PHONY: clean-fake-gcs-server
+clean-fake-gcs-server:
+	@if [ $$(docker ps -q -f name=$(FAKE_GCS_DOCKER_CONTAINER)) ]; then \
+		echo "Stopping container $(FAKE_GCS_DOCKER_CONTAINER)..."; \
+		docker stop $(FAKE_GCS_DOCKER_CONTAINER); \
+	fi
+	@if [ $$(docker ps -aq -f name=$(FAKE_GCS_DOCKER_CONTAINER)) ]; then \
+		echo "Removing container $(FAKE_GCS_DOCKER_CONTAINER)..."; \
+		docker rm $(FAKE_GCS_DOCKER_CONTAINER); \
+	fi
+
+.PHONY: test
+test: start-fake-gcs-server
+	STORAGE_EMULATOR_HOST=$(STORAGE_EMULATOR_HOST) $(GO) test -v ./...
diff --git a/pkg/bucket/bucket.go b/pkg/bucket/bucket.go
@@ -25,19 +25,19 @@ type (
 	ReadWriter struct {
 		client            *storage.Client
 		FileReader        io.Reader
-		srcPrefixedBucket *prefixedBucket
-		dstPrefixedBucket *prefixedBucket
+		srcPrefixedBucket *PrefixedBucket
+		dstPrefixedBucket *PrefixedBucket
 		ReadWriters       []*ReadWriteCloser
 	}
 
 	Completer struct {
 		client            *storage.Client
-		dstPrefixedBucket *prefixedBucket
+		dstPrefixedBucket *PrefixedBucket
 	}
 
-	prefixedBucket struct {
-		bucket string
-		prefix string
+	PrefixedBucket struct {
+		Bucket string
+		Prefix string
 	}
 
 	// ReadWriteCloser contains the name of the object, its reader and a writer.
@@ -56,6 +56,19 @@ type (
 	Option func(*bucketOptions)
 )
 
+// GCSClientOptions is used to set insucure HTTP client for integration tests.
+var GCSClientOptions = []option.ClientOption{}
+
+func gcsClientOptions(token string) []option.ClientOption {
+	return append(GCSClientOptions, option.WithTokenSource(
+		oauth2.StaticTokenSource(
+			&oauth2.Token{
+				AccessToken: token,
+			},
+		),
+	))
+}
+
 // WithReader allows to specify a reader to be used for the bucket.
 func WithReader(reader io.Reader) Option {
 	return func(o *bucketOptions) {
@@ -77,16 +90,7 @@ func NewBucketCompleter(ctx context.Context, downscopedToken string, dstURL stri
 		return nil, ErrTokenRequired
 	}
 
-	client, err := storage.NewClient(
-		ctx,
-		option.WithTokenSource(
-			oauth2.StaticTokenSource(
-				&oauth2.Token{
-					AccessToken: downscopedToken,
-				},
-			),
-		),
-	)
+	client, err := storage.NewClient(ctx, gcsClientOptions(downscopedToken)...)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create storage client: %w", err)
 	}
@@ -105,8 +109,8 @@ func NewBucketCompleter(ctx context.Context, downscopedToken string, dstURL stri
 // Complete writes a .Completed file to the destination bucket to signal that the transfer is complete.
 // It then closes the client.
 func (b *Completer) Complete(ctx context.Context) error {
-	dstBucket := b.client.Bucket(b.dstPrefixedBucket.bucket)
-	completedWriter := dstBucket.Object(fmt.Sprintf("%s/%s", b.dstPrefixedBucket.prefix, CompletedFile)).NewWriter(ctx)
+	dstBucket := b.client.Bucket(b.dstPrefixedBucket.Bucket)
+	completedWriter := dstBucket.Object(fmt.Sprintf("%s/%s", b.dstPrefixedBucket.Prefix, CompletedFile)).NewWriter(ctx)
 	if _, err := completedWriter.Write([]byte{}); err != nil {
 		return fmt.Errorf("failed to write completed file: %w", err)
 	}
@@ -120,8 +124,8 @@ func (b *Completer) Complete(ctx context.Context) error {
 
 // Checks if the .Completed file exists in the destination bucket.
 func (b *Completer) HasCompleted(ctx context.Context) (bool, error) {
-	dstBucket := b.client.Bucket(b.dstPrefixedBucket.bucket)
-	_, err := dstBucket.Object(fmt.Sprintf("%s/%s", b.dstPrefixedBucket.prefix, CompletedFile)).Attrs(ctx)
+	dstBucket := b.client.Bucket(b.dstPrefixedBucket.Bucket)
+	_, err := dstBucket.Object(fmt.Sprintf("%s/%s", b.dstPrefixedBucket.Prefix, CompletedFile)).Attrs(ctx)
 	if errors.Is(err, storage.ErrObjectNotExist) {
 		return false, nil
 	}
@@ -140,16 +144,7 @@ func NewBucketReadWriter(ctx context.Context, downscopedToken string, dstURL str
 		opt(bucketOption)
 	}
 
-	client, err := storage.NewClient(
-		ctx,
-		option.WithTokenSource(
-			oauth2.StaticTokenSource(
-				&oauth2.Token{
-					AccessToken: downscopedToken,
-				},
-			),
-		),
-	)
+	client, err := storage.NewClient(ctx, gcsClientOptions(downscopedToken)...)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create storage client: %w", err)
 	}
@@ -193,7 +188,7 @@ func NewBucketReadWriter(ctx context.Context, downscopedToken string, dstURL str
 }
 
 // bucketFromObjectURL parses a valid objectURL and returns a Bucket.
-func bucketFromObjectURL(objectURL string) (*prefixedBucket, error) {
+func bucketFromObjectURL(objectURL string) (*PrefixedBucket, error) {
 	url, err := url.Parse(objectURL)
 	if err != nil {
 		return nil, err
@@ -203,9 +198,9 @@ func bucketFromObjectURL(objectURL string) (*prefixedBucket, error) {
 		return nil, fmt.Errorf("invalid object URL: %s", objectURL)
 	}
 
-	return &prefixedBucket{
-		bucket: url.Host,
-		prefix: strings.TrimLeft(url.Path, "/"),
+	return &PrefixedBucket{
+		Bucket: url.Host,
+		Prefix: strings.TrimLeft(url.Path, "/"),
 	}, nil
 }
 
@@ -214,10 +209,10 @@ func bucketFromObjectURL(objectURL string) (*prefixedBucket, error) {
 // It then opens a writer for each object under the same name specified by the destinationPrefix.
 func (b *ReadWriter) newObjectReadWriteCloser(ctx context.Context) error {
 	logger := zerolog.Ctx(ctx)
-	query := &storage.Query{Prefix: b.srcPrefixedBucket.prefix + "/"}
+	query := &storage.Query{Prefix: b.srcPrefixedBucket.Prefix + "/"}
 
-	srcBucket := b.client.Bucket(b.srcPrefixedBucket.bucket)
-	dstBucket := b.client.Bucket(b.dstPrefixedBucket.bucket)
+	srcBucket := b.client.Bucket(b.srcPrefixedBucket.Bucket)
+	dstBucket := b.client.Bucket(b.dstPrefixedBucket.Bucket)
 
 	it := srcBucket.Objects(ctx, query)
 	var rwc []*ReadWriteCloser
@@ -227,7 +222,7 @@ func (b *ReadWriter) newObjectReadWriteCloser(ctx context.Context) error {
 		if errors.Is(err, iterator.Done) {
 			break
 		} else if err != nil {
-			logger.Debug().Err(err).Msgf("failed to list objects from source bucket %s", b.srcPrefixedBucket.prefix)
+			logger.Debug().Err(err).Msgf("failed to list objects from source bucket %s", b.srcPrefixedBucket.Prefix)
 			return err
 		}
 
@@ -243,7 +238,7 @@ func (b *ReadWriter) newObjectReadWriteCloser(ctx context.Context) error {
 		rwc = append(rwc, &ReadWriteCloser{
 			name:   blobFromObjectName(obj.Name),
 			Reader: reader,
-			Writer: dstBucket.Object(objectPathWithPrefix(obj.Name, b.dstPrefixedBucket.prefix)).NewWriter(ctx),
+			Writer: dstBucket.Object(objectPathWithPrefix(obj.Name, b.dstPrefixedBucket.Prefix)).NewWriter(ctx),
 		})
 	}
 
@@ -254,8 +249,8 @@ func (b *ReadWriter) newObjectReadWriteCloser(ctx context.Context) error {
 
 // newObjectWriteCloser creates a new writer for the destination bucket.
 func (b *ReadWriter) newObjectWriteCloser(ctx context.Context) *ReadWriteCloser {
-	dstBucket := b.client.Bucket(b.dstPrefixedBucket.bucket)
-	writer := dstBucket.Object(fmt.Sprintf("%s/data_%s.csv", b.dstPrefixedBucket.prefix, shortHex())).NewWriter(ctx)
+	dstBucket := b.client.Bucket(b.dstPrefixedBucket.Bucket)
+	writer := dstBucket.Object(fmt.Sprintf("%s/data_%s.csv", b.dstPrefixedBucket.Prefix, shortHex())).NewWriter(ctx)
 	return &ReadWriteCloser{
 		name:   CompletedFile,
 		Writer: writer,
diff --git a/pkg/bucket/readers.go b/pkg/bucket/readers.go
@@ -26,8 +26,8 @@ type (
 		client            *storage.Client
 		AdvReader         []io.ReadCloser
 		PubReader         []io.ReadCloser
-		AdvPrefixedBucket *prefixedBucket
-		PubPrefixedBucket *prefixedBucket
+		AdvPrefixedBucket *PrefixedBucket
+		PubPrefixedBucket *PrefixedBucket
 		PubFileReader     io.Reader
 	}
 )
@@ -89,7 +89,7 @@ func NewReaders(ctx context.Context, downScopedToken, advURL string, opts ...Opt
 // newObjectReaders lists the objects specified by the advPrefixedBucket and pubPrefixedBucket and opens a reader for each object,
 // except for the .Completed file.
 func (b *Readers) newObjectReaders(ctx context.Context) error {
-	advReaders, err := readersFromPrefixedBucket(ctx, b.client, b.AdvPrefixedBucket)
+	advReaders, err := ReadersFromPrefixedBucket(ctx, b.client, b.AdvPrefixedBucket)
 	if err != nil {
 		return err
 	}
@@ -105,7 +105,7 @@ func (b *Readers) newObjectReaders(ctx context.Context) error {
 		return errors.New("missing publisher bucket URL")
 	}
 
-	pubReaders, err := readersFromPrefixedBucket(ctx, b.client, b.PubPrefixedBucket)
+	pubReaders, err := ReadersFromPrefixedBucket(ctx, b.client, b.PubPrefixedBucket)
 	if err != nil {
 		return err
 	}
@@ -115,11 +115,11 @@ func (b *Readers) newObjectReaders(ctx context.Context) error {
 	return nil
 }
 
-func readersFromPrefixedBucket(ctx context.Context, client *storage.Client, pBucket *prefixedBucket) ([]io.ReadCloser, error) {
+func ReadersFromPrefixedBucket(ctx context.Context, client *storage.Client, pBucket *PrefixedBucket) ([]io.ReadCloser, error) {
 	logger := zerolog.Ctx(ctx)
-	query := &storage.Query{Prefix: pBucket.prefix + "/"}
+	query := &storage.Query{Prefix: pBucket.Prefix + "/"}
 
-	bucket := client.Bucket(pBucket.bucket)
+	bucket := client.Bucket(pBucket.Bucket)
 
 	it := bucket.Objects(ctx, query)
 	var readers []io.ReadCloser
@@ -129,7 +129,7 @@ func readersFromPrefixedBucket(ctx context.Context, client *storage.Client, pBuc
 		if errors.Is(err, iterator.Done) {
 			break
 		} else if err != nil {
-			logger.Debug().Err(err).Msgf("failed to list objects from source bucket %s", pBucket.prefix)
+			logger.Debug().Err(err).Msgf("failed to list objects from source bucket %s", pBucket.Prefix)
 			return nil, err
 		}
 
diff --git a/pkg/cmd/cli/get_test.go b/pkg/cmd/cli/get_test.go
@@ -7,6 +7,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"optable-pair-cli/pkg/internal"
 	"testing"
 	"time"
 
@@ -37,19 +38,19 @@ func TestCleanroomRun(t *testing.T) {
 
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		switch r.URL.Path {
-		case "/admin/api/external/v1/cleanroom/get":
+		case internal.AdminCleanroomGetURL:
 			assertGetRequest(r)
 
 			cleanroom := v1.Cleanroom{}
 			data, err := proto.Marshal(&cleanroom)
 			if err != nil {
 				t.Errorf("Failed to marshal response: %v", err)
 			}
+			w.WriteHeader(http.StatusOK)
 			_, err = w.Write(data)
 			if err != nil {
 				t.Errorf("Failed to write response body: %v", err)
 			}
-			w.WriteHeader(http.StatusOK)
 
 		default:
 			t.Errorf("Unexpected call %s", r.URL.Path)
@@ -84,7 +85,7 @@ func TestCleanroomRun_RequestFail(t *testing.T) {
 
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		switch r.URL.Path {
-		case "/admin/api/external/v1/cleanroom/get":
+		case internal.AdminCleanroomGetURL:
 			w.WriteHeader(http.StatusNotFound)
 
 		default:
diff --git a/pkg/cmd/cli/key_test.go b/pkg/cmd/cli/key_test.go
diff --git a/pkg/cmd/cli/run_it_test.go b/pkg/cmd/cli/run_it_test.go
diff --git a/pkg/internal/client.go b/pkg/internal/client.go
