1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
|
package schema
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestEncodeDecodeVarint(t *testing.T) {
testCases := []struct {
name string
value uint64
}{
{"zero", 0},
{"small", 1},
{"medium", 127},
{"large", 128},
{"very_large", 16384},
{"max_uint32", 4294967295},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
// Encode the value
encoded := encodeVarint(tc.value)
require.NotEmpty(t, encoded)
// Decode it back
decoded, bytesRead := readVarint(encoded)
require.Equal(t, len(encoded), bytesRead, "Should consume all encoded bytes")
assert.Equal(t, tc.value, decoded, "Decoded value should match original")
})
}
}
func TestCreateConfluentEnvelopeWithProtobufIndexes(t *testing.T) {
testCases := []struct {
name string
format Format
schemaID uint32
indexes []int
payload []byte
}{
{
name: "avro_no_indexes",
format: FormatAvro,
schemaID: 123,
indexes: nil,
payload: []byte("avro payload"),
},
{
name: "protobuf_no_indexes",
format: FormatProtobuf,
schemaID: 456,
indexes: nil,
payload: []byte("protobuf payload"),
},
{
name: "protobuf_single_index",
format: FormatProtobuf,
schemaID: 789,
indexes: []int{1},
payload: []byte("protobuf with index"),
},
{
name: "protobuf_multiple_indexes",
format: FormatProtobuf,
schemaID: 101112,
indexes: []int{0, 1, 2, 3},
payload: []byte("protobuf with multiple indexes"),
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
// Create the envelope
envelope := CreateConfluentEnvelope(tc.format, tc.schemaID, tc.indexes, tc.payload)
// Verify basic structure
require.True(t, len(envelope) >= 5, "Envelope should be at least 5 bytes")
assert.Equal(t, byte(0x00), envelope[0], "Magic byte should be 0x00")
// Extract and verify schema ID
extractedSchemaID, ok := ExtractSchemaID(envelope)
require.True(t, ok, "Should be able to extract schema ID")
assert.Equal(t, tc.schemaID, extractedSchemaID, "Schema ID should match")
// Parse the envelope based on format
if tc.format == FormatProtobuf && len(tc.indexes) > 0 {
// Use Protobuf-specific parser with known index count
parsed, ok := ParseConfluentProtobufEnvelopeWithIndexCount(envelope, len(tc.indexes))
require.True(t, ok, "Should be able to parse Protobuf envelope")
assert.Equal(t, tc.format, parsed.Format)
assert.Equal(t, tc.schemaID, parsed.SchemaID)
assert.Equal(t, tc.indexes, parsed.Indexes, "Indexes should match")
assert.Equal(t, tc.payload, parsed.Payload, "Payload should match")
} else {
// Use generic parser
parsed, ok := ParseConfluentEnvelope(envelope)
require.True(t, ok, "Should be able to parse envelope")
assert.Equal(t, tc.schemaID, parsed.SchemaID)
if tc.format == FormatProtobuf && len(tc.indexes) == 0 {
// For Protobuf without indexes, payload should match
assert.Equal(t, tc.payload, parsed.Payload, "Payload should match")
} else if tc.format == FormatAvro {
// For Avro, payload should match (no indexes)
assert.Equal(t, tc.payload, parsed.Payload, "Payload should match")
}
}
})
}
}
func TestProtobufEnvelopeRoundTrip(t *testing.T) {
// Use more realistic index values (typically small numbers for message types)
originalIndexes := []int{0, 1, 2, 3}
originalPayload := []byte("test protobuf message data")
schemaID := uint32(12345)
// Create envelope
envelope := CreateConfluentEnvelope(FormatProtobuf, schemaID, originalIndexes, originalPayload)
// Parse it back with known index count
parsed, ok := ParseConfluentProtobufEnvelopeWithIndexCount(envelope, len(originalIndexes))
require.True(t, ok, "Should be able to parse created envelope")
// Verify all fields
assert.Equal(t, FormatProtobuf, parsed.Format)
assert.Equal(t, schemaID, parsed.SchemaID)
assert.Equal(t, originalIndexes, parsed.Indexes)
assert.Equal(t, originalPayload, parsed.Payload)
assert.Equal(t, envelope, parsed.OriginalBytes)
}
func TestVarintEdgeCases(t *testing.T) {
t.Run("empty_data", func(t *testing.T) {
value, bytesRead := readVarint([]byte{})
assert.Equal(t, uint64(0), value)
assert.Equal(t, 0, bytesRead)
})
t.Run("incomplete_varint", func(t *testing.T) {
// Create an incomplete varint (continuation bit set but no more bytes)
incompleteVarint := []byte{0x80} // Continuation bit set, but no more bytes
value, bytesRead := readVarint(incompleteVarint)
assert.Equal(t, uint64(0), value)
assert.Equal(t, 0, bytesRead)
})
t.Run("max_varint_length", func(t *testing.T) {
// Create a varint that's too long (more than 10 bytes)
tooLongVarint := make([]byte, 11)
for i := 0; i < 10; i++ {
tooLongVarint[i] = 0x80 // All continuation bits
}
tooLongVarint[10] = 0x01 // Final byte
value, bytesRead := readVarint(tooLongVarint)
assert.Equal(t, uint64(0), value)
assert.Equal(t, 0, bytesRead)
})
}
func TestProtobufEnvelopeValidation(t *testing.T) {
t.Run("valid_envelope", func(t *testing.T) {
indexes := []int{1, 2}
envelope := CreateConfluentEnvelope(FormatProtobuf, 123, indexes, []byte("payload"))
parsed, ok := ParseConfluentProtobufEnvelopeWithIndexCount(envelope, len(indexes))
require.True(t, ok)
err := parsed.Validate()
assert.NoError(t, err)
})
t.Run("zero_schema_id", func(t *testing.T) {
indexes := []int{1}
envelope := CreateConfluentEnvelope(FormatProtobuf, 0, indexes, []byte("payload"))
parsed, ok := ParseConfluentProtobufEnvelopeWithIndexCount(envelope, len(indexes))
require.True(t, ok)
err := parsed.Validate()
assert.Error(t, err)
assert.Contains(t, err.Error(), "invalid schema ID: 0")
})
t.Run("empty_payload", func(t *testing.T) {
indexes := []int{1}
envelope := CreateConfluentEnvelope(FormatProtobuf, 123, indexes, []byte{})
parsed, ok := ParseConfluentProtobufEnvelopeWithIndexCount(envelope, len(indexes))
require.True(t, ok)
err := parsed.Validate()
assert.Error(t, err)
assert.Contains(t, err.Error(), "empty payload")
})
}
|
