Fix the acceptance tests

acceptanceTests/Makefile

@@ -1,2 +1,2 @@
 test: ## Run acceptance tests.
-	@go test ./... -timeout 1h
+	@go test ./... -timeout 1h -v

acceptanceTests/go.mod

@@ -3,6 +3,8 @@ module github.com/up9inc/mizu/tests
 go 1.16
 
 require (
+	github.com/gorilla/websocket v1.4.2
+	github.com/stretchr/testify v1.6.1
 	github.com/up9inc/mizu/shared v0.0.0
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
 )

acceptanceTests/go.sum

@@ -75,6 +75,7 @@ github.com/cpuguy83/go-md2man/v2 v2.0.0/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsr
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/daviddengcn/go-colortext v0.0.0-20160507010035-511bcaf42ccd/go.mod h1:dv4zxwHi5C/8AeI+4gX4dCWOIvNi7I6JCSX0HvlKPgE=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
@@ -211,6 +212,7 @@ github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5m
 github.com/googleapis/gnostic v0.4.1/go.mod h1:LRhVm6pbyptWbWbuZ38d1eyptfvIytN3ir6b65WBswg=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
+github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
@@ -312,6 +314,7 @@ github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
@@ -364,6 +367,7 @@ github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoH
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=

acceptanceTests/tap_test.go

@@ -10,10 +10,72 @@ import (
 	"os/exec"
 	"path"
 	"strings"
+	"sync"
 	"testing"
 	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/assert"
 )
 
+// waitTimeout waits for the waitgroup for the specified max timeout.
+// Returns true if waiting timed out.
+func waitTimeout(wg *sync.WaitGroup, timeout time.Duration) bool {
+	c := make(chan struct{})
+	go func() {
+		defer close(c)
+		wg.Wait()
+	}()
+	select {
+	case <-c:
+		return false // completed normally
+	case <-time.After(timeout):
+		return true // timed out
+	}
+}
+
+// checkDBHasEntries checks whether there are any entries in the database
+// before the given timestamp. Returns a slice of non-empty entries if it succeeds.
+func checkDBHasEntries(t *testing.T, timestamp int64, limit int) (entries []map[string]interface{}) {
+	query := fmt.Sprintf("timestamp < %d and limit(%d)", timestamp, limit)
+	webSocketUrl := getWebSocketUrl(defaultApiServerPort)
+
+	c, _, err := websocket.DefaultDialer.Dial(webSocketUrl, nil)
+	assert.Nil(t, err)
+	defer c.Close()
+
+	handleWSConnection := func(wg *sync.WaitGroup) {
+		defer wg.Done()
+		for {
+			_, message, err := c.ReadMessage()
+			if err != nil {
+				return
+			}
+
+			var data map[string]interface{}
+			err = json.Unmarshal([]byte(message), &data)
+			assert.Nil(t, err)
+
+			entries = append(entries, data)
+		}
+	}
+
+	err = c.WriteMessage(websocket.TextMessage, []byte(query))
+	assert.Nil(t, err)
+
+	var wg sync.WaitGroup
+	go handleWSConnection(&wg)
+	wg.Add(1)
+
+	waitTimeout(&wg, 1*time.Second)
+
+	if len(entries) == 0 {
+		t.Error("unexpected entries result - Expected more than 0 entries")
+	}
+
+	return
+}
+
 func TestTap(t *testing.T) {
 	if testing.Short() {
 		t.Skip("ignored acceptance test")
@@ -66,21 +128,11 @@ func TestTap(t *testing.T) {
 			entriesCheckFunc := func() error {
 				timestamp := time.Now().UnixNano() / int64(time.Millisecond)
 
-				entriesUrl := fmt.Sprintf("%v/entries?limit=%v&operator=lt&timestamp=%v", apiServerUrl, entriesCount, timestamp)
+				entries := checkDBHasEntries(t, timestamp, entriesCount)
-				requestResult, requestErr := executeHttpGetRequest(entriesUrl)
+				entry := entries[0]
-				if requestErr != nil {
-					return fmt.Errorf("failed to get entries, err: %v", requestErr)
-				}
-
-				entries := requestResult.([]interface{})
-				if len(entries) == 0 {
-					return fmt.Errorf("unexpected entries result - Expected more than 0 entries")
-				}
-
-				entry := entries[0].(map[string]interface{})
 
 				entryUrl := fmt.Sprintf("%v/entries/%v", apiServerUrl, entry["id"])
-				requestResult, requestErr = executeHttpGetRequest(entryUrl)
+				requestResult, requestErr := executeHttpGetRequest(entryUrl)
 				if requestErr != nil {
 					return fmt.Errorf("failed to get entry, err: %v", requestErr)
 				}
@@ -441,21 +493,11 @@ func TestTapRedact(t *testing.T) {
 	redactCheckFunc := func() error {
 		timestamp := time.Now().UnixNano() / int64(time.Millisecond)
 
-		entriesUrl := fmt.Sprintf("%v/entries?limit=%v&operator=lt&timestamp=%v", apiServerUrl, defaultEntriesCount, timestamp)
+		entries := checkDBHasEntries(t, timestamp, defaultEntriesCount)
-		requestResult, requestErr := executeHttpGetRequest(entriesUrl)
+		firstEntry := entries[0]
-		if requestErr != nil {
-			return fmt.Errorf("failed to get entries, err: %v", requestErr)
-		}
-
-		entries := requestResult.([]interface{})
-		if len(entries) == 0 {
-			return fmt.Errorf("unexpected entries result - Expected more than 0 entries")
-		}
-
-		firstEntry := entries[0].(map[string]interface{})
 
 		entryUrl := fmt.Sprintf("%v/entries/%v", apiServerUrl, firstEntry["id"])
-		requestResult, requestErr = executeHttpGetRequest(entryUrl)
+		requestResult, requestErr := executeHttpGetRequest(entryUrl)
 		if requestErr != nil {
 			return fmt.Errorf("failed to get entry, err: %v", requestErr)
 		}
@@ -556,21 +598,11 @@ func TestTapNoRedact(t *testing.T) {
 	redactCheckFunc := func() error {
 		timestamp := time.Now().UnixNano() / int64(time.Millisecond)
 
-		entriesUrl := fmt.Sprintf("%v/entries?limit=%v&operator=lt&timestamp=%v", apiServerUrl, defaultEntriesCount, timestamp)
+		entries := checkDBHasEntries(t, timestamp, defaultEntriesCount)
-		requestResult, requestErr := executeHttpGetRequest(entriesUrl)
+		firstEntry := entries[0]
-		if requestErr != nil {
-			return fmt.Errorf("failed to get entries, err: %v", requestErr)
-		}
-
-		entries := requestResult.([]interface{})
-		if len(entries) == 0 {
-			return fmt.Errorf("unexpected entries result - Expected more than 0 entries")
-		}
-
-		firstEntry := entries[0].(map[string]interface{})
 
 		entryUrl := fmt.Sprintf("%v/entries/%v", apiServerUrl, firstEntry["id"])
-		requestResult, requestErr = executeHttpGetRequest(entryUrl)
+		requestResult, requestErr := executeHttpGetRequest(entryUrl)
 		if requestErr != nil {
 			return fmt.Errorf("failed to get entry, err: %v", requestErr)
 		}
@@ -671,21 +703,11 @@ func TestTapRegexMasking(t *testing.T) {
 	redactCheckFunc := func() error {
 		timestamp := time.Now().UnixNano() / int64(time.Millisecond)
 
-		entriesUrl := fmt.Sprintf("%v/entries?limit=%v&operator=lt&timestamp=%v", apiServerUrl, defaultEntriesCount, timestamp)
+		entries := checkDBHasEntries(t, timestamp, defaultEntriesCount)
-		requestResult, requestErr := executeHttpGetRequest(entriesUrl)
+		firstEntry := entries[0]
-		if requestErr != nil {
-			return fmt.Errorf("failed to get entries, err: %v", requestErr)
-		}
-
-		entries := requestResult.([]interface{})
-		if len(entries) == 0 {
-			return fmt.Errorf("unexpected entries result - Expected more than 0 entries")
-		}
-
-		firstEntry := entries[0].(map[string]interface{})
 
 		entryUrl := fmt.Sprintf("%v/entries/%v", apiServerUrl, firstEntry["id"])
-		requestResult, requestErr = executeHttpGetRequest(entryUrl)
+		requestResult, requestErr := executeHttpGetRequest(entryUrl)
 		if requestErr != nil {
 			return fmt.Errorf("failed to get entry, err: %v", requestErr)
 		}
@@ -778,20 +800,11 @@ func TestTapIgnoredUserAgents(t *testing.T) {
 	ignoredUserAgentsCheckFunc := func() error {
 		timestamp := time.Now().UnixNano() / int64(time.Millisecond)
 
-		entriesUrl := fmt.Sprintf("%v/entries?limit=%v&operator=lt&timestamp=%v", apiServerUrl, defaultEntriesCount*2, timestamp)
+		entries := checkDBHasEntries(t, timestamp, defaultEntriesCount)
-		requestResult, requestErr := executeHttpGetRequest(entriesUrl)
-		if requestErr != nil {
-			return fmt.Errorf("failed to get entries, err: %v", requestErr)
-		}
-
-		entries := requestResult.([]interface{})
-		if len(entries) == 0 {
-			return fmt.Errorf("unexpected entries result - Expected more than 0 entries")
-		}
 
 		for _, entryInterface := range entries {
-			entryUrl := fmt.Sprintf("%v/entries/%v", apiServerUrl, entryInterface.(map[string]interface{})["id"])
+			entryUrl := fmt.Sprintf("%v/entries/%v", apiServerUrl, entryInterface["id"])
-			requestResult, requestErr = executeHttpGetRequest(entryUrl)
+			requestResult, requestErr := executeHttpGetRequest(entryUrl)
 			if requestErr != nil {
 				return fmt.Errorf("failed to get entry, err: %v", requestErr)
 			}
@@ -986,21 +999,11 @@ func TestDaemonSeeTraffic(t *testing.T) {
 			entriesCheckFunc := func() error {
 				timestamp := time.Now().UnixNano() / int64(time.Millisecond)
 
-				entriesUrl := fmt.Sprintf("%v/entries?limit=%v&operator=lt&timestamp=%v", apiServerUrl, entriesCount, timestamp)
+				entries := checkDBHasEntries(t, timestamp, entriesCount)
-				requestResult, requestErr := executeHttpGetRequest(entriesUrl)
+				entry := entries[0]
-				if requestErr != nil {
-					return fmt.Errorf("failed to get entries, err: %v", requestErr)
-				}
-
-				entries := requestResult.([]interface{})
-				if len(entries) == 0 {
-					return fmt.Errorf("unexpected entries result - Expected more than 0 entries")
-				}
-
-				entry := entries[0].(map[string]interface{})
 
 				entryUrl := fmt.Sprintf("%v/entries/%v", apiServerUrl, entry["id"])
-				requestResult, requestErr = executeHttpGetRequest(entryUrl)
+				requestResult, requestErr := executeHttpGetRequest(entryUrl)
 				if requestErr != nil {
 					return fmt.Errorf("failed to get entry, err: %v", requestErr)
 				}

acceptanceTests/testsUtils.go

@@ -82,13 +82,18 @@ func getApiServerUrl(port uint16) string {
 	return fmt.Sprintf("http://localhost:%v", port)
 }
 
+func getWebSocketUrl(port uint16) string {
+	return fmt.Sprintf("ws://localhost:%v/ws", port)
+}
+
 func getDefaultCommandArgs() []string {
+	headless := "--headless"
 	setFlag := "--set"
 	telemetry := "telemetry=false"
 	agentImage := "agent-image=gcr.io/up9-docker-hub/mizu/ci:0.0.0"
 	imagePullPolicy := "image-pull-policy=Never"
 
-	return []string{setFlag, telemetry, setFlag, agentImage, setFlag, imagePullPolicy}
+	return []string{headless, setFlag, telemetry, setFlag, agentImage, setFlag, imagePullPolicy}
 }
 
 func getDefaultTapCommandArgs() []string {

cli/cmd/tap.go

@@ -113,4 +113,5 @@ func init() {
 	tapCmd.Flags().String(configStructs.EnforcePolicyFile, defaultTapConfig.EnforcePolicyFile, "Yaml file path with policy rules")
 	tapCmd.Flags().String(configStructs.ContractFile, defaultTapConfig.ContractFile, "OAS/Swagger file to validate to monitor the contracts")
 	tapCmd.Flags().Bool(configStructs.DaemonModeTapName, defaultTapConfig.DaemonMode, "Run mizu in daemon mode, detached from the cli")
+	tapCmd.Flags().Bool(configStructs.HeadlessMode, defaultTapConfig.HeadlessMode, "Enable headless mode.")
 }

cli/cmd/tapRunner.go

@@ -4,16 +4,17 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/up9inc/mizu/cli/cmd/goUtils"
 	"io/ioutil"
-	k8serrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"path"
 	"regexp"
 	"strings"
 	"time"
 
+	"github.com/up9inc/mizu/cli/cmd/goUtils"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+
 	"github.com/getkin/kin-openapi/openapi3"
 	"github.com/up9inc/mizu/cli/apiserver"
 	"github.com/up9inc/mizu/cli/config"
@@ -625,7 +626,9 @@ func watchApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provi
 				}
 
 				logger.Log.Infof("Mizu is available at %s\n", url)
+				if !config.Config.Tap.HeadlessMode {
 					uiUtils.OpenBrowser(url)
+				}
 				if err := apiProvider.ReportTappedPods(state.tapperSyncer.CurrentlyTappedPods); err != nil {
 					logger.Log.Debugf("[Error] failed update tapped pods %v", err)
 				}

cli/cmd/viewRunner.go

@@ -58,7 +58,9 @@ func runMizuView() {
 
 	logger.Log.Infof("Mizu is available at %s\n", url)
 
+	if !config.Config.Tap.HeadlessMode {
 		uiUtils.OpenBrowser(url)
+	}
 
 	if isCompatible, err := version.CheckVersionCompatibility(apiServerProvider); err != nil {
 		logger.Log.Errorf("Failed to check versions compatibility %v", err)

cli/config/configStructs/tapConfig.go

@@ -3,9 +3,10 @@ package configStructs
 import (
 	"errors"
 	"fmt"
-	"github.com/up9inc/mizu/shared"
 	"regexp"
 
+	"github.com/up9inc/mizu/shared"
+
 	"github.com/up9inc/mizu/shared/units"
 )
 
@@ -22,6 +23,7 @@ const (
 	EnforcePolicyFile             = "traffic-validation-file"
 	ContractFile                  = "contract"
 	DaemonModeTapName             = "daemon"
+	HeadlessMode                  = "headless"
 )
 
 type TapConfig struct {
@@ -44,6 +46,7 @@ type TapConfig struct {
 	ApiServerResources     shared.Resources `yaml:"api-server-resources"`
 	TapperResources        shared.Resources `yaml:"tapper-resources"`
 	DaemonMode             bool             `yaml:"daemon" default:"false"`
+	HeadlessMode           bool             `yaml:"headless" default:"false"`
 }
 
 func (config *TapConfig) PodRegex() *regexp.Regexp {