kubernetes/staging/src/k8s.io/apiserver/pkg/server/filters/priority-and-fairness_test.go

func TestMain(m *testing.M) { … }

type mockDecision …

const decisionNoQueuingExecute …

const decisionQueuingExecute …

const decisionCancelWait …

const decisionReject …

const decisionSkipFilter …

var defaultRequestWorkEstimator …

type fakeApfFilter …

func (t fakeApfFilter) Handle(ctx context.Context,
	requestDigest utilflowcontrol.RequestDigest,
	noteFn func(fs *flowcontrol.FlowSchema, pl *flowcontrol.PriorityLevelConfiguration, flowDistinguisher string),
	workEstimator func() fcrequest.WorkEstimate,
	queueNoteFn fq.QueueNoteFn,
	execFn func(),
) { … }

func (t fakeApfFilter) Run(stopCh <-chan struct{ … }

func (t fakeApfFilter) Install(c *mux.PathRecorderMux) { … }

func newApfServerWithSingleRequest(t *testing.T, decision mockDecision) *httptest.Server { … }

func newApfServerWithHooks(t *testing.T, decision mockDecision, onExecute, postExecute, postEnqueue, postDequeue func()) *httptest.Server { … }

func newApfServerWithFilter(t *testing.T, flowControlFilter utilflowcontrol.Interface, defaultWaitLimit time.Duration, onExecute, postExecute func()) *httptest.Server { … }

func newApfHandlerWithFilter(t *testing.T, flowControlFilter utilflowcontrol.Interface, defaultWaitLimit time.Duration, onExecute, postExecute func()) http.Handler { … }

func TestApfSkipLongRunningRequest(t *testing.T) { … }

func TestApfRejectRequest(t *testing.T) { … }

func TestApfExemptRequest(t *testing.T) { … }

func TestApfExecuteRequest(t *testing.T) { … }

func TestApfExecuteMultipleRequests(t *testing.T) { … }

func TestApfCancelWaitRequest(t *testing.T) { … }

type fakeWatchApfFilter …

func newFakeWatchApfFilter(capacity int) *fakeWatchApfFilter { … }

func (f *fakeWatchApfFilter) Handle(ctx context.Context,
	requestDigest utilflowcontrol.RequestDigest,
	noteFn func(fs *flowcontrol.FlowSchema, pl *flowcontrol.PriorityLevelConfiguration, flowDistinguisher string),
	_ func() fcrequest.WorkEstimate,
	_ fq.QueueNoteFn,
	execFn func(),
) { … }

func (f *fakeWatchApfFilter) Run(stopCh <-chan struct{ … }

func (t *fakeWatchApfFilter) Install(c *mux.PathRecorderMux) { … }

func (f *fakeWatchApfFilter) wait() error { … }

func TestApfExecuteWatchRequestsWithInitializationSignal(t *testing.T) { … }

func TestApfRejectWatchRequestsWithInitializationSignal(t *testing.T) { … }

func TestApfWatchPanic(t *testing.T) { … }

func TestApfWatchHandlePanic(t *testing.T) { … }

// TestContextClosesOnRequestProcessed ensures that the request context is cancelled
// automatically even if the server doesn't cancel is explicitly.
// This is required to ensure we won't be leaking goroutines that wait for context
// cancelling (e.g. in queueset::StartRequest method).
// Even though in production we are not using httptest.Server, this logic is shared
// across these two.
func TestContextClosesOnRequestProcessed(t *testing.T) { … }

type fakeFilterRequestDigest …

func (f *fakeFilterRequestDigest) Handle(ctx context.Context,
	requestDigest utilflowcontrol.RequestDigest,
	noteFn func(fs *flowcontrol.FlowSchema, pl *flowcontrol.PriorityLevelConfiguration, flowDistinguisher string),
	workEstimator func() fcrequest.WorkEstimate,
	_ fq.QueueNoteFn, _ func(),
) { … }

func TestApfWithRequestDigest(t *testing.T) { … }

func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) { … }

func fmtError(err error) string { … }

func startAPFController(t *testing.T, stopCh <-chan struct{ … }

// returns a started http2 server, with a client function to send request to the server.
func newHTTP2ServerWithClient(handler http.Handler, clientTimeout time.Duration) (*httptest.Server, func(path string) (*http.Response, error)) { … }

type headerMatcher …

// verifies that the expected flow schema and priority level UIDs are attached to the header.
func (m *headerMatcher) inspect(t *testing.T, w http.ResponseWriter, expectedFS, expectedPL string) { … }

// when a request panics, http2 resets the stream with an INTERNAL_ERROR message
func expectResetStreamError(t *testing.T, err error) { … }

func newClientset(t *testing.T, objects ...runtime.Object) clientset.Interface { … }

// builds a chain of handlers that include the panic recovery and timeout filter, so we can simulate the behavior of
// a real apiserver.
// the specified user is added as the authenticated user to the request context.
func newHandlerChain(t *testing.T, handler http.Handler, filter utilflowcontrol.Interface, userName string, requestTimeout time.Duration) http.Handler { … }

func unsyncedInformers(status map[reflect.Type]bool) []string { … }

func newConfiguration(fsName, plName, user string, concurrency int32, queueLength int32) []runtime.Object { … }

// gathers and checks the metrics.
func checkForExpectedMetrics(t *testing.T, expectedMetrics []string) { … }

// gaugeValueMatch ensures that the value of gauge metrics matching the labelFilter is as expected.
func gaugeValueMatch(name string, labelFilter map[string]string, wantValue int) error { … }

func isClientTimeout(err error) bool { … }

func isStreamReset(err error) bool { … }

func TestGetRequestWaitContext(t *testing.T) { … }