chore: update to EDC 0.2.1 (#716)

* chore: update to EDC 0.2.1

* update DEPENDENCIES, add migration, re-enable postgres e2e tests

* remove control-plane dependency

* pr remarks

* DEPENDENCIES

.github/workflows/verify.yaml

@@ -149,17 +149,10 @@ jobs:
           POSTGRES_PASSWORD: password
 
     steps:
-      - name: Issue warning
-        run: |
-          echo "::warning file=verify.yaml,line=138,title=Deactivated PostgreSQL tests::Deactivated the PostgreSQL tests until https://github.com/eclipse-edc/Connector/pull/3336 is merged and packaged in a version"
-
       - uses: actions/checkout@v3
-        if: ${{ false }}  # disabled until https://github.com/eclipse-edc/Connector/pull/3336 is merged and packaged in a version
       - uses: ./.github/actions/setup-java
-        if: ${{ false }}  # disabled until https://github.com/eclipse-edc/Connector/pull/3336 is merged and packaged in a version
 
       - name: Run Postgresql E2E tests
-        if: ${{ false }}  # disabled until https://github.com/eclipse-edc/Connector/pull/3336 is merged and packaged in a version
         run: ./gradlew test -DincludeTags="PostgresqlIntegrationTest" -PverboseTest=true
 
   miw-integration-tests:

core/edr-cache-core/src/main/java/org/eclipse/tractusx/edc/edr/core/defaults/EdrCacheEntryPredicateConverter.java

@@ -14,12 +14,110 @@
 
 package org.eclipse.tractusx.edc.edr.core.defaults;
 
-import org.eclipse.edc.spi.query.BaseCriterionToPredicateConverter;
+import org.eclipse.edc.spi.query.Criterion;
+import org.eclipse.edc.spi.query.CriterionToPredicateConverter;
 import org.eclipse.tractusx.edc.edr.spi.types.EndpointDataReferenceEntry;
+import org.jetbrains.annotations.NotNull;
 
-public class EdrCacheEntryPredicateConverter extends BaseCriterionToPredicateConverter<EndpointDataReferenceEntry> {
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Predicate;
+import java.util.regex.Pattern;
+
+import static java.lang.String.format;
+
+/**
+ * This class is almost a 1:1 copy of the {@code CriterionToPredicateConverterImpl} (except for the {@code property()} method) from the {@code control-plane-core} module.
+ * Pulling in that module is not possible, because that would pull in almost the entire Control Plane
+ */
+public class EdrCacheEntryPredicateConverter implements CriterionToPredicateConverter {
 
     @Override
+    public <T> Predicate<T> convert(Criterion criterion) {
+        var operator = criterion.getOperator().toLowerCase();
+
+        return switch (operator) {
+            case "=" -> equalPredicate(criterion);
+            case "in" -> inPredicate(criterion);
+            case "like" -> likePredicate(criterion);
+            default -> throw new IllegalArgumentException(format("Operator [%s] is not supported by this converter!", criterion.getOperator()));
+        };
+    }
+
+    @NotNull
+    private <T> Predicate<T> equalPredicate(Criterion criterion) {
+        return t -> {
+            var operandLeft = (String) criterion.getOperandLeft();
+            var property = property(operandLeft, t);
+            if (property == null) {
+                return false;
+            }
+
+            if (property.getClass().isEnum() && criterion.getOperandRight() instanceof String) {
+                var enumProperty = (Enum<?>) property;
+                return Objects.equals(enumProperty.name(), criterion.getOperandRight());
+            }
+
+            if (property instanceof Number c1 && criterion.getOperandRight() instanceof Number c2) {
+                // interpret as double to not lose any precision
+                return Double.compare(c1.doubleValue(), c2.doubleValue()) == 0;
+            }
+
+            if (property instanceof List<?> list) {
+                return list.stream().anyMatch(it -> Objects.equals(it, criterion.getOperandRight()));
+            }
+
+            return Objects.equals(property, criterion.getOperandRight());
+        };
+    }
+
+    @NotNull
+    private <T> Predicate<T> inPredicate(Criterion criterion) {
+        return t -> {
+            var operandLeft = (String) criterion.getOperandLeft();
+            var property = property(operandLeft, t);
+            if (property == null) {
+                return false;
+            }
+
+            if (criterion.getOperandRight() instanceof Iterable<?> iterable) {
+                for (var value : iterable) {
+                    if (value.equals(property)) {
+                        return true;
+                    }
+                }
+                return false;
+            } else {
+                throw new IllegalArgumentException("Operator IN requires the right-hand operand to be an " + Iterable.class.getName() + " but was " + criterion.getOperandRight().getClass().getName());
+            }
+
+
+        };
+    }
+
+    @NotNull
+    private <T> Predicate<T> likePredicate(Criterion criterion) {
+        return t -> {
+            var operandLeft = (String) criterion.getOperandLeft();
+            var property = property(operandLeft, t);
+            if (property == null) {
+                return false;
+            }
+
+            if (criterion.getOperandRight() instanceof String operandRight) {
+                var regexPattern = Pattern.quote(operandRight)
+                        .replace("%", "\\E.*\\Q")
+                        .replace("_", "\\E.\\Q");
+
+                regexPattern = "^" + regexPattern + "$";
+
+                return Pattern.compile(regexPattern).matcher(property.toString()).matches();
+            }
+
+            return false;
+        };
+    }
+
     protected Object property(String key, Object object) {
         if (object instanceof EndpointDataReferenceEntry entry) {
             return switch (key) {

edc-extensions/dataplane-proxy/edc-dataplane-proxy-consumer-api/src/test/java/org/eclipse/tractusx/edc/dataplane/proxy/consumer/api/asset/model/ConsumerAssetRequestControllerTest.java

@@ -297,7 +297,7 @@ void requestAsset_shouldReturnData_withDataPlaneUrl() throws IOException {
 
         var flowRequest = captor.getValue();
 
-        assertThat(flowRequest.getSourceDataAddress().getProperty(BASE_URL)).isEqualTo(edr.getEndpoint());
+        assertThat(flowRequest.getSourceDataAddress().getStringProperty(BASE_URL)).isEqualTo(edr.getEndpoint());
 
         assertThat(flowRequest.getProperties().get(QUERY_PARAMS)).isEqualTo(request.get(QUERY_PARAMS));
         assertThat(flowRequest.getProperties().get(PATH)).isEqualTo(request.get(PATH));

edc-extensions/edr/edr-api/src/test/java/org/eclipse/tractusx/edc/api/edr/transform/EndpointDataReferenceToDataAddressTransformerTest.java

@@ -51,10 +51,10 @@ void transform() {
 
         assertThat(dataAddress).isNotNull();
         assertThat(dataAddress.getType()).isEqualTo(EDR_SIMPLE_TYPE);
-        assertThat(dataAddress.getProperty(ID)).isEqualTo(dto.getId());
-        assertThat(dataAddress.getProperty(ENDPOINT)).isEqualTo(dto.getEndpoint());
-        assertThat(dataAddress.getProperty(AUTH_KEY)).isEqualTo(dto.getAuthKey());
-        assertThat(dataAddress.getProperty(AUTH_CODE)).isEqualTo(dto.getAuthCode());
+        assertThat(dataAddress.getStringProperty(ID)).isEqualTo(dto.getId());
+        assertThat(dataAddress.getStringProperty(ENDPOINT)).isEqualTo(dto.getEndpoint());
+        assertThat(dataAddress.getStringProperty(AUTH_KEY)).isEqualTo(dto.getAuthKey());
+        assertThat(dataAddress.getStringProperty(AUTH_CODE)).isEqualTo(dto.getAuthCode());
 
     }
 }

edc-extensions/edr/edr-callback/src/test/java/org/eclipse/tractusx/edc/callback/ContractNegotiationCallbackTest.java

@@ -15,17 +15,19 @@
 package org.eclipse.tractusx.edc.callback;
 
 import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationAccepted;
-import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationConfirmed;
-import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationDeclined;
+import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationAgreed;
 import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationEvent;
-import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationFailed;
+import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationFinalized;
 import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationInitiated;
 import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationOffered;
 import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationRequested;
 import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationTerminated;
+import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationVerified;
+import org.eclipse.edc.connector.contract.spi.types.agreement.ContractAgreement;
 import org.eclipse.edc.connector.spi.transferprocess.TransferProcessService;
 import org.eclipse.edc.connector.transfer.spi.types.TransferProcess;
 import org.eclipse.edc.connector.transfer.spi.types.TransferRequest;
+import org.eclipse.edc.policy.model.Policy;
 import org.eclipse.edc.service.spi.result.ServiceResult;
 import org.eclipse.edc.spi.monitor.Monitor;
 import org.eclipse.edc.spi.types.domain.callback.CallbackAddress;
@@ -61,6 +63,16 @@ public class ContractNegotiationCallbackTest {
 
     ContractNegotiationCallback callback;
 
+    private static <T extends ContractNegotiationEvent, B extends ContractNegotiationEvent.Builder<T, B>> B baseBuilder(B builder) {
+        var callbacks = List.of(CallbackAddress.Builder.newInstance().uri("http://local").events(Set.of("test")).build());
+        return builder
+                .contractNegotiationId("id")
+                .protocol("test")
+                .callbackAddresses(callbacks)
+                .counterPartyAddress("addr")
+                .counterPartyId("provider");
+    }
+
     @BeforeEach
     void setup() {
         callback = new ContractNegotiationCallback(transferProcessService, monitor);
@@ -121,15 +133,31 @@ void invoke_shouldIgnoreOtherEvents(ContractNegotiationEvent event) {
         verifyNoInteractions(transferProcessService);
     }
 
+    @Test
+    void invoke_whenFinalized() {
+        var evt = baseBuilder(ContractNegotiationFinalized.Builder.newInstance().contractAgreement(
+                ContractAgreement.Builder.newInstance()
+                        .providerId("test-provider")
+                        .assetId("test-asset")
+                        .policy(Policy.Builder.newInstance().build())
+                        .id("test-id")
+                        .consumerId("test-consumer")
+                        .build())).build();
+        var message = remoteMessage(evt);
+        when(transferProcessService.initiateTransfer(any())).thenReturn(ServiceResult.success(null));
+
+        callback.invoke(message);
+        verify(transferProcessService).initiateTransfer(any(TransferRequest.class));
+    }
+
     private static class EventInstances implements ArgumentsProvider {
 
         @Override
         public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
             return Stream.of(
                     baseBuilder(ContractNegotiationAccepted.Builder.newInstance()).build(),
-                    baseBuilder(ContractNegotiationConfirmed.Builder.newInstance()).build(),
-                    baseBuilder(ContractNegotiationDeclined.Builder.newInstance()).build(),
-                    baseBuilder(ContractNegotiationFailed.Builder.newInstance()).build(),
+                    baseBuilder(ContractNegotiationAgreed.Builder.newInstance()).build(),
+                    baseBuilder(ContractNegotiationVerified.Builder.newInstance()).build(),
                     baseBuilder(ContractNegotiationInitiated.Builder.newInstance()).build(),
                     baseBuilder(ContractNegotiationOffered.Builder.newInstance()).build(),
                     baseBuilder(ContractNegotiationRequested.Builder.newInstance()).build(),
@@ -138,14 +166,6 @@ public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
 
         }
 
-        private <T extends ContractNegotiationEvent, B extends ContractNegotiationEvent.Builder<T, B>> B baseBuilder(B builder) {
-            var callbacks = List.of(CallbackAddress.Builder.newInstance().uri("http://local").events(Set.of("test")).build());
-            return builder
-                    .contractNegotiationId("id")
-                    .protocol("test")
-                    .callbackAddresses(callbacks)
-                    .counterPartyAddress("addr")
-                    .counterPartyId("provider");
-        }
+
     }
 }

edc-extensions/postgresql-migration/src/main/resources/org/eclipse/tractusx/edc/postgresql/migration/contractnegotiation/V0_0_7__Alter_ContractNegotiation_AddPendingField.sql

@@ -0,0 +1,16 @@
+--
+--  Copyright (c) 2023 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+--
+--  This program and the accompanying materials are made available under the
+--  terms of the Apache License, Version 2.0 which is available at
+--  https://www.apache.org/licenses/LICENSE-2.0
+--
+--  SPDX-License-Identifier: Apache-2.0
+--
+--  Contributors:
+--       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+--
+
+-- alter type of column "type"
+ALTER TABLE edc_contract_negotiation
+    ADD COLUMN pending BOOLEAN DEFAULT FALSE;

edc-extensions/transferprocess-sftp-client/src/main/java/org/eclipse/tractusx/edc/transferprocess/sftp/client/SftpDataSinkFactory.java

@@ -36,15 +36,6 @@ public boolean canHandle(DataFlowRequest request) {
         }
     }
 
-    @Override
-    public @NotNull Result<Boolean> validate(DataFlowRequest request) {
-        if (!canHandle(request)) {
-            return Result.failure(String.format("Invalid DataFlowRequest: %s", request.getId()));
-        }
-
-        return VALID;
-    }
-
     @Override
     public DataSink createSink(DataFlowRequest request) {
         if (!canHandle(request)) {
@@ -65,4 +56,13 @@ public DataSink createSink(DataFlowRequest request) {
 
         return new SftpDataSink(sftpClientWrapper);
     }
+
+    @Override
+    public @NotNull Result<Void> validateRequest(DataFlowRequest request) {
+        if (!canHandle(request)) {
+            return Result.failure(String.format("Invalid DataFlowRequest: %s", request.getId()));
+        }
+
+        return Result.success();
+    }
 }

edc-extensions/transferprocess-sftp-client/src/main/java/org/eclipse/tractusx/edc/transferprocess/sftp/client/SftpDataSourceFactory.java

@@ -33,15 +33,6 @@ public boolean canHandle(DataFlowRequest request) {
         }
     }
 
-    @Override
-    public @NotNull Result<Boolean> validate(DataFlowRequest request) {
-        if (!canHandle(request)) {
-            return Result.failure(String.format("Invalid DataFlowRequest: %s", request.getId()));
-        }
-
-        return VALID;
-    }
-
     @Override
     public DataSource createSource(DataFlowRequest request) {
         if (!canHandle(request)) {
@@ -59,4 +50,12 @@ public DataSource createSource(DataFlowRequest request) {
         SftpClientWrapper sftpClientWrapper = new SftpClientWrapperImpl(sftpClientConfig);
         return new SftpDataSource(sftpClientWrapper);
     }
+
+    @Override
+    public @NotNull Result<Void> validateRequest(DataFlowRequest request) {
+        if (!canHandle(request)) {
+            return Result.failure(String.format("Invalid DataFlowRequest: %s", request.getId()));
+        }
+        return Result.success();
+    }
 }