feat(tests): add e2e tests

Author: sendya

.github/workflows/go.yml

@@ -21,8 +21,14 @@ jobs:
       with:
         go-version: '1.25'
 
-#    - name: Build
-#      run: go build -v ./...
+    - name: Build
+      run: make build
+
+    - name: Run CacheServer
+      run: |
+        ./bin/tavern -c ./test/config.yaml 2>&1 &
+        sleep 2
+        echo "CacheServer is running"
 
     - name: Test
       run: go test -v ./...

internal/constants/global.go

@@ -14,4 +14,5 @@ const (
 	InternalSwapfile         = "i-x-swapfile"
 	InternalFillRangePercent = "i-x-fp"
 	InternalCacheErrCode     = "i-x-ct-code"
+	InternalUpstreamAddr     = "i-x-ups-addr"
 )

pkg/e2e/e2e.go

@@ -0,0 +1,185 @@
+package e2e
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/http/httputil"
+	"net/url"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/omalloc/tavern/internal/constants"
+	"github.com/stretchr/testify/assert"
+)
+
+var (
+	mu        sync.RWMutex
+	localAddr = "127.0.0.1:8888"
+	dump      = atomic.Bool{}
+	dumpReq   = atomic.Bool{}
+
+	manual = atomic.Bool{}
+)
+
+type E2E struct {
+	caseUrl    string
+	srcHandler http.Handler
+	ts         *httptest.Server
+	cs         *http.Client
+	req        *http.Request
+	resp       *http.Response
+	err        error
+}
+
+func New(caseUrl string, srcHandler http.HandlerFunc) *E2E {
+	e := &E2E{
+		caseUrl:    caseUrl,
+		srcHandler: srcHandler,
+	}
+
+	u, err := url.Parse(caseUrl)
+	if err != nil {
+		return e
+	}
+
+	if e.srcHandler == nil {
+		e.srcHandler = http.NotFoundHandler()
+	}
+
+	e.ts = httptest.NewServer(e.srcHandler)
+
+	dialer := &net.Dialer{}
+
+	// replace the default transport with a custom one that uses the local address
+	e.ts.Client().Transport = &http.Transport{
+		DialContext: func(ctx context.Context, network, _ string) (net.Conn, error) {
+			mu.RLock()
+			addr := localAddr
+			mu.RUnlock()
+
+			if strings.HasSuffix(addr, ".sock") {
+				return dialer.DialContext(ctx, "unix", addr)
+			}
+			return dialer.DialContext(ctx, network, addr)
+		},
+	}
+	e.cs = e.ts.Client()
+
+	req, err := http.NewRequest(http.MethodGet, e.caseUrl, nil)
+	if err != nil {
+		panic(err)
+	}
+
+	req.Header.Set("X-Test-Case", u.Path)
+	req.Header.Set("X-Store-Url", u.String())
+
+	e.req = req
+	return e
+}
+
+func (e *E2E) Do(rewrite func(r *http.Request)) (*http.Response, error) {
+	rewrite(e.req)
+
+	method := e.req.Method
+
+	e.req.Header.Set(constants.InternalUpstreamAddr, e.ts.Listener.Addr().String())
+
+	if dumpReq.Load() && method != "PURGE" {
+		DumpReq(e.req)
+	}
+
+	if manual.Load() {
+		fmt.Printf("manual mode wait 20s, src addr %q\n", e.ts.Listener.Addr().String())
+		time.Sleep(time.Second * 20)
+	}
+
+	resp, err := e.cs.Do(e.req)
+	e.resp = resp
+	e.err = err
+
+	// PURGE 不打印响应
+	if dump.Load() && method != "PURGE" {
+		DumpResp(resp)
+	}
+
+	// wait for a while to let the connection close properly
+	// time.Sleep(time.Millisecond * 200)
+
+	return resp, err
+}
+
+func (e *E2E) Close() {
+	e.ts.Close()
+	if e.resp != nil && e.resp.Body != nil {
+		_ = e.resp.Body.Close()
+	}
+}
+
+func SetLocalAddr(addr string) {
+	mu.Lock()
+	defer mu.Unlock()
+
+	localAddr = addr
+}
+
+func SetDump(b bool) {
+	dump.Store(b)
+}
+
+func SetDumpReq(b bool) {
+	dumpReq.Store(b)
+}
+
+func SetManual(b bool) {
+	manual.Store(b)
+}
+
+func DumpReq(req *http.Request) {
+	if req == nil {
+		fmt.Println("request is nil")
+		return
+	}
+
+	buf, err := httputil.DumpRequest(req, false)
+	if err != nil {
+		return
+	}
+	fmt.Println()
+	fmt.Println(string(buf))
+}
+
+func DumpResp(resp *http.Response) {
+	if resp == nil {
+		fmt.Println("response is nil")
+		return
+	}
+
+	buf, err := httputil.DumpResponse(resp, false)
+	if err != nil {
+		return
+	}
+	fmt.Println()
+	fmt.Println(string(buf))
+}
+
+func Purge(t *testing.T, url string) {
+	resp, err := New(url, func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusBadGateway)
+	}).Do(func(r *http.Request) {
+		r.Method = "PURGE"
+	})
+
+	assert.NoError(t, err, "purge should not error")
+
+	assert.Contains(t, []int{http.StatusOK, http.StatusNotFound}, resp.StatusCode, "PURGE should be 404 or 200")
+
+	if dump.Load() {
+		t.Logf("Purge %s success", url)
+	}
+}

pkg/e2e/e2e_file.go

@@ -0,0 +1,84 @@
+package e2e
+
+import (
+	"crypto/md5"
+	"crypto/rand"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/omalloc/tavern/pkg/iobuf"
+)
+
+type MockFile struct {
+	Path string
+	MD5  string
+	Size int
+}
+
+func GenFile(t *testing.T, size int) *MockFile {
+	buf := make([]byte, size)
+	_, _ = rand.Read(buf)
+
+	fpath := filepath.Join(t.TempDir(), fmt.Sprintf("file-%d.bin", size))
+	_ = os.WriteFile(fpath, buf, 0o644)
+
+	return &MockFile{
+		Path: fpath,
+		MD5:  SumMD5(buf),
+		Size: size,
+	}
+}
+
+func SumMD5(buf []byte) string {
+	h := md5.New()
+	_, _ = h.Write(buf)
+	return hex.EncodeToString(h.Sum(nil))
+}
+
+func DiscardBody(resp *http.Response) int64 {
+	if resp == nil || resp.Body == nil {
+		return 0
+	}
+
+	n, _ := io.Copy(io.Discard, resp.Body)
+	resp.Body.Close()
+	return n
+}
+
+func HashBody(resp *http.Response) string {
+	if resp == nil || resp.Body == nil {
+		return ""
+	}
+
+	h := md5.New()
+	_, _ = io.Copy(h, resp.Body)
+	return hex.EncodeToString(h.Sum(nil))
+}
+
+func HashFile(path string, offset, length int) string {
+	f, err := os.OpenFile(path, os.O_RDONLY, 0o644)
+	if err != nil {
+		return ""
+	}
+	defer f.Close()
+
+	_, _ = f.Seek(int64(offset), io.SeekStart)
+
+	h := md5.New()
+	_, _ = io.CopyN(h, f, int64(length))
+	return hex.EncodeToString(h.Sum(nil))
+}
+
+func SplitFile(path string, offset, length int) io.ReadCloser {
+	f, err := os.OpenFile(path, os.O_RDONLY, 0o644)
+	if err != nil {
+		return nil
+	}
+	_, _ = f.Seek(int64(offset), io.SeekStart)
+	return iobuf.RangeReader(f, offset, offset+length-1, offset, offset+length-1)
+}

pkg/e2e/e2e_hander.go

@@ -0,0 +1,47 @@
+package e2e
+
+import (
+	"errors"
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func WrongHit(t *testing.T) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusBadGateway)
+
+		assert.Error(t, errors.New("equal request HIT but NOT HIT"))
+	}
+}
+
+func RespSimpleFile(f *MockFile) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Cache-Control", "max-age=10")
+		w.Header().Set("Content-MD5", f.MD5)
+		w.Header().Set("ETag", f.MD5)
+		w.Header().Set("X-Server", "tavern-e2e/1.0.0")
+
+		http.ServeFile(w, r, f.Path)
+	}
+}
+
+func RespCallbackFile(f *MockFile, cb func(w http.ResponseWriter, r *http.Request)) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Cache-Control", "max-age=10")
+		w.Header().Set("Content-MD5", f.MD5)
+		w.Header().Set("ETag", f.MD5)
+		w.Header().Set("X-Server", "tavern-e2e/1.0.0")
+
+		cb(w, r)
+
+		http.ServeFile(w, r, f.Path)
+	}
+}
+
+func RespCallback(cb func(w http.ResponseWriter, r *http.Request)) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		cb(w, r)
+	}
+}

server/middleware/caching/internal.go

@@ -17,6 +17,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/omalloc/proxy/selector"
 	"github.com/omalloc/tavern/api/defined/v1/storage"
 	"github.com/omalloc/tavern/api/defined/v1/storage/object"
 	"github.com/omalloc/tavern/contrib/log"
@@ -292,6 +293,13 @@ func cloneRequest(req *http.Request) *http.Request {
 	xhttp.CopyHeader(proxyReq.Header, req.Header)
 	xhttp.RemoveHopByHopHeaders(proxyReq.Header)
 
+	// custom upstream addr
+	if upsAddr := req.Header.Get(constants.InternalUpstreamAddr); upsAddr != "" {
+		proxyReq = proxyReq.WithContext(
+			selector.NewPeerContext(context.Background(), selector.NewPeer(selector.NewNode("http", upsAddr, map[string]string{}))),
+		)
+	}
+
 	return proxyReq
 }