test: RunOnListener, local pypi virtual merge, v2 download, waitForStore/Unwrap

internal/proxy/engine_test.go

@@ -318,6 +318,47 @@ func TestBearerTokenParsing(t *testing.T) {
 	}
 }
 
+func TestWaitForStoreCoalesces(t *testing.T) {
+	requireStack(t)
+	ctx := context.Background()
+	r := seed(t, genericRemote("eng-herd"))
+	p := prov(t, models.PackageGeneric)
+
+	// Fire concurrent cold-cache fetches: only one holds the lock, the others
+	// wait on the store (waitForStore) and pick up the result.
+	const n = 4
+	done := make(chan string, n)
+	for i := 0; i < n; i++ {
+		go func() {
+			res, err := testEngine.Fetch(ctx, r, "blob.bin", p)
+			if err != nil {
+				done <- "err:" + err.Error()
+				return
+			}
+			done <- readAll(t, res)
+		}()
+	}
+	for i := 0; i < n; i++ {
+		if got := <-done; got != "immutable blob" {
+			t.Errorf("concurrent fetch got %q", got)
+		}
+	}
+}
+
+func TestUpstreamErrorUnwrap(t *testing.T) {
+	base := context.DeadlineExceeded
+	ue := &UpstreamError{Err: base}
+	if ue.Unwrap() != base {
+		t.Error("Unwrap should return the wrapped error")
+	}
+	if !isNetworkError(ue) {
+		t.Error("UpstreamError should be a network error")
+	}
+	if isNetworkError(context.Canceled) {
+		t.Error("plain error should not be a network error")
+	}
+}
+
 func asProxyError(err error, target **ProxyError) bool {
 	pe, ok := err.(*ProxyError)
 	if ok {

internal/server/server_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -21,6 +22,7 @@ import (
 var (
 	testTS   *httptest.Server // the artifactapi router
 	upstream *httptest.Server // mock upstream the proxy fetches from
+	testSrv  *Server
 )
 
 func TestMain(m *testing.M) {
@@ -72,6 +74,7 @@ func TestMain(m *testing.M) {
 	if err != nil {
 		panic(err)
 	}
+	testSrv = srv
 	testTS = httptest.NewServer(srv.router)
 	upstream = httptest.NewServer(http.HandlerFunc(mockUpstream))
 
@@ -199,6 +202,10 @@ func TestServerLocalUpload(t *testing.T) {
 	if resp.StatusCode != 200 || string(b) != "local payload" {
 		t.Errorf("download local: %d %s", resp.StatusCode, b)
 	}
+	// Also download via the v2 files endpoint.
+	if resp, b := req(t, "GET", "/api/v2/remotes/srv-local/files/dir/hello.bin", ""); resp.StatusCode != 200 || string(b) != "local payload" {
+		t.Errorf("v2 download: %d %s", resp.StatusCode, b)
+	}
 }
 
 func TestServerVirtualMerge(t *testing.T) {
@@ -443,6 +450,60 @@ func TestServerEvents(t *testing.T) {
 	// A timeout is expected for a streaming endpoint; the handler still ran.
 }
 
+func TestRunOnListener(t *testing.T) {
+	requireStack(t)
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	errc := make(chan error, 1)
+	go func() { errc <- testSrv.RunOnListener(ctx, ln) }()
+
+	base := "http://" + ln.Addr().String()
+	ok := false
+	for i := 0; i < 50; i++ {
+		if resp, e := http.Get(base + "/health"); e == nil {
+			resp.Body.Close()
+			ok = resp.StatusCode == 200
+			break
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	if !ok {
+		t.Error("server did not serve /health")
+	}
+	cancel()
+	select {
+	case err := <-errc:
+		if err != nil {
+			t.Errorf("RunOnListener returned error: %v", err)
+		}
+	case <-time.After(12 * time.Second):
+		t.Fatal("RunOnListener did not shut down")
+	}
+}
+
+func TestServerVirtualLocalPyPIMerge(t *testing.T) {
+	requireStack(t)
+	for _, n := range []string{"a", "b"} {
+		req(t, "POST", "/api/v2/remotes", `{"name":"srv-pm-`+n+`","package_type":"pypi","repo_type":"local"}`)
+		defer req(t, "DELETE", "/api/v2/remotes/srv-pm-"+n, "")
+	}
+	put(t, "/api/v2/remotes/srv-pm-a/files/foo-1.0-py3-none-any.whl", []byte("foo"))
+	put(t, "/api/v2/remotes/srv-pm-b/files/bar-2.0-py3-none-any.whl", []byte("bar"))
+	req(t, "POST", "/api/v2/virtuals", `{"name":"srv-pmv","package_type":"pypi","members":["srv-pm-a","srv-pm-b"]}`)
+	defer req(t, "DELETE", "/api/v2/virtuals/srv-pmv", "")
+
+	resp, b := req(t, "GET", "/api/v1/virtual/srv-pmv/simple/", "")
+	if resp.StatusCode != 200 {
+		t.Fatalf("virtual pypi index: %d %s", resp.StatusCode, b)
+	}
+	if s := string(b); !strings.Contains(s, "foo") || !strings.Contains(s, "bar") {
+		t.Errorf("merged local pypi index missing packages: %s", s)
+	}
+}
+
 func TestServerNotFound(t *testing.T) {
 	requireStack(t)
 	if resp, _ := req(t, "GET", "/api/v2/remotes/does-not-exist", ""); resp.StatusCode != 404 {