featurebase/idk/kafka_sasl/source_test.go

//go:build kafka_sasl
// +build kafka_sasl

package kafka_sasl

import (
	"fmt"
	"math/rand"
	"reflect"
	"strconv"
	"testing"
	"time"

	"github.com/featurebasedb/featurebase/v3/idk"
	"github.com/featurebasedb/featurebase/v3/idk/common"
)

func configureSourceTestFlags(source *Source) {
	source.ConfluentCommand = configMapSource
}

var tests = []struct {
	name   string
	header string
	data   []string
	exp    [][]interface{}
}{
	{
		name:   "Flat",
		header: "../kafka_static/testdata/Flat.json",
		data:   []string{`{"first": "hello"}`},
		exp:    [][]interface{}{{"hello"}},
	},
	{
		name:   "Tree",
		header: "../kafka_static/testdata/Tree.json",
		data:   []string{`{"protocol": "TCP", "port": 22, "client": "127.0.0.1", "server": "127.0.0.1", "metadata": {"suspicious": true, "region": "North America"}}`},
		exp: [][]interface{}{
			{"TCP", 22.0, "127.0.0.1", "127.0.0.1", true, "North America"},
		},
	},
}

// TestKafkaStaticSourceIntegration uses a real Kafka, and a static schema.
func TestSaslKafkaStaticSourceIntegration(t *testing.T) {
	t.Parallel()

	if testing.Short() {
		t.Skip()
	}

	rand.Seed(time.Now().UnixNano())
	a := rand.Int()

	key := fmt.Sprintf("%d", a)
	for _, test := range tests {
		test := test
		t.Run(test.name, func(t *testing.T) {
			// t.Parallel()

			topic := "sasl_testKafkaStaticSourceIntegration" + test.name + strconv.Itoa(a)

			// create topic before openning the Source
			adminClient := tCreateConfluentAdmin(t)
			tCreateConfluentTopic(t, adminClient, topic)
			defer adminClient.Close()

			src := NewSource()
			defer src.Close()
			configureSourceTestFlags(src)
			src.Topics = []string{topic}
			src.Group = "group0"
			src.Header = test.header
			cfg, err := common.SetupConfluent(&configMapSource)
			if err != nil {
				t.Fatal(err)
			}
			src.ConfigMap = cfg

			err = src.Open()
			if err != nil {
				t.Fatalf("opening source: %v", err)
			}

			producer := tCreateConfluentProducer(t)

			for j, record := range test.data {
				tWriteConfluentMessage(t, producer, topic, key, record)

				rec, err := src.Record()
				if err != nil {
					t.Fatalf("unexpected error getting record: %v", err)
				}
				if rec == nil {
					t.Fatalf("should have a record")
				}

				data := rec.Data()
				if !reflect.DeepEqual(data, test.exp[j]) {
					t.Errorf("data mismatch exp/got:\n%+v\n%+v", test.exp[j], data)
					if len(data) != len(test.exp[j]) {
						t.Fatalf("mismatched lengths exp/got %d/%d", len(test.exp[j]), len(data))
					}
					for k := range test.exp[j] {
						if !reflect.DeepEqual(test.exp[j][k], data[k]) {
							t.Errorf("Mismatch at %d, exp/got\n%v of %[2]T\n%v of %[3]T", k, test.exp[j][k], data[k])
						}
					}
				}
			}
			defer producer.Close()
		})
	}

}

func TestSaslPathTable(t *testing.T) {
	t.Parallel()

	cases := []struct {
		name               string
		table              idk.PathTable
		allowMissingFields bool
		in                 interface{}
		out                []interface{}
		err                string
	}{
		{
			name: "nil",
			out:  []interface{}{},
		},
		{
			name: "Root",
			table: idk.PathTable{
				{},
			},
			in: 1,
			out: []interface{}{
				1,
			},
		},
		{
			name: "Flat",
			table: idk.PathTable{
				{"x"},
				{"y"},
			},
			in: map[string]interface{}{
				"x": 1,
				"y": 2,
			},
			out: []interface{}{
				1,
				2,
			},
		},
		{
			name: "Missing",
			table: idk.PathTable{
				{"x"},
				{"y"},
			},
			in: map[string]interface{}{
				"x": 1,
			},
			err: "failed to lookup path [y]: element \"y\" at [] is missing",
		},
		{
			name:               "Missing_Allowed",
			allowMissingFields: true,
			table: idk.PathTable{
				{"x"},
				{"y"},
			},
			in: map[string]interface{}{
				"x": 1,
			},
			out: []interface{}{
				1,
				nil,
			},
		},
		{
			name:               "Check Nil Delete",
			allowMissingFields: true,
			table: idk.PathTable{
				{"x"},
				{"y"},
			},
			in: map[string]interface{}{
				"x": nil,
			},
			out: []interface{}{
				idk.DELETE_SENTINEL,
				nil,
			},
		},
		{
			name: "Tree",
			table: idk.PathTable{
				{"x", "y"},
				{"y", "z", "a"},
			},
			in: map[string]interface{}{
				"x": map[string]interface{}{
					"y": []interface{}{1, 2, 3},
					"z": map[string]interface{}{
						"h": "?",
					},
				},
				"y": map[string]interface{}{
					"z": map[string]interface{}{
						"a": "b",
					},
				},
			},
			out: []interface{}{
				[]interface{}{1, 2, 3},
				"b",
			},
		},
		{
			name: "TreeMissing",
			table: idk.PathTable{
				{"x", "y"},
				{"y", "z", "b"},
			},
			in: map[string]interface{}{
				"x": map[string]interface{}{
					"y": []interface{}{1, 2, 3},
					"z": map[string]interface{}{
						"h": "?",
					},
				},
				"y": map[string]interface{}{
					"z": map[string]interface{}{
						"a": "b",
					},
				},
			},
			err: "failed to lookup path [y z b]: element \"b\" at [y z] is missing",
		},
		{
			name: "TreeMissing_Allowed",
			table: idk.PathTable{
				{"x", "y"},
				{"a", "b", "c"},
			},
			allowMissingFields: true,
			in: map[string]interface{}{
				"x": map[string]interface{}{
					"y": []interface{}{1, 2, 3},
					"z": map[string]interface{}{
						"h": "?",
					},
				},
				"a": map[string]interface{}{
					"c": "words",
				},
			},
			out: []interface{}{
				[]interface{}{1, 2, 3},
				nil,
			},
		},
		{
			name: "WrongNodeType",
			table: idk.PathTable{
				{"x", "y"},
			},
			in: map[string]interface{}{
				"x": []interface{}{"z"},
			},
			err: "failed to lookup path [x y]: element at [x] is not an object",
		},
	}

	for _, c := range cases {
		c := c
		t.Run(c.name, func(t *testing.T) {
			// t.Parallel()
			out, err := c.table.Lookup(c.in, c.allowMissingFields)
			if err != nil {
				switch {
				case c.err == "":
					t.Errorf("unexpected error: %v", err)
				case c.err != err.Error():
					t.Errorf("expected error %q but got %q", c.err, err.Error())
				}
			} else {
				switch {
				case c.err != "":
					t.Error("unexpected success")
				case !reflect.DeepEqual(out, c.out):
					t.Errorf("expected %v but got %v", c.out, out)
				}
			}
		})
	}
}