feat: pass export service and jobid as HTTP headers in generic importer (#1473)

Add 2 HTTP headers to `GenericImporter` and `GenericFileImporter`:
- `X-DTP-Export-Service`, passing the `ExportService` from the
`JobMetadata`, i.e. the source of data;
- `X-DTP-Job-Id`, passing the current job UUID as String.

This will help the receiving service in building the internal logic
(i.e. if they want to know which service is sending the data).

TODO:
- [x] this PR: figure out tests (what is the currently recommended
workflow with testing?)
- [ ] next PR: add an optional `recurringJobId` field to
`PortabilityJob` (populated from `*JobStore`) and `JobMetadata` -- to be
sent for recurring transfers only via `X-DTP-Recurring-Job-Id`

Author: alexeyqu

extensions/data-transfer/portability-data-transfer-generic/src/main/java/org/datatransferproject/datatransfer/generic/GenericFileImporter.java

@@ -64,6 +64,8 @@ private <T> boolean importSingleFileItem(
         new Request.Builder()
             .url(endpoint)
             .addHeader("Authorization", format("Bearer %s", authData.getToken()))
+            .addHeader("X-DTP-Export-Service", this.exportService)
+            .addHeader("X-DTP-Job-Id", jobId.toString())
             .post(
                 new MultipartBody.Builder()
                     .setType(MULTIPART_RELATED)

extensions/data-transfer/portability-data-transfer-generic/src/main/java/org/datatransferproject/datatransfer/generic/GenericImporter.java

@@ -33,6 +33,7 @@
 import org.datatransferproject.spi.transfer.provider.Importer;
 import org.datatransferproject.spi.transfer.types.DestinationMemoryFullException;
 import org.datatransferproject.spi.transfer.types.InvalidTokenException;
+import org.datatransferproject.transfer.JobMetadata;
 import org.datatransferproject.types.common.models.ContainerResource;
 import org.datatransferproject.types.transfer.auth.AppCredentials;
 import org.datatransferproject.types.transfer.auth.TokensAndUrlAuthData;
@@ -79,6 +80,7 @@ public String toString() {
   OkHttpClient client = new OkHttpClient();
   ObjectMapper om = new ObjectMapper();
   Map<UUID, OAuthTokenManager> jobTokenManagerMap = new HashMap<>();
+  protected final String exportService;
 
   static final MediaType JSON = MediaType.parse("application/json");
 
@@ -91,6 +93,7 @@ public GenericImporter(
     this.appCredentials = appCredentials;
     this.endpoint = endpoint;
     this.containerSerializer = containerSerializer;
+    this.exportService = JobMetadata.getExportService();
     configureObjectMapper(om);
   }
 
@@ -163,6 +166,8 @@ boolean importSingleItem(UUID jobId, TokensAndUrlAuthData authData, ImportableDa
         new Request.Builder()
             .url(endpoint)
             .addHeader("Authorization", format("Bearer %s", authData.getToken()))
+            .addHeader("X-DTP-Export-Service", this.exportService)
+            .addHeader("X-DTP-Job-Id", jobId.toString())
             .post(RequestBody.create(JSON, om.writeValueAsBytes(dataItem.getJsonData())))
             .build();
 

extensions/data-transfer/portability-data-transfer-generic/src/test/java/org/datatransferproject/datatransfer/generic/GenericFileImporterTest.java

@@ -21,12 +21,15 @@
 import org.datatransferproject.api.launcher.Monitor;
 import org.datatransferproject.spi.cloud.storage.TemporaryPerJobDataStore;
 import org.datatransferproject.spi.transfer.idempotentexecutor.InMemoryIdempotentImportExecutor;
+import org.datatransferproject.transfer.JobMetadata;
 import org.datatransferproject.types.common.models.IdOnlyContainerResource;
 import org.datatransferproject.types.transfer.auth.AppCredentials;
 import org.datatransferproject.types.transfer.auth.TokensAndUrlAuthData;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
 
 public class GenericFileImporterTest {
   private MockWebServer webServer;
@@ -38,16 +41,24 @@ public InputStreamWrapper getStream(UUID jobId, String key) throws IOException {
           return new InputStreamWrapper(new ByteArrayInputStream("Hello world".getBytes()));
         }
       };
+  private static final UUID MOCK_JOB_ID = UUID.fromString("123e4567-e89b-12d3-a456-426614174000");
+  private static final String MOCK_EXPORT_SERVICE = "mockExportService";
+  private MockedStatic<JobMetadata> jobMetadataMock;
 
   @Before
   public void setup() throws IOException {
     webServer = new MockWebServer();
     webServer.start();
+
+    jobMetadataMock = Mockito.mockStatic(JobMetadata.class);
+    jobMetadataMock.when(JobMetadata::getJobId).thenReturn(MOCK_JOB_ID);
+    jobMetadataMock.when(JobMetadata::getExportService).thenReturn(MOCK_EXPORT_SERVICE);
   }
 
   @After
   public void teardown() throws IOException {
     webServer.shutdown();
+    jobMetadataMock.close();
   }
 
   public static MultipartStream getMultipartStream(RecordedRequest request) {
@@ -174,4 +185,39 @@ public void testGenericFileImporterMixedTypes() throws Exception {
             + " file\",\"schemaSource\":\"schemasource\",\"apiVersion\":\"0.1.0\"}",
         jsonRequest.getBody().readUtf8());
   }
+
+  @Test
+  public void testGenericFileImporterPassingJobMetadata() throws Exception {
+    GenericFileImporter<IdOnlyContainerResource, String> importer =
+      new GenericFileImporter<>(
+          container ->
+              Arrays.asList(
+                  new ImportableFileData<>(
+                      new CachedDownloadableItem(container.getId(), container.getId()),
+                      "video/mp4",
+                      new GenericPayload<>(container.getId(), "schemasource"),
+                      container.getId(),
+                      container.getId())),
+          new AppCredentials("key", "secret"),
+          webServer.url("/id").url(),
+          dataStore,
+          monitor);
+    InMemoryIdempotentImportExecutor executor = new InMemoryIdempotentImportExecutor(monitor);
+    webServer.enqueue(new MockResponse().setResponseCode(201).setBody("OK"));
+
+    importer.importItem(
+        MOCK_JOB_ID,
+        executor,
+        new TokensAndUrlAuthData(
+            "accessToken", "refreshToken", webServer.url("/refresh").toString()),
+        new IdOnlyContainerResource("id"));
+
+    assertEquals(1, webServer.getRequestCount());
+
+    RecordedRequest request = webServer.takeRequest();
+    assertEquals("POST", request.getMethod());
+    assertEquals(this.MOCK_EXPORT_SERVICE, request.getHeader("X-DTP-Export-Service"));
+    assertEquals(MOCK_JOB_ID.toString(), request.getHeader("X-DTP-Job-Id"));
+    assertTrue(executor.getErrors().isEmpty());
+  }
 }

extensions/data-transfer/portability-data-transfer-generic/src/test/java/org/datatransferproject/datatransfer/generic/GenericImporterTest.java

@@ -19,6 +19,7 @@
 import org.datatransferproject.spi.cloud.storage.TemporaryPerJobDataStore;
 import org.datatransferproject.spi.transfer.idempotentexecutor.InMemoryIdempotentImportExecutor;
 import org.datatransferproject.spi.transfer.types.DestinationMemoryFullException;
+import org.datatransferproject.transfer.JobMetadata;
 import org.datatransferproject.types.common.models.IdOnlyContainerResource;
 import org.datatransferproject.types.transfer.auth.AppCredentials;
 import org.datatransferproject.types.transfer.auth.TokensAndUrlAuthData;
@@ -30,13 +31,18 @@
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameter;
 import org.junit.runners.Parameterized.Parameters;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
 
 @RunWith(Parameterized.class)
 public class GenericImporterTest {
   private final Monitor monitor = new Monitor() {};
   private final TemporaryPerJobDataStore dataStore = new TemporaryPerJobDataStore() {};
   @Parameter public String importerClass;
   private MockWebServer webServer;
+  private static final UUID MOCK_JOB_ID = UUID.fromString("123e4567-e89b-12d3-a456-426614174000");
+  private static final String MOCK_EXPORT_SERVICE = "mockExportService";
+  private MockedStatic<JobMetadata> jobMetadataMock;
 
   @Parameters(name = "{0}")
   public static Collection<String> strings() {
@@ -52,11 +58,16 @@ static void assertContains(String expected, String actual) {
   public void setup() throws IOException {
     webServer = new MockWebServer();
     webServer.start();
+
+    jobMetadataMock = Mockito.mockStatic(JobMetadata.class);
+    jobMetadataMock.when(JobMetadata::getJobId).thenReturn(MOCK_JOB_ID);
+    jobMetadataMock.when(JobMetadata::getExportService).thenReturn(MOCK_EXPORT_SERVICE);
   }
 
   @After
   public void teardown() throws IOException {
     webServer.shutdown();
+    jobMetadataMock.close();
   }
 
   <C> GenericImporter<IdOnlyContainerResource, C> getImporter(
@@ -351,4 +362,33 @@ public void testGenericImporterDestinationFull() throws Exception {
     assertEquals("itemId", error.title());
     assertContains("Generic importer failed with code (413)", error.exception());
   }
+
+  @Test
+  public void testGenericImporterPassingJobMetadata() throws Exception {
+    InMemoryIdempotentImportExecutor executor = new InMemoryIdempotentImportExecutor(monitor);
+    GenericImporter<IdOnlyContainerResource, String> importer =
+        getImporter(
+            importerClass,
+            container ->
+                    List.of(
+                            new ImportableData<>(
+                                    new GenericPayload<>(container.getId(), "schemasource"),
+                                    container.getId(),
+                                    container.getId())));
+    webServer.setDispatcher(getDispatcher());
+
+    importer.importItem(
+        MOCK_JOB_ID,
+        executor,
+        new TokensAndUrlAuthData(
+            "accessToken", "refreshToken", webServer.url("/refresh").toString()),
+        new IdOnlyContainerResource("id"));
+
+    assertEquals(1, webServer.getRequestCount());
+    RecordedRequest request = webServer.takeRequest();
+    assertEquals("POST", request.getMethod());
+    assertEquals(this.MOCK_EXPORT_SERVICE, request.getHeader("X-DTP-Export-Service"));
+    assertEquals(MOCK_JOB_ID.toString(), request.getHeader("X-DTP-Job-Id"));
+    assertTrue(executor.getErrors().isEmpty());
+  }
 }