fixtures

package
v1.9.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 29, 2026 License: Apache-2.0 Imports: 44 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	Namespace  = "numaflow-system"
	Label      = "numaflow-e2e"
	LabelValue = "true"
	ISBSvcName = "numaflow-e2e"

	LogSourceVertexStarted    = "Start processing source messages"
	SinkVertexStarted         = "Start processing sink messages"
	LogUDFVertexStarted       = "Start processing udf messages"
	LogReduceUDFVertexStarted = "Start processing reduce udf messages"
	LogDaemonStarted          = "Daemon server started successfully"

	LogSourceVertexStartedRustRuntime = "Starting source forwarder"
	LogSinkVertexStartedRustRuntime   = "Starting sink forwarder"
	LogMapVertexStartedRustRuntime    = "Starting map forwarder"
	LogReduceVertexStartedRustRuntime = "Starting reduce forwarder"
	LogRetryAttempt                   = `"retry_attempt":`
	LogRetriesExhaustedDrop           = "Retries exhausted, dropping message"
	LogReceivedNack                   = "received nack"
)

Variables

View Source 
var CheckPodKillSucceeded = func(t *testing.T, output string, err error) {
	assert.Contains(t, output, "deleted")
	assert.NoError(t, err)
}
View Source 
var CheckVertexScaled = func(t *testing.T, output string, err error) {
	assert.Contains(t, output, "scaled")
	assert.NoError(t, err)
}
View Source 
var OutputRegexp = func(rx string) func(t *testing.T, output string, err error) {
	return func(t *testing.T, output string, err error) {
		t.Helper()
		if assert.NoError(t, err, output) {
			assert.Regexp(t, rx, output)
		}
	}
}

Functions

func AssertJsonEqual added in v1.8.0 

func AssertJsonEqual[T string | int | int64 | bool | Json](t *testing.T, json string, path string, expected T)

AssertJsonEqual asserts the path value of the json equals to the expected value

func AssertJsonExists added in v1.8.0 

func AssertJsonExists(t *testing.T, json string, path string)

AssertJsonExists asserts the path exists in the json

func AssertJsonStringContains added in v1.8.0 

func AssertJsonStringContains(t *testing.T, json string, path string, expectedContains string)

func CreateKafkaTopic 

func CreateKafkaTopic(topicName string, partitions int32)

func DaemonPodLogContains added in v0.5.4 

func DaemonPodLogContains(ctx context.Context, kubeClient kubernetes.Interface, namespace, pipelineName, regex string, opts ...PodLogCheckOption) (bool, error)

func DeleteKafkaTopic 

func DeleteKafkaTopic(topic string)

func Exec 

func Exec(name string, args ...string) (string, error)

func ExpectKafkaTopicCount 

func ExpectKafkaTopicCount(topic string, total int, timeout time.Duration)

func FetchServingResult added in v1.5.0 

func FetchServingResult(host, reqId string) string

func GenerateKafkaTopicName added in v1.3.0 

func GenerateKafkaTopicName() string

func GetKafkaCount 

func GetKafkaCount(topic string, count int) int

func HTTPExpect 

func HTTPExpect(t require.TestingT, baseURL string) *httpexpect.Expect

func InvokeE2EAPI 

func InvokeE2EAPI(format string, args ...interface{}) string

func InvokeE2EAPIPOST 

func InvokeE2EAPIPOST(format string, body string, args ...interface{}) string

func ListKafkaTopics added in v1.3.0 

func ListKafkaTopics()

func MonoVertexPodLogContains added in v1.3.0 

func MonoVertexPodLogContains(ctx context.Context, kubeClient kubernetes.Interface, namespace, mvName, regex string, opts ...PodLogCheckOption) (bool, error)

func MonoVertexPodLogNotContains added in v1.5.0 

func MonoVertexPodLogNotContains(ctx context.Context, kubeClient kubernetes.Interface, namespace, mvName, regex string, opts ...PodLogCheckOption) (bool, error)

func PodPortForward 

func PodPortForward(config *rest.Config, namespace, podName string, localPort, remotePort int, stopCh <-chan struct{}) error

func PodsLogContains 

func PodsLogContains(ctx context.Context, kubeClient kubernetes.Interface, namespace, regex string, podList *corev1.PodList, opts ...PodLogCheckOption) bool

func PodsLogNotContains added in v0.5.3 

func PodsLogNotContains(ctx context.Context, kubeClient kubernetes.Interface, namespace, regex string, podList *corev1.PodList, opts ...PodLogCheckOption) bool

func PumpJetstream added in v1.3.0 

func PumpJetstream(streamName string, payload string, count int)

func PumpKafkaTopic 

func PumpKafkaTopic(topic string, n int, opts ...interface{})

func PumpNatsSubject added in v0.7.0 

func PumpNatsSubject(subject string, n int, opts ...interface{})

func SendMessage 

func SendMessage(topic string, key string, message string, partition int)

func SendMessageTo added in v0.7.0 

func SendMessageTo(host string, vertexName string, r HttpPostRequest)

SendMessageTo sends a http post request to a pod in http source vertex.

func SendServingMessage added in v1.5.0 

func SendServingMessage(host, reqId, payload string, sync bool) string

func VertexPodLogContains 

func VertexPodLogContains(ctx context.Context, kubeClient kubernetes.Interface, namespace, pipelineName, vertexName, regex string, opts ...PodLogCheckOption) (bool, error)

func VertexPodLogNotContains added in v0.5.3 

func VertexPodLogNotContains(ctx context.Context, kubeClient kubernetes.Interface, namespace, pipelineName, vertexName, regex string, opts ...PodLogCheckOption) (bool, error)

func WaitForDaemonPodsRunning added in v0.5.4 

func WaitForDaemonPodsRunning(kubeClient kubernetes.Interface, namespace, pipelineName string, timeout time.Duration) error

func WaitForISBSvcReady 

func WaitForISBSvcReady(ctx context.Context, isbSvcClient flowpkg.InterStepBufferServiceInterface, isbSvcName string, timeout time.Duration) error

func WaitForISBSvcStatefulSetReady 

func WaitForISBSvcStatefulSetReady(ctx context.Context, kubeClient kubernetes.Interface, namespace, isbSvcName string, timeout time.Duration) error

func WaitForMonoVertexPodRunning added in v1.3.0 

func WaitForMonoVertexPodRunning(kubeClient kubernetes.Interface, monoVertexClient flowpkg.MonoVertexInterface, namespace, monoVertexName string, timeout time.Duration) error

func WaitForMonoVertexPodScalingTo added in v1.9.0 

func WaitForMonoVertexPodScalingTo(kubeClient kubernetes.Interface, monoVertexClient flowpkg.MonoVertexInterface, namespace, monoVertexName string, timeout time.Duration, size int) error

func WaitForMonoVertexRunning added in v1.3.0 

func WaitForMonoVertexRunning(ctx context.Context, monoVertexClient flowpkg.MonoVertexInterface, monoVertexName string, timeout time.Duration) error

func WaitForMvtxDaemonPodsRunning added in v1.5.0 

func WaitForMvtxDaemonPodsRunning(kubeClient kubernetes.Interface, namespace, mvtx string, timeout time.Duration) error

func WaitForPipelineRunning 

func WaitForPipelineRunning(ctx context.Context, pipelineClient flowpkg.PipelineInterface, pipelineName string, timeout time.Duration) error

func WaitForServingPipelineRunning added in v1.5.0 

func WaitForServingPipelineRunning(ctx context.Context, servingPipelineClient flowpkg.ServingPipelineInterface, servingPipelineName string, timeout time.Duration) error

func WaitForServingServerPodsRunning added in v1.7.0 

func WaitForServingServerPodsRunning(kubeClient kubernetes.Interface, namespace, servingPipelineName string, timeout time.Duration) error

func WaitForVertexPodRunning 

func WaitForVertexPodRunning(kubeClient kubernetes.Interface, vertexClient flowpkg.VertexInterface, namespace, pipelineName, vertexName string, timeout time.Duration) error

func WaitForVertexPodScalingTo added in v0.7.3 

func WaitForVertexPodScalingTo(kubeClient kubernetes.Interface, vertexClient flowpkg.VertexInterface, namespace, pipelineName, vertexName string, timeout time.Duration, size int) error

Types

type CheckFunc added in v0.7.0 

type CheckFunc func() bool

type E2ESuite 

type E2ESuite struct {
	suite.Suite
	// contains filtered or unexported fields
}

func (*E2ESuite) CheckError 

func (s *E2ESuite) CheckError(err error)

func (*E2ESuite) Given 

func (s *E2ESuite) Given() *Given

func (*E2ESuite) SetupSuite 

func (s *E2ESuite) SetupSuite()

func (*E2ESuite) TearDownSuite 

func (s *E2ESuite) TearDownSuite()

type Expect 

type Expect struct {
	// contains filtered or unexported fields
}

func (*Expect) DaemonPodLogContains added in v0.5.4 

func (t *Expect) DaemonPodLogContains(pipelineName, regex string, opts ...PodLogCheckOption) *Expect

func (*Expect) DaemonPodsRunning added in v0.5.4 

func (t *Expect) DaemonPodsRunning() *Expect

func (*Expect) ISBSvcDeleted 

func (t *Expect) ISBSvcDeleted(timeout time.Duration) *Expect

func (*Expect) MonoVertexNumaStable added in v1.9.0 

func (t *Expect) MonoVertexNumaStable(baseline PodRuntimeSnapshot) *Expect

MonoVertexNumaStable verifies that the exact pod captured in the baseline snapshot is still the one running the MonoVertex, and that its numa container has not restarted since the baseline was taken.

func (*Expect) MonoVertexPodLogContains added in v1.3.0 

func (t *Expect) MonoVertexPodLogContains(regex string, opts ...PodLogCheckOption) *Expect

func (*Expect) MonoVertexPodLogNotContains added in v1.5.0 

func (t *Expect) MonoVertexPodLogNotContains(regex string, opts ...PodLogCheckOption) *Expect

func (*Expect) MonoVertexPodRuntimeSnapshot added in v1.9.0 

func (t *Expect) MonoVertexPodRuntimeSnapshot() PodRuntimeSnapshot

func (*Expect) MonoVertexPodsRunning added in v1.3.0 

func (t *Expect) MonoVertexPodsRunning() *Expect

func (*Expect) MonoVertexSizeScaledTo added in v1.9.0 

func (t *Expect) MonoVertexSizeScaledTo(size int) *Expect

func (*Expect) MvtxDaemonPodsRunning added in v1.5.0 

func (t *Expect) MvtxDaemonPodsRunning() *Expect

func (*Expect) RedisSinkContains added in v1.3.0 

func (t *Expect) RedisSinkContains(hashKey string, targetStr string, opts ...SinkCheckOption) *Expect

RedisSinkContains checks if the target string is written to the redis sink hashKey is the hash key environment variable set by the sink targetStr is the target string to check

func (*Expect) RedisSinkListEquals added in v1.8.0 

func (t *Expect) RedisSinkListEquals(keyName string, expectedValues []string, opts ...SinkCheckOption) *Expect

RedisSinkListEquals checks that the Redis list at keyName contains exactly the expectedValues in order. This is used for ordered processing verification where the Rust redis-sink stores values as lists using RPUSH.

func (*Expect) RedisSinkNotContains added in v1.3.0 

func (t *Expect) RedisSinkNotContains(hashKey string, targetStr string, opts ...SinkCheckOption) *Expect

RedisSinkNotContains checks if the target string is not written to the redis sink hashKey is the hash key environment variable set by the sink targetStr is the target string to check

func (*Expect) ServingPodsRunning added in v1.7.0 

func (t *Expect) ServingPodsRunning() *Expect

func (*Expect) VertexNumaStable added in v1.9.0 

func (t *Expect) VertexNumaStable(baseline PodRuntimeSnapshot, vertexName string) *Expect

VertexNumaStable verifies that the exact pod captured in the baseline snapshot is still the one running the vertex, and that its numa container has not restarted since the baseline was taken.

func (*Expect) VertexPodLogContains 

func (t *Expect) VertexPodLogContains(vertexName, regex string, opts ...PodLogCheckOption) *Expect

func (*Expect) VertexPodLogNotContains added in v0.5.3 

func (t *Expect) VertexPodLogNotContains(vertexName, regex string, opts ...PodLogCheckOption) *Expect

func (*Expect) VertexPodRuntimeSnapshot added in v1.9.0 

func (t *Expect) VertexPodRuntimeSnapshot(vertexName string) PodRuntimeSnapshot

func (*Expect) VertexPodsRunning 

func (t *Expect) VertexPodsRunning() *Expect

func (*Expect) VertexSizeScaledTo added in v0.7.3 

func (t *Expect) VertexSizeScaledTo(v string, size int) *Expect

func (*Expect) When 

func (t *Expect) When() *When

type Given 

type Given struct {
	// contains filtered or unexported fields
}

func (*Given) ISBSvc 

func (g *Given) ISBSvc(text string) *Given

ISBSvc creates an ISBSvc based on the parameter, this may be:

1. A file name if it starts with "@" 2. Raw YAML.

func (*Given) MonoVertex added in v1.3.0 

func (g *Given) MonoVertex(text string) *Given

MonoVertex creates a MonoVertex based on the parameter, this may be:

1. A file name if it starts with "@" 2. Raw YAML.

func (*Given) Pipeline 

func (g *Given) Pipeline(text string) *Given

Pipeline creates a Pipeline based on the parameter, this may be:

1. A file name if it starts with "@" 2. Raw YAML.

func (*Given) ServingPipeline added in v1.5.0 

func (g *Given) ServingPipeline(text string) *Given

ServingPipeline creates a Pipeline based on the parameter, this may be:

1. A file name if it starts with "@" 2. Raw YAML.

func (*Given) When 

func (g *Given) When() *When

func (*Given) WithMonoVertex added in v1.3.0 

func (g *Given) WithMonoVertex(mv *dfv1.MonoVertex) *Given

func (*Given) WithPipeline 

func (g *Given) WithPipeline(p *dfv1.Pipeline) *Given

type HttpPostRequest added in v0.7.0 

type HttpPostRequest struct {
	Header map[string]string `json:"header"`
	Body   []byte            `json:"body"`
}

func NewHttpPostRequest added in v0.7.0 

func NewHttpPostRequest() HttpPostRequest

NewHttpPostRequest constructor for HttpPostRequest

func (HttpPostRequest) WithBody added in v0.7.0 

func (b HttpPostRequest) WithBody(body []byte) HttpPostRequest

func (HttpPostRequest) WithHeader added in v0.7.0 

func (b HttpPostRequest) WithHeader(k, v string) HttpPostRequest

type Json added in v1.8.0 

type Json string

type PodLogCheckOption added in v0.5.3 

type PodLogCheckOption func(*podLogCheckOptions)

func PodLogCheckOptionWithContainer added in v0.5.3 

func PodLogCheckOptionWithContainer(c string) PodLogCheckOption

func PodLogCheckOptionWithCount added in v0.5.3 

func PodLogCheckOptionWithCount(c int) PodLogCheckOption

func PodLogCheckOptionWithTimeout added in v0.5.3 

func PodLogCheckOptionWithTimeout(t time.Duration) PodLogCheckOption

type PodRuntimeSnapshot added in v1.9.0 

type PodRuntimeSnapshot struct {
	PodName          string
	UID              types.UID
	NumaRestartCount int32
}

PodRuntimeSnapshot captures pod identity and numa container restart state.

func WaitForMonoVertexPodRuntimeSnapshot added in v1.9.0 

func WaitForMonoVertexPodRuntimeSnapshot(kubeClient kubernetes.Interface, monoVertexClient flowpkg.MonoVertexInterface, namespace, monoVertexName string, timeout time.Duration) (PodRuntimeSnapshot, error)

WaitForMonoVertexPodRuntimeSnapshot waits until the MonoVertex pod is running and reports numa container status.

func WaitForVertexPodRuntimeSnapshot added in v1.9.0 

func WaitForVertexPodRuntimeSnapshot(kubeClient kubernetes.Interface, vertexClient flowpkg.VertexInterface, namespace, pipelineName, vertexName string, timeout time.Duration) (PodRuntimeSnapshot, error)

WaitForVertexPodRuntimeSnapshot waits until the vertex pod is running and reports numa container status. Unlike WaitForVertexPodRunning, it does not require every container in the pod to be ready.

type SinkCheckOption added in v0.7.0 

type SinkCheckOption func(*redisCheckOptions)

func SinkCheckWithContainCount added in v1.2.0 

func SinkCheckWithContainCount(c int) SinkCheckOption

SinkCheckWithContainCount updates the redisCheckOptions to specify count. The count is the expected number of matches for the check.

func SinkCheckWithTimeout added in v1.2.0 

func SinkCheckWithTimeout(t time.Duration) SinkCheckOption

SinkCheckWithTimeout updates the redisCheckOptions to specify timeout. The timeout specifies how long the redis check will wait for expected data to be ready in redis.

type When 

type When struct {
	// contains filtered or unexported fields
}

func (*When) And 

func (w *When) And(block func()) *When

func (*When) CreateISBSvc 

func (w *When) CreateISBSvc() *When

func (*When) CreateMonoVertexAndWait added in v1.3.0 

func (w *When) CreateMonoVertexAndWait() *When

func (*When) CreatePipelineAndWait 

func (w *When) CreatePipelineAndWait() *When

func (*When) CreateServingPipelineAndWait added in v1.5.0 

func (w *When) CreateServingPipelineAndWait() *When

func (*When) DaemonPodPortForward added in v0.5.4 

func (w *When) DaemonPodPortForward(pipelineName string, localPort, remotePort int) *When

func (*When) DeleteISBSvc 

func (w *When) DeleteISBSvc() *When

func (*When) DeleteMonoVertexAndWait added in v1.3.0 

func (w *When) DeleteMonoVertexAndWait() *When

func (*When) DeletePipelineAndWait 

func (w *When) DeletePipelineAndWait() *When

func (*When) DeleteServingPipelineAndWait added in v1.5.0 

func (w *When) DeleteServingPipelineAndWait() *When

func (*When) Exec 

func (w *When) Exec(name string, args []string, block func(t *testing.T, output string, err error)) *When

func (*When) Expect 

func (w *When) Expect() *Expect

func (*When) Given 

func (w *When) Given() *Given

func (*When) MonoVertexPodPortForward added in v1.5.0 

func (w *When) MonoVertexPodPortForward(localPort, remotePort int) *When

func (*When) MvtxDaemonPodPortForward added in v1.5.0 

func (w *When) MvtxDaemonPodPortForward(localPort, remotePort int) *When

func (*When) SendMessageTo added in v0.7.0 

func (w *When) SendMessageTo(pipelineName string, vertexName string, req HttpPostRequest) *When

SendMessageTo sends msg to one of the pods in http source vertex.

func (*When) SendMessageToMvTx added in v1.8.0 

func (w *When) SendMessageToMvTx(monoVertex string, req HttpPostRequest) *When

SendMessageToMvTx sends msg to http source vertex in a MonoVertex.

func (*When) StreamControllerLogs added in v1.3.1 

func (w *When) StreamControllerLogs() *When

func (*When) StreamISBLogs added in v1.3.1 

func (w *When) StreamISBLogs() *When

func (*When) StreamMonoVertexPodLogs added in v1.7.0 

func (w *When) StreamMonoVertexPodLogs(containerName string) *When

func (*When) StreamServingPodLogs added in v1.7.0 

func (w *When) StreamServingPodLogs(containerName string) *When

func (*When) StreamServingVertexPodLogs added in v1.7.0 

func (w *When) StreamServingVertexPodLogs(vertexName, containerName string) *When

func (*When) StreamVertexPodLogs added in v1.3.1 

func (w *When) StreamVertexPodLogs(vertexName, containerName string) *When

func (*When) TerminateAllPodLogs added in v1.2.0 

func (w *When) TerminateAllPodLogs() *When

func (*When) TerminateAllPodPortForwards 

func (w *When) TerminateAllPodPortForwards() *When

func (*When) UXServerPodPortForward added in v1.1.0 

func (w *When) UXServerPodPortForward(localPort, remotePort int) *When

func (*When) VertexPodPortForward 

func (w *When) VertexPodPortForward(vertexName string, localPort, remotePort int) *When

func (*When) Wait 

func (w *When) Wait(timeout time.Duration) *When

func (*When) WaitForISBSvcReady 

func (w *When) WaitForISBSvcReady() *When

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.