Introduction unit tests

pom.xml

@@ -187,6 +187,19 @@
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
+      <dependency>
+        <groupId>org.assertj</groupId>
+        <artifactId>assertj-core</artifactId>
+        <version>3.24.2</version>
+        <scope>test</scope>
+      </dependency>
+      <dependency>
+        <groupId>org.instancio</groupId>
+        <artifactId>instancio-junit</artifactId>
+        <version>3.0.1</version>
+        <scope>test</scope>
+      </dependency>
+
 
         <!-- Testcontainers -->
         <dependency>

src/main/java/kafdrop/controller/ClusterController.java

@@ -89,7 +89,7 @@ public String clusterInfo(Model model,
     final var clusterSummary = kafkaMonitor.getClusterSummary(topics);
 
     final var missingBrokerIds = clusterSummary.getExpectedBrokerIds().stream()
-      .filter(brokerId -> brokers.stream().noneMatch(b -> b.getId() == brokerId))
+      .filter(brokerId -> brokers.stream().noneMatch(b -> b.id() == brokerId))
       .collect(Collectors.toList());
 
     model.addAttribute("brokers", brokers);

src/main/java/kafdrop/model/BrokerVO.java

@@ -18,38 +18,5 @@
 
 package kafdrop.model;
 
-public final class BrokerVO {
-  private final int id;
-  private final String host;
-  private final int port;
-  private final boolean controller;
-  private final String rack;
-
-  public BrokerVO(int id, String host, int port, String rack, boolean controller) {
-    this.id = id;
-    this.host = host;
-    this.port = port;
-    this.rack = rack;
-    this.controller = controller;
-  }
-
-  public int getId() {
-    return id;
-  }
-
-  public String getHost() {
-    return host;
-  }
-
-  public int getPort() {
-    return port;
-  }
-
-  public String getRack() {
-    return rack;
-  }
-
-  public boolean isController() {
-    return controller;
-  }
+public record BrokerVO(int id, String host, int port, String rack, boolean controller) {
 }

src/main/java/kafdrop/model/ClusterDescriptionVO.java

@@ -0,0 +1,10 @@
+package kafdrop.model;
+
+import org.apache.kafka.common.Node;
+
+import java.util.Collection;
+
+public record ClusterDescriptionVO(Collection<Node> nodes,
+                                   Node controller,
+                                   String clusterId) {
+}

src/main/java/kafdrop/service/KafkaHighLevelAdminClient.java

@@ -2,6 +2,7 @@
 
 import jakarta.annotation.PostConstruct;
 import kafdrop.config.KafkaConfiguration;
+import kafdrop.model.ClusterDescriptionVO;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.ConsumerGroupListing;
@@ -55,19 +56,7 @@ public void init() {
     adminClient = AdminClient.create(properties);
   }
 
-  final class ClusterDescription {
-    final Collection<Node> nodes;
-    final Node controller;
-    final String clusterId;
-
-    ClusterDescription(Collection<Node> nodes, Node controller, String clusterId) {
-      this.nodes = nodes;
-      this.controller = controller;
-      this.clusterId = clusterId;
-    }
-  }
-
-  ClusterDescription describeCluster() {
+  ClusterDescriptionVO describeCluster() {
     final var result = adminClient.describeCluster();
     final Collection<Node> nodes;
     final Node controller;
@@ -80,7 +69,7 @@ ClusterDescription describeCluster() {
       throw new KafkaAdminClientException(e);
     }
 
-    return new ClusterDescription(nodes, controller, clusterId);
+    return new ClusterDescriptionVO(nodes, controller, clusterId);
   }
 
   Set<String> listConsumerGroups() {

src/main/java/kafdrop/service/KafkaMonitorImpl.java

@@ -36,10 +36,12 @@
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.Headers;
+import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
+import java.time.LocalDate;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -72,17 +74,17 @@ public KafkaMonitorImpl(KafkaHighLevelConsumer highLevelConsumer, KafkaHighLevel
   @Override
   public List<BrokerVO> getBrokers() {
     final var clusterDescription = highLevelAdminClient.describeCluster();
-    final var brokerVos = new ArrayList<BrokerVO>(clusterDescription.nodes.size());
-    for (var node : clusterDescription.nodes) {
-      final var isController = node.id() == clusterDescription.controller.id();
-      brokerVos.add(new BrokerVO(node.id(), node.host(), node.port(), node.rack(), isController));
-    }
-    return brokerVos;
+    return clusterDescription.nodes().stream()
+      .map(node -> {
+        boolean isController = node.id() == clusterDescription.controller().id();
+        return new BrokerVO(node.id(), node.host(), node.port(), node.rack(), isController);
+      })
+      .toList();
   }
 
   @Override
   public Optional<BrokerVO> getBroker(int id) {
-    return getBrokers().stream().filter(brokerVo -> brokerVo.getId() == id).findAny();
+    return getBrokers().stream().filter(brokerVo -> brokerVo.id() == id).findAny();
   }
 
   @Override

src/test/java/kafdrop/service/KafkaMonitorImplTest.java

@@ -0,0 +1,99 @@
+package kafdrop.service;
+
+import kafdrop.model.BrokerVO;
+import kafdrop.model.ClusterDescriptionVO;
+import kafdrop.model.ClusterSummaryVO;
+import kafdrop.model.TopicVO;
+import org.apache.kafka.common.Node;
+import org.instancio.Instancio;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+public class KafkaMonitorImplTest {
+
+  private KafkaMonitor kafkaMonitor;
+  private KafkaHighLevelConsumer highLevelConsumer = mock(KafkaHighLevelConsumer.class);
+  private KafkaHighLevelAdminClient highLevelAdminClient = mock(KafkaHighLevelAdminClient.class);
+
+  @BeforeEach
+  void setUp() {
+    kafkaMonitor = new KafkaMonitorImpl(
+      highLevelConsumer,
+      highLevelAdminClient
+    );
+  }
+
+  @Test
+  void cluster_should_return_brokers() {
+    mockThreeAdminNodes();
+
+    var brokers = kafkaMonitor.getBrokers();
+
+    assertThat(brokers).containsExactlyInAnyOrder(
+      new BrokerVO(1, "host1", 1, "rack", false),
+      new BrokerVO(2, "host2", 1, "rack", false),
+      new BrokerVO(3, "host3", 1, "rack", true)
+    );
+  }
+
+
+  @Test
+  void getClusterSummmary_returns_expected_summary_for_cluster() {
+    var topics = Instancio.createSet(TopicVO.class);
+    int topicCount = calculateTopicCount(topics);
+    int partitionCount = calculatePartitions(topics);
+    int underReplicatedCount = calculateReplicas(topics);
+    double preferredReplicaPercent = calculatePreferredReplicaRepcent(topics);
+
+    var clusterSummary = kafkaMonitor.getClusterSummary(topics);
+
+    assertThat(clusterSummary)
+      .returns(partitionCount, ClusterSummaryVO::getPartitionCount)
+      .returns(topicCount, ClusterSummaryVO::getTopicCount)
+      .returns(underReplicatedCount, ClusterSummaryVO::getUnderReplicatedCount)
+      .returns(preferredReplicaPercent, ClusterSummaryVO::getPreferredReplicaPercent);
+  }
+
+  private double calculatePreferredReplicaRepcent(Set<TopicVO> set) {
+    return set.stream().map(TopicVO::getPreferredReplicaPercent)
+      .mapToDouble(it -> it)
+      .sum();
+  }
+
+  private int calculateReplicas(Set<TopicVO> set) {
+    return set.stream().map(TopicVO::getUnderReplicatedPartitions)
+      .map(Collection::size)
+      .mapToInt(it -> it)
+      .sum();
+  }
+
+  private int calculatePartitions(Set<TopicVO> set) {
+    return set.stream().map(TopicVO::getPartitions)
+      .map(Collection::size)
+      .mapToInt(it -> it)
+      .sum();
+  }
+
+  private int calculateTopicCount(Set<TopicVO> set) {
+    return set.size();
+  }
+
+  private void mockThreeAdminNodes() {
+    var controller = new Node(3, "host3", 1, "rack");
+    var nodes = List.of(
+      new Node(1, "host1", 1, "rack"),
+      new Node(2, "host2", 1, "rack"),
+      controller
+    );
+    var clusterId = new ClusterDescriptionVO(nodes, controller, "clusterId");
+    given(highLevelAdminClient.describeCluster()).willReturn(clusterId);
+  }
+}