47 lines
1.1 KiB
Go
47 lines
1.1 KiB
Go
package kafka
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
|
|
"github.com/fiskerinc/cloud-services/pkg/kafka/mock"
|
|
"github.com/fiskerinc/cloud-services/pkg/testhelper"
|
|
|
|
"github.com/confluentinc/confluent-kafka-go/v2/kafka"
|
|
)
|
|
|
|
func TestAdmin(t *testing.T) {
|
|
a := mock.Admin{
|
|
MockFunc: func(ctx context.Context, topics []kafka.TopicSpecification, options ...kafka.CreateTopicsAdminOption) ([]kafka.TopicResult, error) {
|
|
return []kafka.TopicResult{{
|
|
Topic: "test",
|
|
Error: kafka.Error{},
|
|
}}, nil
|
|
},
|
|
}
|
|
|
|
Admin = &a
|
|
|
|
err := EnsureTopicsExist([]string{"test"})
|
|
if err != nil {
|
|
t.Errorf(testhelper.TestErrorTemplate, "TestAdmin", nil, err)
|
|
}
|
|
}
|
|
|
|
func TestAdminError(t *testing.T) {
|
|
testError := errors.New("Test error being thrown")
|
|
a := mock.Admin{
|
|
MockFunc: func(ctx context.Context, topics []kafka.TopicSpecification, options ...kafka.CreateTopicsAdminOption) ([]kafka.TopicResult, error) {
|
|
return []kafka.TopicResult{}, testError
|
|
},
|
|
}
|
|
|
|
Admin = &a
|
|
|
|
err := EnsureTopicsExist([]string{"test"})
|
|
if err.Error() != testError.Error() {
|
|
t.Errorf(testhelper.TestErrorTemplate, "TestAdmin", testError, err)
|
|
}
|
|
}
|