pom fixes

leia-core/pom.xml

@@ -61,6 +61,24 @@
             <artifactId>aspectjrt</artifactId>
             <groupId>org.aspectj</groupId>
         </dependency>
+
+        <dependency>
+            <artifactId>mockito-core</artifactId>
+            <groupId>org.mockito</groupId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <groupId>org.mockito</groupId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <artifactId>mockito-inline</artifactId>
+            <groupId>org.mockito</groupId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <properties>

leia-core/src/main/java/com/grookage/leia/core/ingestion/hub/SchemaProcessorHub.java

@@ -23,21 +23,27 @@
 import com.grookage.leia.models.schema.engine.SchemaEvent;
 import com.grookage.leia.models.schema.engine.SchemaEventVisitor;
 import com.grookage.leia.repository.SchemaRepository;
-import lombok.NoArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Optional;
 
 @Slf4j
-@NoArgsConstructor
 public class SchemaProcessorHub {
 
     private final Map<SchemaEvent, SchemaProcessor> processors = Maps.newHashMap();
     private SchemaRepository schemaRepository;
     private VersionIDGenerator versionIDGenerator;
 
+    private SchemaProcessorHub() {
+
+    }
+
+    public static SchemaProcessorHub of() {
+        return new SchemaProcessorHub();
+    }
+
     public SchemaProcessorHub withSchemaRepository(SchemaRepository schemaRepository) {
         Preconditions.checkNotNull(schemaRepository, "Schema Repository can't be null");
         this.schemaRepository = schemaRepository;

leia-core/src/main/java/com/grookage/leia/core/ingestion/processors/ApproveSchemaProcessor.java

@@ -19,7 +19,6 @@
 import com.grookage.leia.core.exception.LeiaErrorCode;
 import com.grookage.leia.core.exception.LeiaException;
 import com.grookage.leia.core.ingestion.utils.ContextUtils;
-import com.grookage.leia.core.ingestion.utils.SchemaUtils;
 import com.grookage.leia.models.schema.SchemaDetails;
 import com.grookage.leia.models.schema.SchemaKey;
 import com.grookage.leia.models.schema.engine.SchemaContext;
@@ -50,7 +49,7 @@ public void process(SchemaContext context) {
         final var schemaKey = context.getContext(SchemaKey.class)
                 .orElseThrow((Supplier<Throwable>) () -> LeiaException.error(LeiaErrorCode.VALUE_NOT_FOUND));
         final var storedSchema = getSchemaRepository().get(schemaKey).orElse(null);
-        if (null == storedSchema) {
+        if (null == storedSchema || storedSchema.getSchemaState() != SchemaState.CREATED) {
             log.error("There are no stored schemas present with namespace {}, version {} and schemaName {}. Please try updating them instead",
                     schemaKey.getNamespace(),
                     schemaKey.getVersion(),
@@ -64,6 +63,6 @@ public void process(SchemaContext context) {
         storedSchema.getSchemaMeta().setUpdatedAt(System.currentTimeMillis());
         storedSchema.setSchemaState(SchemaState.APPROVED);
         getSchemaRepository().update(storedSchema);
-        context.addContext(SchemaDetails.class.getSimpleName(), SchemaUtils.toSchemaDetails(storedSchema));
+        context.addContext(SchemaDetails.class.getSimpleName(), storedSchema);
     }
 }

leia-core/src/main/java/com/grookage/leia/core/ingestion/processors/CreateSchemaProcessor.java

@@ -59,8 +59,8 @@ public void process(SchemaContext context) {
         }
         final var userName = ContextUtils.getUser(context);
         final var email = ContextUtils.getEmail(context);
-        final var storedSchema = SchemaUtils.toStoredSchema(createSchemaRequest, userName, email, getVersionIDGenerator());
-        getSchemaRepository().create(storedSchema);
-        context.addContext(SchemaDetails.class.getSimpleName(), SchemaUtils.toSchemaDetails(storedSchema));
+        final var schemaDetails = SchemaUtils.toSchemaDetails(createSchemaRequest, userName, email, getVersionIDGenerator());
+        getSchemaRepository().create(schemaDetails);
+        context.addContext(SchemaDetails.class.getSimpleName(), schemaDetails);
     }
 }

leia-core/src/main/java/com/grookage/leia/core/ingestion/processors/RejectSchemaProcessor.java

@@ -19,7 +19,6 @@
 import com.grookage.leia.core.exception.LeiaErrorCode;
 import com.grookage.leia.core.exception.LeiaException;
 import com.grookage.leia.core.ingestion.utils.ContextUtils;
-import com.grookage.leia.core.ingestion.utils.SchemaUtils;
 import com.grookage.leia.models.schema.SchemaDetails;
 import com.grookage.leia.models.schema.SchemaKey;
 import com.grookage.leia.models.schema.engine.SchemaContext;
@@ -49,7 +48,7 @@ public void process(SchemaContext context) {
         final var schemaKey = context.getContext(SchemaKey.class)
                 .orElseThrow((Supplier<Throwable>) () -> LeiaException.error(LeiaErrorCode.VALUE_NOT_FOUND));
         final var storedSchema = getSchemaRepository().get(schemaKey).orElse(null);
-        if (null == storedSchema) {
+        if (null == storedSchema || storedSchema.getSchemaState() != SchemaState.CREATED) {
             log.error("There are no stored schemas present with namespace {}, version {} and schemaName {}. Please try updating them instead",
                     schemaKey.getNamespace(),
                     schemaKey.getVersion(),
@@ -63,6 +62,6 @@ public void process(SchemaContext context) {
         storedSchema.getSchemaMeta().setUpdatedAt(System.currentTimeMillis());
         storedSchema.setSchemaState(SchemaState.REJECTED);
         getSchemaRepository().update(storedSchema);
-        context.addContext(SchemaDetails.class.getSimpleName(), SchemaUtils.toSchemaDetails(storedSchema));
+        context.addContext(SchemaDetails.class.getSimpleName(), storedSchema);
     }
 }

leia-core/src/main/java/com/grookage/leia/core/ingestion/processors/UpdateSchemaProcessor.java

@@ -19,7 +19,6 @@
 import com.grookage.leia.core.exception.LeiaErrorCode;
 import com.grookage.leia.core.exception.LeiaException;
 import com.grookage.leia.core.ingestion.utils.ContextUtils;
-import com.grookage.leia.core.ingestion.utils.SchemaUtils;
 import com.grookage.leia.models.schema.SchemaDetails;
 import com.grookage.leia.models.schema.SchemaKey;
 import com.grookage.leia.models.schema.engine.SchemaContext;
@@ -70,6 +69,8 @@ public void process(SchemaContext context) {
         storedSchema.getSchemaMeta().setCreatedBy(userName);
         storedSchema.getSchemaMeta().setCreatedByEmail(email);
         storedSchema.getSchemaMeta().setCreatedAt(System.currentTimeMillis());
-        context.addContext(SchemaDetails.class.getSimpleName(), SchemaUtils.toSchemaDetails(storedSchema));
+        storedSchema.setValidationType(updateSchemaRequest.getValidationType());
+        storedSchema.setSchemaType(updateSchemaRequest.getSchemaType());
+        context.addContext(SchemaDetails.class.getSimpleName(), storedSchema);
     }
 }

leia-core/src/main/java/com/grookage/leia/core/ingestion/utils/SchemaUtils.java

@@ -19,48 +19,35 @@
 import com.grookage.leia.core.ingestion.VersionIDGenerator;
 import com.grookage.leia.models.schema.SchemaDetails;
 import com.grookage.leia.models.schema.SchemaKey;
+import com.grookage.leia.models.schema.SchemaMeta;
 import com.grookage.leia.models.schema.engine.SchemaState;
 import com.grookage.leia.models.schema.ingestion.CreateSchemaRequest;
-import com.grookage.leia.models.storage.StoredSchema;
-import com.grookage.leia.models.storage.StoredSchemaMeta;
 import lombok.experimental.UtilityClass;
 
 @UtilityClass
 public class SchemaUtils {
 
-    public StoredSchema toStoredSchema(final CreateSchemaRequest createSchemaRequest,
-                                       final String userName,
-                                       final String email,
-                                       final VersionIDGenerator versionIDGenerator) {
-        return StoredSchema.builder()
-                .schemaName(createSchemaRequest.getSchemaName())
-                .namespace(createSchemaRequest.getNamespace())
-                .versionId(versionIDGenerator.generateVersionId("V"))
+    public SchemaDetails toSchemaDetails(final CreateSchemaRequest createSchemaRequest,
+                                         final String userName,
+                                         final String email,
+                                         final VersionIDGenerator versionIDGenerator) {
+        return SchemaDetails.builder()
+                .schemaKey(SchemaKey.builder()
+                        .namespace(createSchemaRequest.getNamespace())
+                        .schemaName(createSchemaRequest.getSchemaName())
+                        .version(versionIDGenerator.generateVersionId("V"))
+                        .build()
+                )
                 .schemaState(SchemaState.CREATED)
                 .schemaType(createSchemaRequest.getSchemaType())
                 .description(createSchemaRequest.getDescription())
                 .attributes(createSchemaRequest.getAttributes())
-                .schemaMeta(StoredSchemaMeta.builder()
+                .schemaMeta(SchemaMeta.builder()
                         .createdBy(userName)
                         .createdByEmail(email)
                         .createdAt(System.currentTimeMillis())
                         .build())
                 .validationType(createSchemaRequest.getValidationType())
                 .build();
     }
-
-    public SchemaDetails toSchemaDetails(final StoredSchema storedSchema) {
-        return SchemaDetails.builder()
-                .schemaState(storedSchema.getSchemaState())
-                .description(storedSchema.getDescription())
-                .attributes(storedSchema.getAttributes())
-                .schemaKey(SchemaKey.builder()
-                        .namespace(storedSchema.getNamespace())
-                        .schemaName(storedSchema.getSchemaName())
-                        .version(storedSchema.getVersionId())
-                        .build())
-                .schemaType(storedSchema.getSchemaType())
-                .validationType(storedSchema.getValidationType())
-                .build();
-    }
 }

leia-core/src/main/java/com/grookage/leia/core/retrieval/RepositorySupplier.java

@@ -16,7 +16,6 @@
 
 package com.grookage.leia.core.retrieval;
 
-import com.grookage.leia.core.ingestion.utils.SchemaUtils;
 import com.grookage.leia.models.schema.SchemaRegistry;
 import com.grookage.leia.provider.suppliers.LeiaSupplier;
 import com.grookage.leia.repository.SchemaRepository;
@@ -42,7 +41,7 @@ public void stop() {
     @Override
     public SchemaRegistry get() {
         final var schemaDetails = schemaRepository.getSchemas(
-                Set.of(), Set.of(), SchemaUtils::toSchemaDetails);
+                Set.of(), Set.of());
         final var registry = new SchemaRegistry();
         schemaDetails.forEach(registry::add);
         return registry;

leia-core/src/main/java/com/grookage/leia/core/retrieval/SchemaRetriever.java

@@ -17,7 +17,6 @@
 package com.grookage.leia.core.retrieval;
 
 import com.google.common.base.Preconditions;
-import com.grookage.leia.core.ingestion.utils.SchemaUtils;
 import com.grookage.leia.models.schema.SchemaDetails;
 import com.grookage.leia.models.schema.SchemaKey;
 import com.grookage.leia.models.schema.engine.SchemaState;
@@ -60,26 +59,23 @@ public Optional<SchemaDetails> getSchemaDetails(final SchemaKey schemaKey) {
         if (null != cacheConfig && cacheConfig.isEnabled()) {
             return this.refresher.getConfiguration().getSchemaDetails(schemaKey);
         } else {
-            final var storedSchema = schemaRepository.get(schemaKey);
-            return storedSchema.map(SchemaUtils::toSchemaDetails);
+            return schemaRepository.get(schemaKey);
         }
     }
 
     public List<SchemaDetails> getCurrentSchemaDetails(final Set<String> namespaces) {
         if (null != cacheConfig && cacheConfig.isEnabled()) {
             return this.refresher.getConfiguration().getSchemaDetails(namespaces);
         } else {
-            return schemaRepository.getSchemas(namespaces, Set.of(SchemaState.APPROVED),
-                    SchemaUtils::toSchemaDetails);
+            return schemaRepository.getSchemas(namespaces, Set.of(SchemaState.APPROVED));
         }
     }
 
     public List<SchemaDetails> getAllSchemaDetails(final Set<String> namespaces) {
         if (null != cacheConfig && cacheConfig.isEnabled()) {
             return this.refresher.getConfiguration().getAllSchemaDetails(namespaces);
         } else {
-            return schemaRepository.getSchemas(namespaces, Arrays.stream(SchemaState.values()).collect(Collectors.toSet()),
-                    SchemaUtils::toSchemaDetails);
+            return schemaRepository.getSchemas(namespaces, Arrays.stream(SchemaState.values()).collect(Collectors.toSet()));
         }
     }
 }

leia-core/src/test/java/com/grookage/leia/core/ingestion/SchemaIngestorTest.java

@@ -0,0 +1,122 @@
+/*
+ * Copyright (c) 2024. Koushik R <[email protected]>.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.grookage.leia.core.ingestion;
+
+import com.grookage.leia.core.exception.LeiaException;
+import com.grookage.leia.core.ingestion.hub.SchemaProcessorHub;
+import com.grookage.leia.core.ingestion.processors.SchemaProcessor;
+import com.grookage.leia.core.stubs.StubbedSchemaUpdater;
+import com.grookage.leia.models.ResourceHelper;
+import com.grookage.leia.models.schema.SchemaKey;
+import com.grookage.leia.models.schema.engine.SchemaEvent;
+import com.grookage.leia.models.schema.ingestion.CreateSchemaRequest;
+import com.grookage.leia.models.schema.ingestion.UpdateSchemaRequest;
+import lombok.SneakyThrows;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Optional;
+
+class SchemaIngestorTest {
+
+    private static SchemaProcessorHub schemaProcessorHub;
+    private static SchemaIngestor<StubbedSchemaUpdater> schemaIngestor;
+    private static final StubbedSchemaUpdater schemaUpdater = new StubbedSchemaUpdater();
+
+    @BeforeEach
+    void setup() {
+        schemaProcessorHub = Mockito.mock(SchemaProcessorHub.class);
+        schemaIngestor = new SchemaIngestor<StubbedSchemaUpdater>()
+                .withProcessorHub(schemaProcessorHub).build();
+
+    }
+
+    @Test
+    @SneakyThrows
+    void testCreateSchemaNoProcessor() {
+        Mockito.when(schemaProcessorHub.getProcessor(Mockito.any(SchemaEvent.class)))
+                .thenReturn(Optional.empty());
+        final var createSchemaRequest = ResourceHelper.getResource(
+                "schema/createSchemaRequest.json",
+                CreateSchemaRequest.class
+        );
+        Assertions.assertNotNull(createSchemaRequest);
+        Assertions.assertThrows(LeiaException.class, () -> schemaIngestor.add(schemaUpdater, createSchemaRequest));
+    }
+
+    @Test
+    @SneakyThrows
+    void testCreateSchema() {
+        final var schemaProcessor = Mockito.mock(SchemaProcessor.class);
+        Mockito.when(schemaProcessorHub.getProcessor(Mockito.any(SchemaEvent.class)))
+                .thenReturn(Optional.of(schemaProcessor));
+        final var createSchemaRequest = ResourceHelper.getResource(
+                "schema/createSchemaRequest.json",
+                CreateSchemaRequest.class
+        );
+        Assertions.assertNotNull(createSchemaRequest);
+        schemaIngestor.add(schemaUpdater, createSchemaRequest);
+        Mockito.verify(schemaProcessor, Mockito.times(1)).process(Mockito.any());
+    }
+
+    @Test
+    @SneakyThrows
+    void testUpdateSchema() {
+        final var schemaProcessor = Mockito.mock(SchemaProcessor.class);
+        Mockito.when(schemaProcessorHub.getProcessor(Mockito.any(SchemaEvent.class)))
+                .thenReturn(Optional.of(schemaProcessor));
+        final var updateSchemaRequest = ResourceHelper.getResource(
+                "schema/updateSchemaRequest.json",
+                UpdateSchemaRequest.class
+        );
+        Assertions.assertNotNull(updateSchemaRequest);
+        schemaIngestor.update(schemaUpdater, updateSchemaRequest);
+        Mockito.verify(schemaProcessor, Mockito.times(1)).process(Mockito.any());
+    }
+
+    @Test
+    @SneakyThrows
+    void testApproveSchema() {
+        final var schemaProcessor = Mockito.mock(SchemaProcessor.class);
+        Mockito.when(schemaProcessorHub.getProcessor(Mockito.any(SchemaEvent.class)))
+                .thenReturn(Optional.of(schemaProcessor));
+        final var schemaKey = ResourceHelper.getResource(
+                "schema/schemaKey.json",
+                SchemaKey.class
+        );
+        Assertions.assertNotNull(schemaKey);
+        schemaIngestor.approve(schemaUpdater, schemaKey);
+        Mockito.verify(schemaProcessor, Mockito.times(1)).process(Mockito.any());
+    }
+
+    @Test
+    @SneakyThrows
+    void testRejectSchema() {
+        final var schemaProcessor = Mockito.mock(SchemaProcessor.class);
+        Mockito.when(schemaProcessorHub.getProcessor(Mockito.any(SchemaEvent.class)))
+                .thenReturn(Optional.of(schemaProcessor));
+        final var schemaKey = ResourceHelper.getResource(
+                "schema/schemaKey.json",
+                SchemaKey.class
+        );
+        Assertions.assertNotNull(schemaKey);
+        schemaIngestor.reject(schemaUpdater, schemaKey);
+        Mockito.verify(schemaProcessor, Mockito.times(1)).process(Mockito.any());
+    }
+}