Function TestFetcherOpenParallel

core/remotes/docker/fetcher_test.go:139–294

(t *testing.T)

}

func TestFetcherOpenParallel(t *testing.T) {
	size := int64(3 * 1024 * 1024)
	content := make([]byte, size)
	rand.New(rand.NewSource(1)).Read(content)
	sendContentLength := true

	f := dockerFetcher{
		&dockerBase{
			repository: "nonempty",
			limiter:    nil,
			performances: transfer.ImageResolverPerformanceSettings{
				ConcurrentLayerFetchBuffer: 0,
			},
		},
	}

	cfgConcurency := func(maxConcurrentDls, chunkSize int) {
		f.dockerBase.performances.MaxConcurrentDownloads = maxConcurrentDls
		f.dockerBase.performances.ConcurrentLayerFetchBuffer = chunkSize
		limiter := semaphore.NewWeighted(int64(maxConcurrentDls))
		f.dockerBase.limiter = limiter
	}

	ignoreContentRange := false
	var failTimeoutOnceAfter atomic.Int64
	failAfter := int64(0)
	var forceRange []httpRange
	s := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
		rng, err := parseRange(r.Header.Get("Range"), size)
		if errors.Is(err, errNoOverlap) {
			err = nil
		}
		assert.NoError(t, err)
		if ignoreContentRange {
			rng = nil
		}
		if forceRange != nil {
			rng = forceRange
		}
		if failAfter > 0 && rng[0].start > failAfter {
			rw.WriteHeader(http.StatusInternalServerError)
			return
		}
		timeoutOnceAfter := failTimeoutOnceAfter.Load()
		if timeoutOnceAfter > 0 && rng[0].start > timeoutOnceAfter {
			failTimeoutOnceAfter.Store(0)
			rw.WriteHeader(http.StatusRequestTimeout)
			return
		}

		if len(rng) == 0 {
			if sendContentLength {
				rw.Header().Set("content-length", strconv.Itoa(len(content)))
			}
			_, _ = rw.Write(content)
		} else {
			b := content[rng[0].start : rng[0].start+rng[0].length]
			rw.Header().Set("content-range", rng[0].contentRange(size))