package krabkatest
import "github.com/krabka-io/krabka-streams-go/krabkatest"
Package krabkatest bundles test helpers for the schema and columnar packages: an in-memory schema registry server and an in-process driver for columnar topologies. Neither needs a broker or a real registry.
Index
Types
type ColumnarTestDriver source
type ColumnarTestDriver struct { // contains filtered or unexported fields }
ColumnarTestDriver runs a built columnar topology in-process and queues the produced records per topic. It is not safe for concurrent use.
PipeInput and PipeBatch differ in an important way for BlobCodec topologies: each PipeInput call is its own batch, so per-batch operators such as GroupBy see one record at a time. Use PipeBatch when the test is about batch behavior.
The driver holds no Arrow memory of its own, since every batch is created
and released inside RunBatch.
Output:Example
package main
import (
"fmt"
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/krabka-io/krabka-streams-go/columnar"
"github.com/krabka-io/krabka-streams-go/krabkatest"
)
type echoSerde struct{}
func (echoSerde) Serialize(topic string, value string) ([]byte, error) {
return []byte(value), nil
}
func (echoSerde) Deserialize(topic string, data []byte) (string, error) {
return string(data), nil
}
func main() {
mem := memory.NewGoAllocator()
codec := columnar.NewRowCodec[string](echoSerde{}, columnar.NewJSONRowBridge[string](), mem)
topology := columnar.NewTopology(mem)
source, err := topology.AddSource("source", []string{"in"}, codec)
if err != nil {
panic(err)
}
if _, err := topology.AddSink("sink", "out", codec, source); err != nil {
panic(err)
}
built, err := topology.Build()
if err != nil {
panic(err)
}
driver := krabkatest.NewColumnarTestDriver(built)
if err := driver.PipeInput("in", 0, []byte("a"), []byte("first"), 10); err != nil {
panic(err)
}
record, err := driver.ReadOutput("out")
if err != nil {
panic(err)
}
fmt.Println(string(record.Key), string(record.Value), record.Timestamp)
}
a first 10
func NewColumnarTestDriver source
func NewColumnarTestDriver(topology *columnar.BuiltTopology) *ColumnarTestDriver
NewColumnarTestDriver creates a driver over a built topology.
func (*ColumnarTestDriver) DrainOutput source
func (d *ColumnarTestDriver) DrainOutput(topic string) []columnar.ProduceRecord
DrainOutput removes and returns everything queued for a topic.
func (*ColumnarTestDriver) FailNext source
func (d *ColumnarTestDriver) FailNext(fault error)
FailNext returns one deterministic fault before the next batch evaluation.
func (*ColumnarTestDriver) IsOutputEmpty source
func (d *ColumnarTestDriver) IsOutputEmpty(topic string) bool
IsOutputEmpty reports whether a sink topic's queue is empty.
func (*ColumnarTestDriver) OutputSize source
func (d *ColumnarTestDriver) OutputSize(topic string) int
OutputSize returns the queue depth of a sink topic.
func (*ColumnarTestDriver) PipeBatch source
func (d *ColumnarTestDriver) PipeBatch(topic string, records []columnar.ConsumedRecord) error
PipeBatch runs a whole record list as one batch. Offsets are whatever the records carry.
func (*ColumnarTestDriver) PipeInput source
func (d *ColumnarTestDriver) PipeInput(topic string, partition int, key, value []byte, timestamp int64, headers ...columnar.RecordHeader) error
PipeInput runs one record as a single-record batch. Offsets start at 0 per topic-partition and increment.
func (*ColumnarTestDriver) ReadOutput source
func (d *ColumnarTestDriver) ReadOutput(topic string) (columnar.ProduceRecord, error)
ReadOutput removes and returns the oldest record queued for a topic.
type SchemaRegistryStub source
type SchemaRegistryStub struct { // contains filtered or unexported fields }
SchemaRegistryStub is a real HTTP server implementing the registry endpoints the client uses, backed by in-memory state. It binds to 127.0.0.1 on an ephemeral port.
Schema identity is the triple (schema, schemaType, messageType), so an Avro schema and a JSON schema with the same text receive different IDs. IDs are assigned from 1; identical schemas reuse an ID and the subject's version list grows. Request handling is synchronized, so counts are stable to read from the test.
func NewSchemaRegistryStub source
func NewSchemaRegistryStub() (*SchemaRegistryStub, error)
NewSchemaRegistryStub starts the server.
func (*SchemaRegistryStub) Close source
func (s *SchemaRegistryStub) Close() error
Close stops the server.
func (*SchemaRegistryStub) RequestCount source
func (s *SchemaRegistryStub) RequestCount(method, path string) int
RequestCount counts requests by method and raw path, which is how you assert that prewarming resolved a subject exactly once.
func (*SchemaRegistryStub) URL source
func (s *SchemaRegistryStub) URL() string
URL returns the base URL of the running server.