integration

func CreateGrpcPayload ¶

func CreateGrpcPayload(msg proto.Message) ([]byte, error)

helper function to simulate the gRPC payload framing [1 byte compression flag] [4 bytes message length] [message bytes...]

func ExtProcServerClient ¶

func ExtProcServerClient(
	ctx context.Context,
	t *testing.T,
	port int,
	logger logr.Logger,
) (extProcPb.ExternalProcessor_ProcessClient, *grpc.ClientConn)

ExtProcServerClient returns a ExternalProcessor_ProcessClient listen to localhost on given port.

func GRPCRequestProto ¶

func GRPCRequestProto(prompt, methodName string, stream bool) proto.Message

func GenerateGRPCRequest ¶

func GenerateGRPCRequest(logger logr.Logger, prompt, methodName string, stream bool, filterMetadata []string) *extProcPb.ProcessingRequest

func GenerateRequest ¶

func GenerateRequest(logger logr.Logger, prompt, model string, filterMetadata []string) *extProcPb.ProcessingRequest

GenerateRequest constructs a `ProcessingRequest` containing a JSON-formatted LLM payload. It accepts a filterMetadata slice to inject Envoy Dynamic Metadata (used for subset load balancing).

func GenerateRequestMetadata ¶

func GenerateRequestMetadata(filterMetadata []string) map[string]*structpb.Struct

GenerateRequestMetadata constructs the Envoy Dynamic Metadata structure. This is primarily used to inject "envoy.lb" subset keys for testing logic that depends on specific backend subsets.

func GenerateRequestWithStream ¶

func GenerateRequestWithStream(logger logr.Logger, prompt, model string, filterMetadata []string) *extProcPb.ProcessingRequest

func GenerateStreamedGRPCRequestSet ¶

func GenerateStreamedGRPCRequestSet(
	logger logr.Logger,
	prompt string,
	inferenceObjective string,
	filterMetadata []string,
	methodName string,
) []*extProcPb.ProcessingRequest

GenerateStreamedRequestSet creates a slice of requests simulating an Envoy stream: 1. A Headers frame with standard Inference Extension headers. 2. A Body frame with the gRPC payload.

func GenerateStreamedRequestSet ¶

func GenerateStreamedRequestSet(
	logger logr.Logger,
	prompt, model, targetModel string,
	filterMetadata []string,
) []*extProcPb.ProcessingRequest

GenerateStreamedRequestSet creates a slice of requests simulating an Envoy stream: 1. A Headers frame with standard Inference Extension headers. 2. A Body frame with the JSON payload.

func GetFreePort ¶

func GetFreePort() (int, error)

GetFreePort finds an available IPv4 TCP port on localhost. It works by asking the OS to allocate a port by listening on port 0, capturing the assigned address, and then immediately closing the listener.

Note: There is a theoretical race condition where another process grabs the port between the Close() call and the subsequent usage, but this is generally acceptable in hermetic test environments.

func NewImmediateErrorResponse ¶

func NewImmediateErrorResponse(code envoyTypePb.StatusCode, body string) []*extProcPb.ProcessingResponse

NewImmediateErrorResponse creates a response that immediately terminates the request with a specific HTTP status code and body. Use this for testing Load Shedding (503), Rate Limiting (429), or Bad Request (400) logic.

func NewRequestBufferedResponse ¶

func NewRequestBufferedResponse(
	destinationEndpoint string,
	rewrittenBody []byte,
	otherHeaders ...*envoyCorev3.HeaderValueOption,
) []*extProcPb.ProcessingResponse

NewRequestBufferedResponse creates a complete set of responses for the Request phase. It simulates the EPP deciding to:

1. Modify headers (e.g., set destination endpoint).
2. Replace the entire request body (e.g., rewriting the model name).

It returns two messages: one for the Header response and one for the Body response.

func NewResponseBufferedResponse ¶

func NewResponseBufferedResponse(
	rewrittenBody string,
	eos bool,
	headersToSet ...*envoyCorev3.HeaderValueOption,
) []*extProcPb.ProcessingResponse

NewResponseBufferedResponse creates a complete set of responses for the Response phase. It simulates the EPP modifying the upstream response before sending it to the client. It returns a Header mutation message followed by a Body replacement message.

func NewResponseHeaders ¶

func NewResponseHeaders(headersToSet ...*envoyCorev3.HeaderValueOption) *extProcPb.ProcessingResponse

NewResponseHeaders creates a single response message to modify response headers. Use this when testing header mutations without body changes, or as the first step in a streamed response test.

func NewResponseStreamChunk ¶

func NewResponseStreamChunk(body string, endOfStream bool) *extProcPb.ProcessingResponse

NewResponseStreamChunk creates a single gRPC message representing one chunk of a streaming response. Use this to verify that EPP correctly passes through chunks (e.g., SSE events) or injects specific chunks.

func ReqGRPCLLM ¶

func ReqGRPCLLM(logger logr.Logger, prompt, inferenceObjective, methodName string) []*extProcPb.ProcessingRequest

func ReqGRPCLLMWithStream ¶

func ReqGRPCLLMWithStream(logger logr.Logger, prompt, inferenceObjective, methodName string) []*extProcPb.ProcessingRequest

func ReqHeaderOnly ¶

func ReqHeaderOnly(headers map[string]string) []*extProcPb.ProcessingRequest

ReqHeaderOnly creates a request sequence consisting solely of headers, with no body. It sets `EndOfStream: true` on the headers frame.

Use this for testing non-inference traffic, such as GET requests, health checks, or requests that should bypass the inference processor logic.

func ReqLLM ¶

func ReqLLM(logger logr.Logger, prompt, model, targetModel string) []*extProcPb.ProcessingRequest

ReqLLM creates a sequence of gRPC messages representing a standard, streamed LLM inference request. It generates:

1. A RequestHeaders message containing standard inference headers (Objective, Model Rewrite, Request ID).
2. A RequestBody message containing the JSON payload with EndOfStream=true.

Use this for the majority of "Happy Path" EPP and BBR streaming tests.

func ReqLLMUnary ¶

func ReqLLMUnary(logger logr.Logger, prompt, model string) *extProcPb.ProcessingRequest

ReqLLMUnary creates a single `ProcessingRequest` containing a complete JSON body. This simulates a scenario where Envoy has buffered the request body before sending it to the external processor (unary mode).

Use this for tests where `streaming: false` or when testing legacy buffered behavior.

func ReqLLMWithStream ¶

func ReqLLMWithStream(logger logr.Logger, prompt, model, targetModel string) []*extProcPb.ProcessingRequest

func ReqRaw ¶

func ReqRaw(headers map[string]string, bodyChunks ...string) []*extProcPb.ProcessingRequest

ReqRaw creates a custom sequence of gRPC messages with specific headers and arbitrary body chunks. This is a lower-level helper useful for testing edge cases, such as:

• Invalid JSON bodies (to test error handling).
• Fragmentation (split bodies) to ensure the processor handles accumulation correctly.
• Protocol attacks (e.g., missing headers).

func SendRequest ¶

func SendRequest(
	t *testing.T,
	client extProcPb.ExternalProcessor_ProcessClient,
	req *extProcPb.ProcessingRequest,
) (*extProcPb.ProcessingResponse, error)

SendRequest is a helper for Unary (One-Shot) test scenarios. It sends a single request message and waits for exactly one response.

func StartExtProcServer ¶

func StartExtProcServer(
	ctx context.Context,
	t *testing.T,
	serverRunner func(context.Context) error,
	port int,
	logger logr.Logger,
) (extProcPb.ExternalProcessor_ProcessClient, *grpc.ClientConn)

StartExtProcServer handles the lifecycle of starting a gRPC server in the background and connecting to it. It guarantees that the server is listening on the specified port before returning.

serverRunner: A function that blocks until the server exits (e.g. Runnable.Start). port: The port the server is configured to listen on.

func StreamedRequest ¶

func StreamedRequest(
	t *testing.T,
	client extProcPb.ExternalProcessor_ProcessClient,
	requests []*extProcPb.ProcessingRequest,
	expectedResponses int,
) ([]*extProcPb.ProcessingResponse, error)

StreamedRequest is a helper for Full-Duplex Streaming test scenarios. It performs the following actions:

1. Sends all requests in the provided slice to the server.
2. Listens for responses on the stream until 'expectedResponses' count is reached.
3. Enforces a 10-second timeout to prevent deadlocks if the server hangs.
4. Handles io.EOF gracefully (server closed stream).