gateway/tests/integration.go

//go:build legacy
// +build legacy

// Package tests implements utility functions to help with API Gateway testing.
package tests

import (
	"bytes"
	"compress/gzip"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"net/textproto"
	"os"
	"os/exec"
	"path"
	"reflect"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/santhosh-tekuri/jsonschema/v6"
)

var (
	defaultBinPath     *string = flag.String("gateway_bin_path", ".././gateway", "The default path to the gateway bin")
	defaultSpecsPath   *string = flag.String("gateway_specs_path", "./fixtures/specs", "The default path to the specs folder")
	defaultBackendPort *int    = flag.Int("gateway_backend_port", 8081, "The port for the mocked backend api")
	defaultCfgPath     *string = flag.String(
		"gateway_config_path",
		"fixtures/gateway.json",
		"The default path to the gateway config",
	)
	defaultDelay *time.Duration = flag.Duration(
		"gateway_delay",
		200*time.Millisecond,
		"The delay for the delayed backend endpoint",
	)
	defaultEnvironPatterns *string = flag.String(
		"gateway_envar_pattern",
		"",
		"Comma separated list of patterns to use to filter the envars to pass (set to \".*\" to pass everything)",
	)
	notFollowRedirects                = flag.Bool("client_not_follow_redirects", false, "The test http client should not follow http redirects")
	defaultStartupWait *time.Duration = flag.Duration(
		"gateway_startup_wait",
		1500*time.Millisecond,
		"The time to wait to let servers startup before start testing",
	)
	// POLL for readiness; do not guess at it. Empty here meant waitForStartup
	// fell back to sleeping gateway_startup_wait (1.5s) and then testing whatever
	// was listening — so on a loaded machine every single spec failed with
	// `connection refused` and the suite reported ~60 unrelated failures for one
	// cause. The readiness probe already existed and was simply switched off.
	// Port 8080 is the harness config's port and is hardcoded in every spec URL.
	defaultReadyURL *string = flag.String(
		"gateway_ready_url",
		"http://localhost:8080/healthz",
		"The url to check for system under test readiness.",
	)
	// A ceiling, not a delay: waitForReady returns as soon as the probe answers
	// 200, so a generous budget costs a fast machine nothing and keeps a slow or
	// loaded one honest.
	defaultReadyURLWait *time.Duration = flag.Duration(
		"gateway_ready_url_wait",
		30*time.Second,
		"The maximum time to wait for the ready url to return a 200 Ok response.",
	)
)

// TestCase defines a single case to be tested
type TestCase struct {
	Name string     `json:"name"`
	Err  string     `json:"error"`
	In   Input      `json:"in"`
	Out  Output     `json:"out"`
	Next []TestCase `json:"next"`
}

// Input is the definition of the request to send in a given TestCase
type Input struct {
	URL    string            `json:"url"`
	Method string            `json:"method"`
	Header map[string]string `json:"header"`
	Body   interface{}       `json:"body"`
}

// Output contains the data required to verify the response received in a given TestCase
type Output struct {
	StatusCode int                    `json:"status_code"`
	Body       interface{}            `json:"body"`
	Header     map[string][]string    `json:"header"`
	Schema     map[string]interface{} `json:"schema"`
}

// CmdBuilder defines an interface for building the cmd to be managed by the Runner
type CmdBuilder interface {
	New(*Config) *exec.Cmd
}

// BackendBuilder defines an interface for building a server as a backend for the tests
type BackendBuilder interface {
	New(*Config) http.Server
}

// GenericServer defines an interface to launch a server that
// could be an http.Server, a different type, or a wrapper
// around multiple servers.
type GenericServer interface {
	Close() error
	ListenAndServe() error
}

// ComposableBackendBuilder allows us to return
// a more generic interface for any kind of server.
type ComposableBackendBuilder interface {
	NewGenericServer(*Config) GenericServer
}

// Config contains options for running a test.
type Config struct {
	BinPath         string
	CfgPath         string
	SpecsPath       string
	EnvironPatterns string
	BackendPort     int
	Delay           time.Duration
	HttpClient      *http.Client
	StartupWait     time.Duration
	ReadyURL        string
	ReadyURLWait    time.Duration
}

func (c *Config) getBinPath() string {
	if c.BinPath != "" {
		return c.BinPath
	}
	return *defaultBinPath
}

func (c *Config) getCfgPath() string {
	if c.CfgPath != "" {
		return c.CfgPath
	}
	return *defaultCfgPath
}

func (c *Config) getSpecsPath() string {
	if c.SpecsPath != "" {
		return c.SpecsPath
	}
	return *defaultSpecsPath
}

func (c *Config) getBackendPort() int {
	if c.BackendPort != 0 {
		return c.BackendPort
	}
	return *defaultBackendPort
}

func (c *Config) getDelay() time.Duration {
	if c.Delay != 0 {
		return c.Delay
	}
	return *defaultDelay
}

func (c *Config) getReadyURL() string {
	if c.ReadyURL != "" {
		return c.ReadyURL
	}
	return *defaultReadyURL
}

func (c *Config) getReadyURLWait() time.Duration {
	if c.ReadyURLWait != 0 {
		return c.ReadyURLWait
	}
	return *defaultReadyURLWait
}

func (c *Config) getStartupWait() time.Duration {
	if c.StartupWait != 0 {
		return c.StartupWait
	}
	return *defaultStartupWait
}

func (c *Config) getEnvironPatterns() string {
	if c.EnvironPatterns != "" {
		return c.EnvironPatterns
	}
	return *defaultEnvironPatterns
}

func (c *Config) getHttpClient() *http.Client {
	if c.HttpClient != nil {
		return c.HttpClient
	}
	return defaultHttpClient()

}

func defaultHttpClient() *http.Client {
	if *notFollowRedirects {
		return &http.Client{
			CheckRedirect: func(_ *http.Request, _ []*http.Request) error {
				return http.ErrUseLastResponse
			},
		}
	}
	return http.DefaultClient
}

var defaultConfig Config

// NewIntegration sets up a runner for the integration test and returns it with the parsed specs from the specs folder
// and an error signaling if something went wrong. It uses the default values for any nil argument
func NewIntegration(cfg *Config, cb CmdBuilder, bb BackendBuilder) (*Runner, []TestCase, error) {
	if cfg == nil {
		cfg = &defaultConfig
	}

	if cb == nil {
		cb = defaultCmdBuilder
	}
	cmd := cb.New(cfg)

	var tcs []TestCase
	if err := cmd.Start(); err != nil {
		return nil, tcs, err
	}
	closeFuncs := []func(){
		func() {
			cmd.Process.Kill()
		},
	}

	go func() { fmt.Println(cmd.Wait()) }()

	var err error
	tcs, err = testCases(*cfg)
	if err != nil {
		cmd.Process.Kill()
		return nil, tcs, err
	}

	if bb == nil {
		bb = DefaultBackendBuilder
	}

	var backend GenericServer
	cbb, ok := bb.(ComposableBackendBuilder)
	if ok {
		backend = cbb.NewGenericServer(cfg)
	} else {
		httpServer := bb.New(cfg)
		backend = &httpServer
	}

	closeFuncs = append(closeFuncs, func() { backend.Close() })

	go func() {
		if err := backend.ListenAndServe(); err != nil {
			log.Printf("backend closed: %v\n", err)
		}
	}()

	// wait for system under test and backend server to be ready
	if err := waitForStartup(cfg.getReadyURL(), cfg.getReadyURLWait(), cfg.getStartupWait()); err != nil {
		return nil, tcs, err
	}

	return &Runner{
		closeFuncs: closeFuncs,
		once:       new(sync.Once),
		httpClient: cfg.getHttpClient(),
	}, tcs, nil
}

// waitForStartup checks the ready endpoint if provided for the provided time
// and then and additional startupWait time for the backend services to
// be ready.
func waitForStartup(readyURL string, readyURLWait, startupWait time.Duration) error {
	if readyURL != "" {
		return waitForReady(readyURL, readyURLWait)
	}

	if startupWait <= 0 {
		startupWait = 1500 * time.Millisecond
	}
	<-time.After(startupWait)
	return nil
}

func waitForReady(readyURL string, readyURLWait time.Duration) error {
	if readyURLWait < time.Second {
		readyURLWait = time.Second
	}
	deadline := time.Now().Add(readyURLWait)
	resp, err := http.Get(readyURL)

	for (resp == nil || resp.StatusCode != 200) && time.Now().Before(deadline) {
		<-time.After(time.Second)
		resp, err = http.Get(readyURL)
	}

	if resp != nil && resp.StatusCode == 200 {
		// stderr: this is progress, not program output. ExampleNewIntegration
		// asserts on the harness's STDOUT, so a diagnostic printed there fails
		// the example for saying something true.
		fmt.Fprintf(os.Stderr, "system under test %s is ready\n", readyURL)
		return nil
	}

	if err != nil {
		return fmt.Errorf("cannot check %s is ready: %s", readyURL, err.Error())
	}

	if resp != nil {
		return fmt.Errorf("cannot check %s is ready: expecting 200 status code, got %d",
			readyURL, resp.StatusCode)
	}

	return fmt.Errorf("cannot check %s is ready", readyURL)
}

// Runner handles the integration test execution, by dealing with the request generation, response verification
// and the final shutdown
type Runner struct {
	closeFuncs []func()
	once       *sync.Once
	httpClient *http.Client
}

// Close shuts down the mocked backend server and the process of the instance
// under test
func (i *Runner) Close() {
	i.once.Do(func() {
		for _, closeF := range i.closeFuncs {
			closeF()
		}
	})

}

// Check runs a test case, returning an error if something goes wrong
func (i *Runner) Check(tc TestCase) error {
	req, err := newRequest(tc.In)
	if err != nil {
		return err
	}
	resp, err := i.httpClient.Do(req)
	if err != nil && err.Error() != tc.Err {
		return err
	}

	if err != nil {
		return nil
	}

	return assertResponse(resp, tc.Out)
}

type responseError struct {
	errMessage []string
}

func (m responseError) Error() string {
	return "wrong response:\n\t" + strings.Join(m.errMessage, "\n\t")
}

func assertResponse(actual *http.Response, expected Output) error {
	var errMsgs []string
	if actual.StatusCode != expected.StatusCode {
		errMsgs = append(errMsgs, fmt.Sprintf("unexpected status code. have: %d, want: %d", actual.StatusCode, expected.StatusCode))
	}

	for k, vs := range expected.Header {
		k = textproto.CanonicalMIMEHeaderKey(k)
		hs, ok := actual.Header[k]
		isEqual := reflect.DeepEqual(vs, hs)
		if ok && isEqual {
			continue
		}

		if ok {
			errMsgs = append(errMsgs, fmt.Sprintf("unexpected value for header %s. have: %s, want: %s", k, hs, vs))
			continue
		}

		if vs[0] != "" {
			errMsgs = append(errMsgs, fmt.Sprintf("header %s not present: %+v", k, actual.Header))
		}
	}

	var body interface{}
	var bodyBytes []byte
	if actual.Body != nil {
		// check if the body is compressed with gzip
		var r io.Reader = actual.Body
		if ce, ok := actual.Header["Content-Encoding"]; ok && len(ce) > 0 && (ce[0] == "gzip" || ce[0] == "x-gzip") {
			if gr, err := gzip.NewReader(actual.Body); err == nil {
				r = gr
			}
		}

		b, err := io.ReadAll(r)
		if err != nil {
			return err
		}
		_ = actual.Body.Close()

		switch expected.Body.(type) {
		case string:
			body = string(b)
		default:
			_ = json.Unmarshal(b, &body)
		}
		bodyBytes = b
	}

	if len(expected.Schema) != 0 {
		if len(bodyBytes) == 0 {
			return responseError{
				errMessage: append(errMsgs, "cannot validate empty body"),
			}
		}
		s, err := json.Marshal(expected.Schema)
		if err != nil {
			return responseError{
				errMessage: append(errMsgs, fmt.Sprintf("problem marshaling the user provided json-schema: %s", err)),
			}
		}
		doc, err := jsonschema.UnmarshalJSON(bytes.NewReader(s))
		if err != nil {
			return responseError{
				errMessage: append(errMsgs, fmt.Sprintf("problem unmarshaling the user provided json-schema: %s", err)),
			}
		}

		c := jsonschema.NewCompiler()
		c.AddResource("./schema.json", doc)
		schema, err := c.Compile("./schema.json")
		if err != nil {
			return responseError{
				errMessage: append(errMsgs, fmt.Sprintf("problem generating json-schema schema: %s", err)),
			}
		}

		b, err := jsonschema.UnmarshalJSON(bytes.NewReader(bodyBytes))
		if err != nil {
			return responseError{
				errMessage: append(errMsgs, fmt.Sprintf("problem unmarshaling the body: %s", err)),
			}
		}
		if err := schema.Validate(b); err != nil {
			return responseError{
				errMessage: append(errMsgs, fmt.Sprintf("problem validating the body: %s", sanitizeValidationError(err))),
			}
		}
	} else if expected.Body != nil {
		if !reflect.DeepEqual(body, expected.Body) {
			errMsgs = append(errMsgs, fmt.Sprintf("unexpected body.\n\t\thave: %v\n\t\twant: %v", body, expected.Body))
		}
	}
	if len(errMsgs) == 0 {
		return nil
	}

	return responseError{
		errMessage: errMsgs,
	}
}

// sanitizeValidationError drops the leading "jsonschema validation failed with
// <url>" line so fixture output stays stable across schema resource names.
func sanitizeValidationError(e error) string {
	s := e.Error()
	if strings.HasPrefix(s, "jsonschema validation failed with") {
		if ss := strings.SplitN(s, "\n", 2); len(ss) == 2 {
			return ss[1]
		}
	}
	return s
}

func testCases(cfg Config) ([]TestCase, error) {
	var tcs []TestCase
	content, err := readSpecs(cfg.getSpecsPath())
	if err != nil {
		return tcs, err
	}

	for name, c := range content {
		tc, err := parseTestCase(name, c)
		if err != nil {
			return tcs, err
		}
		tcs = append(tcs, tc)
	}

	return tcs, nil
}

func parseTestCase(name string, in []byte) (TestCase, error) {
	tc := TestCase{}
	if err := json.Unmarshal(in, &tc); err != nil {
		return tc, err
	}
	tc.Name = name

	return tc, nil
}

func newRequest(in Input) (*http.Request, error) {
	var body io.Reader

	if in.Body != nil {
		var b []byte
		switch in.Body.(type) {
		case string:
			b = []byte(in.Body.(string))
		default:
			b, _ = json.Marshal(in.Body)
		}
		body = bytes.NewBuffer(b)
	}

	req, err := http.NewRequest(in.Method, in.URL, body)
	if err != nil {
		return nil, err
	}

	if host, ok := in.Header["Host"]; ok {
		req.Host = host
	}

	for k, v := range in.Header {
		req.Header.Add(k, v)
	}
	return req, nil
}

func readSpecs(dirPath string) (map[string][]byte, error) {
	data := map[string][]byte{}
	files, err := os.ReadDir(dirPath)
	if err != nil {
		return data, err
	}

	for _, file := range files {
		if !strings.HasSuffix(file.Name(), ".json") {
			continue
		}
		content, err := os.ReadFile(path.Join(dirPath, file.Name()))
		if err != nil {
			return data, err
		}
		data[file.Name()[:len(file.Name())-5]] = content
	}
	return data, nil
}

var defaultCmdBuilder gatewayCmdBuilder

type gatewayCmdBuilder struct{}

func (k gatewayCmdBuilder) New(cfg *Config) *exec.Cmd {
	cmd := exec.Command(cfg.getBinPath(), "run", "-d", "-c", cfg.getCfgPath())
	cmd.Env = k.getEnviron(cfg)
	return cmd
}

func (gatewayCmdBuilder) getEnviron(cfg *Config) []string {
	environ := []string{"USAGE_DISABLE=1", "AUTH_ENABLED=false"}

	var patterns []*regexp.Regexp
	for _, pattern := range strings.Split(cfg.getEnvironPatterns(), ",") {
		re, err := regexp.Compile(pattern)
		if err != nil {
			continue
		}
		patterns = append(patterns, re)
	}

	for _, candidate := range os.Environ() {
		for _, pattern := range patterns {
			if pattern.MatchString(candidate) {
				environ = append(environ, candidate)
				break
			}
		}
	}

	return environ
}

var DefaultBackendBuilder mockBackendBuilder

type mockBackendBuilder struct{}

func (mockBackendBuilder) New(cfg *Config) http.Server {
	mux := http.NewServeMux()

	mux.HandleFunc("/param_forwarding/", checkXForwardedFor(http.HandlerFunc(echoEndpoint)))
	mux.HandleFunc("/xml", checkXForwardedFor(http.HandlerFunc(xmlEndpoint)))
	mux.HandleFunc("/collection/", checkXForwardedFor(http.HandlerFunc(collectionEndpoint)))
	mux.HandleFunc("/delayed/", checkXForwardedFor(delayedEndpoint(cfg.getDelay(), http.HandlerFunc(echoEndpoint))))
	mux.HandleFunc("/redirect/", checkXForwardedFor(http.HandlerFunc(redirectEndpoint)))
	mux.HandleFunc("/rss/", checkXForwardedFor(http.HandlerFunc(rssEndpoint)))
	mux.HandleFunc("/jwk/symmetric", http.HandlerFunc(symmetricJWKEndpoint))

	return http.Server{ // skipcq: GO-S2112
		Addr:    fmt.Sprintf(":%v", cfg.getBackendPort()),
		Handler: mux,
	}
}

func collectionEndpoint(rw http.ResponseWriter, r *http.Request) {
	rw.Header().Add("Content-Type", "application/json")
	var res []interface{}

	for i := 0; i < 10; i++ {
		res = append(res, map[string]interface{}{
			"path": r.URL.Path,
			"i":    i,
		})
	}

	json.NewEncoder(rw).Encode(res)
}

func checkXForwardedFor(h http.Handler) http.HandlerFunc {
	return func(rw http.ResponseWriter, r *http.Request) {
		if ip := net.ParseIP(r.Header.Get("X-Forwarded-For")); ip == nil || !ip.IsLoopback() {
			http.Error(rw, "invalid X-Forwarded-For", 400)
			return
		}
		h.ServeHTTP(rw, r)
	}
}

func delayedEndpoint(d time.Duration, h http.Handler) http.HandlerFunc {
	return func(rw http.ResponseWriter, req *http.Request) {
		<-time.After(d)
		h.ServeHTTP(rw, req)
	}
}

// rssEndpoint backs the /show/{id} spec, which pins the `rss` decoder composed
// with `group` + `allow`. Both of its backends used to be http://showrss.info/ —
// a third party, over plaintext HTTP — so the suite's verdict depended on a site
// nobody here operates. The channel title is the only field the spec allows
// through, so the feed only has to be well-formed RSS with a distinguishable
// title per path.
func rssEndpoint(rw http.ResponseWriter, r *http.Request) {
	title := "personal feed"
	if strings.HasPrefix(r.URL.Path, "/rss/schedule/") {
		title = "schedule feed"
	}
	rw.Header().Add("Content-Type", "application/rss+xml; charset=utf-8")
	fmt.Fprintf(rw, `<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0"><channel><title>%s</title><link>http://127.0.0.1/</link>
<description>gateway integration fixture</description></channel></rss>`, title)
}

func xmlEndpoint(rw http.ResponseWriter, _ *http.Request) {
	rw.Header().Add("Content-Type", "application/xml; charset=utf-8")
	rw.Write([]byte(`<?xml version="1.0" encoding="UTF-8"?>
<user type="admin">
  <name>Elliot</name>
  <social>
    <facebook>https://facebook.com</facebook>
    <twitter>https://twitter.com</twitter>
    <youtube>https://youtube.com</youtube>
  </social>
</user>`))
}

func echoEndpoint(rw http.ResponseWriter, r *http.Request) {
	rw.Header().Add("Content-Type", "application/json")
	rw.Header().Add("Set-Cookie", "test1=test1")
	rw.Header().Add("Set-Cookie", "test2=test2")
	r.Header.Del("X-Forwarded-For")
	resp := map[string]interface{}{
		"path":    r.URL.Path,
		"query":   r.URL.Query(),
		"headers": r.Header,
		"foo":     42,
	}

	if r.URL.Query().Get("dump_body") == "1" {
		b, _ := io.ReadAll(r.Body)
		r.Body.Close()
		resp["body"] = string(b)
	}

	json.NewEncoder(rw).Encode(resp)
}

func redirectEndpoint(rw http.ResponseWriter, r *http.Request) {
	u := r.URL
	u.Path = "/param_forwarding/"

	status, ok2 := r.URL.Query()["status"]
	code := 301
	if !ok2 || status[0] != "301" {
		var err error
		code, err = strconv.Atoi(status[0])
		if err != nil {
			http.Error(rw, err.Error(), http.StatusBadRequest)
			return
		}
	}
	http.Redirect(rw, r, u.String(), code)
}

func symmetricJWKEndpoint(rw http.ResponseWriter, _ *http.Request) {
	rw.Header().Add("Content-Type", "application/json")
	rw.Write([]byte(`{
  "keys": [
    {
      "kty": "oct",
      "alg": "A128KW",
      "k": "GawgguFyGrWKav7AX4VKUg",
      "kid": "sim1"
    },
    {
      "kty": "oct",
      "k": "AyM1SysPpbyDfgZld3umj1qzKObwVMkoqQ-EstJQLr_T-1qS0gZH75aKtMN3Yj0iPS4hcgUuTwjAzZr1Z9CAow",
      "kid": "sim2",
      "alg": "HS256"
    }
  ]
}`))
}