diff --git a/src/main/java/io/github/jpmorganchase/fusion/Fusion.java b/src/main/java/io/github/jpmorganchase/fusion/Fusion.java index 03d4459..0d7f331 100644 --- a/src/main/java/io/github/jpmorganchase/fusion/Fusion.java +++ b/src/main/java/io/github/jpmorganchase/fusion/Fusion.java @@ -2,6 +2,10 @@ import static io.github.jpmorganchase.fusion.filter.DatasetFilter.filterDatasets; +import com.google.gson.Gson; +import com.google.gson.JsonArray; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; import io.github.jpmorganchase.fusion.api.APIManager; import io.github.jpmorganchase.fusion.api.FusionAPIManager; import io.github.jpmorganchase.fusion.api.exception.APICallException; @@ -11,6 +15,7 @@ import io.github.jpmorganchase.fusion.builders.APIConfiguredBuilders; import io.github.jpmorganchase.fusion.builders.Builders; import io.github.jpmorganchase.fusion.http.Client; +import io.github.jpmorganchase.fusion.http.HttpResponse; import io.github.jpmorganchase.fusion.http.JdkClient; import io.github.jpmorganchase.fusion.model.*; import io.github.jpmorganchase.fusion.oauth.credential.BearerTokenCredentials; @@ -36,14 +41,20 @@ import java.util.Map; import java.util.Objects; import lombok.Builder; +import lombok.extern.slf4j.Slf4j; /** * Class representing the Fusion API, providing methods that correspond to available API endpoints */ +@Slf4j +@SuppressWarnings({"LombokSetterMayBeUsed", "LombokGetterMayBeUsed"}) public class Fusion { private static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd"); + @SuppressWarnings({"FieldCanBeLocal", "FieldMayBeFinal"}) + private static int defaultPageSize = -1; + private final APIManager api; private String defaultCatalog; private final String defaultPath; @@ -195,6 +206,91 @@ private Map> callForMap(String url) { return responseParser.parseResourcesUntyped(json); } + /** + * Makes paginated API calls and aggregates all results transparently. + * This method handles the pagination logic internally, making multiple API calls + * as needed and combining the results into a single response string. + * + * @param url the API endpoint URL + * @return aggregated JSON response containing all pages of data + */ + private String callAPIWithPagination(String url) { + log.debug("Starting paginated request to URL: {}", url); + + Map headers = new HashMap<>(); + headers.put("x-jpmc-paginate", "true"); + if (defaultPageSize > 0) { + log.debug("Using page size: {}", defaultPageSize); + headers.put("x-jpmc-page-size", String.valueOf(defaultPageSize)); + } + + Gson gson = new Gson(); + JsonArray aggregatedResources = new JsonArray(); + String nextToken = null; + int pageCount = 0; + + do { + pageCount++; + if (nextToken != null) { + headers.put("x-jpmc-next-token", nextToken); + log.debug("Fetching page {} with next token", pageCount); + } else { + log.debug("Fetching page {}", pageCount); + } + + HttpResponse response = this.api.callAPIWithResponse(url, headers); + String pageJson = response.getBody(); + + JsonObject pageObject = JsonParser.parseString(pageJson).getAsJsonObject(); + if (pageObject.has("resources") && pageObject.get("resources").isJsonArray()) { + JsonArray pageResources = pageObject.getAsJsonArray("resources"); + int pageResourceCount = pageResources.size(); + pageResources.forEach(aggregatedResources::add); + log.debug("Retrieved {} resources from page {}", pageResourceCount, pageCount); + } + + nextToken = getHeaderValue(response.getHeaders(), "x-jpmc-next-token"); + + if (nextToken != null && !nextToken.isEmpty()) { + log.debug("Next token received, more pages available"); + } + + } while (nextToken != null && !nextToken.isEmpty()); + + log.debug( + "Pagination complete. Total pages fetched: {}, Total resources: {}", + pageCount, + aggregatedResources.size()); + + JsonObject result = new JsonObject(); + result.add("resources", aggregatedResources); + return gson.toJson(result); + } + + /** + * Gets a header value from the response headers map (case-insensitive). + * + * @param headers the response headers map + * @param headerName the header name to look for + * @return the header value, or null if not found + */ + @SuppressWarnings("SameParameterValue") + private String getHeaderValue(Map> headers, String headerName) { + if (headers == null || headerName == null) { + return null; + } + + for (Map.Entry> entry : headers.entrySet()) { + if (entry.getKey() != null && entry.getKey().equalsIgnoreCase(headerName)) { + List values = entry.getValue(); + if (values != null && !values.isEmpty()) { + return values.get(0); + } + } + } + return null; + } + /** * Get a list of the catalogs available to the API account. * @@ -203,7 +299,8 @@ private Map> callForMap(String url) { * @throws OAuthException if a token could not be retrieved for authentication */ public Map listCatalogs() { - String json = this.api.callAPI(rootURL.concat("catalogs")); + String url = rootURL.concat("catalogs"); + String json = callAPIWithPagination(url); return responseParser.parseCatalogResponse(json); } @@ -223,7 +320,7 @@ public Map> catalogResources(String catalogName) { /** * Get a filtered list of the data products in the specified catalog *

- * Note that as of current version this search capability is not yet implemented + * Note that as of the current version, this search capability is not yet implemented * * @param catalogName identifier of the catalog to be queried * @param contains a search keyword. @@ -235,7 +332,7 @@ public Map> catalogResources(String catalogName) { public Map listProducts(String catalogName, String contains, boolean idContains) { // TODO: unimplemented logic implied by the method parameters String url = String.format("%1scatalogs/%2s/products", this.rootURL, catalogName); - String json = this.api.callAPI(url); + String json = callAPIWithPagination(url); return responseParser.parseDataProductResponse(json); } @@ -266,7 +363,7 @@ public Map listProducts() { /** * Get a filtered list of the datasets in the specified catalog *

- * Note that as of current version this search capability is not yet implemented + * Note that as of the current version, this search capability is not yet implemented * * @param catalogName identifier of the catalog to be queried * @param contains a search keyword. @@ -277,7 +374,7 @@ public Map listProducts() { */ public Map listDatasets(String catalogName, String contains, boolean idContains) { String url = String.format("%1scatalogs/%2s/datasets", this.rootURL, catalogName); - String json = this.api.callAPI(url); + String json = callAPIWithPagination(url); return filterDatasets(responseParser.parseDatasetResponse(json, catalogName), contains, idContains); } @@ -400,7 +497,7 @@ public Map> datasetResources(String dataset) { */ public Map listDatasetMembers(String catalogName, String dataset) { String url = String.format("%1scatalogs/%2s/datasets/%3s/datasetseries", this.rootURL, catalogName, dataset); - String json = this.api.callAPI(url); + String json = callAPIWithPagination(url); return responseParser.parseDatasetSeriesResponse(json); } @@ -460,7 +557,7 @@ public Map> datasetMemberResources(String dataset, S */ public Map listAttributes(String catalogName, String dataset) { String url = String.format("%1scatalogs/%2s/datasets/%3s/attributes", this.rootURL, catalogName, dataset); - String json = this.api.callAPI(url); + String json = callAPIWithPagination(url); return responseParser.parseAttributeResponse(json, catalogName, dataset); } @@ -502,11 +599,10 @@ public Map> attributeResources(String catalogName, S * @throws OAuthException if a token could not be retrieved for authentication */ public Map listDistributions(String catalogName, String dataset, String seriesMember) { - String url = String.format( "%1scatalogs/%2s/datasets/%3s/datasetseries/%4s/distributions", this.rootURL, catalogName, dataset, seriesMember); - String json = this.api.callAPI(url); + String json = callAPIWithPagination(url); return responseParser.parseDistributionResponse(json); } diff --git a/src/main/java/io/github/jpmorganchase/fusion/api/APIManager.java b/src/main/java/io/github/jpmorganchase/fusion/api/APIManager.java index 7374196..955047a 100644 --- a/src/main/java/io/github/jpmorganchase/fusion/api/APIManager.java +++ b/src/main/java/io/github/jpmorganchase/fusion/api/APIManager.java @@ -3,10 +3,12 @@ import io.github.jpmorganchase.fusion.api.exception.APICallException; import io.github.jpmorganchase.fusion.api.operations.APIDownloadOperations; import io.github.jpmorganchase.fusion.api.operations.APIUploadOperations; +import io.github.jpmorganchase.fusion.http.HttpResponse; import java.io.UnsupportedEncodingException; import java.net.MalformedURLException; import java.net.URL; import java.net.URLEncoder; +import java.util.Map; public interface APIManager extends APIDownloadOperations, APIUploadOperations { @@ -19,6 +21,16 @@ public interface APIManager extends APIDownloadOperations, APIUploadOperations { */ String callAPI(String apiPath) throws APICallException; + /** + * Sends a GET request to the specified API endpoint with custom headers and returns the full HTTP response. + * + * @param apiPath the API endpoint path to which the GET request will be sent + * @param headers additional HTTP headers to include in the request + * @return the full {@code HttpResponse} including headers and body + * @throws APICallException if the response status indicates an error or the request fails + */ + HttpResponse callAPIWithResponse(String apiPath, Map headers) throws APICallException; + String callAPIToPost(String apiPath) throws APICallException; /** diff --git a/src/main/java/io/github/jpmorganchase/fusion/api/FusionAPIManager.java b/src/main/java/io/github/jpmorganchase/fusion/api/FusionAPIManager.java index 41e9368..28abf7f 100644 --- a/src/main/java/io/github/jpmorganchase/fusion/api/FusionAPIManager.java +++ b/src/main/java/io/github/jpmorganchase/fusion/api/FusionAPIManager.java @@ -63,6 +63,38 @@ public String callAPI(String apiPath) throws APICallException { return response.getBody(); } + /** + * Sends a GET request to the specified API endpoint with custom headers and returns the full HTTP response. + * + *

This method constructs the necessary authorization headers using a bearer token from + * the {@code tokenProvider}, merges them with the provided custom headers, and sends a GET + * request to the specified {@code apiPath} using the {@code httpClient}. It checks the HTTP + * response status for errors and returns the full response including headers. + * + * @param apiPath the API endpoint path to which the GET request will be sent + * @param customHeaders additional HTTP headers to include in the request + * @return the full {@code HttpResponse} including status, headers, and body + * @throws APICallException if the response status indicates an error or the request fails + */ + @Override + public HttpResponse callAPIWithResponse(String apiPath, Map customHeaders) + throws APICallException { + Map requestHeaders = new HashMap<>(); + requestHeaders.put("Authorization", "Bearer " + tokenProvider.getSessionBearerToken()); + + if (customHeaders != null) { + customHeaders.forEach((key, value) -> { + if (!"Authorization".equalsIgnoreCase(key)) { + requestHeaders.put(key, value); + } + }); + } + + HttpResponse response = httpClient.get(APIManager.encodeUrl(apiPath), requestHeaders); + checkResponseStatus(response); + return response; + } + @Override public String callAPIToPost(String apiPath) throws APICallException { Map requestHeaders = new HashMap<>(); diff --git a/src/test/java/io/github/jpmorganchase/fusion/FusionTest.java b/src/test/java/io/github/jpmorganchase/fusion/FusionTest.java index ce0cee9..eca8408 100644 --- a/src/test/java/io/github/jpmorganchase/fusion/FusionTest.java +++ b/src/test/java/io/github/jpmorganchase/fusion/FusionTest.java @@ -8,6 +8,7 @@ import io.github.jpmorganchase.fusion.api.APIManager; import io.github.jpmorganchase.fusion.http.Client; +import io.github.jpmorganchase.fusion.http.HttpResponse; import io.github.jpmorganchase.fusion.model.*; import io.github.jpmorganchase.fusion.oauth.credential.BearerTokenCredentials; import io.github.jpmorganchase.fusion.parsing.APIResponseParser; @@ -109,9 +110,17 @@ private Map setupDatasetTest(String catalog) throws Exception { .title("Title datasetOne") .build()); - when(apiManager.callAPI(String.format("%1scatalogs/%2s/datasets", config.getRootURL(), catalog))) - .thenReturn("{\"key\":value}"); - when(responseParser.parseDatasetResponse("{\"key\":value}", catalog)).thenReturn(stubResponse); + HttpResponse httpResponse = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"key\":\"value\"}]}") + .headers(new HashMap<>()) + .build(); + + when(apiManager.callAPIWithResponse( + eq(String.format("%1scatalogs/%2s/datasets", config.getRootURL(), catalog)), anyMap())) + .thenReturn(httpResponse); + when(responseParser.parseDatasetResponse("{\"resources\":[{\"key\":\"value\"}]}", catalog)) + .thenReturn(stubResponse); return stubResponse; } @@ -141,9 +150,17 @@ private Map setupProductTest(String catalog) throws Excepti Map stubResponse = new HashMap<>(); stubResponse.put("first", DataProduct.builder().identifier("product1").build()); - when(apiManager.callAPI(String.format("%1scatalogs/%2s/products", config.getRootURL(), catalog))) - .thenReturn("{\"key\":value}"); - when(responseParser.parseDataProductResponse("{\"key\":value}")).thenReturn(stubResponse); + HttpResponse httpResponse = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"key\":\"value\"}]}") + .headers(new HashMap<>()) + .build(); + + when(apiManager.callAPIWithResponse( + eq(String.format("%1scatalogs/%2s/products", config.getRootURL(), catalog)), anyMap())) + .thenReturn(httpResponse); + when(responseParser.parseDataProductResponse("{\"resources\":[{\"key\":\"value\"}]}")) + .thenReturn(stubResponse); return stubResponse; } @@ -155,10 +172,20 @@ public void testDatasetSeriesInteraction() throws Exception { Map stubResponse = new HashMap<>(); stubResponse.put("first", DatasetSeries.builder().identifier("dataset1").build()); - when(apiManager.callAPI(String.format( - "%1scatalogs/%2s/datasets/%3s/datasetseries", config.getRootURL(), "common", "sample_dataset"))) - .thenReturn("{\"key\":value}"); - when(responseParser.parseDatasetSeriesResponse("{\"key\":value}")).thenReturn(stubResponse); + HttpResponse httpResponse = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"key\":\"value\"}]}") + .headers(new HashMap<>()) + .build(); + + when(apiManager.callAPIWithResponse( + eq(String.format( + "%1scatalogs/%2s/datasets/%3s/datasetseries", + config.getRootURL(), "common", "sample_dataset")), + anyMap())) + .thenReturn(httpResponse); + when(responseParser.parseDatasetSeriesResponse("{\"resources\":[{\"key\":\"value\"}]}")) + .thenReturn(stubResponse); Map actualResponse = f.listDatasetMembers("sample_dataset"); assertThat(actualResponse, is(equalTo(stubResponse))); @@ -171,10 +198,19 @@ public void testAttributeInteraction() throws Exception { Map stubResponse = new HashMap<>(); stubResponse.put("first", Attribute.builder().identifier("attribute1").build()); - when(apiManager.callAPI(String.format( - "%1scatalogs/%2s/datasets/%3s/attributes", config.getRootURL(), "common", "sample_dataset"))) - .thenReturn("{\"key\":value}"); - when(responseParser.parseAttributeResponse("{\"key\":value}", "common", "sample_dataset")) + HttpResponse httpResponse = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"key\":\"value\"}]}") + .headers(new HashMap<>()) + .build(); + + when(apiManager.callAPIWithResponse( + eq(String.format( + "%1scatalogs/%2s/datasets/%3s/attributes", + config.getRootURL(), "common", "sample_dataset")), + anyMap())) + .thenReturn(httpResponse); + when(responseParser.parseAttributeResponse("{\"resources\":[{\"key\":\"value\"}]}", "common", "sample_dataset")) .thenReturn(stubResponse); Map actualResponse = f.listAttributes("sample_dataset"); @@ -208,11 +244,20 @@ public void testDistributionInteraction() throws Exception { stubResponse.put( "first", Distribution.builder().identifier("attribute1").build()); - when(apiManager.callAPI(String.format( - "%1scatalogs/%2s/datasets/%3s/datasetseries/%4s/distributions", - config.getRootURL(), "common", "sample_dataset", "20230308"))) - .thenReturn("{\"key\":value}"); - when(responseParser.parseDistributionResponse("{\"key\":value}")).thenReturn(stubResponse); + HttpResponse httpResponse = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"attribute1\"}]}") + .headers(new HashMap<>()) + .build(); + + when(apiManager.callAPIWithResponse( + eq(String.format( + "%1scatalogs/%2s/datasets/%3s/datasetseries/%4s/distributions", + config.getRootURL(), "common", "sample_dataset", "20230308")), + anyMap())) + .thenReturn(httpResponse); + when(responseParser.parseDistributionResponse("{\"resources\":[{\"identifier\":\"attribute1\"}]}")) + .thenReturn(stubResponse); Map actualResponse = f.listDistributions("sample_dataset", "20230308"); assertThat(actualResponse, is(equalTo(stubResponse))); @@ -588,9 +633,16 @@ public void testListCatalogsInteraction() throws Exception { Map stubResponse = new HashMap<>(); stubResponse.put("first", Catalog.builder().identifier("catalog1").build()); - when(apiManager.callAPI(String.format("%1scatalogs", config.getRootURL()))) - .thenReturn("{\"key\":value}"); - when(responseParser.parseCatalogResponse("{\"key\":value}")).thenReturn(stubResponse); + HttpResponse httpResponse = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"catalog1\"}]}") + .headers(new HashMap<>()) + .build(); + + when(apiManager.callAPIWithResponse(eq(String.format("%1scatalogs", config.getRootURL())), anyMap())) + .thenReturn(httpResponse); + when(responseParser.parseCatalogResponse("{\"resources\":[{\"identifier\":\"catalog1\"}]}")) + .thenReturn(stubResponse); Map actualResponse = f.listCatalogs(); assertThat(actualResponse, is(equalTo(stubResponse))); @@ -980,4 +1032,275 @@ public void testDownloadWithNonExistentFilesThrowsException() throws Exception { equalTo( "The following requested files do not exist in catalog=common, dataset=sample_dataset, series=20230308, distribution=csv: nonexistent_file, another_missing_file"))); } + + @Test + public void testListDatasetsWithPagination() throws Exception { + Fusion f = stubFusion(); + + Map> headers1 = new HashMap<>(); + headers1.put("x-jpmc-next-token", Collections.singletonList("token123")); + + HttpResponse httpResponse1 = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"dataset1\",\"description\":\"First\"}]}") + .headers(headers1) + .build(); + + HttpResponse httpResponse2 = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"dataset2\",\"description\":\"Second\"}]}") + .headers(new HashMap<>()) + .build(); + + when(apiManager.callAPIWithResponse( + eq(String.format("%1scatalogs/%2s/datasets", config.getRootURL(), "common")), + argThat(headers -> !headers.containsKey("x-jpmc-next-token")))) + .thenReturn(httpResponse1); + + when(apiManager.callAPIWithResponse( + eq(String.format("%1scatalogs/%2s/datasets", config.getRootURL(), "common")), + argThat(headers -> "token123".equals(headers.get("x-jpmc-next-token"))))) + .thenReturn(httpResponse2); + + Map stubResponse = new HashMap<>(); + stubResponse.put("dataset1", Dataset.builder().identifier("dataset1").build()); + stubResponse.put("dataset2", Dataset.builder().identifier("dataset2").build()); + + String aggregatedJson = + "{\"resources\":[{\"identifier\":\"dataset1\",\"description\":\"First\"},{\"identifier\":\"dataset2\",\"description\":\"Second\"}]}"; + when(responseParser.parseDatasetResponse(aggregatedJson, "common")).thenReturn(stubResponse); + + Map actualResponse = f.listDatasets("common"); + assertThat(actualResponse.size(), is(equalTo(2))); + assertThat(actualResponse.containsKey("dataset1"), is(true)); + assertThat(actualResponse.containsKey("dataset2"), is(true)); + + verify(apiManager, times(2)).callAPIWithResponse(anyString(), anyMap()); + } + + @Test + public void testListProductsWithPagination() throws Exception { + Fusion f = stubFusion(); + + Map> headers1 = new HashMap<>(); + headers1.put("x-jpmc-next-token", Collections.singletonList("productToken")); + + HttpResponse httpResponse1 = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"product1\"}]}") + .headers(headers1) + .build(); + + HttpResponse httpResponse2 = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"product2\"}]}") + .headers(new HashMap<>()) + .build(); + + when(apiManager.callAPIWithResponse( + eq(String.format("%1scatalogs/%2s/products", config.getRootURL(), "common")), + argThat(headers -> !headers.containsKey("x-jpmc-next-token")))) + .thenReturn(httpResponse1); + + when(apiManager.callAPIWithResponse( + eq(String.format("%1scatalogs/%2s/products", config.getRootURL(), "common")), + argThat(headers -> "productToken".equals(headers.get("x-jpmc-next-token"))))) + .thenReturn(httpResponse2); + + Map stubResponse = new HashMap<>(); + stubResponse.put( + "product1", DataProduct.builder().identifier("product1").build()); + stubResponse.put( + "product2", DataProduct.builder().identifier("product2").build()); + + String aggregatedJson = "{\"resources\":[{\"identifier\":\"product1\"},{\"identifier\":\"product2\"}]}"; + when(responseParser.parseDataProductResponse(aggregatedJson)).thenReturn(stubResponse); + + Map actualResponse = f.listProducts("common"); + assertThat(actualResponse.size(), is(equalTo(2))); + + verify(apiManager, times(2)).callAPIWithResponse(anyString(), anyMap()); + } + + @Test + public void testListAttributesWithPagination() throws Exception { + Fusion f = stubFusion(); + + Map> headers1 = new HashMap<>(); + headers1.put("x-jpmc-next-token", Collections.singletonList("attrToken")); + + HttpResponse httpResponse1 = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"attr1\"}]}") + .headers(headers1) + .build(); + + HttpResponse httpResponse2 = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"attr2\"}]}") + .headers(new HashMap<>()) + .build(); + + when(apiManager.callAPIWithResponse( + eq(String.format( + "%1scatalogs/%2s/datasets/%3s/attributes", + config.getRootURL(), "common", "sample_dataset")), + argThat(headers -> !headers.containsKey("x-jpmc-next-token")))) + .thenReturn(httpResponse1); + + when(apiManager.callAPIWithResponse( + eq(String.format( + "%1scatalogs/%2s/datasets/%3s/attributes", + config.getRootURL(), "common", "sample_dataset")), + argThat(headers -> "attrToken".equals(headers.get("x-jpmc-next-token"))))) + .thenReturn(httpResponse2); + + Map stubResponse = new HashMap<>(); + stubResponse.put("attr1", Attribute.builder().identifier("attr1").build()); + stubResponse.put("attr2", Attribute.builder().identifier("attr2").build()); + + String aggregatedJson = "{\"resources\":[{\"identifier\":\"attr1\"},{\"identifier\":\"attr2\"}]}"; + when(responseParser.parseAttributeResponse(aggregatedJson, "common", "sample_dataset")) + .thenReturn(stubResponse); + + Map actualResponse = f.listAttributes("common", "sample_dataset"); + assertThat(actualResponse.size(), is(equalTo(2))); + + verify(apiManager, times(2)).callAPIWithResponse(anyString(), anyMap()); + } + + @Test + public void testListDatasetMembersWithPagination() throws Exception { + Fusion f = stubFusion(); + + Map> headers1 = new HashMap<>(); + headers1.put("x-jpmc-next-token", Collections.singletonList("seriesToken")); + + HttpResponse httpResponse1 = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"series1\"}]}") + .headers(headers1) + .build(); + + HttpResponse httpResponse2 = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"series2\"}]}") + .headers(new HashMap<>()) + .build(); + + when(apiManager.callAPIWithResponse( + eq(String.format( + "%1scatalogs/%2s/datasets/%3s/datasetseries", + config.getRootURL(), "common", "sample_dataset")), + argThat(headers -> !headers.containsKey("x-jpmc-next-token")))) + .thenReturn(httpResponse1); + + when(apiManager.callAPIWithResponse( + eq(String.format( + "%1scatalogs/%2s/datasets/%3s/datasetseries", + config.getRootURL(), "common", "sample_dataset")), + argThat(headers -> "seriesToken".equals(headers.get("x-jpmc-next-token"))))) + .thenReturn(httpResponse2); + + Map stubResponse = new HashMap<>(); + stubResponse.put( + "series1", DatasetSeries.builder().identifier("series1").build()); + stubResponse.put( + "series2", DatasetSeries.builder().identifier("series2").build()); + + when(responseParser.parseDatasetSeriesResponse(anyString())).thenReturn(stubResponse); + + Map actualResponse = f.listDatasetMembers("common", "sample_dataset"); + assertThat(actualResponse.size(), is(equalTo(2))); + + verify(apiManager, times(2)).callAPIWithResponse(anyString(), anyMap()); + } + + @Test + public void testListCatalogsWithPagination() throws Exception { + Fusion f = stubFusion(); + + Map> headers1 = new HashMap<>(); + headers1.put("x-jpmc-next-token", Collections.singletonList("catalogToken")); + + HttpResponse httpResponse1 = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"catalog1\"}]}") + .headers(headers1) + .build(); + + HttpResponse httpResponse2 = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"catalog2\"}]}") + .headers(new HashMap<>()) + .build(); + + when(apiManager.callAPIWithResponse( + eq(String.format("%1scatalogs", config.getRootURL())), + argThat(headers -> !headers.containsKey("x-jpmc-next-token")))) + .thenReturn(httpResponse1); + + when(apiManager.callAPIWithResponse( + eq(String.format("%1scatalogs", config.getRootURL())), + argThat(headers -> "catalogToken".equals(headers.get("x-jpmc-next-token"))))) + .thenReturn(httpResponse2); + + Map stubResponse = new HashMap<>(); + stubResponse.put("catalog1", Catalog.builder().identifier("catalog1").build()); + stubResponse.put("catalog2", Catalog.builder().identifier("catalog2").build()); + + String aggregatedJson = "{\"resources\":[{\"identifier\":\"catalog1\"},{\"identifier\":\"catalog2\"}]}"; + when(responseParser.parseCatalogResponse(aggregatedJson)).thenReturn(stubResponse); + + Map actualResponse = f.listCatalogs(); + assertThat(actualResponse.size(), is(equalTo(2))); + + verify(apiManager, times(2)).callAPIWithResponse(anyString(), anyMap()); + } + + @Test + public void testListDistributionsWithPagination() throws Exception { + Fusion f = stubFusion(); + + Map> headers1 = new HashMap<>(); + headers1.put("x-jpmc-next-token", Collections.singletonList("distToken")); + + HttpResponse httpResponse1 = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"csv\"}]}") + .headers(headers1) + .build(); + + HttpResponse httpResponse2 = HttpResponse.builder() + .statusCode(200) + .body("{\"resources\":[{\"identifier\":\"parquet\"}]}") + .headers(new HashMap<>()) + .build(); + + when(apiManager.callAPIWithResponse( + eq(String.format( + "%1scatalogs/%2s/datasets/%3s/datasetseries/%4s/distributions", + config.getRootURL(), "common", "sample_dataset", "20230308")), + argThat(headers -> !headers.containsKey("x-jpmc-next-token")))) + .thenReturn(httpResponse1); + + when(apiManager.callAPIWithResponse( + eq(String.format( + "%1scatalogs/%2s/datasets/%3s/datasetseries/%4s/distributions", + config.getRootURL(), "common", "sample_dataset", "20230308")), + argThat(headers -> "distToken".equals(headers.get("x-jpmc-next-token"))))) + .thenReturn(httpResponse2); + + Map stubResponse = new HashMap<>(); + stubResponse.put("csv", Distribution.builder().identifier("csv").build()); + stubResponse.put("parquet", Distribution.builder().identifier("parquet").build()); + + String aggregatedJson = "{\"resources\":[{\"identifier\":\"csv\"},{\"identifier\":\"parquet\"}]}"; + when(responseParser.parseDistributionResponse(aggregatedJson)).thenReturn(stubResponse); + + Map actualResponse = f.listDistributions("common", "sample_dataset", "20230308"); + assertThat(actualResponse.size(), is(equalTo(2))); + + verify(apiManager, times(2)).callAPIWithResponse(anyString(), anyMap()); + } }