Adding basic UTs for different modules and also enabling jacoco plugin for code coverage.

buildSrc/src/main/groovy/com.flipkart.varadhi.java-common-conventions.gradle

@@ -3,6 +3,7 @@ import org.gradle.api.tasks.testing.logging.TestExceptionFormat
 plugins {
  id 'idea'
  id 'java'
+ id 'jacoco'
 }
 
 group = projectGroupId
@@ -33,12 +34,6 @@ sourceSets {
  }
 }
 
-tasks.register('testE2E', Test) {
- testClassesDirs = sourceSets.testE2E.output.classesDirs
- classpath = sourceSets.testE2E.runtimeClasspath
- outputs.upToDateWhen { false } //this will force testE2E to execute always on target invocation.
-}
-
 configurations {
  testE2EImplementation.extendsFrom testImplementation
  testE2ERuntime.extendsFrom testRuntime
@@ -79,6 +74,7 @@ dependencies {
  implementation("io.micrometer:micrometer-registry-jmx:$micrometer_version")
 
  testImplementation('org.junit.jupiter:junit-jupiter:5.9.1')
+ testImplementation('org.mockito:mockito-junit-jupiter:5.4.0')
  testImplementation("io.vertx:vertx-junit5:$vertx_version")
 
  testE2EImplementation("javax.ws.rs:javax.ws.rs-api:$ws_rs_version")
@@ -96,6 +92,7 @@ dependencies {
 
  // common test framework
  testImplementation('org.junit.jupiter:junit-jupiter')
+ testImplementation('org.mockito:mockito-junit-jupiter')
 }
 
 java {
@@ -104,19 +101,32 @@ java {
  }
 }
 
+tasks.register('testE2E', Test) {
+ testClassesDirs = sourceSets.testE2E.output.classesDirs
+ classpath = sourceSets.testE2E.runtimeClasspath
+ outputs.upToDateWhen { false } //this will force testE2E to execute always on target invocation.
+}
+
+tasks.register('copyDependencies', Copy) {
+ into layout.buildDirectory.dir('dependencies')
+ from configurations.runtimeClasspath
+}
+
 tasks.withType(JavaCompile).configureEach {
  options.release.set(17)
 }
 
-
 tasks.withType(JavaCompile).configureEach {
  options.compilerArgs.add("-parameters")
 }
 
-
-tasks.register('copyDependencies', Copy) {
- into layout.buildDirectory.dir('dependencies')
- from configurations.runtimeClasspath
+tasks.withType(JacocoReport).configureEach {
+ reports {
+ html.required = true
+ }
+ getSourceDirectories().from(sourceSets.main.allSource.srcDirs)
+ getClassDirectories().from(sourceSets.main.output)
+ getExecutionData().from(layout.buildDirectory.files().findAll { it.name.endsWith('.exec') })
 }
 
 tasks.withType(Test).configureEach {

common/src/main/java/com/flipkart/varadhi/utils/YamlLoader.java

@@ -3,7 +3,6 @@
 import com.flipkart.varadhi.exceptions.VaradhiException;
 import org.yaml.snakeyaml.Yaml;
 
-import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 
@@ -12,7 +11,7 @@ public class YamlLoader {
  public static <T> T loadConfig(String configFile, Class<T> clazz) {
  try {
  return new Yaml().loadAs(Files.readString(Path.of(configFile)), clazz);
- } catch (IOException e) {
+ } catch (Exception e) {
  throw new VaradhiException(String.format("Failed to load config file: %s as %s.", configFile, clazz), e);
  }
  }

common/src/test/java/com/flipkart/varadhi/utils/JsonMapperTest.java

@@ -2,8 +2,10 @@
 
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.exc.InvalidDefinitionException;
 import com.fasterxml.jackson.databind.jsontype.NamedType;
 import com.flipkart.varadhi.exceptions.VaradhiException;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
@@ -39,6 +41,13 @@ public void testJsonDeserialize_InvalidPolymorphicData() {
  assertTrue(exception.getMessage().contains("Could not resolve type id 'InvalidType'"));
  }
 
+ @Test
+ public void testJsonSerializeFailure() {
+ CantSerialize obj = new CantSerialize();
+ Exception exception = assertThrows(VaradhiException.class, () -> JsonMapper.jsonSerialize(obj));
+ Assertions.assertEquals(InvalidDefinitionException.class, exception.getCause().getClass());
+ }
+
  @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "@vehicleType")
  private abstract static class Vehicle {
  private String manufacturer;
@@ -57,4 +66,12 @@ public Car(String manufacturer, String model, int year) {
  setManufacturer(manufacturer);
  }
  }
+
+ private static class CantSerialize {
+ private int foobar;
+
+ private void CantSerialize() {
+
+ }
+ }
 }

common/src/test/java/com/flipkart/varadhi/utils/YamlLoaderTest.java

@@ -0,0 +1,69 @@
+package com.flipkart.varadhi.utils;
+
+
+import com.flipkart.varadhi.exceptions.VaradhiException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+public class YamlLoaderTest {
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ public void testLoadConfig_ValidFile() throws IOException {
+ // Create a temporary YAML config file
+ Path configFile = tempDir.resolve("config.yaml");
+ String yamlContent = "message: Hello, World!";
+ Files.write(configFile, yamlContent.getBytes());
+
+ // Load the config using YamlLoader
+ Config config = YamlLoader.loadConfig(configFile.toString(), Config.class);
+
+ // Verify the loaded config
+ Assertions.assertNotNull(config);
+ Assertions.assertEquals("Hello, World!", config.getMessage());
+ }
+
+ @Test
+ public void testLoadConfig_InvalidFile() {
+ // Non-existent file path
+ String configFile = "nonexistent.yaml";
+
+ // Verify that VaradhiException is thrown when loading the config
+ Assertions.assertThrows(VaradhiException.class, () -> {
+ YamlLoader.loadConfig(configFile, Config.class);
+ });
+ }
+
+ @Test
+ public void testLoadConfig_InvalidYamlContent() throws IOException {
+ // Create a temporary YAML config file with invalid content
+ Path configFile = tempDir.resolve("config.yaml");
+ String yamlContent = "invalid_yaml_content";
+ Files.write(configFile, yamlContent.getBytes());
+
+ // Verify that VaradhiException is thrown when loading the config
+ Assertions.assertThrows(VaradhiException.class, () -> {
+ YamlLoader.loadConfig(configFile.toString(), Config.class);
+ });
+ }
+
+ // Sample config class for testing
+ public static class Config {
+ private String message;
+
+ public String getMessage() {
+ return message;
+ }
+
+ public void setMessage(String message) {
+ this.message = message;
+ }
+ }
+}

core/build.gradle

@@ -6,4 +6,5 @@ dependencies {
  implementation(project(':entities'))
  implementation(project(':common'))
  implementation('com.fasterxml.jackson.core:jackson-databind')
+ testImplementation(project(':pulsar'))
 }

core/src/test/java/com/flipkart/varadhi/services/VaradhiTopicServiceTest.java

@@ -0,0 +1,70 @@
+package com.flipkart.varadhi.services;
+
+import com.flipkart.varadhi.db.MetaStore;
+import com.flipkart.varadhi.entities.*;
+import com.flipkart.varadhi.exceptions.VaradhiException;
+import com.flipkart.varadhi.pulsar.entities.PulsarStorageTopic;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.mockito.Mockito.*;
+
+public class VaradhiTopicServiceTest {
+
+ StorageTopicFactory<StorageTopic> storageTopicFactory;
+ private VaradhiTopicFactory varadhiTopicFactory;
+ private StorageTopicService<StorageTopic> storageTopicService;
+ private MetaStore metaStore;
+ private VaradhiTopicService varadhiTopicService;
+
+ @BeforeEach
+ public void setUp() {
+ storageTopicService = mock(StorageTopicService.class);
+ metaStore = mock(MetaStore.class);
+ storageTopicFactory = mock(StorageTopicFactory.class);
+ varadhiTopicFactory = spy(new VaradhiTopicFactory(storageTopicFactory));
+ varadhiTopicService = new VaradhiTopicService(storageTopicService, metaStore);
+ //TODO::check it, not a circular dependency though
+ PulsarStorageTopic pTopic = new PulsarStorageTopic("public.default.testTopic.Main.local", 1);
+ doReturn(pTopic).when(storageTopicFactory).getTopic("public.default.testTopic.Main.local", null);
+ }
+
+ @Test
+ public void createVaradhiTopic() {
+ TopicResource topicResource = new TopicResource("testTopic", 1, "testProject", true, false, null);
+ VaradhiTopic varadhiTopic = varadhiTopicFactory.get(topicResource);
+ varadhiTopicService.create(varadhiTopic);
+ verify(metaStore, times(1)).createVaradhiTopic(varadhiTopic);
+ StorageTopic st = varadhiTopic.getInternalTopics().get(InternalTopic.TopicKind.Main).getStorageTopic();
+ verify(storageTopicService, times(1)).create(st);
+ verify(storageTopicFactory, times(1)).getTopic(st.getName(), null);
+ }
+
+ @Test
+ public void createVaradhiTopicWhenMetaStoreFails() {
+ TopicResource topicResource = new TopicResource("testTopic", 1, "testProject", true, false, null);
+ VaradhiTopic varadhiTopic = varadhiTopicFactory.get(topicResource);
+ StorageTopic st = varadhiTopic.getInternalTopics().get(InternalTopic.TopicKind.Main).getStorageTopic();
+ doThrow(new VaradhiException("Some error")).when(metaStore).createVaradhiTopic(varadhiTopic);
+ Exception exception =
+ Assertions.assertThrows(VaradhiException.class, () -> varadhiTopicService.create(varadhiTopic));
+ verify(metaStore, times(1)).createVaradhiTopic(varadhiTopic);
+ verify(storageTopicService, never()).create(st);
+ Assertions.assertEquals(exception.getClass(), VaradhiException.class);
+ }
+
+ @Test
+ public void createVaradhiTopicWhenStorageTopicServiceFails() {
+ TopicResource topicResource = new TopicResource("testTopic", 1, "testProject", true, false, null);
+ VaradhiTopic varadhiTopic = varadhiTopicFactory.get(topicResource);
+ StorageTopic st = varadhiTopic.getInternalTopics().get(InternalTopic.TopicKind.Main).getStorageTopic();
+ doThrow(new VaradhiException("Some error")).when(storageTopicService).create(st);
+ Exception exception =
+ Assertions.assertThrows(VaradhiException.class, () -> varadhiTopicService.create(varadhiTopic));
+ verify(metaStore, times(1)).createVaradhiTopic(varadhiTopic);
+ verify(storageTopicService, times(1)).create(st);
+ Assertions.assertEquals(exception.getClass(), VaradhiException.class);
+ }
+}
+

entities/build.gradle

@@ -4,4 +4,5 @@ plugins {
 dependencies {
  implementation(project(':common'))
  implementation('com.fasterxml.jackson.core:jackson-databind:2.14.2')
+ testImplementation(project(':pulsar'))
 }

entities/src/test/java/com/flipkart/varadhi/entities/VaradhiTopicFactoryTest.java

@@ -0,0 +1,37 @@
+package com.flipkart.varadhi.entities;
+
+import com.flipkart.varadhi.pulsar.entities.PulsarStorageTopic;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.mockito.Mockito.*;
+
+public class VaradhiTopicFactoryTest {
+ private VaradhiTopicFactory varadhiTopicFactory;
+ private StorageTopicFactory<StorageTopic> storageTopicFactory;
+
+ @BeforeEach
+ public void setUp() {
+ storageTopicFactory = mock(StorageTopicFactory.class);
+ varadhiTopicFactory = new VaradhiTopicFactory(storageTopicFactory);
+ //TODO::check it, not a circular dependency though
+ PulsarStorageTopic pTopic = new PulsarStorageTopic("public.default.testTopic.Main.local", 1);
+ doReturn(pTopic).when(storageTopicFactory).getTopic("public.default.testTopic.Main.local", null);
+ }
+
+ @Test
+ public void getTopic() {
+ TopicResource topicResource = new TopicResource("testTopic", 1, "testProject", true, false, null);
+ VaradhiTopic varadhiTopic = varadhiTopicFactory.get(topicResource);
+ Assertions.assertNotNull(varadhiTopic);
+ Assertions.assertEquals(1, varadhiTopic.getInternalTopics().size());
+ InternalTopic it = varadhiTopic.getInternalTopics().get(InternalTopic.TopicKind.Main);
+ StorageTopic st = it.getStorageTopic();
+ Assertions.assertEquals(it.getStatus(), InternalTopic.ProduceStatus.Active);
+ Assertions.assertEquals(it.getRegion(), "local");
+ Assertions.assertNull(it.getSourceRegion());
+ Assertions.assertNotNull(st);
+ verify(storageTopicFactory, times(1)).getTopic("public.default.testTopic.Main.local", null);
+ }
+}

lombok.config

@@ -0,0 +1,7 @@
+# This tells lombok this directory is the root,
+# no need to look somewhere else for java code.
+config.stopBubbling = true
+# This will add the @lombok.Generated annotation
+# to all the code generated by Lombok,
+# so it can be excluded from coverage by jacoco.
+lombok.addLombokGeneratedAnnotation = true

pulsar/src/main/java/com/flipkart/varadhi/pulsar/PulsarStackProvider.java

@@ -5,6 +5,7 @@
 import com.flipkart.varadhi.entities.StorageTopic;
 import com.flipkart.varadhi.entities.StorageTopicFactory;
 import com.flipkart.varadhi.exceptions.InvalidStateException;
+import com.flipkart.varadhi.pulsar.config.PulsarClientOptions;
 import com.flipkart.varadhi.pulsar.config.PulsarConfig;
 import com.flipkart.varadhi.pulsar.entities.PulsarStorageTopic;
 import com.flipkart.varadhi.pulsar.entities.PulsarTopicFactory;
@@ -13,12 +14,17 @@
 import com.flipkart.varadhi.services.MessagingStackProvider;
 import com.flipkart.varadhi.services.StorageTopicService;
 import com.flipkart.varadhi.utils.YamlLoader;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+import java.util.concurrent.TimeUnit;
 
 
 public class PulsarStackProvider implements MessagingStackProvider {
  private PulsarTopicService pulsarTopicService;
  private PulsarTopicFactory pulsarTopicFactory;
  private volatile boolean initialised = false;
+ private TimeUnit timeUnit = TimeUnit.MILLISECONDS;
 
  public void init(MessagingStackOptions messagingStackOptions, ObjectMapper mapper) {
  if (!initialised) {
@@ -27,7 +33,8 @@ public void init(MessagingStackOptions messagingStackOptions, ObjectMapper mappe
  PulsarConfig pulsarConfig =
  YamlLoader.loadConfig(messagingStackOptions.getConfigFile(), PulsarConfig.class);
  pulsarTopicFactory = new PulsarTopicFactory();
- pulsarTopicService = new PulsarTopicService(pulsarConfig.getPulsarClientOptions());
+ PulsarAdmin pulsarAdmin = getPulsarAdminClient(pulsarConfig.getPulsarClientOptions());
+ pulsarTopicService = new PulsarTopicService(pulsarAdmin);
  registerSubtypes(mapper);
  initialised = true;
  }
@@ -52,4 +59,18 @@ public <T extends StorageTopic> StorageTopicService<T> getStorageTopicService()
  private void registerSubtypes(ObjectMapper mapper) {
  mapper.registerSubtypes(new NamedType(PulsarStorageTopic.class, "Pulsar"));
  }
+
+ PulsarAdmin getPulsarAdminClient(PulsarClientOptions pulsarClientOptions) {
+ try {
+ //TODO::Add authentication to the pulsar clients. It should be optional however.
+ return PulsarAdmin.builder()
+ .serviceHttpUrl(pulsarClientOptions.getPulsarUrl())
+ .connectionTimeout(pulsarClientOptions.getConnectTimeout(), timeUnit)
+ .requestTimeout(pulsarClientOptions.getRequestTimeout(), timeUnit)
+ .readTimeout(pulsarClientOptions.getReadTimeout(), timeUnit)
+ .build();
+ } catch (PulsarClientException e) {
+ throw new RuntimeException(e);
+ }
+ }
 }