fixtures

Documentation

Index

• Constants
• func CreateKafkaTopic() string
• func DeleteKafkaTopic(topic string) string
• func Exec(name string, args ...string) (string, error)
• func ExpectKafkaTopicCount(topic string, total int, timeout time.Duration)
• func GetKafkaCount(topic string, count int) int
• func HTTPExpect(t require.TestingT, baseURL string) *httpexpect.Expect
• func InvokeE2EAPI(format string, args ...interface{}) string
• func InvokeE2EAPIPOST(format string, body string, args ...interface{}) string
• func PodPortForward(config *rest.Config, namespace, podName string, localPort, remotePort int, ...) error
• func PodsLogContains(ctx context.Context, kubeClient kubernetes.Interface, namespace, regex string, ...) bool
• func PumpKafkaTopic(topic string, n int, opts ...interface{})
• func SendMessage(topic string, key string, message string)
• func ValidateMessage(topic string, key string, position int)
• func VertexPodLogContains(ctx context.Context, kubeClient kubernetes.Interface, ...) (bool, error)
• func WaitForISBSvcReady(ctx context.Context, isbSvcClient flowpkg.InterStepBufferServiceInterface, ...) error
• func WaitForISBSvcStatefulSetReady(ctx context.Context, kubeClient kubernetes.Interface, ...) error
• func WaitForPipelineRunning(ctx context.Context, pipelineClient flowpkg.PipelineInterface, ...) error
• func WaitForVertexPodRunning(kubeClient kubernetes.Interface, namespace, pipelineName, vertexName string, ...) error
• type E2ESuite
  • func (s *E2ESuite) CheckError(err error)
  • func (s *E2ESuite) Given() *Given
  • func (s *E2ESuite) SetupSuite()
  • func (s *E2ESuite) StartPortForward(podName string, port int) (stopPortForward func())
  • func (s *E2ESuite) TearDownSuite()
• type Expect
  • func (t *Expect) ISBSvcDeleted(timeout time.Duration) *Expect
  • func (t *Expect) VertexPodLogContains(vertexName, containerName, regex string) *Expect
  • func (t *Expect) VertexPodsRunning() *Expect
  • func (t *Expect) When() *When
• type Given
  • func (g *Given) ISBSvc(text string) *Given
  • func (g *Given) Pipeline(text string) *Given
  • func (g *Given) When() *When
  • func (g *Given) WithPipeline(p *dfv1.Pipeline) *Given
• type When
  • func (w *When) And(block func()) *When
  • func (w *When) CreateISBSvc() *When
  • func (w *When) CreatePipelineAndWait() *When
  • func (w *When) DeleteISBSvc() *When
  • func (w *When) DeletePipelineAndWait() *When
  • func (w *When) Exec(name string, args []string, block func(t *testing.T, output string, err error)) *When
  • func (w *When) Expect() *Expect
  • func (w *When) Given() *Given
  • func (w *When) TerminateAllPodPortForwards() *When
  • func (w *When) VertexPodPortForward(vertexName string, localPort, remotePort int) *When
  • func (w *When) Wait(timeout time.Duration) *When
  • func (w *When) WaitForISBSvcReady() *When

Constants

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

Functions

func CreateKafkaTopic

func CreateKafkaTopic() string

func DeleteKafkaTopic

func DeleteKafkaTopic(topic string) string

func Exec

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

func ExpectKafkaTopicCount

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

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 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, containerName string, timeout time.Duration) bool

func PumpKafkaTopic

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

func SendMessage

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

func ValidateMessage

func ValidateMessage(topic string, key string, position int)

func VertexPodLogContains

func VertexPodLogContains(ctx context.Context, kubeClient kubernetes.Interface, namespace, pipelineName, vertexName, containerName, regex string, timeout time.Duration) (bool, 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 WaitForPipelineRunning

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

func WaitForVertexPodRunning

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

Types

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) StartPortForward

func (s *E2ESuite) StartPortForward(podName string, port int) (stopPortForward func())

func (*E2ESuite) TearDownSuite

func (s *E2ESuite) TearDownSuite()

type Expect

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

func (*Expect) ISBSvcDeleted

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

func (*Expect) VertexPodLogContains

func (t *Expect) VertexPodLogContains(vertexName, containerName, regex string) *Expect

func (*Expect) VertexPodsRunning

func (t *Expect) VertexPodsRunning() *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

creates an ISBSvc 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

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) WithPipeline

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

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) CreatePipelineAndWait

func (w *When) CreatePipelineAndWait() *When

func (*When) DeleteISBSvc

func (w *When) DeleteISBSvc() *When

func (*When) DeletePipelineAndWait

func (w *When) DeletePipelineAndWait() *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) TerminateAllPodPortForwards

func (w *When) TerminateAllPodPortForwards() *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