fix linting test and add a custom acc test func

kafka/client.go

@@ -605,7 +605,7 @@ func (c *Client) getKafkaTopics() ([]Topic, error) {
 		return nil, err
 	}
 	topicList := make([]Topic, len(topics))
-	for i, _ := range topicList {
+	for i := range topicList {
 		topicList[i], err = c.ReadTopic(topics[i], true)
 		if err != nil {
 			return nil, err

kafka/data_source_kafka_topics.go

@@ -2,9 +2,9 @@ package kafka
 
 import (
 	"context"
+	"fmt"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
-	"time"
 )
 
 func kafkaTopicsDataSource() *schema.Resource {
@@ -60,13 +60,13 @@ func dataSourceTopicsRead(ctx context.Context, d *schema.ResourceData, meta inte
 	if err != nil {
 		return diag.FromErr(err)
 	}
-	d.SetId(time.Now().UTC().String())
+	d.SetId(fmt.Sprint(len(topics)))
 	return diags
 }
 
 func flattenTopicsData(topicList *[]Topic) []interface{} {
 	if topicList != nil {
-		topics := make([]interface{}, len(*topicList), len(*topicList))
+		topics := make([]interface{}, len(*topicList))
 		for i, topic := range *topicList {
 			topicObj := make(map[string]interface{})
 			topicObj["topic_name"] = topic.Name

kafka/data_source_kafka_topics_test.go

@@ -1,48 +1,77 @@
 package kafka
 
 import (
+	"fmt"
+	"github.com/hashicorp/go-uuid"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
 	"testing"
 
 	r "github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
 )
 
 func TestAcc_Topics(t *testing.T) {
+	u, err := uuid.GenerateUUID()
+	if err != nil {
+		t.Fatal(err)
+	}
+	topicName := fmt.Sprintf("syslog-%s", u)
+
 	bs := testBootstrapServers[0]
-	// Should be only one topic in a brand new kafka cluster
-	expectedTopic := "__confluent.support.metrics"
 	r.Test(t, r.TestCase{
 		ProviderFactories: overrideProviderFactory(),
 		Steps: []r.TestStep{
 			{
-				Config: cfg(t, bs, testDataSourceKafkaTopics),
+				Config: cfg(t, bs, fmt.Sprintf(testDataSourceKafkaTopics, topicName)),
 				Check: r.ComposeTestCheckFunc(
-					r.TestCheckOutput("partitions", "1"),
-					r.TestCheckOutput("replication_factor", "3"),
-					r.TestCheckOutput("topic_name", expectedTopic),
-					r.TestCheckOutput("retention_ms", "31536000000"),
+					testDatasourceTopics,
 				),
 			},
 		},
 	})
 }
 
 const testDataSourceKafkaTopics = `
-data "kafka_topics" "test" {
+resource "kafka_topic" "test" {
+  name               = "%[1]s"
+  replication_factor = 1
+  partitions         = 1
+  config = {
+    "retention.ms" = "22222"
+  }
 }
-
-output "partitions" {
- value = data.kafka_topics.test.list[0].partitions
-}
-
-output "replication_factor" {
- value = data.kafka_topics.test.list[0].replication_factor
+data "kafka_topics" "test" {
+ depends_on = [kafka_topic.test]
 }
+`
 
-output "topic_name" {
- value = data.kafka_topics.test.list[0].topic_name
-}
+func testDatasourceTopics(s *terraform.State) error {
+	resourceState := s.Modules[0].Resources["data.kafka_topics.test"]
+	if resourceState == nil {
+		return fmt.Errorf("resource not found in state")
+	}
+	instanceState := resourceState.Primary
+	client := testProvider.Meta().(*LazyClient)
+	expectedTopics, err := client.GetKafkaTopics()
+	if err != nil {
+		return fmt.Errorf(err.Error())
+	}
+	for i := 0; i < len(expectedTopics); i++ {
+		expectedTopicName := instanceState.Attributes[fmt.Sprintf("list.%d.topic_name", i)]
+		expectedTopicOutput, err := client.ReadTopic(expectedTopicName, true)
+		if err != nil {
+			return fmt.Errorf(err.Error())
+		}
 
-output "retention_ms" {
- value = data.kafka_topics.test.list[0].config["retention.ms"]
+		if instanceState.Attributes[fmt.Sprintf("list.%d.partitions", i)] != fmt.Sprint(expectedTopicOutput.Partitions) {
+			return fmt.Errorf("expected %d for topic %s partition, got %s", expectedTopicOutput.Partitions, expectedTopicOutput.Name, instanceState.Attributes[fmt.Sprintf("list.%d.partitions", i)])
+		}
+		if instanceState.Attributes[fmt.Sprintf("list.%d.replication_factor", i)] != fmt.Sprint(expectedTopicOutput.ReplicationFactor) {
+			return fmt.Errorf("expected %d for topic %s replication factor, got %s", expectedTopicOutput.ReplicationFactor, expectedTopicOutput.Name, instanceState.Attributes[fmt.Sprintf("list.%d.replication_factor", i)])
+		}
+		retentionMs := expectedTopicOutput.Config["retention.ms"]
+		if instanceState.Attributes[fmt.Sprintf("list.%d.config.retention.ms", i)] != *retentionMs {
+			return fmt.Errorf("expected %s for topic %s config retention.ms, got %s", *retentionMs, expectedTopicOutput.Name, instanceState.Attributes[fmt.Sprintf("list.%d.config.retention.ms", i)])
+		}
+	}
+	return nil
 }
-`