Core: Refactor: Test suite for freshness-aware loading

core/src/test/java/org/apache/iceberg/rest/RequestMatcher.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.rest;
+
+import static org.mockito.ArgumentMatchers.argThat;
+
+import java.util.Map;
+import java.util.Objects;
+
+class RequestMatcher {
+  private RequestMatcher() {}
+
+  public static HTTPRequest matches(HTTPRequest.HTTPMethod method) {
+    return argThat(req -> req.method() == method);
+  }
+
+  static HTTPRequest matches(HTTPRequest.HTTPMethod method, String path) {
+    return argThat(req -> req.method() == method && req.path().equals(path));
+  }
+
+  public static HTTPRequest matches(
+      HTTPRequest.HTTPMethod method, String path, Map<String, String> headers) {
+    return argThat(
+        req ->
+            req.method() == method
+                && req.path().equals(path)
+                && req.headers().equals(HTTPHeaders.of(headers)));
+  }
+
+  public static HTTPRequest matches(
+      HTTPRequest.HTTPMethod method,
+      String path,
+      Map<String, String> headers,
+      Map<String, String> parameters) {
+    return argThat(
+        req ->
+            req.method() == method
+                && req.path().equals(path)
+                && req.headers().equals(HTTPHeaders.of(headers))
+                && req.queryParameters().equals(parameters));
+  }
+
+  public static HTTPRequest matches(
+      HTTPRequest.HTTPMethod method,
+      String path,
+      Map<String, String> headers,
+      Map<String, String> parameters,
+      Object body) {
+    return argThat(
+        req ->
+            req.method() == method
+                && req.path().equals(path)
+                && req.headers().equals(HTTPHeaders.of(headers))
+                && req.queryParameters().equals(parameters)
+                && Objects.equals(req.body(), body));
+  }
+
+  public static HTTPRequest containsHeaders(
+      HTTPRequest.HTTPMethod method, String path, Map<String, String> headers) {
+    return argThat(
+        req ->
+            req.method() == method
+                && req.path().equals(path)
+                && req.headers().entries().containsAll(HTTPHeaders.of(headers).entries()));
+  }
+}

core/src/test/java/org/apache/iceberg/rest/TestBaseWithRESTServer.java

@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.rest;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.File;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Consumer;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SessionCatalog;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.rest.responses.ErrorResponse;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.handler.gzip.GzipHandler;
+import org.eclipse.jetty.servlet.ServletContextHandler;
+import org.eclipse.jetty.servlet.ServletHolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mockito;
+
+public abstract class TestBaseWithRESTServer {
+  protected static final ObjectMapper MAPPER = RESTObjectMapper.mapper();
+  protected static final Namespace NS = Namespace.of("ns");
+  protected static final SessionCatalog.SessionContext DEFAULT_SESSION_CONTEXT =
+      new SessionCatalog.SessionContext(
+          UUID.randomUUID().toString(),
+          "user",
+          ImmutableMap.of("credential", "user:12345"),
+          ImmutableMap.of());
+
+  protected InMemoryCatalog backendCatalog;
+  protected RESTCatalogAdapter adapterForRESTServer;
+  protected Server httpServer;
+  protected RESTCatalog restCatalog;
+  protected ParserContext parserContext;
+
+  @TempDir private Path temp;
+
+  @BeforeEach
+  public void before() throws Exception {
+    File warehouse = temp.toFile();
+    this.backendCatalog = new InMemoryCatalog();
+    this.backendCatalog.initialize(
+        "in-memory",
+        ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, warehouse.getAbsolutePath()));
+
+    adapterForRESTServer = createAdapterForServer();
+
+    ServletContextHandler servletContext =
+        new ServletContextHandler(ServletContextHandler.NO_SESSIONS);
+    servletContext.addServlet(
+        new ServletHolder(new RESTCatalogServlet(adapterForRESTServer)), "/*");
+    servletContext.setHandler(new GzipHandler());
+
+    this.httpServer = new Server(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
+    httpServer.setHandler(servletContext);
+    httpServer.start();
+
+    restCatalog = initCatalog(catalogName(), ImmutableMap.of());
+  }
+
+  @AfterEach
+  public void after() throws Exception {
+    if (restCatalog != null) {
+      restCatalog.close();
+    }
+
+    if (backendCatalog != null) {
+      backendCatalog.close();
+    }
+
+    if (httpServer != null) {
+      httpServer.stop();
+      httpServer.join();
+    }
+  }
+
+  protected RESTCatalogAdapter createAdapterForServer() {
+    return Mockito.spy(
+        new RESTCatalogAdapter(backendCatalog) {
+          @Override
+          public <T extends RESTResponse> T execute(
+              HTTPRequest request,
+              Class<T> responseType,
+              Consumer<ErrorResponse> errorHandler,
+              Consumer<Map<String, String>> responseHeaders) {
+            Object body = roundTripSerialize(request.body(), "request");
+            HTTPRequest req = ImmutableHTTPRequest.builder().from(request).body(body).build();
+            T response = super.execute(req, responseType, errorHandler, responseHeaders);
+            return roundTripSerialize(response, "response");
+          }
+        });
+  }
+
+  protected abstract String catalogName();
+
+  @SuppressWarnings("unchecked")
+  protected <T> T roundTripSerialize(T payload, String description) {
+    if (payload != null) {
+      try {
+        if (payload instanceof RESTMessage) {
+          return (T) MAPPER.readValue(MAPPER.writeValueAsString(payload), payload.getClass());
+        } else {
+          // use Map so that Jackson doesn't try to instantiate ImmutableMap from payload.getClass()
+          return (T) MAPPER.readValue(MAPPER.writeValueAsString(payload), Map.class);
+        }
+      } catch (JsonProcessingException e) {
+        throw new RuntimeException(
+            String.format("Failed to serialize and deserialize %s: %s", description, payload), e);
+      }
+    }
+    return null;
+  }
+
+  private RESTCatalog initCatalog(String catalogName, Map<String, String> additionalProperties) {
+    RESTCatalog catalog =
+        new RESTCatalog(
+            DEFAULT_SESSION_CONTEXT,
+            (config) ->
+                HTTPClient.builder(config)
+                    .uri(config.get(CatalogProperties.URI))
+                    .withHeaders(RESTUtil.configHeaders(config))
+                    .build());
+    catalog.setConf(new Configuration());
+    Map<String, String> properties =
+        ImmutableMap.of(
+            CatalogProperties.URI,
+            httpServer.getURI().toString(),
+            CatalogProperties.FILE_IO_IMPL,
+            "org.apache.iceberg.inmemory.InMemoryFileIO");
+    catalog.initialize(
+        catalogName,
+        ImmutableMap.<String, String>builder()
+            .putAll(properties)
+            .putAll(additionalProperties)
+            .build());
+
+    return catalog;
+  }
+}