chore: testing-5

Author: saksham-datazip

drivers/kafka/internal/kafka_test.go

@@ -19,14 +19,14 @@ func TestKafkaIntegration(t *testing.T) {
 			name: "JSON-Format",
 			cfg: &testutils.IntegrationTest{
 				TestConfig:                      testutils.GetTestConfig(string(constants.Kafka),"Json"),
-				Namespace:                        "json-topic",
+				Namespace:                        "topics",
 				ExpectedData:                     ExpectedKafkaJSONData,
 				ExpectedUpdatedData:              ExpectedKafkaUpdatedJSONData,
 				DestinationDataTypeSchema:        KafkaToDestinationJSONSchema,
 				UpdatedDestinationDataTypeSchema: EvolvedKafkaToDestinationJSONSchema,
 				DefaultCDCColumnsSchema:          ExpectedKafkaDefaultCDCColumnsSchema,
 				ExecuteQuery:                     ExecuteQueryForJson,
-				DestinationDB:                    "kafka_topics",
+				DestinationDB:                    "kafka_json_topics",
 				PartitionRegex:                   "/{int_value,identity}",
 				FilterConfig: `{
 					"logical_operator": "And",
@@ -49,14 +49,14 @@ func TestKafkaIntegration(t *testing.T) {
 			name: "AVRO-Format",
 			cfg: &testutils.IntegrationTest{
 				TestConfig:                       testutils.GetTestConfig(string(constants.Kafka),"Avro"),
-				Namespace:                        "Avro-topic",
+				Namespace:                        "topics",
 				ExpectedData:                     ExpectedKafkaAvroData,
 				ExpectedUpdatedData:              ExpectedKafkaUpdatedAvroData,
 				DestinationDataTypeSchema:        KafkaToDestinationAvroSchema,
 				UpdatedDestinationDataTypeSchema: EvolvedKafkaToDestinationAvroSchema,
 				DefaultCDCColumnsSchema:          ExpectedKafkaDefaultCDCColumnsSchema,
 				ExecuteQuery:                     ExecuteQueryForAvro,
-				DestinationDB:                    "kafka_topics",
+				DestinationDB:                    "kafka_avro_topics",
 				PartitionRegex:                   "/{int_value,identity}",
 				FilterConfig: `{
 					"logical_operator": "And",

drivers/kafka/internal/kafka_test_utils.go

@@ -20,7 +20,8 @@ import (
 
 var (
 	// Message key and value for JSON
-	key            = []byte("test-key")
+	Jsonkey            = []byte("json-key")
+	Avrokey            = []byte("avro-key")
 	value          = []byte(`{"int_value": 100,"float_value": 99.99,"boolean_true": true,"boolean_false": false,"timestamp_value": "2026-03-22T14:30:00Z","string_value": "test_string"}`)
 	evolved_value  = []byte(`{"int_value": 100,"float_value": 99.99,"boolean_true": true,"boolean_false": false,"timestamp_value": "2026-03-22T14:30:00Z","string_value": "test_string", "id_int": 101}`)
 	filtervalue1   = []byte(`{"int_value": 99,"float_value": 99.99}`)
@@ -128,23 +129,23 @@ func ExecuteQueryForJson(ctx context.Context, t *testing.T, streams []string, op
 	case "add":
 		for partition := 0; partition < partitionCount; partition++ {
 			writeMessagesWithRetry(ctx, t, writer, kafka.Message{
-				Key:   key,
+				Key:   Jsonkey,
 				Value: value,
 			})
 		}
 		writeMessagesWithRetry(ctx, t, writer, kafka.Message{
-			Key:   key,
+			Key:   Jsonkey,
 			Value: filtervalue1,
 		})
 		writeMessagesWithRetry(ctx, t, writer, kafka.Message{
-			Key:   key,
+			Key:   Jsonkey,
 			Value: filtervalue2,
 		})
 		t.Logf("Added 7 messages to topic '%s' (one per partition and two for filters)", streams[0])
 	case "evolve-schema":
 		for partition := 0; partition < partitionCount; partition++ {
 			writeMessagesWithRetry(ctx, t, writer, kafka.Message{
-				Key:       key,
+				Key:       Jsonkey,
 				Value:     evolved_value,
 				Partition: partition,
 			})
@@ -203,8 +204,8 @@ func ExecuteQueryForAvro(ctx context.Context, t *testing.T, streams []string, op
 		confluentBaseMsg := encodeConfluentBinary(schemaID, binaryData)
 		confluentFilterMsg := encodeConfluentBinary(schemaID, binaryDataFilter)
 		err = writer.WriteMessages(ctx,
-			kafka.Message{Topic: streams[0], Key: []byte("avro-key"), Value: confluentBaseMsg},
-			kafka.Message{Topic: streams[0], Key: []byte("avro-key"), Value: confluentFilterMsg},
+			kafka.Message{Key: Avrokey, Value: confluentBaseMsg},
+			kafka.Message{Key: Avrokey, Value: confluentFilterMsg},
 		)
 		require.NoError(t, err)
 	case "evolve-schema":
@@ -216,8 +217,7 @@ func ExecuteQueryForAvro(ctx context.Context, t *testing.T, streams []string, op
 		require.NoError(t, err)
 		confluentMsg := encodeConfluentBinary(schemaID, binaryData)
 		err = writer.WriteMessages(ctx, kafka.Message{
-			Topic: streams[0],
-			Key:   []byte("avro-key"),
+			Key:   Avrokey,
 			Value: confluentMsg,
 		})
 		require.NoError(t, err)

utils/testutils/test_utils.go

@@ -75,6 +75,7 @@ type TestConfig struct {
 	HostTestDataPath       string
 	HostCatalogPath        string
 	HostTestCatalogPath    string
+	DataFormat             string
 }
 
 // history stores the RPS values and the last updated time for a given mode.
@@ -184,6 +185,7 @@ func GetTestConfig(driver string,dataFormat string) *TestConfig {
 	return &TestConfig{
 		Driver:                 driver,
 		HostRootPath:           rootPath,
+		DataFormat:             dataFormat,
 		HostTestDataPath:       fmt.Sprintf(hostTestDataPath, driver, ""),
 		HostTestCatalogPath:    fmt.Sprintf(hostTestDataPath, driver, "test_streams.json"),
 		HostCatalogPath:        fmt.Sprintf(hostTestDataPath, driver, "streams.json"),
@@ -813,6 +815,9 @@ func (cfg *IntegrationTest) TestIntegration(t *testing.T) {
 	t.Logf("Root Project directory: %s", cfg.TestConfig.HostRootPath)
 	t.Logf("Test data directory: %s", cfg.TestConfig.HostTestDataPath)
 	currentTestTable := fmt.Sprintf("%s_test_table_olake", cfg.TestConfig.Driver)
+	if cfg.TestConfig.DataFormat != "" {
+		currentTestTable = fmt.Sprintf("%s_%s_test_table_olake", cfg.TestConfig.Driver, cfg.TestConfig.DataFormat)
+	}
 
 	t.Run("Discover", func(t *testing.T) {
 		req := testcontainers.ContainerRequest{