diff --git a/aggregation-manager-job/pom.xml b/aggregation-manager-job/pom.xml
index f36bfdaae9b567ca491289d2225f7cd633deb8db..3ec206f322539b2f45aefa8ba3738a984d81239c 100644
--- a/aggregation-manager-job/pom.xml
+++ b/aggregation-manager-job/pom.xml
@@ -50,6 +50,10 @@
<artifactId>ozg-cloud-spring-boot-starter</artifactId>
<groupId>de.ozgcloud.api-lib</groupId>
</dependency>
+ <dependency>
+ <groupId>de.ozgcloud.aggregation</groupId>
+ <artifactId>aggregation-manager-interface</artifactId>
+ </dependency>
<!-- Spring -->
<dependency>
<groupId>org.springframework.boot</groupId>
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/Aggregation.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/Aggregation.java
new file mode 100644
index 0000000000000000000000000000000000000000..50368c67cbb3360bdac7474a7f58a5e7089e6d3a
--- /dev/null
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/Aggregation.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation;
+
+import java.util.stream.Stream;
+
+import de.ozgcloud.aggregation.warehouse.DocumentEntry;
+import lombok.Builder;
+
+@Builder
+public record Aggregation(String aggregationName, Stream<DocumentEntry> documentEntries) {
+}
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationDataLoader.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationDataLoader.java
new file mode 100644
index 0000000000000000000000000000000000000000..454b0adc0c7f1e84cf9633d8f15e453ea4ab3e44
--- /dev/null
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationDataLoader.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation;
+
+import de.ozgcloud.aggregation.transformation.AggregationMapping;
+
+public interface AggregationDataLoader {
+
+ void loadIntoTarget(Aggregation aggregation);
+
+ AggregationMapping.Scope getScope();
+}
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationDataLoaderRegistry.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationDataLoaderRegistry.java
new file mode 100644
index 0000000000000000000000000000000000000000..614e35a422db7f486d6fd53a8aaa5ed4878b063d
--- /dev/null
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationDataLoaderRegistry.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.springframework.stereotype.Component;
+
+import de.ozgcloud.aggregation.transformation.AggregationMapping;
+import de.ozgcloud.common.errorhandling.TechnicalException;
+
+@Component
+public class AggregationDataLoaderRegistry {
+
+ private final Map<AggregationMapping.Scope, AggregationDataLoader> loadersByScope;
+
+ AggregationDataLoaderRegistry(List<AggregationDataLoader> loaders) {
+ this.loadersByScope = toLoadersByScope(loaders);
+ }
+
+ public boolean hasLoader(AggregationMapping.Scope scope) {
+ return loadersByScope.containsKey(scope);
+ }
+
+ public AggregationDataLoader getLoader(AggregationMapping.Scope scope) {
+ return Optional.ofNullable(loadersByScope.get(scope)).orElseThrow(() -> new TechnicalException("No data loader for scope " + scope));
+ }
+
+ static Map<AggregationMapping.Scope, AggregationDataLoader> toLoadersByScope(List<AggregationDataLoader> loaders) {
+ try {
+ return loaders.stream().collect(Collectors.toMap(AggregationDataLoader::getScope, Function.identity()));
+ } catch (IllegalStateException e) {
+ throw new TechnicalException("Multiple loaders exist for single scope.", e);
+ }
+ }
+}
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerGrpcConfiguration.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerGrpcConfiguration.java
index eec32d5d22db5bb94650f73344808f20fc3e5931..3994100601b36455b0cddfbaf5f3dcf965ebfdfa 100644
--- a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerGrpcConfiguration.java
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerGrpcConfiguration.java
@@ -23,9 +23,13 @@
*/
package de.ozgcloud.aggregation;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
+import de.ozgcloud.aggregation.data.AggregationDataServiceGrpc;
+import de.ozgcloud.aggregation.extern.AggregationDataRemoteService;
+import de.ozgcloud.aggregation.extern.GrpcAggregationDataMapper;
import de.ozgcloud.apilib.vorgang.OzgCloudVorgangService;
import de.ozgcloud.apilib.vorgang.grpc.GrpcOzgCloudVorgangService;
import de.ozgcloud.apilib.vorgang.grpc.OzgCloudVorgangMapper;
@@ -37,9 +41,13 @@ import net.devh.boot.grpc.client.inject.GrpcClientBean;
@Configuration
@GrpcClientBean(clazz = VorgangServiceGrpc.VorgangServiceBlockingStub.class, beanName = "vorgangServiceBlockingStub", client = @GrpcClient("vorgang-manager"))
+@GrpcClientBean(clazz = AggregationDataServiceGrpc.AggregationDataServiceStub.class, beanName = "aggregationDataServiceStub", client = @GrpcClient("aggregation-manager"))
public class AggregationManagerGrpcConfiguration {
+
@GrpcClient("vorgang-manager")
VorgangServiceGrpc.VorgangServiceBlockingStub vorgangServiceBlockingStub;
+ @GrpcClient("aggregation-manager")
+ AggregationDataServiceGrpc.AggregationDataServiceStub aggregationDataServiceStub;
@Bean
OzgCloudVorgangService grpcOzgCloudVorgangService(OzgCloudVorgangMapper vorgangMapper,
@@ -49,4 +57,9 @@ public class AggregationManagerGrpcConfiguration {
aggregationCallContext);
}
+ @Bean
+ @ConditionalOnProperty("grpc.client.aggregation-manager.address")
+ AggregationDataRemoteService aggregationDataRemoteService(AggregationManagerProperties properties, GrpcAggregationDataMapper grpcAggregationDataMapper) {
+ return new AggregationDataRemoteService(aggregationDataServiceStub, properties, grpcAggregationDataMapper);
+ }
}
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerConfiguration.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerProperties.java
similarity index 79%
rename from aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerConfiguration.java
rename to aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerProperties.java
index 00541877a3dbf271f7e375b6abf291d861e785cc..c54be5d11eeba10534f55e0ab8669afcaffb9fa1 100644
--- a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerConfiguration.java
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerProperties.java
@@ -24,15 +24,21 @@
package de.ozgcloud.aggregation;
import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.validation.annotation.Validated;
import lombok.Getter;
import lombok.Setter;
+import lombok.extern.log4j.Log4j2;
-@ConfigurationProperties(prefix = "aggregation-manager")
+@ConfigurationProperties(prefix = "ozgcloud")
+@Configuration
+@Validated
@Getter
@Setter
-public class AggregationManagerConfiguration {
+@Log4j2
+public class AggregationManagerProperties {
private int fetchingBatchSize;
-
+ private String mandant;
}
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerRunner.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerRunner.java
index 3d5596e8748fbdd4c3cf9ebcc884faa7d40fa79b..8b0544e0d812837ccb7969ada028c97c9ee06247 100644
--- a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerRunner.java
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationManagerRunner.java
@@ -23,8 +23,9 @@
*/
package de.ozgcloud.aggregation;
-import java.util.Objects;
+import java.util.Arrays;
import java.util.UUID;
+import java.util.stream.Stream;
import org.apache.logging.log4j.ThreadContext;
import org.springframework.beans.factory.annotation.Lookup;
@@ -48,41 +49,50 @@ public abstract class AggregationManagerRunner implements CommandLineRunner {
private static final String MDC_EXECUTION = "execution";
- private final AggregationManagerConfiguration config;
+ private final AggregationManagerProperties aggregationManagerProperties;
private final TransformationProperties transformationProperties;
private final TransformationService transformationService;
+ private final AggregationDataLoaderRegistry loaderRegistry;
@Override
public void run(String... args) throws TransformationException {
- var identifier = transformationProperties.getIdentifier();
- var aggregationMappings = transformationProperties.getAggregationMappings();
- if (Objects.isNull(aggregationMappings) || aggregationMappings.isEmpty()) {
- runWithDefaultTransformation(transformationService.load(identifier, null));
- } else {
- aggregationMappings.stream()
- .forEach(aggregationMapping -> runWithTransformation(transformationService.load(identifier, aggregationMapping),
- aggregationMapping));
- }
+ getScopesWithoutConfiguredTransformations().forEach(this::runWithDefaultTransformation);
+ runWithConfiguredTransformations();
+ }
+
+ Stream<AggregationMapping.Scope> getScopesWithoutConfiguredTransformations() {
+ return Arrays.stream(AggregationMapping.Scope.values()).filter(loaderRegistry::hasLoader).filter(this::hasNoConfiguredTransformations);
+ }
+
+ boolean hasNoConfiguredTransformations(AggregationMapping.Scope scope) {
+ return transformationProperties.getAggregationMappings().stream().map(AggregationMapping::getScope).noneMatch(scope::equals);
+ }
+
+ void runWithDefaultTransformation(AggregationMapping.Scope scope) throws TransformationException {
+ runWithTransformation(
+ transformationService.load(transformationProperties.getIdentifier(), null),
+ createAggregator(scope));
}
- void runWithDefaultTransformation(Transformation transformation) {
- runWithTransformation(transformation, null);
+ void runWithConfiguredTransformations() {
+ transformationProperties.getAggregationMappings().forEach(aggregationMapping -> runWithTransformation(
+ transformationService.load(transformationProperties.getIdentifier(), aggregationMapping),
+ createAggregator(aggregationMapping.getScope()).withAggregationMapping(aggregationMapping)));
}
- void runWithTransformation(Transformation transformation, AggregationMapping aggregationMapping) {
+ void runWithTransformation(Transformation transformation, Aggregator aggregator) {
try (Execution execution = new Execution(transformation)) {
ThreadContext.put(MDC_EXECUTION, execution.id.toString());
- prepareAggregator(execution, aggregationMapping).aggregate();
+ aggregator.withExecution(execution).aggregate();
} finally {
ThreadContext.remove(MDC_EXECUTION);
}
}
- Aggregator prepareAggregator(Execution execution, AggregationMapping aggregationMapping) {
+ Aggregator createAggregator(AggregationMapping.Scope scope) {
return createAggregator()
- .withExecution(execution)
- .withAggregationMapping(aggregationMapping)
- .withBatchSize(config.getFetchingBatchSize());
+ .withBatchSize(aggregationManagerProperties.getFetchingBatchSize())
+ .withLoader(loaderRegistry.getLoader(scope));
}
@Lookup
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/Aggregator.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/Aggregator.java
index 79cf481fe8871d780ceec13bb1b7564037d2fb0c..3049875ba35003057a653cbf82d8019748d8741c 100644
--- a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/Aggregator.java
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/Aggregator.java
@@ -22,7 +22,6 @@ import de.ozgcloud.aggregation.transformation.AggregationMapping.FormIdentifier;
import de.ozgcloud.aggregation.transformation.TransformationException;
import de.ozgcloud.aggregation.transformation.VorgangMapper;
import de.ozgcloud.aggregation.warehouse.DocumentEntry;
-import de.ozgcloud.aggregation.warehouse.WarehouseRepository;
import de.ozgcloud.apilib.vorgang.OzgCloudVorgang;
import de.ozgcloud.apilib.vorgang.OzgCloudVorgangQuery;
import de.ozgcloud.apilib.vorgang.OzgCloudVorgangQuery.FormIdentification;
@@ -42,13 +41,13 @@ class Aggregator {
private static final Predicate<Batch> HAS_NEXT_BATCH = x -> !x.items.isEmpty();
private final OzgCloudVorgangService vorgangService;
- private final WarehouseRepository repository;
private final VorgangMapper vorgangMapper;
private Execution execution;
private FormIdentifier formIdentifier;
- private String collectionName = DocumentEntry.COLLECTION;
+ private String aggregationName;
private int batchSize = 100;
+ private AggregationDataLoader loader;
public Aggregator withExecution(Execution execution) {
this.execution = execution;
@@ -58,7 +57,7 @@ class Aggregator {
public Aggregator withAggregationMapping(AggregationMapping aggregationMapping) {
if (Objects.nonNull(aggregationMapping)) {
this.formIdentifier = aggregationMapping.getFormIdentifier();
- this.collectionName = aggregationMapping.getName();
+ this.aggregationName = aggregationMapping.getName();
}
return this;
}
@@ -68,8 +67,13 @@ class Aggregator {
return this;
}
+ public Aggregator withLoader(AggregationDataLoader loader) {
+ this.loader = loader;
+ return this;
+ }
+
public void aggregate() {
- loadVorgaengeIntoRepository(Stream.concat(extractBatchesOfVorgaengeFromDataSource(), extractBatchesOfDeletedVorgaengeFromDataSource()));
+ loadVorgaenge(Stream.concat(extractBatchesOfVorgaengeFromDataSource(), extractBatchesOfDeletedVorgaengeFromDataSource()));
}
Stream<Batch> extractBatchesOfVorgaengeFromDataSource() {
@@ -129,13 +133,12 @@ class Aggregator {
return UUID.randomUUID();
}
- void loadVorgaengeIntoRepository(Stream<Batch> batches) {
- repository.clearCollection(collectionName);
- batches.map(this::transformBatchToDocumentEntries).forEach(this::loadDocumentEntriesIntoRepository);
+ void loadVorgaenge(Stream<Batch> batches) {
+ loader.loadIntoTarget(new Aggregation(aggregationName, batches.flatMap(this::transformBatchToDocumentEntries)));
}
- List<DocumentEntry> transformBatchToDocumentEntries(Batch batch) {
- return batch.items.stream().map(vorgang -> transformWithinBatch(batch, vorgang)).filter(Objects::nonNull).toList();
+ Stream<DocumentEntry> transformBatchToDocumentEntries(Batch batch) {
+ return batch.items.stream().map(vorgang -> transformWithinBatch(batch, vorgang)).filter(Objects::nonNull);
}
DocumentEntry transformWithinBatch(Batch batch, OzgCloudVorgang vorgang) {
@@ -149,11 +152,6 @@ class Aggregator {
}
}
- void loadDocumentEntriesIntoRepository(List<DocumentEntry> entries) {
- LOG.atDebug().log("store documents: {}", () -> entries.stream().map(DocumentEntry::getId).toList());
- repository.saveAllInCollection(entries, collectionName);
- }
-
record Batch(Execution execution, UUID id, Page page, List<OzgCloudVorgang> items) {
}
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/TransformationProperties.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/TransformationProperties.java
index aadc1cac0a74c6ceec10de99559e7c6862b2a84c..25c70cafbd354f0f5c345c28654a0efa8662de22 100644
--- a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/TransformationProperties.java
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/TransformationProperties.java
@@ -23,15 +23,15 @@
*/
package de.ozgcloud.aggregation;
+import java.util.ArrayList;
import java.util.List;
-import jakarta.validation.Valid;
-
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.validation.annotation.Validated;
import de.ozgcloud.aggregation.transformation.AggregationMapping;
+import jakarta.validation.Valid;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.log4j.Log4j2;
@@ -49,7 +49,7 @@ public class TransformationProperties {
* field mappings should be applied
*/
@Valid
- private List<AggregationMapping> aggregationMappings;
+ private List<AggregationMapping> aggregationMappings = new ArrayList<>();
/*
* mapping definition for the entry id
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteLoader.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteLoader.java
new file mode 100644
index 0000000000000000000000000000000000000000..81701adf987258c089a22fa88a36fcd8e7436bf4
--- /dev/null
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteLoader.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.extern;
+
+import java.util.concurrent.ExecutionException;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+import de.ozgcloud.aggregation.Aggregation;
+import de.ozgcloud.aggregation.AggregationDataLoader;
+import de.ozgcloud.aggregation.transformation.AggregationMapping;
+import de.ozgcloud.common.errorhandling.TechnicalException;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+
+@Component
+@ConditionalOnProperty("grpc.client.aggregation-manager.address")
+@RequiredArgsConstructor
+@Log4j2
+public class AggregationDataRemoteLoader implements AggregationDataLoader {
+
+ private final AggregationDataRemoteService service;
+
+ @Override
+ public void loadIntoTarget(Aggregation aggregation) {
+ try {
+ service.sendAggregationData(aggregation).get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new TechnicalException("Waiting for sending aggregation data to complete was interrupted.", e);
+ } catch (ExecutionException e) {
+ throw new TechnicalException("Error on sending aggregation data.", e);
+ }
+ }
+
+ @Override
+ public AggregationMapping.Scope getScope() {
+ return AggregationMapping.Scope.EXTERN;
+ }
+
+}
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteService.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteService.java
new file mode 100644
index 0000000000000000000000000000000000000000..b0b91dcbd5829c257ac1a915ea7a94a9de67a99b
--- /dev/null
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteService.java
@@ -0,0 +1,139 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.extern;
+
+import java.util.Iterator;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.stream.Stream;
+
+import de.ozgcloud.aggregation.Aggregation;
+import de.ozgcloud.aggregation.AggregationManagerProperties;
+import de.ozgcloud.aggregation.data.AggregationDataServiceGrpc;
+import de.ozgcloud.aggregation.data.GrpcAggregationData;
+import de.ozgcloud.aggregation.data.GrpcSendAggregationDataRequest;
+import de.ozgcloud.aggregation.data.GrpcSendAggregationDataResponse;
+import de.ozgcloud.aggregation.warehouse.DocumentEntry;
+import io.grpc.stub.ClientCallStreamObserver;
+import io.grpc.stub.ClientResponseObserver;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import net.devh.boot.grpc.client.inject.GrpcClient;
+
+@RequiredArgsConstructor
+public class AggregationDataRemoteService {
+
+ @GrpcClient("aggregation-manager")
+ private final AggregationDataServiceGrpc.AggregationDataServiceStub serviceStub;
+ private final AggregationManagerProperties properties;
+ private final GrpcAggregationDataMapper grpcAggregationDataMapper;
+
+ public Future<Void> sendAggregationData(Aggregation aggregation) {
+ var responseObserver = buildSendAggregationDataResponseObserver(aggregation);
+ serviceStub.sendAggregationData(responseObserver);
+ return responseObserver.getResponseFuture();
+ }
+
+ SendAggregationDataResponseObserver buildSendAggregationDataResponseObserver(Aggregation aggregation) {
+ var requestData = new RequestData(properties.getMandant(), aggregation.aggregationName(),
+ toGrpcAggregationDataStream(aggregation.documentEntries()).iterator());
+ return new SendAggregationDataResponseObserver(properties.getFetchingBatchSize(), requestData);
+ }
+
+ Stream<GrpcAggregationData> toGrpcAggregationDataStream(Stream<DocumentEntry> documentEntries) {
+ return documentEntries.map(grpcAggregationDataMapper::toGrpcAggregationData);
+ }
+
+ @RequiredArgsConstructor
+ static class SendAggregationDataResponseObserver
+ implements ClientResponseObserver<GrpcSendAggregationDataRequest, GrpcSendAggregationDataResponse> {
+
+ private final int batchSize;
+ private final RequestData requestData;
+ @Getter
+ private final CompletableFuture<Void> responseFuture = new CompletableFuture<>();
+
+ @Override
+ public void beforeStart(ClientCallStreamObserver<GrpcSendAggregationDataRequest> requestObserver) {
+ requestObserver.setOnReadyHandler(buildOnReadyHandler(requestObserver));
+ }
+
+ @Override
+ public void onNext(GrpcSendAggregationDataResponse value) {
+
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ responseFuture.completeExceptionally(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ responseFuture.complete(null);
+ }
+
+ SendAggregationDataOnReadyHandler buildOnReadyHandler(ClientCallStreamObserver<GrpcSendAggregationDataRequest> requestObserver) {
+ return new SendAggregationDataOnReadyHandler(requestObserver, batchSize, requestData);
+ }
+ }
+
+ @RequiredArgsConstructor
+ static class SendAggregationDataOnReadyHandler implements Runnable {
+
+ private final ClientCallStreamObserver<GrpcSendAggregationDataRequest> requestObserver;
+ private final int batchSize;
+ private final RequestData requestData;
+
+ @Override
+ public void run() {
+ while (requestObserver.isReady() && requestData.aggregationDataIterator.hasNext()) {
+ requestObserver.onNext(buildRequest());
+ }
+ if (!requestData.aggregationDataIterator.hasNext()) {
+ requestObserver.onCompleted();
+ }
+ }
+
+ GrpcSendAggregationDataRequest buildRequest() {
+ var builder = GrpcSendAggregationDataRequest.newBuilder()
+ .setName(requestData.aggregationName())
+ .setMandant(requestData.mandant());
+ addAggregationData(builder);
+ return builder.build();
+ }
+
+ private void addAggregationData(GrpcSendAggregationDataRequest.Builder builder) {
+ var elementsAdded = 0;
+ while (requestData.aggregationDataIterator.hasNext() && elementsAdded++ < batchSize) {
+ builder.addAggregationData(requestData.aggregationDataIterator.next());
+ }
+ }
+ }
+
+ @Builder
+ record RequestData(String mandant, String aggregationName, Iterator<GrpcAggregationData> aggregationDataIterator) {
+ }
+}
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/extern/GrpcAggregationDataMapper.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/extern/GrpcAggregationDataMapper.java
new file mode 100644
index 0000000000000000000000000000000000000000..33bdee45a8f99724ce130f8be3a95046fe4f4e3e
--- /dev/null
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/extern/GrpcAggregationDataMapper.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.extern;
+
+import org.mapstruct.Mapper;
+import org.mapstruct.Mapping;
+
+import de.ozgcloud.aggregation.data.GrpcAggregationData;
+import de.ozgcloud.aggregation.warehouse.DocumentEntry;
+
+@Mapper
+public interface GrpcAggregationDataMapper {
+
+ @Mapping(target = "eingangDatum", source = "eingangsdatum")
+ @Mapping(target = "vorgangName", source = "vorgangsname")
+ GrpcAggregationData toGrpcAggregationData(DocumentEntry documentEntry);
+}
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/transformation/AggregationMapping.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/transformation/AggregationMapping.java
index f32b019c01de61bd0c2cba4d87b0d2b3e935f3aa..71d00b3d52542aa9a2caf5b0300085cfa804c540 100644
--- a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/transformation/AggregationMapping.java
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/transformation/AggregationMapping.java
@@ -28,7 +28,6 @@ import java.util.List;
import jakarta.validation.Valid;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
-
import lombok.Builder;
import lombok.Getter;
import lombok.Singular;
@@ -57,7 +56,7 @@ public class AggregationMapping {
}
public enum Scope {
- LAND, MANDANT;
+ EXTERN, INTERN;
}
@Getter
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/AggregationDataWarehouseLoader.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/AggregationDataWarehouseLoader.java
new file mode 100644
index 0000000000000000000000000000000000000000..3ef137ad0027ab2416afae8dc81ce81f80a2d43a
--- /dev/null
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/AggregationDataWarehouseLoader.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.warehouse;
+
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Component;
+
+import de.ozgcloud.aggregation.Aggregation;
+import de.ozgcloud.aggregation.AggregationDataLoader;
+import de.ozgcloud.aggregation.transformation.AggregationMapping;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+
+@Component
+@RequiredArgsConstructor
+@Log4j2
+public class AggregationDataWarehouseLoader implements AggregationDataLoader {
+
+ private final WarehouseRepository repository;
+
+ @Override
+ public void loadIntoTarget(Aggregation aggregation) {
+ var collectionName = getCollectionName(aggregation);
+ repository.clearCollection(collectionName);
+ repository.saveAllInCollection(aggregation.documentEntries(), collectionName);
+ }
+
+ String getCollectionName(Aggregation aggregation) {
+ return StringUtils.isNotBlank(aggregation.aggregationName()) ? aggregation.aggregationName() : DocumentEntry.COLLECTION;
+ }
+
+ @Override
+ public AggregationMapping.Scope getScope() {
+ return AggregationMapping.Scope.INTERN;
+ }
+}
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepository.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepository.java
index 0a312165a18ef10eab5ce9178e3c4ba6f4efc396..a2abb1d85de61ec7e9e45170cab347f944c289ba 100644
--- a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepository.java
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepository.java
@@ -1,12 +1,12 @@
package de.ozgcloud.aggregation.warehouse;
-import java.util.List;
+import java.util.stream.Stream;
interface CustomWarehouseRepository {
DocumentEntry saveInCollection(DocumentEntry documentEntry, String collectionName);
- List<DocumentEntry> saveAllInCollection(Iterable<DocumentEntry> documentEntries, String collectionName);
+ void saveAllInCollection(Stream<DocumentEntry> documentEntries, String collectionName);
void clearCollection(String collectionName);
}
diff --git a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepositoryImpl.java b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepositoryImpl.java
index a5169d606f6a6055943538d8ac02f8a0477716b1..f2ce884b4e31f245599b2e5c427ef47becc18805 100644
--- a/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepositoryImpl.java
+++ b/aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepositoryImpl.java
@@ -1,7 +1,6 @@
package de.ozgcloud.aggregation.warehouse;
-import java.util.List;
-import java.util.stream.StreamSupport;
+import java.util.stream.Stream;
import org.springframework.data.mongodb.core.MongoTemplate;
@@ -13,8 +12,8 @@ class CustomWarehouseRepositoryImpl implements CustomWarehouseRepository {
private final MongoTemplate mongoTemplate;
@Override
- public List<DocumentEntry> saveAllInCollection(Iterable<DocumentEntry> documentEntries, String collectionName) {
- return StreamSupport.stream(documentEntries.spliterator(), false).map(entry -> saveInCollection(entry, collectionName)).toList();
+ public void saveAllInCollection(Stream<DocumentEntry> documentEntries, String collectionName) {
+ documentEntries.forEach(entry -> saveInCollection(entry, collectionName));
}
@Override
diff --git a/aggregation-manager-job/src/main/resources/application-local.yml b/aggregation-manager-job/src/main/resources/application-local.yml
index f971794e9d5962d901fa845a6421360ae558671a..a038ff206e2a99435dd80fa4a1cfd7df9a427c8e 100644
--- a/aggregation-manager-job/src/main/resources/application-local.yml
+++ b/aggregation-manager-job/src/main/resources/application-local.yml
@@ -14,10 +14,6 @@ spring:
database: aggregation-manager-job
ozgcloud:
- vorgang-manager:
- address: static://127.0.0.1:9090
- negotiationType: PLAINTEXT
-
-aggregation-manager:
- fetching-batch-size: 5
+ fetching-batch-size: 5
+ mandant: "Landeshauptstadt Kiel"
diff --git a/aggregation-manager-job/src/main/resources/application.yml b/aggregation-manager-job/src/main/resources/application.yml
index 10ba491e28a6e9b6f87c6acb015035784a31d3fb..879916463132e5cc9e8380de050b048df348bf9d 100644
--- a/aggregation-manager-job/src/main/resources/application.yml
+++ b/aggregation-manager-job/src/main/resources/application.yml
@@ -39,6 +39,7 @@ spring:
application:
name: OzgCloud_AggregationManager
-aggregation-manager:
+ozgcloud:
fetching-batch-size: 100
- identifier: "id.value"
+ aggregation:
+ identifier: "id.value"
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregationDataLoaderRegistryTest.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregationDataLoaderRegistryTest.java
new file mode 100644
index 0000000000000000000000000000000000000000..92cde4b15573002348204af710305aa358bce038
--- /dev/null
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregationDataLoaderRegistryTest.java
@@ -0,0 +1,151 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.mockito.MockedStatic;
+
+import de.ozgcloud.aggregation.transformation.AggregationMapping;
+import de.ozgcloud.common.errorhandling.TechnicalException;
+import de.ozgcloud.common.test.ReflectionTestUtils;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+class AggregationDataLoaderRegistryTest {
+
+ private final AggregationMapping.Scope registeredScope = AggregationMapping.Scope.INTERN;
+ private final AggregationMapping.Scope notRegisteredScope = AggregationMapping.Scope.EXTERN;
+
+ private final AggregationDataLoader loader = new TestLoader(registeredScope);
+ private AggregationDataLoaderRegistry registry;
+
+ @BeforeEach
+ void init() {
+ registry = spy(new AggregationDataLoaderRegistry(List.of(loader)));
+ }
+
+ @Nested
+ class TestConstructor {
+
+ private final List<AggregationDataLoader> loaders = List.of(new TestLoader(AggregationMapping.Scope.INTERN));
+
+ @Test
+ void shouldConvertToLoadersByScope() {
+ try (MockedStatic<AggregationDataLoaderRegistry> mocked = mockStatic(AggregationDataLoaderRegistry.class)) {
+ new AggregationDataLoaderRegistry(loaders);
+ mocked.verify(() -> AggregationDataLoaderRegistry.toLoadersByScope(loaders));
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void setLoadersByScope() {
+ var loadersByScope = Map.of(AggregationMapping.Scope.INTERN, new TestLoader(AggregationMapping.Scope.INTERN));
+ try (MockedStatic<AggregationDataLoaderRegistry> mocked = mockStatic(AggregationDataLoaderRegistry.class)) {
+ mocked.when(() -> AggregationDataLoaderRegistry.toLoadersByScope(loaders)).thenReturn(loadersByScope);
+
+ AggregationDataLoaderRegistry registry = new AggregationDataLoaderRegistry(loaders);
+
+ var loadersByScopeFieldValue = ReflectionTestUtils.getField(registry, "loadersByScope", Map.class);
+ assertThat(loadersByScopeFieldValue).isEqualTo(loadersByScope);
+ }
+ }
+ }
+
+ @Nested
+ class TestHasLoader {
+
+ @Test
+ void shouldReturnFalseIfNoLoaderRegisteredForScope() {
+ var hasLoader = registry.hasLoader(notRegisteredScope);
+
+ assertThat(hasLoader).isFalse();
+ }
+
+ @Test
+ void shouldReturnTrueIfLoaderRegisteredForScope() {
+ var hasLoader = registry.hasLoader(registeredScope);
+
+ assertThat(hasLoader).isTrue();
+ }
+ }
+
+ @Nested
+ class TestGetLoader {
+
+ @Test
+ void shouldReturnLoaderThatSupportsScope() {
+ var loaderSupportingScope = registry.getLoader(registeredScope);
+
+ assertThat(loaderSupportingScope).isSameAs(loader);
+ }
+
+ @Test
+ void shouldThrowExceptionIfNoLoaderRegisteredForScope() {
+ assertThatThrownBy(() -> registry.getLoader(notRegisteredScope)).isInstanceOf(TechnicalException.class);
+ }
+ }
+
+ @Nested
+ class TestToLoadersByScope {
+
+ @ParameterizedTest
+ @EnumSource
+ void shouldAddLoader(AggregationMapping.Scope scope) {
+ var loader = new TestLoader(scope);
+ var loadersByScope = AggregationDataLoaderRegistry.toLoadersByScope(List.of(loader));
+
+ assertThat(loadersByScope).containsEntry(scope, loader);
+ }
+
+ @Test
+ void shouldThrowExceptionIfMultipleLoadersHaveSameScope() {
+ var scope = AggregationMapping.Scope.EXTERN;
+ assertThatThrownBy(() -> AggregationDataLoaderRegistry.toLoadersByScope(List.of(new TestLoader(scope), new TestLoader(scope))))
+ .isInstanceOf(TechnicalException.class);
+ }
+ }
+
+ @RequiredArgsConstructor
+ static class TestLoader implements AggregationDataLoader {
+
+ @Getter
+ private final AggregationMapping.Scope scope;
+
+ @Override
+ public void loadIntoTarget(Aggregation aggregation) {
+
+ }
+ }
+}
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregationManagerRunnerTest.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregationManagerRunnerTest.java
index ef2a1f26bb2fdfbcdf47a5dcd7d2f6292f63b661..70483569a3c3da7975005b77bbcd7b67f03091b5 100644
--- a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregationManagerRunnerTest.java
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregationManagerRunnerTest.java
@@ -27,13 +27,16 @@ import static org.assertj.core.api.Assertions.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
-import java.util.Collections;
import java.util.List;
+import java.util.stream.Stream;
import org.apache.commons.lang3.RandomUtils;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.mockito.AdditionalMatchers;
import org.mockito.ArgumentMatcher;
import org.mockito.InjectMocks;
import org.mockito.Mock;
@@ -46,24 +49,17 @@ import de.ozgcloud.aggregation.transformation.AggregationMapping;
import de.ozgcloud.aggregation.transformation.AggregationMappingTestFactory;
import de.ozgcloud.aggregation.transformation.Transformation;
import de.ozgcloud.aggregation.transformation.TransformationService;
-import de.ozgcloud.aggregation.transformation.VorgangMapper;
-import de.ozgcloud.aggregation.warehouse.WarehouseRepository;
-import de.ozgcloud.apilib.vorgang.OzgCloudVorgangService;
class AggregationManagerRunnerTest {
@Mock
- private OzgCloudVorgangService vorgangService;
- @Mock
- private AggregationManagerConfiguration config;
+ private AggregationManagerProperties aggregationManagerProperties;
@Mock
private TransformationProperties transformationProperties;
@Mock
private TransformationService transformationService;
@Mock
- private WarehouseRepository repository;
- @Mock
- private VorgangMapper vorgangMapper;
+ private AggregationDataLoaderRegistry loaderRegistry;
@Mock
private static Aggregator aggregator;
@Spy
@@ -72,126 +68,226 @@ class AggregationManagerRunnerTest {
@Nested
class TestRun {
- private final String identifier = LoremIpsum.getInstance().getWords(1);
@BeforeEach
- void mock() {
- when(transformationProperties.getIdentifier()).thenReturn(identifier);
- doNothing().when(runner).runWithTransformation(any(), any());
+ void init() {
+ doNothing().when(runner).runWithConfiguredTransformations();
}
@Test
- void shouldGetAggregationMappings() {
+ void shouldGetScopesWithoutConfiguredTransformations() {
runner.run();
- verify(transformationProperties).getAggregationMappings();
+ verify(runner).getScopesWithoutConfiguredTransformations();
+ }
+
+ @ParameterizedTest
+ @EnumSource
+ void shouldRunWithDefaultTransformationForScopesWithoutConfiguredOnes(AggregationMapping.Scope scope) {
+ doReturn(Stream.of(scope)).when(runner).getScopesWithoutConfiguredTransformations();
+ doNothing().when(runner).runWithDefaultTransformation(any());
+
+ runner.run();
+
+ verify(runner).runWithDefaultTransformation(scope);
}
@Test
- void shouldGetIdentifier() {
+ void shouldRunWithConfiguredTransformations() {
runner.run();
- verify(transformationProperties).getIdentifier();
+ verify(runner).runWithConfiguredTransformations();
}
+ }
+
+ @Nested
+ class TestGetScopesWithoutConfiguredTransformations {
+
+ @ParameterizedTest
+ @EnumSource
+ void shouldCheckForLoader(AggregationMapping.Scope scope) {
+ runner.getScopesWithoutConfiguredTransformations().toList();
- @Nested
- class TestOnAggregationMappingConfigured {
- private final List<AggregationMapping> aggregationMappings = List.of(AggregationMappingTestFactory.create(),
- AggregationMappingTestFactory.create());
+ verify(loaderRegistry).hasLoader(scope);
+ }
- @Mock
- private Transformation firstTransformation;
- @Mock
- private Transformation secondTransformation;
+ @ParameterizedTest
+ @EnumSource
+ void shouldCheckForConfiguredTransformationsWithScope(AggregationMapping.Scope scope) {
+ when(loaderRegistry.hasLoader(any())).thenReturn(true);
- @BeforeEach
- void mock() {
- when(transformationProperties.getAggregationMappings()).thenReturn(aggregationMappings);
- when(transformationService.load(any(), any())).thenReturn(firstTransformation, secondTransformation);
- }
+ runner.getScopesWithoutConfiguredTransformations().toList();
- @Test
- void shouldLoadTransformationForEachAggregationMapping() {
- runner.run();
+ verify(runner).hasNoConfiguredTransformations(scope);
+ }
- aggregationMappings.forEach(mapping -> verify(transformationService).load(identifier, mapping));
- }
+ @ParameterizedTest
+ @EnumSource
+ void shouldReturnScopesWithoutTransformations(AggregationMapping.Scope scope) {
+ doReturn(true).when(runner).hasNoConfiguredTransformations(scope);
+ doReturn(false).when(runner).hasNoConfiguredTransformations(AdditionalMatchers.not(eq(scope)));
+ when(loaderRegistry.hasLoader(any())).thenReturn(true);
- @Test
- void shouldRunWithTransformationForEachTransformation() {
- runner.run();
+ var scopes = runner.getScopesWithoutConfiguredTransformations();
- verify(runner).runWithTransformation(firstTransformation, aggregationMappings.get(0));
- verify(runner).runWithTransformation(secondTransformation, aggregationMappings.get(1));
- }
+ assertThat(scopes).containsExactly(scope);
}
- @Nested
- class TestOnAggregationMappingsNull {
- @Mock
- private Transformation transformation;
+ @Test
+ void shouldOmitScopesWithoutLoaders() {
+ lenient().doReturn(true).when(runner).hasNoConfiguredTransformations(any());
+ when(loaderRegistry.hasLoader(any())).thenReturn(false);
- @BeforeEach
- void mock() {
- when(transformationProperties.getAggregationMappings()).thenReturn(null);
- when(transformationService.load(any(), any())).thenReturn(transformation);
- }
+ var scopes = runner.getScopesWithoutConfiguredTransformations();
- @Test
- void shouldLoadTransformation() {
- runner.run();
+ assertThat(scopes).isEmpty();
+ }
+ }
- verify(transformationService).load(identifier, null);
- }
+ @Nested
+ class TestHasNoConfiguredTransformations {
- @Test
- void shouldCallRunWithDefaultTransformation() {
- runner.run();
+ @Test
+ void shouldGetAggregationMappings() {
+ runner.hasNoConfiguredTransformations(AggregationMapping.Scope.INTERN);
- verify(runner).runWithDefaultTransformation(transformation);
- }
+ verify(transformationProperties).getAggregationMappings();
}
- @Nested
- class TestOnAggregationMappingsEmpty {
- @Mock
- private Transformation transformation;
+ @Test
+ void shouldReturnFalseIfThereAreConfiguredTransformations() {
+ when(transformationProperties.getAggregationMappings()).thenReturn(List.of(AggregationMappingTestFactory.create()));
- @BeforeEach
- void mock() {
- when(transformationProperties.getAggregationMappings()).thenReturn(Collections.emptyList());
- when(transformationService.load(any(), any())).thenReturn(transformation);
- }
+ var hasNoTransformations = runner.hasNoConfiguredTransformations(AggregationMappingTestFactory.SCOPE);
- @Test
- void shouldLoadTransformation() {
- runner.run();
+ assertThat(hasNoTransformations).isFalse();
+ }
- verify(transformationService).load(identifier, null);
- }
+ @Test
+ void shouldReturnTrueIfThereAreNoConfiguredTransformations() {
+ when(transformationProperties.getAggregationMappings()).thenReturn(List.of(AggregationMappingTestFactory.createBuilder().scope(
+ AggregationMapping.Scope.INTERN).build()));
- @Test
- void shouldCallRunWithDefaultTransformation() {
- runner.run();
+ var hasNoTransformations = runner.hasNoConfiguredTransformations(AggregationMapping.Scope.EXTERN);
- verify(runner).runWithDefaultTransformation(transformation);
- }
+ assertThat(hasNoTransformations).isTrue();
}
}
@Nested
class TestRunWithDefaultTransformation {
+ private final AggregationMapping.Scope scope = AggregationMapping.Scope.INTERN;
+ private final String identifier = LoremIpsum.getInstance().getWords(1);
+
@Mock
private Transformation transformation;
+ @BeforeEach
+ void init() {
+ when(transformationProperties.getIdentifier()).thenReturn(identifier);
+ when(transformationService.load(any(), any())).thenReturn(transformation);
+ doReturn(aggregator).when(runner).createAggregator(any());
+ doNothing().when(runner).runWithTransformation(any(), any());
+ }
+
+ @Test
+ void shouldGetIdentifier() {
+ runner.runWithDefaultTransformation(scope);
+
+ verify(transformationProperties).getIdentifier();
+ }
+
+ @Test
+ void shouldLoadTransformation() {
+ runner.runWithDefaultTransformation(scope);
+
+ verify(transformationService).load(identifier, null);
+ }
+
+ @Test
+ void shouldCreateAggregatorForScope() {
+ runner.runWithDefaultTransformation(scope);
+
+ verify(runner).createAggregator(scope);
+ }
+
@Test
- void shouldCallRunWithTransformation() {
+ void shouldRunWithTransformation() {
+ runner.runWithDefaultTransformation(scope);
+
+ verify(runner).runWithTransformation(transformation, aggregator);
+ }
+ }
+
+ @Nested
+ class TestRunWithConfiguredTransformations {
+
+ private final AggregationMapping.Scope firstScope = AggregationMapping.Scope.INTERN;
+ @Mock
+ private Transformation firstTransformation;
+ @Mock
+ private Aggregator firstAggregator;
+
+ private final AggregationMapping.Scope secondScope = AggregationMapping.Scope.EXTERN;
+ @Mock
+ private Transformation secondTransformation;
+ @Mock
+ private Aggregator secondAggregator;
+
+ private final String identifier = LoremIpsum.getInstance().getWords(1);
+ private final List<AggregationMapping> aggregationMappings = List.of(
+ AggregationMappingTestFactory.createBuilder().scope(firstScope).build(),
+ AggregationMappingTestFactory.createBuilder().scope(secondScope).build());
+
+ @BeforeEach
+ void mock() {
+ when(transformationProperties.getAggregationMappings()).thenReturn(aggregationMappings);
+ when(transformationProperties.getIdentifier()).thenReturn(identifier);
+ when(transformationService.load(any(), any())).thenReturn(firstTransformation, secondTransformation);
doNothing().when(runner).runWithTransformation(any(), any());
+ doReturn(firstAggregator).when(runner).createAggregator(firstScope);
+ doReturn(secondAggregator).when(runner).createAggregator(secondScope);
+ when(firstAggregator.withAggregationMapping(any())).thenReturn(firstAggregator);
+ when(secondAggregator.withAggregationMapping(any())).thenReturn(secondAggregator);
+ }
+
+ @Test
+ void shouldGetIdentifier() {
+ runner.runWithConfiguredTransformations();
+
+ verify(transformationProperties, times(2)).getIdentifier();
+ }
- runner.runWithDefaultTransformation(transformation);
+ @Test
+ void shouldLoadTransformationForEachAggregationMapping() {
+ runner.runWithConfiguredTransformations();
+
+ aggregationMappings.forEach(mapping -> verify(transformationService).load(identifier, mapping));
+ }
+
+ @Test
+ void shouldCreateAggregatorForScope() {
+ runner.runWithConfiguredTransformations();
+
+ verify(runner).createAggregator(firstScope);
+ verify(runner).createAggregator(secondScope);
+ }
+
+ @Test
+ void shouldSetAggregationMappingInAggregator() {
+ runner.runWithConfiguredTransformations();
+
+ verify(firstAggregator).withAggregationMapping(aggregationMappings.getFirst());
+ verify(secondAggregator).withAggregationMapping(aggregationMappings.getLast());
+ }
+
+ @Test
+ void shouldRunWithTransformationForEachTransformation() {
+ runner.runWithConfiguredTransformations();
- verify(runner).runWithTransformation(eq(transformation), isNull());
+ verify(runner).runWithTransformation(firstTransformation, firstAggregator);
+ verify(runner).runWithTransformation(secondTransformation, secondAggregator);
}
}
@@ -200,19 +296,18 @@ class AggregationManagerRunnerTest {
@Mock
private Transformation transformation;
- private final AggregationMapping aggregationMapping = AggregationMappingTestFactory.create();
private final ArgumentMatcher<Execution> hasTransformation = execution -> execution.getTransformation().equals(transformation);
@BeforeEach
- void mock() {
- doReturn(aggregator).when(runner).prepareAggregator(any(), any());
+ void init() {
+ when(aggregator.withExecution(any())).thenReturn(aggregator);
}
@Test
- void shouldCallPrepareAggregator() {
+ void shouldSetExecution() {
runWithTransformation();
- verify(runner).prepareAggregator(argThat(hasTransformation), eq(aggregationMapping));
+ verify(aggregator).withExecution(argThat(hasTransformation));
}
@Test
@@ -223,67 +318,67 @@ class AggregationManagerRunnerTest {
}
private void runWithTransformation() {
- runner.runWithTransformation(transformation, aggregationMapping);
+ runner.runWithTransformation(transformation, aggregator);
}
}
@Nested
- class TestPrepareAggregator {
+ class TestCreateAggregator {
- @Mock
- private Execution execution;
- private final AggregationMapping aggregationMapping = AggregationMappingTestFactory.create();
+ private final AggregationMapping.Scope scope = AggregationMapping.Scope.INTERN;
private final int batchSize = RandomUtils.insecure().randomInt();
+ @Mock
+ private AggregationDataLoader loader;
@BeforeEach
- void mock() {
- when(config.getFetchingBatchSize()).thenReturn(batchSize);
- when(aggregator.withExecution(any())).thenReturn(aggregator);
- when(aggregator.withAggregationMapping(any())).thenReturn(aggregator);
+ void init() {
+ when(aggregationManagerProperties.getFetchingBatchSize()).thenReturn(batchSize);
+ when(loaderRegistry.getLoader(any())).thenReturn(loader);
when(aggregator.withBatchSize(anyInt())).thenReturn(aggregator);
+ when(aggregator.withLoader(any())).thenReturn(aggregator);
}
@Test
void shouldGetBatchSize() {
- runner.prepareAggregator(execution, aggregationMapping);
+ runner.createAggregator(scope);
- verify(config).getFetchingBatchSize();
+ verify(aggregationManagerProperties).getFetchingBatchSize();
}
@Test
- void shouldSetExecution() {
- runner.prepareAggregator(execution, aggregationMapping);
+ void shouldSetBatchSize() {
+ runner.createAggregator(scope);
- verify(aggregator).withExecution(execution);
+ verify(aggregator).withBatchSize(batchSize);
}
@Test
- void shouldSetAggregationMapping() {
- runner.prepareAggregator(execution, aggregationMapping);
+ void shouldGetLoaderFromRegistry() {
+ runner.createAggregator(scope);
- verify(aggregator).withAggregationMapping(aggregationMapping);
+ verify(loaderRegistry).getLoader(scope);
}
@Test
- void shouldSetBatchSize() {
- runner.prepareAggregator(execution, aggregationMapping);
+ void shouldSetLoader() {
+ runner.createAggregator(scope);
- verify(aggregator).withBatchSize(batchSize);
+ verify(aggregator).withLoader(loader);
}
@Test
void shouldReturnAggregator() {
- var result = runner.prepareAggregator(execution, aggregationMapping);
+ var created = runner.createAggregator(scope);
- assertThat(result).isEqualTo(aggregator);
+ assertThat(created).isEqualTo(aggregator);
}
}
static class AggregationManagerRunnerImpl extends AggregationManagerRunner {
- public AggregationManagerRunnerImpl(AggregationManagerConfiguration config, TransformationProperties transformationProperties,
- TransformationService transformationService) {
- super(config, transformationProperties, transformationService);
+ public AggregationManagerRunnerImpl(AggregationManagerProperties aggregationManagerProperties, TransformationProperties transformationProperties,
+ TransformationService transformationService, AggregationDataLoaderRegistry loaderRegistry) {
+ super(aggregationManagerProperties, transformationProperties, transformationService, loaderRegistry);
}
@Override
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregationTestFactory.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregationTestFactory.java
new file mode 100644
index 0000000000000000000000000000000000000000..4f99f08685f79c8f7e4df114648e1f6ee63a9b22
--- /dev/null
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregationTestFactory.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation;
+
+import java.util.stream.Stream;
+
+import com.thedeanda.lorem.LoremIpsum;
+
+import de.ozgcloud.aggregation.warehouse.DocumentEntry;
+import de.ozgcloud.aggregation.warehouse.DocumentEntryTestFactory;
+
+public class AggregationTestFactory {
+
+ public static final String AGGREGATION_NAME = LoremIpsum.getInstance().getWords(1);
+ public static final DocumentEntry DOCUMENT_ENTRY = DocumentEntryTestFactory.create();
+
+ public static Aggregation create() {
+ return createBuilder().build();
+ }
+
+ public static Aggregation.AggregationBuilder createBuilder() {
+ return Aggregation.builder()
+ .aggregationName(AGGREGATION_NAME)
+ .documentEntries(Stream.of(DOCUMENT_ENTRY));
+ }
+}
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregatorTest.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregatorTest.java
index 2d66f74b8f51b0eb4abbd382b62aa780b98dce54..684c5df30efeddec375bc3eecf52e9523b87c23e 100644
--- a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregatorTest.java
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregatorTest.java
@@ -82,7 +82,7 @@ class AggregatorTest {
private final AggregationMapping aggregationMapping = AggregationMappingTestFactory.create();
private final FormIdentifier formIdentifier = AggregationMappingTestFactory.FORM_IDENTIFIER;
- private final String collectionName = AggregationMappingTestFactory.NAME;
+ private final String aggregationName = AggregationMappingTestFactory.NAME;
@Test
void shouldReturnSelf() {
@@ -99,10 +99,10 @@ class AggregatorTest {
}
@Test
- void shouldSetCollectionName() {
+ void shouldSetAggregationName() {
var result = aggregator.withAggregationMapping(aggregationMapping);
- assertThat(ReflectionTestUtils.getField(result, "collectionName")).isEqualTo(collectionName);
+ assertThat(ReflectionTestUtils.getField(result, "aggregationName")).isEqualTo(aggregationName);
}
}
@@ -124,10 +124,10 @@ class AggregatorTest {
}
@Test
- void shouldNotSetCollectionName() {
+ void shouldNotSetAggregationName() {
var result = aggregator.withAggregationMapping(null);
- assertThat(ReflectionTestUtils.getField(result, "collectionName")).isEqualTo(DocumentEntry.COLLECTION);
+ assertThat(ReflectionTestUtils.getField(result, "aggregationName")).isNull();
}
}
}
@@ -152,6 +152,27 @@ class AggregatorTest {
}
}
+ @Nested
+ class TestWithLoader {
+
+ @Mock
+ private AggregationDataLoader loader;
+
+ @Test
+ void shouldReturnSelf() {
+ var result = aggregator.withLoader(loader);
+
+ assertThat(result).isSameAs(aggregator);
+ }
+
+ @Test
+ void shouldSetLoader() {
+ aggregator.withLoader(loader);
+
+ assertThat(ReflectionTestUtils.getField(aggregator, "loader")).isSameAs(loader);
+ }
+ }
+
@Nested
class TestAggregate {
@Mock
@@ -165,7 +186,7 @@ class AggregatorTest {
void setUp() {
doReturn(Stream.of(batchOfVorgaenge)).when(aggregator).extractBatchesOfVorgaengeFromDataSource();
doReturn(Stream.of(batchOfDeletedVorgaenge)).when(aggregator).extractBatchesOfDeletedVorgaengeFromDataSource();
- doNothing().when(aggregator).loadVorgaengeIntoRepository(any());
+ doNothing().when(aggregator).loadVorgaenge(any());
}
@Test
@@ -186,7 +207,7 @@ class AggregatorTest {
void shouldLoadVorgaengeIntoRepository() {
aggregator.aggregate();
- verify(aggregator).loadVorgaengeIntoRepository(batchStreamCaptor.capture());
+ verify(aggregator).loadVorgaenge(batchStreamCaptor.capture());
assertThat(batchStreamCaptor.getValue()).containsExactly(batchOfVorgaenge, batchOfDeletedVorgaenge);
}
}
@@ -485,67 +506,55 @@ class AggregatorTest {
}
@Nested
- class TestLoadVorgaengeIntoRepository {
+ class TestLoadVorgaenge {
@Mock
private Execution execution;
@Mock
private Batch batch;
+ @Mock
+ private AggregationDataLoader loader;
private final DocumentEntry documentEntry = DocumentEntryTestFactory.create();
private final AggregationMapping aggregationMapping = AggregationMappingTestFactory.create();
@Captor
- private ArgumentCaptor<List<DocumentEntry>> documentEntriesCaptor;
+ private ArgumentCaptor<Aggregation> aggregationCaptor;
@BeforeEach
void init() {
- aggregator = aggregator.withAggregationMapping(aggregationMapping);
- doReturn(List.of(documentEntry)).when(aggregator).transformBatchToDocumentEntries(any());
- doNothing().when(aggregator).loadDocumentEntriesIntoRepository(any());
+ aggregator = aggregator.withAggregationMapping(aggregationMapping).withLoader(loader);
}
@Test
- void shouldDropCollection() {
- loadVorgaengeIntoRepository();
+ void shouldLoadIntoTarget() {
+ loadVorgaenge();
- verify(repository).clearCollection(AggregationMappingTestFactory.NAME);
+ verify(loader).loadIntoTarget(any(Aggregation.class));
}
@Test
void shouldTransform() {
- loadVorgaengeIntoRepository();
+ doReturn(Stream.of(documentEntry)).when(aggregator).transformBatchToDocumentEntries(any());
+ loadVorgaenge();
+
+ getArgumentOfLoadIntoTarget().documentEntries().toList();
verify(aggregator).transformBatchToDocumentEntries(batch);
}
@Test
- void shouldLoadIntoRepository() {
- loadVorgaengeIntoRepository();
+ void shouldSetAggregationName() {
+ loadVorgaenge();
- verify(aggregator).loadDocumentEntriesIntoRepository(documentEntriesCaptor.capture());
- assertThat(documentEntriesCaptor.getValue()).containsExactly(documentEntry);
+ assertThat(getArgumentOfLoadIntoTarget().aggregationName()).isEqualTo(AggregationMappingTestFactory.NAME);
}
- private void loadVorgaengeIntoRepository() {
- aggregator.loadVorgaengeIntoRepository(Stream.of(batch));
- }
- }
-
- @Nested
- class TestLoadDocumentEntriesIntoRepository {
-
- private final List<DocumentEntry> documentEntries = List.of(DocumentEntryTestFactory.create());
- private final AggregationMapping aggregationMapping = AggregationMappingTestFactory.create();
-
- @BeforeEach
- void init() {
- aggregator = aggregator.withAggregationMapping(aggregationMapping);
+ private Aggregation getArgumentOfLoadIntoTarget() {
+ verify(loader).loadIntoTarget(aggregationCaptor.capture());
+ return aggregationCaptor.getValue();
}
- @Test
- void shouldSaveDocumentEntriesInCollection() {
- aggregator.loadDocumentEntriesIntoRepository(documentEntries);
-
- verify(repository).saveAllInCollection(documentEntries, AggregationMappingTestFactory.NAME);
+ private void loadVorgaenge() {
+ aggregator.loadVorgaenge(Stream.of(batch));
}
}
}
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteLoaderTest.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteLoaderTest.java
new file mode 100644
index 0000000000000000000000000000000000000000..3c7b0b029486e68ae01ee3186ea5e0ae3a99fbf7
--- /dev/null
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteLoaderTest.java
@@ -0,0 +1,139 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.extern;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Spy;
+
+import de.ozgcloud.aggregation.Aggregation;
+import de.ozgcloud.aggregation.AggregationTestFactory;
+import de.ozgcloud.aggregation.transformation.AggregationMapping;
+import de.ozgcloud.common.errorhandling.TechnicalException;
+import lombok.SneakyThrows;
+
+class AggregationDataRemoteLoaderTest {
+
+ @Mock
+ private AggregationDataRemoteService remoteService;
+ @Spy
+ @InjectMocks
+ private AggregationDataRemoteLoader loader;
+
+ @Nested
+ class TestLoadIntoTarget {
+
+ private final Aggregation aggregation = AggregationTestFactory.create();
+ @Mock
+ private Future<Void> future;
+
+ @BeforeEach
+ void init() {
+ when(remoteService.sendAggregationData(any())).thenReturn(future);
+ }
+
+ @Test
+ @SneakyThrows
+ void shouldSendAggregationData() {
+ loader.loadIntoTarget(aggregation);
+
+ verify(remoteService).sendAggregationData(aggregation);
+ }
+
+ @Test
+ @SneakyThrows
+ void shouldGetFromFuture() {
+ loader.loadIntoTarget(aggregation);
+
+ verify(future).get();
+ }
+
+ @Nested
+ class TestOnInterruptedException {
+
+ private final InterruptedException exception = new InterruptedException();
+
+ @BeforeEach
+ @SneakyThrows
+ void mock() {
+ when(future.get()).thenThrow(exception);
+ }
+
+ @Test
+ void shouldThrowTechnicalException() {
+ assertThrows(TechnicalException.class, () -> loader.loadIntoTarget(aggregation));
+ }
+
+ @Test
+ void shouldInterruptThread() {
+ try {
+ loader.loadIntoTarget(aggregation);
+ } catch (TechnicalException e) {
+ // expected
+ }
+
+ assertThat(Thread.currentThread().isInterrupted()).isTrue();
+ }
+ }
+
+ @Nested
+ class TestOnExecutionException {
+
+ private final ExecutionException exception = new ExecutionException(new Exception());
+
+ @BeforeEach
+ @SneakyThrows
+ void mock() {
+ when(future.get()).thenThrow(exception);
+ }
+
+ @Test
+ void shouldThrowTechnicalException() {
+ assertThrows(TechnicalException.class, () -> loader.loadIntoTarget(aggregation));
+ }
+ }
+ }
+
+ @Nested
+ class TestGetScope {
+
+ @Test
+ void shouldReturnExternScope() {
+ var scope = loader.getScope();
+
+ assertThat(scope).isEqualTo(AggregationMapping.Scope.EXTERN);
+ }
+ }
+}
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteServiceTest.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteServiceTest.java
new file mode 100644
index 0000000000000000000000000000000000000000..8d344615e8ba3a5f7876fd24f4a5a1dcaeb04f12
--- /dev/null
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteServiceTest.java
@@ -0,0 +1,479 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.extern;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Spy;
+
+import com.thedeanda.lorem.LoremIpsum;
+
+import de.ozgcloud.aggregation.Aggregation;
+import de.ozgcloud.aggregation.AggregationManagerProperties;
+import de.ozgcloud.aggregation.AggregationTestFactory;
+import de.ozgcloud.aggregation.data.AggregationDataServiceGrpc;
+import de.ozgcloud.aggregation.data.GrpcAggregationData;
+import de.ozgcloud.aggregation.data.GrpcSendAggregationDataRequest;
+import de.ozgcloud.aggregation.extern.AggregationDataRemoteService.SendAggregationDataResponseObserver;
+import de.ozgcloud.aggregation.warehouse.DocumentEntry;
+import de.ozgcloud.common.test.ReflectionTestUtils;
+import io.grpc.stub.ClientCallStreamObserver;
+import lombok.SneakyThrows;
+
+class AggregationDataRemoteServiceTest {
+
+ @Mock
+ private AggregationDataServiceGrpc.AggregationDataServiceStub serviceStub;
+ @Mock
+ private AggregationManagerProperties properties;
+ @Mock
+ private GrpcAggregationDataMapper grpcAggregationDataMapper;
+ @InjectMocks
+ @Spy
+ private AggregationDataRemoteService service;
+
+ @Nested
+ class TestSendAggregationData {
+
+ private final Aggregation aggregation = AggregationTestFactory.create();
+
+ @Mock
+ private SendAggregationDataResponseObserver responseObserver;
+ @Mock
+ private CompletableFuture<Void> responseFuture;
+
+ @BeforeEach
+ void init() {
+ doReturn(responseObserver).when(service).buildSendAggregationDataResponseObserver(any());
+ when(responseObserver.getResponseFuture()).thenReturn(responseFuture);
+ }
+
+ @Test
+ void shouldBuildResponseObserver() {
+ sendAggregationData();
+
+ verify(service).buildSendAggregationDataResponseObserver(aggregation);
+ }
+
+ @Test
+ void shouldCallServiceStub() {
+ sendAggregationData();
+
+ verify(serviceStub).sendAggregationData(responseObserver);
+ }
+
+ @Test
+ void shouldReturnFutureFromResponseObserver() {
+ var future = sendAggregationData();
+
+ assertThat(future).isSameAs(responseFuture);
+ }
+
+ private Future<Void> sendAggregationData() {
+ return service.sendAggregationData(aggregation);
+ }
+ }
+
+ @Nested
+ class TestBuildSendAggregationDataResponseObserver {
+
+ private static final int BATCH_SIZE = new Random().nextInt(100);
+ @Captor
+ private ArgumentCaptor<Stream<DocumentEntry>> documentEntriesCaptor;
+
+ @BeforeEach
+ void init() {
+ doReturn(Stream.of(GrpcAggregationDataTestFactory.create())).when(service).toGrpcAggregationDataStream(any());
+ when(properties.getFetchingBatchSize()).thenReturn(BATCH_SIZE);
+ }
+
+ @Test
+ void shouldGetMandant() {
+ service.buildSendAggregationDataResponseObserver(AggregationTestFactory.create());
+
+ verify(properties).getMandant();
+ }
+
+ @Test
+ void shouldCreateGrpcAggregationDataStream() {
+ service.buildSendAggregationDataResponseObserver(AggregationTestFactory.create());
+
+ verify(service).toGrpcAggregationDataStream(documentEntriesCaptor.capture());
+ assertThat(documentEntriesCaptor.getValue()).containsExactly(AggregationTestFactory.DOCUMENT_ENTRY);
+ }
+
+ @Test
+ void shouldGetBatchSize() {
+ service.buildSendAggregationDataResponseObserver(AggregationTestFactory.create());
+
+ verify(properties).getFetchingBatchSize();
+ }
+
+ @Test
+ void shouldReturnResponseObserver() {
+ var builtResponseObserver = service.buildSendAggregationDataResponseObserver(AggregationTestFactory.create());
+
+ assertThat(builtResponseObserver).isNotNull();
+ }
+
+ @Nested
+ class TestBuiltResponseObserver {
+
+ private static final String MANDANT = LoremIpsum.getInstance().getWords(1);
+
+ @BeforeEach
+ void init() {
+ when(properties.getMandant()).thenReturn(MANDANT);
+ }
+
+ @Test
+ void shouldHaveBatchSize() {
+ var builtResponseObserver = service.buildSendAggregationDataResponseObserver(AggregationTestFactory.create());
+
+ assertThat(getBatchSize(builtResponseObserver)).isEqualTo(BATCH_SIZE);
+ }
+
+ @Test
+ void shouldHaveMandant() {
+ var builtResponseObserver = service.buildSendAggregationDataResponseObserver(AggregationTestFactory.create());
+
+ assertThat(getRequestData(builtResponseObserver).mandant()).isEqualTo(MANDANT);
+ }
+
+ @Test
+ void shouldHaveAggregationName() {
+ var builtResponseObserver = service.buildSendAggregationDataResponseObserver(AggregationTestFactory.create());
+
+ assertThat(getRequestData(builtResponseObserver).aggregationName()).isEqualTo(AggregationTestFactory.AGGREGATION_NAME);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void shouldHaveGrpcAggregationData() {
+ var builtResponseObserver = service.buildSendAggregationDataResponseObserver(AggregationTestFactory.create());
+
+ var aggregationData = IteratorUtils.toList(getRequestData(builtResponseObserver).aggregationDataIterator());
+ assertThat(aggregationData).containsExactly(GrpcAggregationDataTestFactory.create());
+ }
+
+ private int getBatchSize(SendAggregationDataResponseObserver responseObserver) {
+ return ReflectionTestUtils.getField(responseObserver, "batchSize", Integer.class);
+ }
+
+ private AggregationDataRemoteService.RequestData getRequestData(SendAggregationDataResponseObserver responseObserver) {
+ return ReflectionTestUtils.getField(responseObserver, "requestData", AggregationDataRemoteService.RequestData.class);
+ }
+ }
+ }
+
+ @Nested
+ class TestToGrpcAggregationDataStream {
+
+ @BeforeEach
+ void init() {
+ when(grpcAggregationDataMapper.toGrpcAggregationData(AggregationTestFactory.DOCUMENT_ENTRY)).thenReturn(
+ GrpcAggregationDataTestFactory.create());
+ }
+
+ @Test
+ void shouldMapDocumentEntry() {
+ service.toGrpcAggregationDataStream(Stream.of(AggregationTestFactory.DOCUMENT_ENTRY)).toList();
+
+ verify(grpcAggregationDataMapper).toGrpcAggregationData(AggregationTestFactory.DOCUMENT_ENTRY);
+ }
+
+ @Test
+ void shouldReturnMappedDocumentEntries() {
+ var mappedDocumentEntries = service.toGrpcAggregationDataStream(Stream.of(AggregationTestFactory.DOCUMENT_ENTRY));
+
+ assertThat(mappedDocumentEntries.toList()).containsExactly(GrpcAggregationDataTestFactory.create());
+ }
+ }
+
+ @Nested
+ class SendAggregationDataResponseObserverTest {
+
+ private static final int BATCH_SIZE = 2;
+
+ @Mock
+ private AggregationDataRemoteService.RequestData requestData;
+ @Spy
+ private SendAggregationDataResponseObserver responseObserver = new SendAggregationDataResponseObserver(BATCH_SIZE, requestData);
+
+ @Nested
+ class TestBeforeStart {
+
+ @Mock
+ private ClientCallStreamObserver<GrpcSendAggregationDataRequest> requestObserver;
+ @Mock
+ private AggregationDataRemoteService.SendAggregationDataOnReadyHandler onReadyHandler;
+
+ @BeforeEach
+ void init() {
+ doReturn(onReadyHandler).when(responseObserver).buildOnReadyHandler(any());
+ }
+
+ @Test
+ void shouldBuildOnReadyHandler() {
+ responseObserver.beforeStart(requestObserver);
+
+ verify(responseObserver).buildOnReadyHandler(requestObserver);
+ }
+
+ @Test
+ void shouldSetOnReadyHandler() {
+ responseObserver.beforeStart(requestObserver);
+
+ verify(requestObserver).setOnReadyHandler(onReadyHandler);
+ }
+ }
+
+ @Nested
+ class TestOnError {
+
+ @Test
+ @SneakyThrows
+ void shouldCompleteExceptionally() {
+ var error = new Throwable();
+
+ responseObserver.onError(error);
+
+ assertThatException().isThrownBy(() -> responseObserver.getResponseFuture().get()).withCause(error);
+ }
+ }
+
+ @Nested
+ class TestOnCompleted {
+
+ @Test
+ void shouldCompleteFuture() {
+ responseObserver.onCompleted();
+
+ assertThat(responseObserver.getResponseFuture().isDone()).isTrue();
+ }
+ }
+ }
+
+ @Nested
+ class SendAggregationDataOnReadyHandlerTest {
+
+ private static final int BATCH_SIZE = 2;
+ private static final String MANDANT = LoremIpsum.getInstance().getWords(1);
+
+ @Mock
+ private ClientCallStreamObserver<GrpcSendAggregationDataRequest> requestObserver;
+
+ @Nested
+ class TestRun {
+
+ @Mock
+ private GrpcSendAggregationDataRequest request;
+
+ @Test
+ void shouldCheckIfStreamIsReady() {
+ createOnReadyHandler(createAggregationData(1)).run();
+
+ verify(requestObserver).isReady();
+ }
+
+ @Test
+ void shouldNotCallOnNextWhenStreamIsNotReady() {
+ when(requestObserver.isReady()).thenReturn(false);
+
+ createOnReadyHandler(createAggregationData(1)).run();
+
+ verify(requestObserver, never()).onNext(any());
+ }
+
+ @Test
+ void shouldNotCallOnNextWhenHasNoMoreData() {
+ when(requestObserver.isReady()).thenReturn(true);
+
+ createOnReadyHandler(createAggregationData(0)).run();
+
+ verify(requestObserver, never()).onNext(any());
+ }
+
+ @Test
+ void shouldBuildRequest() {
+ when(requestObserver.isReady()).thenReturn(true);
+ var onReadyHandler = spy(createOnReadyHandler(createAggregationData(1)));
+
+ onReadyHandler.run();
+
+ verify(onReadyHandler).buildRequest();
+ }
+
+ @Test
+ void shouldCallOnNextWithBuiltRequest() {
+ when(requestObserver.isReady()).thenReturn(true).thenReturn(false);
+ var onReadyHandler = spy(createOnReadyHandler(createAggregationData(1)));
+ doReturn(request).when(onReadyHandler).buildRequest();
+
+ onReadyHandler.run();
+
+ verify(requestObserver).onNext(request);
+ }
+
+ @Test
+ void shouldCallOnNextUntilAllDataWasSent() {
+ when(requestObserver.isReady()).thenReturn(true);
+
+ createOnReadyHandler(createAggregationData(BATCH_SIZE + 1)).run();
+
+ verify(requestObserver, times(2)).onNext(any());
+ }
+
+ @Test
+ void shouldCompleteRequest() {
+ when(requestObserver.isReady()).thenReturn(true);
+
+ createOnReadyHandler(createAggregationData(BATCH_SIZE + 1)).run();
+
+ verify(requestObserver).onCompleted();
+ }
+
+ @Test
+ void shouldNotCompleteRequestIfNotAllDataWasSent() {
+ when(requestObserver.isReady()).thenReturn(true).thenReturn(false);
+
+ createOnReadyHandler(createAggregationData(BATCH_SIZE + 1)).run();
+
+ verify(requestObserver, never()).onCompleted();
+ }
+ }
+
+ @Nested
+ class TestBuildRequest {
+
+ @Test
+ void shouldSetName() {
+ var request = createOnReadyHandler().buildRequest();
+
+ assertThat(request.getName()).isEqualTo(AggregationTestFactory.AGGREGATION_NAME);
+ }
+
+ @Test
+ void shouldSetMandant() {
+ var request = createOnReadyHandler().buildRequest();
+
+ assertThat(request.getMandant()).isEqualTo(MANDANT);
+ }
+
+ @Nested
+ class OnHasNoAggregationData {
+
+ @Test
+ void shouldAggregationDataBeEmpty() {
+ var request = createOnReadyHandler(List.of()).buildRequest();
+
+ assertThat(request.getAggregationDataCount()).isEqualTo(0);
+ }
+ }
+
+ @Nested
+ class OnHasLessAggregationDataThenBatchSize {
+
+ @Test
+ void shouldAddAllAggregationData() {
+ var aggregationData = createAggregationData(BATCH_SIZE - 1);
+
+ var request = createOnReadyHandler(aggregationData).buildRequest();
+
+ assertThat(request.getAggregationDataList()).hasSameElementsAs(aggregationData);
+ }
+ }
+
+ @Nested
+ class OnHasBatchSizeOfAggregationData {
+
+ @Test
+ void shouldAddAllAggregationData() {
+ var aggregationData = createAggregationData(BATCH_SIZE);
+
+ var request = createOnReadyHandler(aggregationData).buildRequest();
+
+ assertThat(request.getAggregationDataList()).hasSameElementsAs(aggregationData);
+ }
+ }
+
+ @Nested
+ class OnHasMoreAggregationDataThenBatchSize {
+
+ @Test
+ void shouldHaveExactlyBatchSizeOfAggregationData() {
+ var aggregationData = createAggregationData(BATCH_SIZE + 1);
+
+ var request = createOnReadyHandler(aggregationData).buildRequest();
+
+ assertThat(request.getAggregationDataCount()).isEqualTo(BATCH_SIZE);
+ }
+
+ @Test
+ void shouldContainOnlyFirstBatchSizeOfElements() {
+ var aggregationData = createAggregationData(BATCH_SIZE + 1);
+
+ var request = createOnReadyHandler(aggregationData).buildRequest();
+
+ assertThat(request.getAggregationDataList()).containsExactlyElementsOf(aggregationData.subList(0, BATCH_SIZE));
+ }
+ }
+ }
+
+ private List<GrpcAggregationData> createAggregationData(int count) {
+ return IntStream.range(1, count + 1)
+ .mapToObj(idx -> GrpcAggregationDataTestFactory.createBuilder().setVorgangName("vorgang " + idx).build())
+ .toList();
+ }
+
+ private AggregationDataRemoteService.SendAggregationDataOnReadyHandler createOnReadyHandler() {
+ return createOnReadyHandler(List.of());
+ }
+
+ private AggregationDataRemoteService.SendAggregationDataOnReadyHandler createOnReadyHandler(List<GrpcAggregationData> aggregationData) {
+ return new AggregationDataRemoteService.SendAggregationDataOnReadyHandler(requestObserver, BATCH_SIZE,
+ AggregationDataRemoteService.RequestData.builder()
+ .mandant(MANDANT)
+ .aggregationName(AggregationTestFactory.AGGREGATION_NAME)
+ .aggregationDataIterator(aggregationData.iterator())
+ .build());
+ }
+ }
+}
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/GrpcAggregationDataMapperTest.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/GrpcAggregationDataMapperTest.java
new file mode 100644
index 0000000000000000000000000000000000000000..7f552a7853b964f1e4ed94937b2618cbfdb145ec
--- /dev/null
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/GrpcAggregationDataMapperTest.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.extern;
+
+import static org.assertj.core.api.Assertions.*;
+
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.mapstruct.factory.Mappers;
+
+import de.ozgcloud.aggregation.data.GrpcObject;
+import de.ozgcloud.aggregation.warehouse.DocumentEntryTestFactory;
+
+class GrpcAggregationDataMapperTest {
+
+ private final GrpcAggregationDataMapper mapper = Mappers.getMapper(GrpcAggregationDataMapper.class);
+
+ @Nested
+ class TestToGrpcAggregationData {
+
+ @Test
+ void shouldMapId() {
+ var grpcData = mapper.toGrpcAggregationData(DocumentEntryTestFactory.create());
+
+ assertThat(grpcData.getId()).isEqualTo(DocumentEntryTestFactory.ID);
+ }
+
+ @Test
+ void shouldMapStatus() {
+ var grpcData = mapper.toGrpcAggregationData(DocumentEntryTestFactory.create());
+
+ assertThat(grpcData.getStatus()).isEqualTo(DocumentEntryTestFactory.STATUS);
+ }
+
+ @Test
+ void shouldMapEingangDatum() {
+ var grpcData = mapper.toGrpcAggregationData(DocumentEntryTestFactory.create());
+
+ assertThat(grpcData.getEingangDatum()).isEqualTo(DocumentEntryTestFactory.EINGANGSDATUM.toString());
+ }
+
+ @Test
+ void shouldMapVorgangName() {
+ var grpcData = mapper.toGrpcAggregationData(DocumentEntryTestFactory.create());
+
+ assertThat(grpcData.getVorgangName()).isEqualTo(DocumentEntryTestFactory.VORGANG_NAME);
+ }
+
+ @Test
+ void shouldHaveDefaultPayload() {
+ var grpcData = mapper.toGrpcAggregationData(DocumentEntryTestFactory.create());
+
+ assertThat(grpcData.getPayload()).isEqualTo(GrpcObject.getDefaultInstance());
+ }
+ }
+}
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/GrpcAggregationDataTestFactory.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/GrpcAggregationDataTestFactory.java
new file mode 100644
index 0000000000000000000000000000000000000000..d3ba13a1f713ea51fb2b6e05f1dd18848b6a6619
--- /dev/null
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/GrpcAggregationDataTestFactory.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.extern;
+
+import de.ozgcloud.aggregation.data.GrpcAggregationData;
+import de.ozgcloud.aggregation.data.GrpcAggregationData.Builder;
+import de.ozgcloud.aggregation.warehouse.DocumentEntryTestFactory;
+
+public class GrpcAggregationDataTestFactory {
+
+ public static GrpcAggregationData create() {
+ return createBuilder().build();
+ }
+
+ public static Builder createBuilder() {
+ return GrpcAggregationData.newBuilder()
+ .setId(DocumentEntryTestFactory.ID)
+ .setStatus(DocumentEntryTestFactory.STATUS)
+ .setEingangDatum(DocumentEntryTestFactory.EINGANGSDATUM.toString())
+ .setVorgangName(DocumentEntryTestFactory.VORGANG_NAME);
+ }
+}
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/SpringContextITCase.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/SpringContextITCase.java
new file mode 100644
index 0000000000000000000000000000000000000000..b5cd918ad5550a6da6d17ffee03d5d0c613c8777
--- /dev/null
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/SpringContextITCase.java
@@ -0,0 +1,88 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.extern;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.TestPropertySource;
+
+import de.ozgcloud.aggregation.AggregationDataLoader;
+import de.ozgcloud.aggregation.AggregationDataLoaderRegistry;
+import de.ozgcloud.aggregation.transformation.AggregationMapping;
+import de.ozgcloud.common.test.ITCase;
+
+class SpringContextITCase {
+
+ @Nested
+ @ITCase
+ @TestPropertySource(properties = { "grpc.client.aggregation-manager.address=static://127.0.0.1:9090" })
+ class OnAggregationManagerAddressSet {
+
+ @Autowired
+ private List<AggregationDataLoader> loaders;
+ @Autowired
+ private AggregationDataLoaderRegistry registry;
+
+ @Test
+ void shouldHaveOneLoader() {
+ assertThat(getLoadersWithScopeExtern(loaders)).singleElement().isInstanceOf(AggregationDataRemoteLoader.class);
+ }
+
+ @Test
+ void shouldRegister() {
+ assertThat(registry.hasLoader(AggregationMapping.Scope.EXTERN)).isTrue();
+ }
+ }
+
+ @Nested
+ @ITCase
+ class OnAggregationManagerAddressNotSet {
+
+ @Autowired
+ private List<AggregationDataLoader> loaders;
+ @Autowired
+ private AggregationDataLoaderRegistry registry;
+
+ @Test
+ void shouldNotHaveLoader() {
+ assertThat(getLoadersWithScopeExtern(loaders)).isEmpty();
+ }
+
+ @Test
+ void shouldNotRegister() {
+ assertThat(registry.hasLoader(AggregationMapping.Scope.EXTERN)).isFalse();
+ }
+ }
+
+ private List<AggregationDataLoader> getLoadersWithScopeExtern(List<AggregationDataLoader> loaders) {
+ return loaders.stream()
+ .filter(loader -> loader.getScope() == AggregationMapping.Scope.EXTERN)
+ .toList();
+ }
+}
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/transformation/AggregationMappingTestFactory.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/transformation/AggregationMappingTestFactory.java
index 44a5c25ce128edf4625a5b95ab27d772bb621c46..8986c7903621e7d0b8fa5282c0a4b36111c6f26d 100644
--- a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/transformation/AggregationMappingTestFactory.java
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/transformation/AggregationMappingTestFactory.java
@@ -34,6 +34,7 @@ public class AggregationMappingTestFactory {
public static final FormIdentifier FORM_IDENTIFIER = FormIdentifierTestFactory.create();
public static final FieldMapping MAPPING = FieldMappingTestFactory.create();
public static final String NAME = LoremIpsum.getInstance().getWords(1);
+ public static final AggregationMapping.Scope SCOPE = AggregationMapping.Scope.INTERN;
public static AggregationMapping create() {
return createBuilder().build();
@@ -43,6 +44,7 @@ public class AggregationMappingTestFactory {
return AggregationMapping.builder()
.formIdentifier(FORM_IDENTIFIER)
.fieldMapping(MAPPING)
- .name(NAME);
+ .name(NAME)
+ .scope(SCOPE);
}
}
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/warehouse/AggregationDataWarehouseLoaderTest.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/warehouse/AggregationDataWarehouseLoaderTest.java
new file mode 100644
index 0000000000000000000000000000000000000000..13488366a7a590b2f704f80814f332762e8af894
--- /dev/null
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/warehouse/AggregationDataWarehouseLoaderTest.java
@@ -0,0 +1,122 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.warehouse;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Spy;
+
+import com.thedeanda.lorem.LoremIpsum;
+
+import de.ozgcloud.aggregation.Aggregation;
+import de.ozgcloud.aggregation.AggregationTestFactory;
+import de.ozgcloud.aggregation.transformation.AggregationMapping;
+
+class AggregationDataWarehouseLoaderTest {
+
+ @Mock
+ private WarehouseRepository repository;
+ @Spy
+ @InjectMocks
+ private AggregationDataWarehouseLoader loader;
+
+ @Nested
+ class TestLoadIntoTarget {
+
+ private static final String COLLECTION_NAME = LoremIpsum.getInstance().getWords(1);
+ private final Aggregation aggregation = AggregationTestFactory.create();
+
+ @Captor
+ private ArgumentCaptor<Stream<DocumentEntry>> documentEntriesCaptor;
+
+ @BeforeEach
+ void init() {
+ doReturn(COLLECTION_NAME).when(loader).getCollectionName(aggregation);
+ }
+
+ @Test
+ void shouldGetCollectionName() {
+ loader.loadIntoTarget(aggregation);
+
+ verify(loader).getCollectionName(aggregation);
+ }
+
+ @Test
+ void shouldClearCollection() {
+ loader.loadIntoTarget(aggregation);
+
+ verify(repository).clearCollection(COLLECTION_NAME);
+ }
+
+ @Test
+ void shouldSaveDocumentEntriesInCollection() {
+ loader.loadIntoTarget(aggregation);
+
+ verify(repository).saveAllInCollection(documentEntriesCaptor.capture(), eq(COLLECTION_NAME));
+ assertThat(documentEntriesCaptor.getValue()).containsExactly(AggregationTestFactory.DOCUMENT_ENTRY);
+ }
+ }
+
+ @Nested
+ class TestGetCollectionName {
+
+ @Test
+ void shouldReturnAggregationName() {
+ var collectionName = loader.getCollectionName(AggregationTestFactory.create());
+
+ assertThat(collectionName).isEqualTo(AggregationTestFactory.AGGREGATION_NAME);
+ }
+
+ @ParameterizedTest
+ @NullAndEmptySource
+ void shouldReturnCollectionNameFromDocumentEntry(String aggregationName) {
+ var collectionName = loader.getCollectionName(AggregationTestFactory.createBuilder().aggregationName(aggregationName).build());
+
+ assertThat(collectionName).isEqualTo(DocumentEntry.COLLECTION);
+ }
+ }
+
+ @Nested
+ class TestGetScope {
+
+ @Test
+ void shouldReturnInternScope() {
+ var scope = loader.getScope();
+
+ assertThat(scope).isEqualTo(AggregationMapping.Scope.INTERN);
+ }
+ }
+}
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepositoryImplTest.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepositoryImplTest.java
index 194a372bc2330c39c339baab737c0bf879fba26e..171257c6467a613678a8ad73542b16308ca77165 100644
--- a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepositoryImplTest.java
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepositoryImplTest.java
@@ -1,7 +1,6 @@
package de.ozgcloud.aggregation.warehouse;
import static org.assertj.core.api.Assertions.*;
-import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
import java.util.List;
@@ -33,20 +32,10 @@ class CustomWarehouseRepositoryImplTest {
@Test
void shouldCallSaveInCollection() {
- repository.saveAllInCollection(documentEntries, collectionName);
+ repository.saveAllInCollection(documentEntries.stream(), collectionName);
verify(repository).saveInCollection(documentEntry, collectionName);
}
-
- @Test
- void shouldReturnSavedDocumentEntry() {
- var savedDocumentEntry = DocumentEntryTestFactory.create();
- doReturn(savedDocumentEntry).when(repository).saveInCollection(any(), any());
-
- var returnedDocumentEntries = repository.saveAllInCollection(documentEntries, collectionName);
-
- assertThat(returnedDocumentEntries).containsExactly(savedDocumentEntry);
- }
}
@Nested
diff --git a/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/warehouse/SpringContextITCase.java b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/warehouse/SpringContextITCase.java
new file mode 100644
index 0000000000000000000000000000000000000000..5504e78722cfdc81f172c125461de833b5ef3514
--- /dev/null
+++ b/aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/warehouse/SpringContextITCase.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.warehouse;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import de.ozgcloud.aggregation.AggregationDataLoader;
+import de.ozgcloud.aggregation.AggregationDataLoaderRegistry;
+import de.ozgcloud.aggregation.transformation.AggregationMapping;
+import de.ozgcloud.common.test.ITCase;
+
+@ITCase
+class SpringContextITCase {
+
+ @Autowired
+ private List<AggregationDataLoader> loaders;
+ @Autowired
+ private AggregationDataLoaderRegistry registry;
+
+ @Test
+ void shouldHaveLoader() {
+ assertThat(getLoadersWithScopeIntern(loaders)).singleElement().isInstanceOf(AggregationDataWarehouseLoader.class);
+ }
+
+ @Test
+ void shouldRegister() {
+ assertThat(registry.hasLoader(AggregationMapping.Scope.INTERN)).isTrue();
+ }
+
+ private List<AggregationDataLoader> getLoadersWithScopeIntern(List<AggregationDataLoader> loaders) {
+ return loaders.stream()
+ .filter(loader -> loader.getScope() == AggregationMapping.Scope.INTERN)
+ .toList();
+ }
+}
diff --git a/aggregation-manager-job/src/test/resources/application-itcase.yml b/aggregation-manager-job/src/test/resources/application-itcase.yml
index b370e11fc17d7a792a21b82b7cb7bb69357f3710..c8eef97bb8ab02bde5fc7aed63f23e383d5aea3c 100644
--- a/aggregation-manager-job/src/test/resources/application-itcase.yml
+++ b/aggregation-manager-job/src/test/resources/application-itcase.yml
@@ -10,14 +10,11 @@ spring:
banner-mode: "off"
ozgcloud:
- vorgang-manager:
- address: static://127.0.0.1:9090
- negotiationType: PLAINTEXT
command:
line:
runner:
enabled: false
-
-aggregation-manager:
+ mandant: "Landeshauptstadt Kiel"
fetching-batch-size: 2
- identifier: "id.value"
+ aggregation:
+ identifier: "id.value"