Refactor base stages

internal/assertions/apiversions_response_assertion.go

@@ -3,7 +3,7 @@ package assertions
 import (
 	"fmt"
 
-	kafkaapi "github.com/codecrafters-io/kafka-tester/protocol/api"
+	"github.com/codecrafters-io/kafka-tester/protocol/kafkaapi"
 	"github.com/codecrafters-io/tester-utils/logger"
 )
 

internal/assertions/response_header_assertion.go

@@ -3,7 +3,7 @@ package assertions
 import (
 	"fmt"
 
-	"github.com/codecrafters-io/kafka-tester/protocol/api/headers"
+	"github.com/codecrafters-io/kafka-tester/protocol/kafkaapi/headers"
 	"github.com/codecrafters-io/tester-utils/logger"
 )
 

internal/assertions_legacy/apiversions_response_assertion.go

@@ -0,0 +1,99 @@
+package assertions_legacy
+
+import (
+	"fmt"
+
+	"github.com/codecrafters-io/kafka-tester/protocol/kafkaapi_legacy"
+	"github.com/codecrafters-io/tester-utils/logger"
+)
+
+var apiKeyNames = map[int16]string{
+	1:  "FETCH",
+	18: "API_VERSIONS",
+	75: "DESCRIBE_TOPIC_PARTITIONS",
+}
+
+var errorCodes = map[int]string{
+	0:   "NO_ERROR",
+	3:   "UNKNOWN_TOPIC_OR_PARTITION",
+	35:  "UNSUPPORTED_VERSION",
+	100: "UNKNOWN_TOPIC_ID",
+}
+
+type ApiVersionsResponseAssertion struct {
+	ActualValue   kafkaapi_legacy.ApiVersionsResponse
+	ExpectedValue kafkaapi_legacy.ApiVersionsResponse
+}
+
+func NewApiVersionsResponseAssertion(actualValue kafkaapi_legacy.ApiVersionsResponse, expectedValue kafkaapi_legacy.ApiVersionsResponse) *ApiVersionsResponseAssertion {
+	return &ApiVersionsResponseAssertion{
+		ActualValue:   actualValue,
+		ExpectedValue: expectedValue,
+	}
+}
+
+func (a *ApiVersionsResponseAssertion) assertBody(logger *logger.Logger) error {
+	expectedErrorCodeName, ok := errorCodes[int(a.ExpectedValue.Body.ErrorCode)]
+	if !ok {
+		panic(fmt.Sprintf("CodeCrafters Internal Error: Expected %d to be in errorCodes map", a.ExpectedValue.Body.ErrorCode))
+	}
+	if a.ActualValue.Body.ErrorCode != a.ExpectedValue.Body.ErrorCode {
+		return fmt.Errorf("Expected ErrorCode to be %d (%s), got %d", a.ExpectedValue.Body.ErrorCode, expectedErrorCodeName, a.ActualValue.Body.ErrorCode)
+	}
+	logger.Successf("✓ ErrorCode: %d (%s)", a.ActualValue.Body.ErrorCode, expectedErrorCodeName)
+
+	if err := a.assertAPIKeysArray(logger); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (a *ApiVersionsResponseAssertion) assertAPIKeysArray(logger *logger.Logger) error {
+	if len(a.ActualValue.Body.ApiKeys) < len(a.ExpectedValue.Body.ApiKeys) {
+		return fmt.Errorf("Expected API keys array to include atleast %d keys, got %d", len(a.ExpectedValue.Body.ApiKeys), len(a.ActualValue.Body.ApiKeys))
+	}
+	logger.Successf("✓ API keys array length: %d", len(a.ActualValue.Body.ApiKeys))
+
+	for _, expectedApiVersionKey := range a.ExpectedValue.Body.ApiKeys {
+		found := false
+
+		for _, actualApiVersionKey := range a.ActualValue.Body.ApiKeys {
+			if actualApiVersionKey.ApiKey == expectedApiVersionKey.ApiKey {
+				found = true
+				if actualApiVersionKey.MinVersion > expectedApiVersionKey.MaxVersion {
+					return fmt.Errorf("Expected min version %v to be < max version %v for %s", actualApiVersionKey.MinVersion, expectedApiVersionKey.MaxVersion, apiKeyNames[expectedApiVersionKey.ApiKey])
+				}
+
+				// anything above or equal to expected minVersion is fine
+				if actualApiVersionKey.MinVersion < expectedApiVersionKey.MinVersion {
+					return fmt.Errorf("Expected API version %v to be supported for %s, got %v", expectedApiVersionKey.MinVersion, apiKeyNames[expectedApiVersionKey.ApiKey], actualApiVersionKey.MinVersion)
+				}
+				logger.Successf("✓ MinVersion for %s is <= %v & >= %v", apiKeyNames[expectedApiVersionKey.ApiKey], expectedApiVersionKey.MaxVersion, expectedApiVersionKey.MinVersion)
+
+				if actualApiVersionKey.MaxVersion < expectedApiVersionKey.MaxVersion {
+					return fmt.Errorf("Expected API version %v to be supported for %s, got %v", expectedApiVersionKey.MaxVersion, apiKeyNames[expectedApiVersionKey.ApiKey], actualApiVersionKey.MaxVersion)
+				}
+				logger.Successf("✓ MaxVersion for %s is >= %v", apiKeyNames[expectedApiVersionKey.ApiKey], expectedApiVersionKey.MaxVersion)
+			}
+		}
+
+		if !found {
+			return fmt.Errorf("Expected APIVersionsResponseKey array to include API key %d (%s)", expectedApiVersionKey.ApiKey, apiKeyNames[expectedApiVersionKey.ApiKey])
+		}
+	}
+
+	return nil
+}
+
+func (a *ApiVersionsResponseAssertion) Run(logger *logger.Logger) error {
+	if err := NewResponseHeaderAssertion(a.ActualValue.Header, a.ExpectedValue.Header).Run(logger); err != nil {
+		return err
+	}
+
+	if err := a.assertBody(logger); err != nil {
+		return err
+	}
+
+	return nil
+}

internal/assertions/describe_topic_partitions_response_assertion.go → internal/assertions_legacy/describe_topic_partitions_response_assertion.go

@@ -1,16 +1,16 @@
-package assertions
+package assertions_legacy
 
 import (
 	"fmt"
 
 	"github.com/codecrafters-io/kafka-tester/protocol"
-	kafkaapi "github.com/codecrafters-io/kafka-tester/protocol/api"
+	"github.com/codecrafters-io/kafka-tester/protocol/kafkaapi_legacy"
 	"github.com/codecrafters-io/tester-utils/logger"
 )
 
 type DescribeTopicPartitionsResponseAssertion struct {
-	ActualValue             kafkaapi.DescribeTopicPartitionsResponse
-	ExpectedValue           kafkaapi.DescribeTopicPartitionsResponse
+	ActualValue             kafkaapi_legacy.DescribeTopicPartitionsResponse
+	ExpectedValue           kafkaapi_legacy.DescribeTopicPartitionsResponse
 	excludedBodyFields      []string
 	excludedTopicFields     []string
 	excludedPartitionFields []string
@@ -20,7 +20,7 @@ var DTP_EXCLUDABLE_BODY_FIELDS = []string{"ThrottleTimeMs", "Topics"}
 var DTP_EXCLUDABLE_TOPIC_FIELDS = []string{"ErrorCode", "Name", "TopicID", "Partitions"}
 var DTP_EXCLUDABLE_PARTITION_FIELDS = []string{"ErrorCode", "PartitionIndex"}
 
-func NewDescribeTopicPartitionsResponseAssertion(actualValue kafkaapi.DescribeTopicPartitionsResponse, expectedValue kafkaapi.DescribeTopicPartitionsResponse) *DescribeTopicPartitionsResponseAssertion {
+func NewDescribeTopicPartitionsResponseAssertion(actualValue kafkaapi_legacy.DescribeTopicPartitionsResponse, expectedValue kafkaapi_legacy.DescribeTopicPartitionsResponse) *DescribeTopicPartitionsResponseAssertion {
 	return &DescribeTopicPartitionsResponseAssertion{
 		ActualValue:             actualValue,
 		ExpectedValue:           expectedValue,
@@ -123,7 +123,7 @@ func (a *DescribeTopicPartitionsResponseAssertion) assertTopics(logger *logger.L
 	return nil
 }
 
-func (a *DescribeTopicPartitionsResponseAssertion) assertPartitions(expectedPartitions []kafkaapi.DescribeTopicPartitionsResponsePartition, actualPartitions []kafkaapi.DescribeTopicPartitionsResponsePartition, topicPartitionIndex int, logger *logger.Logger) error {
+func (a *DescribeTopicPartitionsResponseAssertion) assertPartitions(expectedPartitions []kafkaapi_legacy.DescribeTopicPartitionsResponsePartition, actualPartitions []kafkaapi_legacy.DescribeTopicPartitionsResponsePartition, topicPartitionIndex int, logger *logger.Logger) error {
 	if len(actualPartitions) != len(expectedPartitions) {
 		return fmt.Errorf("Expected partitions.length to be %d, got %d", len(expectedPartitions), len(actualPartitions))
 	}

internal/assertions/fetch_response_assertion.go → internal/assertions_legacy/fetch_response_assertion.go

@@ -1,23 +1,23 @@
-package assertions
+package assertions_legacy
 
 import (
 	"bytes"
 	"fmt"
 
 	"github.com/codecrafters-io/kafka-tester/protocol"
-	kafkaapi "github.com/codecrafters-io/kafka-tester/protocol/api"
-	"github.com/codecrafters-io/kafka-tester/protocol/serializer"
+	"github.com/codecrafters-io/kafka-tester/protocol/kafkaapi_legacy"
+	"github.com/codecrafters-io/kafka-tester/protocol/serializer_legacy"
 	"github.com/codecrafters-io/tester-utils/bytes_diff_visualizer"
 	"github.com/codecrafters-io/tester-utils/logger"
 )
 
 type FetchResponseAssertion struct {
-	ActualValue             kafkaapi.FetchResponse
-	ExpectedValue           kafkaapi.FetchResponse
+	ActualValue             kafkaapi_legacy.FetchResponse
+	ExpectedValue           kafkaapi_legacy.FetchResponse
 	excludedPartitionFields []string
 }
 
-func NewFetchResponseAssertion(actualValue kafkaapi.FetchResponse, expectedValue kafkaapi.FetchResponse, logger *logger.Logger) *FetchResponseAssertion {
+func NewFetchResponseAssertion(actualValue kafkaapi_legacy.FetchResponse, expectedValue kafkaapi_legacy.FetchResponse, logger *logger.Logger) *FetchResponseAssertion {
 	return &FetchResponseAssertion{
 		ActualValue:             actualValue,
 		ExpectedValue:           expectedValue,
@@ -78,7 +78,7 @@ func (a *FetchResponseAssertion) assertTopics(logger *logger.Logger) error {
 	return nil
 }
 
-func (a *FetchResponseAssertion) assertPartitions(expectedPartitions []kafkaapi.PartitionResponse, actualPartitions []kafkaapi.PartitionResponse, logger *logger.Logger) error {
+func (a *FetchResponseAssertion) assertPartitions(expectedPartitions []kafkaapi_legacy.PartitionResponse, actualPartitions []kafkaapi_legacy.PartitionResponse, logger *logger.Logger) error {
 	if len(actualPartitions) != len(expectedPartitions) {
 		return fmt.Errorf("Expected partitions.length to be %d, got %d", len(expectedPartitions), len(actualPartitions))
 	}
@@ -121,7 +121,7 @@ func (a *FetchResponseAssertion) assertPartitions(expectedPartitions []kafkaapi.
 	return nil
 }
 
-func (a *FetchResponseAssertion) assertRecordBatches(expectedRecordBatches []kafkaapi.RecordBatch, actualRecordBatches []kafkaapi.RecordBatch, logger *logger.Logger) error {
+func (a *FetchResponseAssertion) assertRecordBatches(expectedRecordBatches []kafkaapi_legacy.RecordBatch, actualRecordBatches []kafkaapi_legacy.RecordBatch, logger *logger.Logger) error {
 	if len(actualRecordBatches) != len(expectedRecordBatches) {
 		return fmt.Errorf("Expected recordBatches.length to be %d, got %d", len(expectedRecordBatches), len(actualRecordBatches))
 	}
@@ -150,7 +150,7 @@ func (a *FetchResponseAssertion) assertRecordBatches(expectedRecordBatches []kaf
 	return nil
 }
 
-func (a *FetchResponseAssertion) assertRecords(expectedRecords []kafkaapi.Record, actualRecords []kafkaapi.Record, logger *logger.Logger) error {
+func (a *FetchResponseAssertion) assertRecords(expectedRecords []kafkaapi_legacy.Record, actualRecords []kafkaapi_legacy.Record, logger *logger.Logger) error {
 	if len(actualRecords) != len(expectedRecords) {
 		return fmt.Errorf("Expected records.length to be %d, got %d", len(expectedRecords), len(actualRecords))
 	}
@@ -169,22 +169,22 @@ func (a *FetchResponseAssertion) assertRecords(expectedRecords []kafkaapi.Record
 }
 
 func (a *FetchResponseAssertion) assertRecordBatchBytes(logger *logger.Logger) error {
-	actualRecordBatches := kafkaapi.RecordBatches{}
+	actualRecordBatches := kafkaapi_legacy.RecordBatches{}
 	for _, topic := range a.ActualValue.TopicResponses {
 		for _, partition := range topic.PartitionResponses {
 			actualRecordBatches = append(actualRecordBatches, partition.RecordBatches...)
 		}
 	}
 
-	expectedRecordBatches := kafkaapi.RecordBatches{}
+	expectedRecordBatches := kafkaapi_legacy.RecordBatches{}
 	for _, topic := range a.ExpectedValue.TopicResponses {
 		for _, partition := range topic.PartitionResponses {
 			expectedRecordBatches = append(expectedRecordBatches, partition.RecordBatches...)
 		}
 	}
 
-	expectedRecordBatchBytes := serializer.GetEncodedBytes(expectedRecordBatches)
-	actualRecordBatchBytes := serializer.GetEncodedBytes(actualRecordBatches)
+	expectedRecordBatchBytes := serializer_legacy.GetEncodedBytes(expectedRecordBatches)
+	actualRecordBatchBytes := serializer_legacy.GetEncodedBytes(actualRecordBatches)
 	// Byte Comparison for expected v actual RecordBatch bytes
 	// As we write them to disk, and expect users to not change the values
 	// we can use a simple byte comparison here.

internal/assertions_legacy/response_header_assertion.go

@@ -0,0 +1,33 @@
+package assertions_legacy
+
+import (
+	"fmt"
+
+	"github.com/codecrafters-io/kafka-tester/protocol/kafkaapi_legacy/headers_legacy"
+	"github.com/codecrafters-io/tester-utils/logger"
+)
+
+type ResponseHeaderAssertion struct {
+	ActualValue   headers_legacy.ResponseHeader
+	ExpectedValue headers_legacy.ResponseHeader
+}
+
+func NewResponseHeaderAssertion(actualValue headers_legacy.ResponseHeader, expectedValue headers_legacy.ResponseHeader) *ResponseHeaderAssertion {
+	return &ResponseHeaderAssertion{
+		ActualValue:   actualValue,
+		ExpectedValue: expectedValue,
+	}
+}
+
+func (a *ResponseHeaderAssertion) assertCorrelationId(logger *logger.Logger) error {
+	if a.ActualValue.CorrelationId != a.ExpectedValue.CorrelationId {
+		return fmt.Errorf("Expected correlation_id to be %d, got %d", a.ExpectedValue.CorrelationId, a.ActualValue.CorrelationId)
+	}
+	logger.Successf("✓ correlation_id: %v", a.ActualValue.CorrelationId)
+
+	return nil
+}
+
+func (a *ResponseHeaderAssertion) Run(logger *logger.Logger) error {
+	return a.assertCorrelationId(logger)
+}

internal/assertions/utils.go → internal/assertions_legacy/utils.go

@@ -1,4 +1,4 @@
-package assertions
+package assertions_legacy
 
 import "fmt"