test: add TCK tests

Author: ndr-brt

build.gradle.kts

@@ -12,12 +12,14 @@ version = "0.0.11-SNAPSHOT"
 
 repositories {
     mavenCentral()
+    mavenLocal()
 }
 
 dependencies {
     implementation("com.fasterxml.jackson.core:jackson-databind:2.21.3")
     implementation("com.nimbusds:nimbus-jose-jwt:10.9")
     implementation("jakarta.ws.rs:jakarta.ws.rs-api:4.0.0")
+    runtimeOnly("org.eclipse.parsson:parsson:1.1.9")
 
     testImplementation(platform("org.junit:junit-bom:6.1.0"))
     testImplementation("org.junit.jupiter:junit-jupiter")
@@ -37,6 +39,7 @@ dependencies {
     testImplementation("org.testcontainers:testcontainers-junit-jupiter:2.0.5")
     testImplementation("org.testcontainers:testcontainers-postgresql:2.0.5")
     testImplementation("org.postgresql:postgresql:42.7.11")
+    testImplementation("org.eclipse.dataspacetck.dps:dps-tck:1.1.0")
 }
 
 tasks.test {

src/main/java/org/eclipse/dataplane/Dataplane.java

@@ -63,13 +63,15 @@
 import java.util.Set;
 import java.util.UUID;
 
+import static com.fasterxml.jackson.annotation.JsonInclude.Include.NON_NULL;
 import static com.fasterxml.jackson.databind.DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES;
 import static jakarta.ws.rs.core.HttpHeaders.AUTHORIZATION;
-import static java.util.Collections.emptyMap;
 
 public class Dataplane {
 
-    private final ObjectMapper objectMapper = new ObjectMapper().configure(FAIL_ON_UNKNOWN_PROPERTIES, false);
+    private final ObjectMapper objectMapper = new ObjectMapper()
+            .configure(FAIL_ON_UNKNOWN_PROPERTIES, false)
+            .setDefaultPropertyInclusion(NON_NULL);
     private DataFlowStore dataFlowStore = new InMemoryDataFlowStore(objectMapper);
     private ControlPlaneStore controlPlaneStore = new InMemoryControlPlaneStore(objectMapper);
     private String id;
@@ -274,8 +276,8 @@ public Result<Void> notifyCompleted(String dataFlowId) {
         return dataFlowStore.findById(dataFlowId)
                 .compose(dataFlow -> {
                     dataFlow.transitionToCompleted();
-
-                    return notifyControlPlane("completed", dataFlow, emptyMap());
+                    var message = new DataFlowStatusMessage(dataFlowId, dataFlow.getState().name(), null, null);
+                    return notifyControlPlane("completed", dataFlow, message);
                 });
     }
 
@@ -374,7 +376,7 @@ private Result<Void> notifyControlPlane(String action, DataFlow dataFlow, Object
                             })
                             .onSuccess(authorizationHeader -> requestBuilder.header(AUTHORIZATION, authorizationHeader));
 
-                    return httpClient.send(requestBuilder.build(), HttpResponse.BodyHandlers.discarding());
+                    return httpClient.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofString());
                 })
                 .compose(response -> {
                     var successful = response.statusCode() >= 200 && response.statusCode() < 300;

src/main/java/org/eclipse/dataplane/domain/DataAddress.java

@@ -14,19 +14,21 @@
 
 package org.eclipse.dataplane.domain;
 
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 import java.util.List;
 
+@JsonIgnoreProperties(value = {"@type"})
 public record DataAddress(
-        @JsonProperty("@type") String type,
         String endpointType,
         String endpoint,
         List<EndpointProperty> endpointProperties
 ) {
 
-    public DataAddress(String endpointType, String endpoint, List<EndpointProperty> endpointProperties) {
-        this("DataAddress", endpointType, endpoint, endpointProperties);
+    @JsonProperty("@type")
+    public String getType() {
+        return "DataAddress";
     }
 
     public record EndpointProperty(

src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlow.java

@@ -132,7 +132,11 @@ public void transitionToTerminated(String reason) {
     }
 
     public boolean isPush() {
-        return transferType.split("-")[1].equalsIgnoreCase("PUSH");
+        return transferType.substring(transferType.lastIndexOf('-') + 1).equalsIgnoreCase("push");
+    }
+
+    public boolean isPull() {
+        return !isPush();
     }
 
     public boolean isInitiating() {
@@ -147,10 +151,6 @@ public boolean isStarted() {
         return state == State.STARTED;
     }
 
-    public boolean isPull() {
-        return transferType.split("-")[1].equalsIgnoreCase("PULL");
-    }
-
     public void setDataAddress(DataAddress dataAddress) {
         this.dataAddress = dataAddress;
     }

src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStartedNotificationMessage.java

@@ -17,6 +17,7 @@
 import org.eclipse.dataplane.domain.DataAddress;
 
 public record DataFlowStartedNotificationMessage(
+        String messageId,
         DataAddress dataAddress
 ) {
 }

src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStatusMessage.java

@@ -17,10 +17,16 @@
 
 import org.eclipse.dataplane.domain.DataAddress;
 
+import java.util.UUID;
+
 public record DataFlowStatusMessage(
+        String messageId,
         String dataFlowId,
         String state,
         DataAddress dataAddress,
         String error
 ) {
+    public DataFlowStatusMessage(String dataFlowId, String state, DataAddress dataAddress, String error) {
+        this(UUID.randomUUID().toString(), dataFlowId, state, dataAddress, error);
+    }
 }

src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStatusResponseMessage.java

@@ -15,7 +15,7 @@
 package org.eclipse.dataplane.domain.dataflow;
 
 public record DataFlowStatusResponseMessage(
-        String dataflowId,
+        String dataFlowId,
         String state
 ) {
 }

src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowSuspendMessage.java

@@ -15,6 +15,7 @@
 package org.eclipse.dataplane.domain.dataflow;
 
 public record DataFlowSuspendMessage(
+        String messageId,
         String reason
 ) {
 }

src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowTerminateMessage.java

@@ -15,6 +15,7 @@
 package org.eclipse.dataplane.domain.dataflow;
 
 public record DataFlowTerminateMessage(
+        String messageId,
         String reason
 ) {
 }

src/main/java/org/eclipse/dataplane/port/exception/DataFlowNotifyControlPlaneFailed.java

@@ -18,15 +18,15 @@
 
 public class DataFlowNotifyControlPlaneFailed extends Exception {
     private final String action;
-    private final HttpResponse<Void> response;
+    private final HttpResponse<String> response;
 
-    public DataFlowNotifyControlPlaneFailed(String action, HttpResponse<Void> response) {
-        super("control-plane responded with %s".formatted(response.statusCode()));
+    public DataFlowNotifyControlPlaneFailed(String action, HttpResponse<String> response) {
+        super("control-plane responded with %s: %s".formatted(response.statusCode(), response.body()));
         this.action = action;
         this.response = response;
     }
 
-    public HttpResponse<Void> getResponse() {
+    public HttpResponse<String> getResponse() {
         return response;
     }