improve e2e test

Author: paullatzelsperger

Requests/ControlPlane Management/Get open dataflow by ID.bru

@@ -0,0 +1,37 @@
+meta {
+  name: Get open dataflow by ID
+  type: http
+  seq: 9
+}
+
+get {
+  url: {{CONSUMER_DP}}/api/proxy/flows
+  body: json
+  auth: none
+}
+
+body:text {
+  {
+      "@context": [
+      "https://w3id.org/edc/connector/management/v2"
+      ],
+      "@type": "QuerySpec"
+  }
+}
+
+script:post-response {
+  const body = res.getBody()
+  const map = new Map(Object.entries(body));
+  // just use the first entry in the map
+  const [flowId, address] = map.entries().next().value;
+  
+  expect(flowId).not.to.be.undefined
+  bru.setVar("flowId",flowId)
+  
+  
+}
+
+settings {
+  encodeUrl: true
+  timeout: 0
+}

k8s/issuer/application/issuerservice-seed-job.yaml

@@ -22,15 +22,15 @@ metadata:
   labels:
     app: issuerservice-seed
     platform: edcv
-    type: edcv-job
+    type: edc-job
 spec:
   backoffLimit: 5
   template:
     metadata:
       labels:
         app: issuerservice-seed
         platform: edcv
-        type: edcv-job
+        type: edc-job
     spec:
       restartPolicy: OnFailure
       initContainers:

launchers/dataplane/src/main/java/org/eclipse/edc/mvd/dataplane/data/ConsumerProxyController.java

@@ -35,11 +35,17 @@ public ConsumerProxyController(ConsumerDataHandler consumerDataHandler) {
     }
 
     @GET
-    @Path("/{flowId}")
+    @Path("/{flowId}/data")
     public String handleDataFlow(@PathParam("flowId") String flowId) {
         return consumerDataHandler.downloadData(flowId).orElseThrow(ExceptionMapper.MAP_TO_WSRS);
     }
 
+    @GET
+    @Path("/{flowId}")
+    public DataAddress getFlowMetadata(@PathParam("flowId") String flowId) {
+        return consumerDataHandler.getFlow(flowId);
+    }
+
     @GET
     public Map<String, DataAddress> getAll() {
         return consumerDataHandler.getAllFlows();

launchers/dataplane/src/main/java/org/eclipse/edc/mvd/dataplane/signaling/ConsumerDataHandler.java

@@ -78,4 +78,8 @@ public Result<String> downloadData(String flowId) {
     public Map<String, DataAddress> getAllFlows() {
         return ongoingTransfers;
     }
+
+    public DataAddress getFlow(String flowId) {
+        return ongoingTransfers.get(flowId);
+    }
 }

tests/end2end/src/test/java/org/eclipse/edc/demo/tests/transfer/TransferEndToEndTest.java

@@ -45,14 +45,11 @@
 public class TransferEndToEndTest {
     // Management API base URL of the consumer connector, goes through Ingress controller
     private static final String CONSUMER_MANAGEMENT_URL = "http://cp.consumer.localhost:8080";
-    // Catalog Query API URL of the consumer connector, goes through ingress controller
-    private static final String CONSUMER_CATALOG_URL = "http://127.0.0.1/consumer/fc";
     // DSP service URL of the provider, not reachable outside the cluster
     private static final String PROVIDER_DSP_URL = "http://controlplane.provider.svc.cluster.local:8082/api/dsp/2025-1";
     // DID of the provider company
     private static final String PROVIDER_ID = "did:web:identityhub.provider.svc.cluster.local%3A7083:provider";
-    // public API endpoint of the provider-qna connector, goes through the ingress controller
-    private static final String PROVIDER_PUBLIC_URL = "http://dp.provider.localhost:8080/public";
+    private static final String CONSUMER_PROXY_URL = "http://dp.consumer.localhost:8080/api/proxy";
     private static final String PROVIDER_MANAGEMENT_URL = "http://cp.provider.localhost:8080";
 
     private final TypeTransformerRegistry transformerRegistry = new TypeTransformerRegistryImpl();
@@ -69,23 +66,23 @@ private static RequestSpecification baseRequest() {
     @DisplayName("Tests a successful End-to-End contract negotiation and data transfer")
     @Test
     void transferData_hasPermission_shouldTransferData() {
-        System.out.println("Waiting for Provider dataplane to come online");
-        // wait until provider's dataplane is available
-        await().atMost(TEST_TIMEOUT_DURATION)
-                .pollDelay(TEST_POLL_DELAY)
-                .untilAsserted(() -> {
-                    var jp = baseRequest()
-                            .get(PROVIDER_MANAGEMENT_URL + "/api/mgmt/v4/dataplanes")
-                            .then()
-                            .statusCode(200)
-                            .log().ifValidationFails()
-                            .extract().body().jsonPath();
-
-                    var state = jp.getString("state");
-                    assertThat(state).isEqualTo("[REGISTERED]");
-                });
-
-        System.out.println("Provider dataplane is online, fetching catalog");
+//        System.out.println("Waiting for Provider dataplane to come online");
+//        // wait until provider's dataplane is available
+//        await().atMost(TEST_TIMEOUT_DURATION)
+//                .pollDelay(TEST_POLL_DELAY)
+//                .untilAsserted(() -> {
+//                    var jp = baseRequest()
+//                            .get(PROVIDER_MANAGEMENT_URL + "/api/mgmt/v4/dataplanes")
+//                            .then()
+//                            .statusCode(200)
+//                            .log().ifValidationFails()
+//                            .extract().body().jsonPath();
+//
+//                    var state = jp.getString("state");
+//                    assertThat(state).isEqualTo("[REGISTERED]");
+//                });
+//
+//        System.out.println("Provider dataplane is online, fetching catalog");
 
         var catalogRequestBody = Json.createObjectBuilder()
                 .add("@context", Json.createObjectBuilder().add("edc", "https://w3id.org/edc/connector/management/v2"))
@@ -178,36 +175,36 @@ void transferData_hasPermission_shouldTransferData() {
                     assertThat(jp.getString("state")).contains("STARTED");
                 });
 
-        System.out.printf("Fetch EDR with ID %s%n", transferProcessId);
+        System.out.printf("Fetch dataflow with ID %s from custom proxy%n", transferProcessId);
         // fetch EDR for transfer processs
         var endpoint = new AtomicReference<String>();
         var token = new AtomicReference<String>();
         await().atMost(TEST_TIMEOUT_DURATION)
                 .pollDelay(TEST_POLL_DELAY)
                 .untilAsserted(() -> {
                     var jp = baseRequest()
-                            .get(CONSUMER_MANAGEMENT_URL + "/api/mgmt/v3/edrs/%s/dataaddress".formatted(transferProcessId))
+                            .get(CONSUMER_PROXY_URL + "/flows/%s".formatted(transferProcessId))
                             .then()
                             .log().ifValidationFails()
                             .statusCode(200)
-                            .onFailMessage("Expected to find an EDR with transfer ID %s but did not!".formatted(transferProcessId))
+                            .onFailMessage("Expected to find a DataFlow ID %s but did not!".formatted(transferProcessId))
                             .extract().body().jsonPath();
 
                     endpoint.set(jp.getString("endpoint"));
-                    token.set(jp.getString("authorization"));
+//                    token.set(jp.getString("authorization"));
 
-                    assertThat(endpoint.get()).isNotNull().endsWith("/api/public");
-                    assertThat(token.get()).isNotNull();
+                    assertThat(endpoint.get()).isNotNull().endsWith("/api/public/data/source");
+//                    assertThat(token.get()).isNotNull();
                 });
 
         //download exemplary JSON data from public endpoint
         var response = given()
-                .header("Authorization", token.get())
-                .get(PROVIDER_PUBLIC_URL + "/api/public")
+//                .header("Authorization", token.get())
+                .get(CONSUMER_PROXY_URL + "/flows/%s/data".formatted(transferProcessId))
                 .then()
                 .log().ifError()
                 .statusCode(200)
-                .extract().body().asString();
+                .extract().body().as(Object[].class);
 
         assertThat(response).isNotEmpty();
     }