testutil

package
v0.0.0-...-c816e34 Latest Latest
Warning

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

 Go to latest Published: Jul 9, 2026 License: Apache-2.0 Imports: 41 Imported by: 0

Documentation

Overview

Package testutil — OTLP gRPC test receiver.

FakeOTLP is an in-process OTLP gRPC server for verifying telemetry export in tests. It accepts trace, metric, and log Export RPCs, records the received payloads, and exposes counts (and the raw payloads) for assertions.

Usage: 

r := testutil.NewFakeOTLP(t)
t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://"+r.Addr())
shutdown, _, _ := observability.InitProvider(ctx, "svc", observability.ProviderConfig{...})
... emit spans/metrics/logs ...
_ = shutdown(ctx)  // forces a final flush
assert.Equal(t, expected, r.SpanCount())

Always call shutdown before asserting counts — the OTel SDK batches exports and only drains on shutdown (or after the batch timeout).

NewFakeOTLPTLS is the TLS variant — it mints an ephemeral self-signed cert and exposes it as a PEM file via CertFile(t), which the test points the SDK at with OTEL_EXPORTER_OTLP_CERTIFICATE (the same custom-CA path a real operator uses for a private gateway). Each Export RPC's gRPC metadata is also captured (LastTraceHeaders / LastMetricHeaders / LastLogHeaders) for asserting auth-header propagation.

Index

Constants

View Source 
const TestJWTSecret = "test-secret-at-least-32-chars-long!"

TestJWTSecret is a shared HMAC secret for test JWT generation.

Variables

This section is empty.

Functions

func AssertBodyContains 

func AssertBodyContains(t *testing.T, rec *httptest.ResponseRecorder, expectedStatus int, expectedSubstring string)

AssertBodyContains checks that rec has the expected status code and that the response body contains the expected substring.

func AssertBodyEquals 

func AssertBodyEquals(t *testing.T, rec *httptest.ResponseRecorder, expectedStatus int, expectedBody string)

AssertBodyEquals checks that rec has the expected status code and that the response body equals the expected string.

func AssertJSONContains 

func AssertJSONContains(t *testing.T, rec *httptest.ResponseRecorder, expectedStatus int, expectedKeys map[string]any)

AssertJSONContains checks that rec has the expected status code and that the response body contains the expected key-value pairs.

func AssertJSONErrorResponse 

func AssertJSONErrorResponse(t *testing.T, w *httptest.ResponseRecorder)

assertJSONErrorResponse verifies that a recorded response is a well-formed JSON error body with the exact headers writeJSONError guarantees: Content-Type: application/json and X-Content-Type-Options: nosniff. Pinned strict so any handler that bypasses writeJSONError and emits a different Content-Type fails the assertion across every call site.

func AssertJSONResponse 

func AssertJSONResponse(t *testing.T, rec *httptest.ResponseRecorder, expectedStatus int, expected any)

AssertJSONResponse checks that rec has the expected status code and that the response body, decoded as JSON, matches expected (compared as Go values).

func MakeExpiredJWT 

func MakeExpiredJWT(t *testing.T, claims map[string]any) string

MakeExpiredJWT creates a signed HS256 JWT that expired 1 hour ago.

func MakeJWT 

func MakeJWT(t *testing.T, claims map[string]any) string

MakeJWT creates a signed HS256 JWT with the given claims merged with standard exp/iat.

func NewJetStream 

func NewJetStream(t *testing.T) jetstream.JetStream

NewJetStream starts an in-process NATS server with JetStream and returns a connected jetstream.JetStream handle. Cleanup is registered on t.

Lives in testutil so packages that need KV-backed tests (pipes.Store, policy.Store) get one shared embedded-server setup instead of each duplicating the natsserver / nats.Connect / jetstream.New dance.

func NewTestSchemaRegistry 

func NewTestSchemaRegistry(tables ...*discovery.TableSchema) *discovery.SchemaRegistry

NewTestSchemaRegistry creates a SchemaRegistry pre-loaded with the given table schemas. Does not require a real ClickHouse connection.

func NopLogger 

func NopLogger() *slog.Logger

NopLogger returns a *slog.Logger that discards all output. Use in tests to suppress noisy log output from embedded NATS, etc.

func RunRoleMatrix 

func RunRoleMatrix(t *testing.T, cases []RoleCase, invoke func(t *testing.T, tc RoleCase) *httptest.ResponseRecorder)

RunRoleMatrix runs each case as a parallel subtest, building and invoking the handler through invoke and asserting the response matches WantForbidden. A forbidden case must return 403 with the standard JSON error envelope (`application/json`, `nosniff`, an `error` field); a permitted case must return neither 403 nor 404. invoke owns the resource/request wiring (and any panic recovery for handlers that reach a nil backend on the allowed path).

Types

type CapturedRequest 

type CapturedRequest struct {
	Method string
	URL    string
	Header http.Header
	Body   []byte
}

CapturedRequest is a frozen snapshot of an outgoing HTTP request.

type FakeOTLP 

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

FakeOTLP is a single gRPC server that implements the trace, metric, and log Export RPCs and captures every received payload (and each request's gRPC metadata). Cleanup is registered on the *testing.T automatically.

func NewFakeOTLP 

func NewFakeOTLP(t *testing.T) *FakeOTLP

NewFakeOTLP binds the receiver to 127.0.0.1 on a random port and starts serving plaintext gRPC. The server is stopped automatically when the test ends.

func NewFakeOTLPTLS 

func NewFakeOTLPTLS(t *testing.T) *FakeOTLP

NewFakeOTLPTLS is the TLS variant: mints an ephemeral self-signed cert (SAN 127.0.0.1) and exposes it as a PEM file via CertFile(t).

func (*FakeOTLP) Addr 

func (r *FakeOTLP) Addr() string

Addr returns the listener address (e.g. "127.0.0.1:42891") suitable for OTEL_EXPORTER_OTLP_ENDPOINT (prefix with http:// for plaintext or https:// for the TLS variant), typically set via t.Setenv before InitProvider.

func (*FakeOTLP) CertFile 

func (r *FakeOTLP) CertFile(t *testing.T) string

CertFile writes the receiver's self-signed certificate to a temp PEM file and returns its path, suitable for OTEL_EXPORTER_OTLP_CERTIFICATE. Only valid for a server constructed via NewFakeOTLPTLS.

func (*FakeOTLP) LastLogHeaders 

func (r *FakeOTLP) LastLogHeaders() metadata.MD

LastLogHeaders returns a copy of the gRPC metadata captured from the most recent log Export RPC, or nil if none.

func (*FakeOTLP) LastMetricHeaders 

func (r *FakeOTLP) LastMetricHeaders() metadata.MD

LastMetricHeaders returns a copy of the gRPC metadata captured from the most recent metric Export RPC, or nil if none.

func (*FakeOTLP) LastTraceHeaders 

func (r *FakeOTLP) LastTraceHeaders() metadata.MD

LastTraceHeaders returns a copy of the gRPC metadata captured from the most recent trace Export RPC, or nil if none.

func (*FakeOTLP) LogCount 

func (r *FakeOTLP) LogCount() int

LogCount returns the total number of log records received across all RPCs.

func (*FakeOTLP) LogCountAtLevel 

func (r *FakeOTLP) LogCountAtLevel(minSeverity int32) int

LogCountAtLevel returns the count of log records whose severity number is >= minSeverity. OTel severity numbers: DEBUG=5, INFO=9, WARN=13, ERROR=17.

func (*FakeOTLP) MetricCount 

func (r *FakeOTLP) MetricCount() int

MetricCount returns the total number of metric records received, flattened across resource and scope groupings.

func (*FakeOTLP) Reset 

func (r *FakeOTLP) Reset()

Reset clears all captured payloads. Useful between test phases.

func (*FakeOTLP) SpanCount 

func (r *FakeOTLP) SpanCount() int

SpanCount returns the total number of spans received across all RPCs. Spans are flattened across resource and scope groupings.

type MockCache 

type MockCache struct {
	cache.Cache   // Embed to satisfy remaining interface methods silently
	InvNamespaces []cache.Namespace
	InvErr        error
	// contains filtered or unexported fields
}

MockCache implements cache.Cache and records invalidated namespaces for testing.

func (*MockCache) GetNamespaces 

func (m *MockCache) GetNamespaces() []cache.Namespace

func (*MockCache) Invalidate 

func (m *MockCache) Invalidate(_ context.Context, namespaces []cache.Namespace) (uint64, error)

type MockConsumer 

type MockConsumer struct {
	jetstream.Consumer

	InfoVal *jetstream.ConsumerInfo
	InfoErr error
}

MockConsumer embeds jetstream.Consumer so unused methods panic.

func (*MockConsumer) Info 

func (m *MockConsumer) Info(_ context.Context) (*jetstream.ConsumerInfo, error)

type MockDeduplicator 

type MockDeduplicator struct {
	Err error // if set, CheckAndMark returns this error
	// contains filtered or unexported fields
}

MockDeduplicator implements dedupe.Deduplicator for testing.

func NewMockDeduplicator 

func NewMockDeduplicator() *MockDeduplicator

func (*MockDeduplicator) CheckAndMark 

func (m *MockDeduplicator) CheckAndMark(_ context.Context, eventID string) (bool, error)

func (*MockDeduplicator) Close 

func (m *MockDeduplicator) Close() error

func (*MockDeduplicator) Stats 

func (m *MockDeduplicator) Stats() map[string]int64

type MockJetStream 

type MockJetStream struct {
	jetstream.JetStream

	StreamFn     func(ctx context.Context, name string) (jetstream.Stream, error)
	ConsumerFn   func(ctx context.Context, stream, consumer string) (jetstream.Consumer, error)
	PublishMsgFn func(ctx context.Context, msg *nats.Msg, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)

	PubErr error // if set, PublishMsg returns this error (and skips recording)
	// contains filtered or unexported fields
}

MockJetStream embeds jetstream.JetStream so unused methods panic. Override only what your test touches: StreamFn / ConsumerFn for sweeper-style code, PublishMsg recorder (Published + PubErr) for DLQ-style code.

func (*MockJetStream) Consumer 

func (m *MockJetStream) Consumer(ctx context.Context, stream, consumer string) (jetstream.Consumer, error)

func (*MockJetStream) PublishMsg 

func (m *MockJetStream) PublishMsg(ctx context.Context, msg *nats.Msg, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)

func (*MockJetStream) Published 

func (m *MockJetStream) Published() []*nats.Msg

Published returns a snapshot of PublishMsg calls recorded so far.

func (*MockJetStream) Stream 

func (m *MockJetStream) Stream(ctx context.Context, name string) (jetstream.Stream, error)

type MockJetStreamMsg 

type MockJetStreamMsg struct {
	MsgData    []byte
	MsgSubject string
	MsgHeaders nats.Header

	// Configurable errors returned from the matching ack-family methods.
	AckErr       error
	NakErr       error
	DoubleAckErr error

	// State flags — atomic because production calls these from goroutines.
	Acked       atomic.Bool
	Naked       atomic.Bool
	DoubleAcked atomic.Bool
}

MockJetStreamMsg implements jetstream.Msg for unit tests. Only the methods production code actually exercises (Data/Subject/Headers/Ack/Nak/DoubleAck) are overridden — calls to anything else panic, which is the loud-fail signal we want from a unit test mock.

func (*MockJetStreamMsg) Ack 

func (m *MockJetStreamMsg) Ack() error

func (*MockJetStreamMsg) Data 

func (m *MockJetStreamMsg) Data() []byte

func (*MockJetStreamMsg) DoubleAck 

func (m *MockJetStreamMsg) DoubleAck(_ context.Context) error

func (*MockJetStreamMsg) Headers 

func (m *MockJetStreamMsg) Headers() nats.Header

func (*MockJetStreamMsg) InProgress 

func (m *MockJetStreamMsg) InProgress() error

func (*MockJetStreamMsg) Metadata 

func (m *MockJetStreamMsg) Metadata() (*jetstream.MsgMetadata, error)

func (*MockJetStreamMsg) Nak 

func (m *MockJetStreamMsg) Nak() error

func (*MockJetStreamMsg) NakWithDelay 

func (m *MockJetStreamMsg) NakWithDelay(_ time.Duration) error

The remaining jetstream.Msg methods are intentionally unimplemented — they panic to signal "the production code touched a path this test wasn't expecting." Implement on demand.

func (*MockJetStreamMsg) Reply 

func (m *MockJetStreamMsg) Reply() string

func (*MockJetStreamMsg) Subject 

func (m *MockJetStreamMsg) Subject() string

func (*MockJetStreamMsg) Term 

func (m *MockJetStreamMsg) Term() error

func (*MockJetStreamMsg) TermWithReason 

func (m *MockJetStreamMsg) TermWithReason(string) error

type MockPublisher 

type MockPublisher struct {
	Messages []PublishedMessage
	Err      error // if set, Publish returns this error
	// contains filtered or unexported fields
}

MockPublisher records all published messages for test assertions.

func (*MockPublisher) Close 

func (m *MockPublisher) Close() error

func (*MockPublisher) LastMessage 

func (m *MockPublisher) LastMessage() *PublishedMessage

LastMessage returns the most recently published message, or nil.

func (*MockPublisher) Publish 

func (m *MockPublisher) Publish(_ context.Context, subject string, data []byte, opts ...mq.PublishOpt) error

type MockRoundTripper 

type MockRoundTripper struct {
	Fn func(req *http.Request) (*http.Response, error)
	// contains filtered or unexported fields
}

MockRoundTripper intercepts HTTP requests for the components that take a custom *http.Client (today: the ClickHouse inserter). Fn lets a test hand-roll a response; if Fn is nil, returns 200 OK.

Every request is captured (with body slurped + replaced so the handler can still read it) — useful for asserting request shape after the fact.

func (*MockRoundTripper) Captured 

func (m *MockRoundTripper) Captured() []CapturedRequest

func (*MockRoundTripper) Hits 

func (m *MockRoundTripper) Hits() int32

func (*MockRoundTripper) RoundTrip 

func (m *MockRoundTripper) RoundTrip(req *http.Request) (*http.Response, error)

type MockStream 

type MockStream struct {
	jetstream.Stream

	InfoVal *jetstream.StreamInfo
	InfoErr error

	Msgs     map[uint64]*jetstream.RawStreamMsg
	GetMsgFn func(ctx context.Context, seq uint64, opts ...jetstream.GetMsgOpt) (*jetstream.RawStreamMsg, error)

	Purged   []jetstream.StreamPurgeOpt
	PurgeErr error
	// contains filtered or unexported fields
}

MockStream embeds jetstream.Stream so unused methods panic. Provide Msgs for static GetMsg lookup, or GetMsgFn for richer behaviour. Purge calls are recorded into Purged.

func (*MockStream) GetMsg 

func (m *MockStream) GetMsg(ctx context.Context, seq uint64, opts ...jetstream.GetMsgOpt) (*jetstream.RawStreamMsg, error)

func (*MockStream) Info 

func (m *MockStream) Info(_ context.Context, _ ...jetstream.StreamInfoOpt) (*jetstream.StreamInfo, error)

func (*MockStream) Purge 

func (m *MockStream) Purge(_ context.Context, opts ...jetstream.StreamPurgeOpt) error

type MockSubscriber 

type MockSubscriber struct {
	Err     error
	Handler func(msg *mq.Message) error
}

MockSubscriber implements mq.Subscriber for testing.

func (*MockSubscriber) Close 

func (m *MockSubscriber) Close() error

func (*MockSubscriber) Subscribe 

func (m *MockSubscriber) Subscribe(_ context.Context, _, _ string, handler func(msg *mq.Message) error) error

type PublishedMessage 

type PublishedMessage struct {
	Subject string
	Data    []byte
}

PublishedMessage records a single publish call.

type RoleCase 

type RoleCase struct {
	Name         string
	AllowedRoles []string
	Role         string // role placed in context when SetRole is true
	SetRole      bool   // false → no role in context at all (e.g. no token / invalid token)
	// WantForbidden is true when the request must be rejected with 403.
	WantForbidden bool
}

RoleCase is one row of a per-resource role-authorization matrix: a resource restricted to AllowedRoles, observed under a given request role.

func StandardRoleMatrix 

func StandardRoleMatrix() []RoleCase

StandardRoleMatrix is the canonical set of (AllowedRoles, observed-role) cases every AllowedRoles-style gate must cover. Authorization is allowlist membership: the observed role must appear in AllowedRoles, the admin role (policy.AdminRole — "admin" by default, which is what these cases assume) always passes, and an empty/absent role matches nothing — so a resource with no AllowedRoles authorizes nobody but admin. The empty/absent-role rows are the ones that fail open when a gate only enforces its allowlist for a non-empty role — the class of bug behind #159. Keeping them in one shared place means a handler that takes AllowedRoles without running this matrix looks obviously under-tested in review.

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.