pkg-proxy/internal/handler/profile_test.go

package handler

import (
	"context"
	"errors"
	"net/http"
	"net/http/httptest"
	"reflect"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/git-pkgs/cooldown"
	"github.com/git-pkgs/proxy/internal/denylist"
	"github.com/git-pkgs/proxy/internal/scanner"
	registryclient "github.com/git-pkgs/registries/client"
)

func TestWithCooldownCopiesProxyFields(t *testing.T) {
	p, _, _, _ := setupTestProxy(t)
	p.RouteFor = func(string, string) *ArtifactRoute { return nil }
	p.Cooldown = profilePolicy(t, "1d")
	p.ShortestCooldown = 24 * time.Hour
	p.Denylist = &denylist.Policy{}
	p.CacheMetadata = true
	p.MetadataTTL = time.Hour
	p.MetadataMaxSize = 1024
	p.MetadataStreamMaxSize = 2048
	p.GradleReadOnly = true
	p.HostedEnabled = true
	p.GradleMaxUploadSize = 4096
	p.NPMFullMetadata = true
	p.DirectServe = true
	p.DirectServeTTL = time.Minute
	p.DirectServeBaseURL = "https://storage.example.com"
	p.RateLimits = registryclient.NewRateLimitGate(http.DefaultTransport)
	p.HostedNPMScopes = []string{"@example"}
	p.HostedComposerVendors = []string{"example"}
	p.StreamArtifacts = true
	p.Scanners = &scanner.Group{}
	p.ScanSigningKey = []byte("signing-key")
	p.ScanFetchBaseURL = "http://scanner.example.com"
	p.SetMetadataRewriteCacheSize(1024)
	p.inFlight = map[string]*inflightFetch{"artifact": {}}
	p.detachSlots = make(chan struct{}, 1)
	p.inFlightMeta = map[string]*inflightMetadata{"metadata": {}}
	p.publishIndexes.entries = map[string]publishIndexEntry{"document": {}}
	p.cooldownWatch.active = map[*cooldownRecorder]struct{}{{}: {}}

	policy := profilePolicy(t, "3d")
	view := p.WithCooldown(policy, 72*time.Hour)
	if view.Cooldown != policy || view.ShortestCooldown != 72*time.Hour || view.shared != p {
		t.Fatal("view did not select its policy and shared proxy")
	}
	original, copied := reflect.ValueOf(p).Elem(), reflect.ValueOf(view).Elem()
	for i := range original.NumField() {
		name := original.Type().Field(i).Name
		before, after := original.Field(i), copied.Field(i)
		switch name {
		case "Cooldown", "ShortestCooldown", "shared":
			continue
		case "fetchMu", "inFlight", "detachSlots", "metaMu", "inFlightMeta", "publishIndexes", "cooldownWatch":
			if !after.IsZero() {
				t.Errorf("view field %s is not empty", name)
			}
			continue
		}
		if before.IsZero() {
			t.Errorf("populate Proxy.%s to check whether WithCooldown copies it", name)
			continue
		}
		switch before.Kind() {
		case reflect.Func, reflect.Slice:
			if before.Pointer() != after.Pointer() {
				t.Errorf("WithCooldown did not copy %s", name)
			}
			if before.Kind() == reflect.Slice && (before.Len() != after.Len() || before.Cap() != after.Cap()) {
				t.Errorf("WithCooldown changed slice bounds for %s", name)
			}
		default:
			if !before.Equal(after) {
				t.Errorf("WithCooldown did not copy %s", name)
			}
		}
	}
}

func TestProfilesShareDownloadPublishIndexes(t *testing.T) {
	for _, c := range downloadCooldownCases() {
		if c.ecosystem != "conda" && c.ecosystem != "helm" {
			continue
		}
		t.Run(c.ecosystem, func(t *testing.T) {
			var calls atomic.Int64
			times := map[string]time.Time{cooldownRecentVersion: time.Now().Add(-4 * 24 * time.Hour)}
			upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				calls.Add(1)
				c.metadata(w, r, times)
			}))
			t.Cleanup(upstream.Close)
			p, _, _, fetcher := setupTestProxy(t)
			p.HTTPClient = upstream.Client()
			p.Cooldown = profilePolicy(t, "1d")
			fetcher.fetchErr = errors.New("artifact must be served from the cache")
			c.seedCachedArtifact(t, p, cooldownRecentVersion)
			permissive := p.WithCooldown(profilePolicy(t, "3d"), 72*time.Hour)
			strict := p.WithCooldown(profilePolicy(t, "7d"), 168*time.Hour)
			nested := permissive.WithCooldown(profilePolicy(t, "5d"), 120*time.Hour)
			for _, request := range []struct {
				proxy *Proxy
				want  int
			}{
				{permissive, http.StatusOK},
				{strict, http.StatusNotFound},
				{p, http.StatusOK},
				{nested, http.StatusNotFound},
			} {
				w := serveGet(t, c.handler(request.proxy, upstream.URL), c.path(cooldownRecentVersion))
				wantStatus(t, w, request.want, "")
			}
			if n := calls.Load(); n != 1 {
				t.Errorf("index downloads across profiles = %d, want 1", n)
			}
			if n := len(p.publishIndexes.entries); n != 1 {
				t.Errorf("shared publish indexes = %d, want 1", n)
			}
			if fetcher.fetchCalled {
				t.Error("cached artifact was fetched again")
			}
		})
	}
}

func profilePolicy(t *testing.T, duration string) *cooldown.Policy {
	t.Helper()
	policy, err := cooldown.NewPolicy(&cooldown.Config{Default: duration}, nil)
	if err != nil {
		t.Fatal(err)
	}
	return policy
}

func TestProfilesCoalesceUncachedMetadata(t *testing.T) {
	u := newMetadataUpstream(t, http.StatusOK)
	p := metadataTestProxy(t, u)
	views := []*Proxy{p.WithCooldown(profilePolicy(t, "3d"), 72*time.Hour), p.WithCooldown(profilePolicy(t, "7d"), 168*time.Hour)}
	first := startMetadataCalls(views[0], context.Background(), 1, u.URL, contentTypeJSON)
	<-u.entered
	second := startMetadataCalls(views[1], context.Background(), 1, u.URL, contentTypeJSON)
	key := metadataCoalesceKey("npm", "left-pad", u.URL, contentTypeJSON, "", false)
	var release sync.Once
	t.Cleanup(func() { release.Do(func() { close(u.release) }) })
	waitForMetadataWaiters(t, p, key, 1)
	release.Do(func() { close(u.release) })
	for _, calls := range [][]chan metadataCall{first, second} {
		result := <-calls[0]
		if result.err != nil || !strings.Contains(string(result.body), "accept") {
			t.Fatalf("metadata = %s, error %v", result.body, result.err)
		}
	}
	if n := u.calls.Load(); n != 1 {
		t.Fatalf("upstream requests = %d, want 1", n)
	}
}

func TestProfilesCoalesceArtifactRequests(t *testing.T) {
	proxy, db, _, _ := setupTestProxy(t)
	if err := db.SetVersionPublishedAt("pkg:cargo/demo@1.0.0", "pkg:cargo/demo", time.Now().Add(-10*24*time.Hour)); err != nil {
		t.Fatal(err)
	}
	fetcher := &countingFetcher{content: "crate bytes", delay: fetchHoldTime}
	proxy.Fetcher = fetcher
	views := []*Proxy{proxy.WithCooldown(profilePolicy(t, "3d"), 72*time.Hour), proxy.WithCooldown(profilePolicy(t, "7d"), 168*time.Hour)}
	var wg sync.WaitGroup
	start := make(chan struct{})
	for range 4 {
		for _, view := range views {
			wg.Go(func() {
				<-start
				h := NewCargoHandler(view, "http://proxy.test", "https://index.crates.io", "https://static.crates.io/crates")
				w := httptest.NewRecorder()
				h.Routes().ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/crates/demo/1.0.0/download", nil))
				if w.Code != http.StatusOK || w.Body.String() != "crate bytes" {
					t.Errorf("artifact response %d: %s", w.Code, w.Body)
				}
			})
		}
	}
	close(start)
	wg.Wait()
	if n := fetcher.calls.Load(); n != 1 {
		t.Fatalf("upstream fetches = %d, want 1", n)
	}
}

type profileClockPolicy struct {
	*cooldown.Policy
	now func() time.Time
}

func (p profileClockPolicy) IsAllowed(ecosystem, packagePURL string, publishedAt time.Time) bool {
	return p.Evaluate(ecosystem, packagePURL, publishedAt, p.now()).Allowed
}

func TestProfileExpiryReleasesVersionsIndependently(t *testing.T) {
	u, proxy, _ := gemCacheSetup(t)
	clock := &testClock{t: time.Now()}
	proxy.now = clock.now
	proxy.SetMetadataRewriteCacheSize(1 << 20)
	view := proxy.WithCooldown(profileClockPolicy{profilePolicy(t, "3d"), clock.now}, 72*time.Hour)
	h := NewGemHandlerWithUpstream(view, "http://proxy.test/cooldown/3d", u.URL).Routes()
	strictView := proxy.WithCooldown(profileClockPolicy{profilePolicy(t, "7d"), clock.now}, 168*time.Hour)
	strict := NewGemHandlerWithUpstream(strictView, "http://proxy.test/cooldown/7d", u.URL).Routes()
	first := getWith(t, h, gemInfoPath, nil)
	if first.Code != http.StatusOK || first.Body.String() != gemFiltered {
		t.Fatalf("initial response %d: %s", first.Code, first.Body)
	}
	if first := getWith(t, strict, gemInfoPath, nil); first.Code != http.StatusOK || first.Body.String() != gemFiltered {
		t.Fatalf("initial strict response %d: %s", first.Code, first.Body)
	}
	clock.advance(4 * 24 * time.Hour)
	second := getWith(t, h, gemInfoPath, nil)
	if second.Code != http.StatusOK || !strings.Contains(second.Body.String(), "2.0.0 |") {
		t.Fatalf("version did not leave cooldown: %d %s", second.Code, second.Body)
	}
	if second := getWith(t, strict, gemInfoPath, nil); second.Code != http.StatusOK || second.Body.String() != gemFiltered {
		t.Fatalf("strict profile released version early: %d %s", second.Code, second.Body)
	}
	if n, _ := u.counts(gemInfoPath); n != 1 {
		t.Errorf("expiry refetched raw metadata %d times", n)
	}
}