feat: mark the transfer as completed on EDR expiration

eclipse-tractusx · Oct 31, 2023 · c77707c · c77707c
1 parent b0cde40
commit c77707c
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 4 deletions.
diff --git a/core/edr-core/build.gradle.kts b/core/edr-core/build.gradle.kts
@@ -28,7 +28,6 @@ dependencies {
     implementation(project(":spi:edr-spi"))
     implementation(project(":spi:core-spi"))
 
-
     testImplementation(libs.edc.junit)
     testImplementation(libs.awaitility)
     testImplementation(testFixtures(project(":spi:edr-spi")))

diff --git a/...allback/src/main/java/org/eclipse/tractusx/edc/callback/TransferProcessLocalCallback.java b/...allback/src/main/java/org/eclipse/tractusx/edc/callback/TransferProcessLocalCallback.java
@@ -148,6 +148,16 @@ private void cleanOldEdr(String assetId, String agreementId) {
             monitor.debug(format("Expiring EDR for transfer process %s", entry.getTransferProcessId()));
             entry.transitionToExpired();
             edrCache.update(entry);
+
+            var transferProcess = transferProcessStore.findById(entry.getTransferProcessId());
+
+            if (transferProcess != null && transferProcess.canBeCompleted()) {
+                transferProcess.transitionCompleting();
+                transferProcessStore.save(transferProcess);
+            } else {
+                monitor.info(format("Cannot terminate transfer process with id: %s", entry.getTransferProcessId()));
+            }
+
         }));
     }
 

diff --git a/...ts/e2e-tests/src/test/java/org/eclipse/tractusx/edc/tests/edr/AbstractRenewalEdrTest.java b/...ts/e2e-tests/src/test/java/org/eclipse/tractusx/edc/tests/edr/AbstractRenewalEdrTest.java
@@ -20,6 +20,7 @@
 import okhttp3.mockwebserver.MockWebServer;
 import org.assertj.core.api.Condition;
 import org.eclipse.edc.connector.transfer.spi.event.TransferProcessStarted;
+import org.eclipse.edc.connector.transfer.spi.types.TransferProcessStates;
 import org.eclipse.edc.policy.model.Operator;
 import org.eclipse.tractusx.edc.lifecycle.Participant;
 import org.junit.jupiter.api.AfterEach;
@@ -38,6 +39,7 @@
 import static org.assertj.core.api.Assertions.anyOf;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
+import static org.awaitility.pollinterval.FibonacciPollInterval.fibonacci;
 import static org.eclipse.edc.spi.CoreConstants.EDC_NAMESPACE;
 import static org.eclipse.tractusx.edc.edr.spi.types.EndpointDataReferenceEntryStates.EXPIRED;
 import static org.eclipse.tractusx.edc.edr.spi.types.EndpointDataReferenceEntryStates.NEGOTIATED;
@@ -108,21 +110,35 @@ void negotiateEdr_shouldRenewTheEdr() throws IOException {
 
         assertThat(expectedEvents).usingRecursiveFieldByFieldElementComparator().containsAll(events);
 
-        JsonArrayBuilder edrCaches = Json.createArrayBuilder();
+        JsonArrayBuilder edrCachesBuilder = Json.createArrayBuilder();
 
         await().atMost(ASYNC_TIMEOUT)
                 .pollInterval(ASYNC_POLL_INTERVAL)
                 .untilAsserted(() -> {
                     var localEdrCaches = SOKRATES.getEdrEntriesByAssetId(assetId);
                     assertThat(localEdrCaches).hasSizeGreaterThan(1);
-                    localEdrCaches.forEach(edrCaches::add);
+                    localEdrCaches.forEach(edrCachesBuilder::add);
                 });
 
+        var edrCaches = edrCachesBuilder.build();
 
-        assertThat(edrCaches.build())
+        assertThat(edrCaches)
                 .extracting(json -> json.asJsonObject().getJsonString("tx:edrState").getString())
                 .areAtMost(1, anyOf(stateCondition(NEGOTIATED.name(), "Negotiated"), stateCondition(REFRESHING.name(), "Refreshing")))
                 .areAtLeast(1, stateCondition(EXPIRED.name(), "Expired"));
+
+        var transferProcessId = edrCaches.stream()
+                .filter(json -> json.asJsonObject().getJsonString("tx:edrState").getString().equals(EXPIRED.name()))
+                .map(json -> json.asJsonObject().getJsonString("edc:transferProcessId").getString())
+                .findFirst()
+                .orElseThrow();
+
+        await().pollInterval(fibonacci())
+                .atMost(ASYNC_TIMEOUT)
+                .untilAsserted(() -> {
+                    var tpState = SOKRATES.getTransferProcessState(transferProcessId);
+                    assertThat(tpState).isNotNull().isEqualTo(TransferProcessStates.COMPLETED.toString());
+                });
     }