Fix the acceptance tests after the merger of #279 (#443)

* Enable acceptance tests

* Fix the acceptance tests

* Move `--headless` from `getDefaultCommandArgs` to `getDefaultTapCommandArgs`

* Fix rest of the failing acceptance tests

* Revert "Enable acceptance tests"

This reverts commit 3f919e865a.

* Revert "Revert "Enable acceptance tests""

This reverts commit c0bfe54b70.

* Ignore `--headless` in `mizu view`

* Make all non-informative things informative

* Remove `github.com/stretchr/testify` dependency from the acceptance tests

* Move the helper methods `waitTimeout` and `checkDBHasEntries` from `tap_test.go` to `testsUtils.go`

* Split `checkDBHasEntries` method into `getDBEntries` and `assertEntriesAtLeast` methods

* Revert "Revert "Revert "Enable acceptance tests"""

This reverts commit c13342671c.

* Revert "Revert "Revert "Revert "Enable acceptance tests""""

This reverts commit 0f8c436926.

* Make `getDBEntries` and `checkEntriesAtLeast` methods return errors instead

* Revert "Revert "Revert "Revert "Revert "Enable acceptance tests"""""

This reverts commit 643fdde009.

acceptanceTests/Makefile

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

acceptanceTests/config_test.go

@@ -2,11 +2,12 @@ package acceptanceTests
 
 import (
 	"fmt"
-	"gopkg.in/yaml.v3"
 	"io/ioutil"
 	"os"
 	"os/exec"
 	"testing"
+
+	"gopkg.in/yaml.v3"
 )
 
 type tapConfig struct {

acceptanceTests/go.mod

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

acceptanceTests/go.sum

@@ -211,6 +211,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=

acceptanceTests/tap_test.go

@@ -66,21 +66,18 @@ 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, err := getDBEntries(timestamp, entriesCount, 1*time.Second)
-				requestResult, requestErr := executeHttpGetRequest(entriesUrl)
+				if err != nil {
-				if requestErr != nil {
+					return err
-					return fmt.Errorf("failed to get entries, err: %v", requestErr)
 				}
-
+				err = checkEntriesAtLeast(entries, 1)
-				entries := requestResult.([]interface{})
+				if err != nil {
-				if len(entries) == 0 {
+					return err
-					return fmt.Errorf("unexpected entries result - Expected more than 0 entries")
 				}
-
+				entry := entries[0]
-				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,38 +438,26 @@ 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, err := getDBEntries(timestamp, defaultEntriesCount, 1*time.Second)
-		requestResult, requestErr := executeHttpGetRequest(entriesUrl)
+		if err != nil {
-		if requestErr != nil {
+			return err
-			return fmt.Errorf("failed to get entries, err: %v", requestErr)
 		}
-
+		err = checkEntriesAtLeast(entries, 1)
-		entries := requestResult.([]interface{})
+		if err != nil {
-		if len(entries) == 0 {
+			return err
-			return fmt.Errorf("unexpected entries result - Expected more than 0 entries")
 		}
-
+		firstEntry := entries[0]
-		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)
 		}
 
-		data := requestResult.(map[string]interface{})["data"].(map[string]interface{})
+		entry := requestResult.(map[string]interface{})["data"].(map[string]interface{})
-		entryJson := data["entry"].(string)
+		request := entry["request"].(map[string]interface{})
 
-		var entry map[string]interface{}
+		headers := request["_headers"].([]interface{})
-		if parseErr := json.Unmarshal([]byte(entryJson), &entry); parseErr != nil {
-			return fmt.Errorf("failed to parse entry, err: %v", parseErr)
-		}
-
-		entryRequest := entry["request"].(map[string]interface{})
-		entryPayload := entryRequest["payload"].(map[string]interface{})
-		entryDetails := entryPayload["details"].(map[string]interface{})
-
-		headers := entryDetails["_headers"].([]interface{})
 		for _, headerInterface := range headers {
 			header := headerInterface.(map[string]interface{})
 			if header["name"].(string) != "User-Agent" {
@@ -485,7 +470,7 @@ func TestTapRedact(t *testing.T) {
 			}
 		}
 
-		postData := entryDetails["postData"].(map[string]interface{})
+		postData := request["postData"].(map[string]interface{})
 		textDataStr := postData["text"].(string)
 
 		var textData map[string]string
@@ -556,38 +541,26 @@ 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, err := getDBEntries(timestamp, defaultEntriesCount, 1*time.Second)
-		requestResult, requestErr := executeHttpGetRequest(entriesUrl)
+		if err != nil {
-		if requestErr != nil {
+			return err
-			return fmt.Errorf("failed to get entries, err: %v", requestErr)
 		}
-
+		err = checkEntriesAtLeast(entries, 1)
-		entries := requestResult.([]interface{})
+		if err != nil {
-		if len(entries) == 0 {
+			return err
-			return fmt.Errorf("unexpected entries result - Expected more than 0 entries")
 		}
-
+		firstEntry := entries[0]
-		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)
 		}
 
-		data := requestResult.(map[string]interface{})["data"].(map[string]interface{})
+		entry := requestResult.(map[string]interface{})["data"].(map[string]interface{})
-		entryJson := data["entry"].(string)
+		request := entry["request"].(map[string]interface{})
 
-		var entry map[string]interface{}
+		headers := request["_headers"].([]interface{})
-		if parseErr := json.Unmarshal([]byte(entryJson), &entry); parseErr != nil {
-			return fmt.Errorf("failed to parse entry, err: %v", parseErr)
-		}
-
-		entryRequest := entry["request"].(map[string]interface{})
-		entryPayload := entryRequest["payload"].(map[string]interface{})
-		entryDetails := entryPayload["details"].(map[string]interface{})
-
-		headers := entryDetails["_headers"].([]interface{})
 		for _, headerInterface := range headers {
 			header := headerInterface.(map[string]interface{})
 			if header["name"].(string) != "User-Agent" {
@@ -600,7 +573,7 @@ func TestTapNoRedact(t *testing.T) {
 			}
 		}
 
-		postData := entryDetails["postData"].(map[string]interface{})
+		postData := request["postData"].(map[string]interface{})
 		textDataStr := postData["text"].(string)
 
 		var textData map[string]string
@@ -671,38 +644,26 @@ 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, err := getDBEntries(timestamp, defaultEntriesCount, 1*time.Second)
-		requestResult, requestErr := executeHttpGetRequest(entriesUrl)
+		if err != nil {
-		if requestErr != nil {
+			return err
-			return fmt.Errorf("failed to get entries, err: %v", requestErr)
 		}
-
+		err = checkEntriesAtLeast(entries, 1)
-		entries := requestResult.([]interface{})
+		if err != nil {
-		if len(entries) == 0 {
+			return err
-			return fmt.Errorf("unexpected entries result - Expected more than 0 entries")
 		}
-
+		firstEntry := entries[0]
-		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)
 		}
 
-		data := requestResult.(map[string]interface{})["data"].(map[string]interface{})
+		entry := requestResult.(map[string]interface{})["data"].(map[string]interface{})
-		entryJson := data["entry"].(string)
+		request := entry["request"].(map[string]interface{})
 
-		var entry map[string]interface{}
+		postData := request["postData"].(map[string]interface{})
-		if parseErr := json.Unmarshal([]byte(entryJson), &entry); parseErr != nil {
-			return fmt.Errorf("failed to parse entry, err: %v", parseErr)
-		}
-
-		entryRequest := entry["request"].(map[string]interface{})
-		entryPayload := entryRequest["payload"].(map[string]interface{})
-		entryDetails := entryPayload["details"].(map[string]interface{})
-
-		postData := entryDetails["postData"].(map[string]interface{})
 		textData := postData["text"].(string)
 
 		if textData != "[REDACTED]" {
@@ -778,38 +739,27 @@ 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, err := getDBEntries(timestamp, defaultEntriesCount, 1*time.Second)
-		requestResult, requestErr := executeHttpGetRequest(entriesUrl)
+		if err != nil {
-		if requestErr != nil {
+			return err
-			return fmt.Errorf("failed to get entries, err: %v", requestErr)
 		}
-
+		err = checkEntriesAtLeast(entries, 1)
-		entries := requestResult.([]interface{})
+		if err != nil {
-		if len(entries) == 0 {
+			return err
-			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)
 			}
 
-			data := requestResult.(map[string]interface{})["data"].(map[string]interface{})
+			entry := requestResult.(map[string]interface{})["data"].(map[string]interface{})
-			entryJson := data["entry"].(string)
+			request := entry["request"].(map[string]interface{})
 
-			var entry map[string]interface{}
+			headers := request["_headers"].([]interface{})
-			if parseErr := json.Unmarshal([]byte(entryJson), &entry); parseErr != nil {
+			for _, headerInterface := range headers {
-				return fmt.Errorf("failed to parse entry, err: %v", parseErr)
-			}
-
-			entryRequest := entry["request"].(map[string]interface{})
-			entryPayload := entryRequest["payload"].(map[string]interface{})
-			entryDetails := entryPayload["details"].(map[string]interface{})
-
-			entryHeaders := entryDetails["_headers"].([]interface{})
-			for _, headerInterface := range entryHeaders {
 				header := headerInterface.(map[string]interface{})
 				if header["name"].(string) != ignoredUserAgentCustomHeader {
 					continue
@@ -986,21 +936,18 @@ 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, err := getDBEntries(timestamp, entriesCount, 1*time.Second)
-				requestResult, requestErr := executeHttpGetRequest(entriesUrl)
+				if err != nil {
-				if requestErr != nil {
+					return err
-					return fmt.Errorf("failed to get entries, err: %v", requestErr)
 				}
-
+				err = checkEntriesAtLeast(entries, 1)
-				entries := requestResult.([]interface{})
+				if err != nil {
-				if len(entries) == 0 {
+					return err
-					return fmt.Errorf("unexpected entries result - Expected more than 0 entries")
 				}
-
+				entry := entries[0]
-				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

@@ -11,22 +11,24 @@ import (
 	"os/exec"
 	"path"
 	"strings"
+	"sync"
 	"syscall"
 	"testing"
 	"time"
 
+	"github.com/gorilla/websocket"
 	"github.com/up9inc/mizu/shared"
 )
 
 const (
-	longRetriesCount     = 100
+	longRetriesCount      = 100
-	shortRetriesCount    = 10
+	shortRetriesCount     = 10
-	defaultApiServerPort = shared.DefaultApiServerPort
+	defaultApiServerPort  = shared.DefaultApiServerPort
-	defaultNamespaceName = "mizu-tests"
+	defaultNamespaceName  = "mizu-tests"
-	defaultServiceName   = "httpbin"
+	defaultServiceName    = "httpbin"
-	defaultEntriesCount  = 50
+	defaultEntriesCount   = 50
 	waitAfterTapPodsReady = 3 * time.Second
-	cleanCommandTimeout  = 1 * time.Minute
+	cleanCommandTimeout   = 1 * time.Minute
 )
 
 type PodDescriptor struct {
@@ -36,7 +38,7 @@ type PodDescriptor struct {
 
 func isPodDescriptorInPodArray(pods []map[string]interface{}, podDescriptor PodDescriptor) bool {
 	for _, pod := range pods {
-		podNamespace :=  pod["namespace"].(string)
+		podNamespace := pod["namespace"].(string)
 		podName := pod["name"].(string)
 
 		if podDescriptor.Namespace == podNamespace && strings.Contains(podName, podDescriptor.Name) {
@@ -82,6 +84,10 @@ 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 {
 	setFlag := "--set"
 	telemetry := "telemetry=false"
@@ -92,10 +98,11 @@ func getDefaultCommandArgs() []string {
 }
 
 func getDefaultTapCommandArgs() []string {
+	headless := "--headless"
 	tapCommand := "tap"
 	defaultCmdArgs := getDefaultCommandArgs()
 
-	return append([]string{tapCommand}, defaultCmdArgs...)
+	return append([]string{tapCommand, headless}, defaultCmdArgs...)
 }
 
 func getDefaultTapCommandArgsWithDaemonMode() []string {
@@ -256,11 +263,11 @@ func runMizuClean() error {
 	}()
 
 	select {
-	case err = <- commandDone:
+	case err = <-commandDone:
 		if err != nil {
 			return err
 		}
-	case <- time.After(cleanCommandTimeout):
+	case <-time.After(cleanCommandTimeout):
 		return errors.New("clean command timed out")
 	}
 
@@ -311,6 +318,77 @@ func daemonCleanup(t *testing.T, viewCmd *exec.Cmd) {
 	}
 }
 
+// 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 {
+	channel := make(chan struct{})
+	go func() {
+		defer close(channel)
+		wg.Wait()
+	}()
+	select {
+	case <-channel:
+		return false // completed normally
+	case <-time.After(timeout):
+		return true // timed out
+	}
+}
+
+// checkEntriesAtLeast checks whether the number of entries greater than or equal to n
+func checkEntriesAtLeast(entries []map[string]interface{}, n int) error {
+	if len(entries) < n {
+		return fmt.Errorf("Unexpected entries result - Expected more than %d entries", n-1)
+	}
+	return nil
+}
+
+// getDBEntries retrieves the entries from the database before the given timestamp.
+// Also limits the results according to the limit parameter.
+// Timeout for the WebSocket connection is defined by the timeout parameter.
+func getDBEntries(timestamp int64, limit int, timeout time.Duration) (entries []map[string]interface{}, err error) {
+	query := fmt.Sprintf("timestamp < %d and limit(%d)", timestamp, limit)
+	webSocketUrl := getWebSocketUrl(defaultApiServerPort)
+
+	var connection *websocket.Conn
+	connection, _, err = websocket.DefaultDialer.Dial(webSocketUrl, nil)
+	if err != nil {
+		return
+	}
+	defer connection.Close()
+
+	handleWSConnection := func(wg *sync.WaitGroup) {
+		defer wg.Done()
+		for {
+			_, message, err := connection.ReadMessage()
+			if err != nil {
+				return
+			}
+
+			var data map[string]interface{}
+			if err = json.Unmarshal([]byte(message), &data); err != nil {
+				return
+			}
+
+			if data["messageType"] == "entry" {
+				entries = append(entries, data)
+			}
+		}
+	}
+
+	err = connection.WriteMessage(websocket.TextMessage, []byte(query))
+	if err != nil {
+		return
+	}
+
+	var wg sync.WaitGroup
+	go handleWSConnection(&wg)
+	wg.Add(1)
+
+	waitTimeout(&wg, timeout)
+
+	return
+}
+
 func Contains(slice []string, containsValue string) bool {
 	for _, sliceValue := range slice {
 		if sliceValue == containsValue {

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)
-				uiUtils.OpenBrowser(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/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 {