Revert "[CYB-207] use https for mock server for all tests."

This reverts commit 9a2e01e.

Author: carolynduby

flink-cyber/flink-enrichment/flink-enrichment-lookup-rest/src/main/java/com/cloudera/cyber/enrichment/rest/AsyncHttpRequest.java

@@ -26,6 +26,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.function.Predicate;
 import java.util.List;
 
 import static com.cloudera.cyber.DataQualityMessageLevel.ERROR;

flink-cyber/flink-enrichment/flink-enrichment-lookup-rest/src/main/java/com/cloudera/cyber/enrichment/rest/RestLookupJob.java

@@ -20,6 +20,7 @@
 import com.fasterxml.jackson.databind.jsontype.BasicPolymorphicTypeValidator;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.streaming.api.TimeCharacteristic;
 import org.apache.flink.streaming.api.datastream.AsyncDataStream;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -29,7 +30,6 @@
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.ArrayList;
-import java.util.Base64;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -60,7 +60,7 @@ private static DataStream<Message> enrich(DataStream<Message> source, List<RestE
             Preconditions.checkNotNull(config.getSources(),"specify a list of source names or ANY to match any source");
             Preconditions.checkArgument(!config.getSources().isEmpty(), "specify a list of source names or ANY to match any source");
             AsyncHttpRequest asyncHttpRequest = new AsyncHttpRequest(config);
-            String processId = "rest-" + Base64.getEncoder().encodeToString(md5(source + config.getEndpointTemplate()));
+            String processId = "rest-" + md5(source + config.getEndpointTemplate());
 
             return AsyncDataStream.unorderedWait(
                     in,
@@ -72,6 +72,7 @@ private static DataStream<Message> enrich(DataStream<Message> source, List<RestE
 
     protected StreamExecutionEnvironment createPipeline(ParameterTool params) throws IOException {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
         env.enableCheckpointing(5000);
         FlinkUtils.setupEnv(env, params);
 

flink-cyber/flink-enrichment/flink-enrichment-lookup-rest/src/test/java/com/cloudera/cyber/enrichment/rest/AsyncHttpRequestTest.java

@@ -32,14 +32,16 @@
 
 public class AsyncHttpRequestTest extends RestRequestTest {
 
+    private static MockRestServer mockRestServer;
+
     @BeforeClass
-    public static void createMockServices() {
-        startMockServer();
+    public static void startMockServer() {
+        mockRestServer = new MockRestServer(false);
     }
 
     @AfterClass
-    public static void stopMockServices() {
-        stopMockServer();
+    public static void stopMockServer() {
+        mockRestServer.close();
     }
 
     private Message buildMessage(String source, String fieldName, String fieldValue) {
@@ -131,7 +133,7 @@ public void testGetError() throws Exception {
         }};
 
         String mockHostAndPort = mockRestServer.getMockHostAndPort();
-        String url = String.format("%s://%s/asset?id=56", MockRestServer.ENCRYPTED_PROTOCOL, mockHostAndPort);
+        String url = String.format("http://%s/asset?id=56", mockHostAndPort);
         String exceptionMessage = String.format(RestRequest.REST_REQUEST_HTTP_FAILURE, "HTTP/1.1 503 Service Unavailable");
         List<DataQualityMessage> expectedQualityMessages = Collections.singletonList(DataQualityMessage.builder().
                 feature(AsyncHttpRequest.REST_ENRICHMENT_FEATURE).
@@ -177,7 +179,7 @@ private void testGetAsset(String fieldName, String assetId, Map<String, Object>
     }
 
     private void testGetAsset(ArrayList<String> enrichmentSources, String messageSource, String fieldName, String assetId, Map<String, Object> expectedExtensions, Collection<DataQualityMessage> expectedDataQualityMessages) throws Exception {
-        AsyncHttpRequest  request = new AsyncHttpRequest(mockRestServer.configureGetAssetRequest(MockRestServer.ENCRYPTED_PROTOCOL).sources(enrichmentSources).build());
+        AsyncHttpRequest  request = new AsyncHttpRequest(mockRestServer.configureGetAssetRequest().sources(enrichmentSources).build());
         request.open(new Configuration());
 
         AsyncResult result = new AsyncResult();

flink-cyber/flink-enrichment/flink-enrichment-lookup-rest/src/test/java/com/cloudera/cyber/enrichment/rest/GetRestRequestTest.java

@@ -15,30 +15,25 @@
 import com.cloudera.cyber.enrichment.rest.impl.MockRestServer;
 import com.google.common.collect.Lists;
 import lombok.extern.slf4j.Slf4j;
-import org.junit.AfterClass;
 import org.junit.Assert;
-import org.junit.BeforeClass;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import java.util.HashMap;
 import java.util.Map;
 
 @Slf4j
+// tests will run either with or without tls when junit runs derived classes
+@Ignore
 public class GetRestRequestTest extends RestRequestTest {
 
-    @BeforeClass
-    public static void createMockServices() {
-        startMockServer();
-    }
-
-    @AfterClass
-    public static void stopMockServices() {
-        stopMockServer();
+    public static void createMockService(boolean enableTlsMutualAuth) {
+        startMockServer(enableTlsMutualAuth);
     }
 
     @Test
     public void testSimpleGet() throws Exception {
-        RestEnrichmentConfig config = mockRestServer.configureGetUserRequest(MockRestServer.ENCRYPTED_PROTOCOL).build();
+        RestEnrichmentConfig config = mockRestServer.configureGetUserRequest().build();
 
         Map<String, String> variables = new HashMap<String, String>() {{
             put("name", "Chris");
@@ -49,7 +44,7 @@ public void testSimpleGet() throws Exception {
 
     @Test
     public void testPropertiesNull() throws Exception {
-        RestEnrichmentConfig config = mockRestServer.getBuilder(null, MockRestServer.ENCRYPTED_PROTOCOL).sources(Lists.newArrayList(MockRestServer.USER_SOURCE)).
+        RestEnrichmentConfig config = mockRestServer.getBuilder(null).sources(Lists.newArrayList(MockRestServer.USER_SOURCE)).
                 endpointTemplate(String.format("%s://%s/user?name=${name}", mockRestServer.getMockProtocol(), mockRestServer.getMockHostAndPort())).build();
 
         Map<String, String> variables = new HashMap<String, String>() {{
@@ -62,18 +57,18 @@ public void testPropertiesNull() throws Exception {
 
     @Test
     public void testBasicAuthGet() throws Exception {
-        RestRequestResult result = makeAssetRequest(MockRestServer.ASSET_ID, mockRestServer);
+        RestRequestResult result = makeAssetRequest(MockRestServer.ASSET_ID);
         Assert.assertEquals(MockRestServer.ASSET_LOCATION, result.getExtensions().get(MockRestServer.ASSET_LOCATION_PROPERTY));
     }
 
     @Test
     public void testServerError() throws Exception {
-        RestRequestResult result = makeAssetRequest(MockRestServer.SERVER_ERROR_ASSET_ID, mockRestServer);
-        verifyErrorResult(result,  "Rest request url='%s://%s/asset?id=56' entity='null' failed 'Rest request failed due to 'HTTP/1.1 503 Service Unavailable'.'", mockRestServer);
+        RestRequestResult result = makeAssetRequest(MockRestServer.SERVER_ERROR_ASSET_ID);
+        verifyErrorResult(result,  "Rest request url='%s://%s/asset?id=56' entity='null' failed 'Rest request failed due to 'HTTP/1.1 503 Service Unavailable'.'");
     }
 
-    private RestRequestResult makeAssetRequest(String assetId, MockRestServer mockRestServer) throws Exception {
-        RestEnrichmentConfig config = mockRestServer.configureGetAssetRequest(MockRestServer.ENCRYPTED_PROTOCOL).build();
+    private RestRequestResult makeAssetRequest(String assetId) throws Exception {
+        RestEnrichmentConfig config = mockRestServer.configureGetAssetRequest().build();
         Map<String, String> variables = new HashMap<String, String>() {{
             put("id", assetId);
         }};

flink-cyber/flink-enrichment/flink-enrichment-lookup-rest/src/test/java/com/cloudera/cyber/enrichment/rest/NonSecureGetRequestTest.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2020 - 2022 Cloudera. All Rights Reserved.
+ *
+ * This file is licensed 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.
+ *
+ * This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. Refer to the License for the specific permissions and
+ * limitations governing your use of the file.
+ */
+
+package com.cloudera.cyber.enrichment.rest;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+public class NonSecureGetRequestTest extends GetRestRequestTest {
+    @BeforeClass
+    public static void createMockService() {
+        createMockService(false);
+    }
+
+    @AfterClass
+    public static void stopMockServer() {
+        mockRestServer.close();
+    }
+
+}

flink-cyber/flink-enrichment/flink-enrichment-lookup-rest/src/test/java/com/cloudera/cyber/enrichment/rest/NonSecurePostRestRequestTest.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2020 - 2022 Cloudera. All Rights Reserved.
+ *
+ * This file is licensed 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.
+ *
+ * This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. Refer to the License for the specific permissions and
+ * limitations governing your use of the file.
+ */
+
+package com.cloudera.cyber.enrichment.rest;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+public class NonSecurePostRestRequestTest extends PostRestRequestTest {
+
+    @BeforeClass
+    public static void createMockService() {
+        createMockService(false);
+    }
+
+    @AfterClass
+    public static void stopMockServer() {
+        mockRestServer.close();
+    }
+
+}

flink-cyber/flink-enrichment/flink-enrichment-lookup-rest/src/test/java/com/cloudera/cyber/enrichment/rest/PostRestRequestTest.java

@@ -15,10 +15,8 @@
 import com.cloudera.cyber.enrichment.rest.impl.MockRestServer;
 import com.google.common.collect.Lists;
 import org.apache.http.client.methods.HttpPost;
-import org.hamcrest.Matcher;
-import org.junit.AfterClass;
 import org.junit.Assert;
-import org.junit.BeforeClass;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import javax.annotation.Nonnull;
@@ -30,29 +28,22 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executors;
 
-import static org.hamcrest.Matchers.containsString;
-import static org.junit.Assert.assertThat;
-
+// tests will run either with or without tls when junit runs derived classes
+@Ignore
 public class PostRestRequestTest extends RestRequestTest {
     private static RestEnrichmentConfig modelResultPostRequest;
 
-    @BeforeClass
-    public static void createMockServices() {
-        startMockServer();
-        modelResultPostRequest = mockRestServer.configureModelPostRequest(MockRestServer.ENCRYPTED_PROTOCOL).build();
-    }
-
-    @AfterClass
-    public static void stopMockServices() {
-        stopMockServer();
+    public static void createMockService(boolean enableTlsMutualAuth) {
+        startMockServer(enableTlsMutualAuth);
+        modelResultPostRequest = mockRestServer.configureModelPostRequest().build();
     }
 
     @Test
     public void testSimplePostCModel() throws Exception {
         testSuccessfulModelPosts(modelResultPostRequest);
     }
 
-    private void testSuccessfulModelPosts(RestEnrichmentConfig config) throws Exception {
+    protected void testSuccessfulModelPosts(RestEnrichmentConfig config) throws Exception {
         for(MockRestServer.ExpectedModelResult expectedModelResult : MockRestServer.expectedModelResults) {
             if (expectedModelResult.isSuccess()) {
                 testDomain(config, expectedModelResult.getDomainName(), expectedModelResult.isLegit());
@@ -63,24 +54,24 @@ private void testSuccessfulModelPosts(RestEnrichmentConfig config) throws Except
     @Test
     public void testHttp200StatusButResultIndicatesError() throws Exception {
         String expectedErrorFormat = "Rest request url='%s://%s/model' entity='{\"accessKey\":\"mup8kz1hsl3erczwepbt8jupamita6y6\",\"request\":{\"domain\":\"unsuccessfuldomain\"}}' failed 'Rest request returned a success code but the content indicated failure.'";
-        testHttpFailedRequest(MockRestServer.UNSUCCESSFUL_DOMAIN, expectedErrorFormat, modelResultPostRequest, mockRestServer);
-    }
+        testHttpFailedRequest(MockRestServer.UNSUCCESSFUL_DOMAIN, expectedErrorFormat);
 
+    }
     @Test
     public void testHttp400Exception() throws Exception {
         String expectedErrorFormat = "Rest request url='%s://%s/model' entity='{\"accessKey\":\"mup8kz1hsl3erczwepbt8jupamita6y6\",\"request\":{\"domain\":\"clienterror\"}}' failed 'Rest request failed due to 'HTTP/1.1 400 Bad Request'.'";
-        testHttpFailedRequest(MockRestServer.CLIENT_ERROR_DOMAIN, expectedErrorFormat, modelResultPostRequest, mockRestServer);
+        testHttpFailedRequest(MockRestServer.CLIENT_ERROR_DOMAIN, expectedErrorFormat);
     }
 
-    private void testHttpFailedRequest(String domainName, String errorMessageFormat, RestEnrichmentConfig modelResultPostRequest, MockRestServer mockRestServer) throws Exception {
+    public void testHttpFailedRequest(String domainName, String errorMessageFormat) throws Exception {
         RestRequest request = modelResultPostRequest.createRestEnrichmentRequest();
         Map<String, String> variables = new HashMap<String, String>() {
             {
                 put(MockRestServer.DOMAIN_EXTENSION_NAME, domainName);
             }
         };
         RestRequestResult result = request.getResult(true, variables).get();
-        verifyErrorResult(result, errorMessageFormat, mockRestServer);
+        verifyErrorResult(result, errorMessageFormat);
     }
 
     private static class MockThrowingPostRequest extends PostRestRequest {
@@ -96,52 +87,30 @@ protected void addEntityToRequest(@Nonnull HttpPost postRequest, @Nonnull String
 
     @Test
     public void testUnsupportedCharacterSetException() throws Exception {
+
         MockThrowingPostRequest throwingPostRequest = new MockThrowingPostRequest(modelResultPostRequest);
         RestRequestKey key = throwingPostRequest.getKey(new HashMap<String, String>() {{ put(MockRestServer.DOMAIN_EXTENSION_NAME, "mydomain");}});
         CompletableFuture<RestRequestResult> future = new MockThrowingPostRequest(modelResultPostRequest).asyncLoad(key, Executors.newFixedThreadPool(1));
         RestRequestResult result = future.get();
         String expectedErrorFormat = "Rest request url='%s://%s/model' entity='{\"accessKey\":\"mup8kz1hsl3erczwepbt8jupamita6y6\",\"request\":{\"domain\":\"mydomain\"}}' failed 'Default UnsupportedEndcodingException message'";
-        verifyErrorResult(result, expectedErrorFormat, mockRestServer);
+        verifyErrorResult(result, expectedErrorFormat);
     }
 
     @Test
     public void testUndefinedVariableWithAnyMatchIgnoresKeyErrors() throws Exception {
-        RestEnrichmentConfig anyMatchRequest = mockRestServer.configureModelPostRequest(MockRestServer.ENCRYPTED_PROTOCOL).sources(Lists.newArrayList(AsyncHttpRequest.ANY_SOURCE_NAME)).build();
+        RestEnrichmentConfig anyMatchRequest = mockRestServer.configureModelPostRequest().sources(Lists.newArrayList(AsyncHttpRequest.ANY_SOURCE_NAME)).build();
         // accessing an undefined field in a key with any source match does not report an error
         testDomain(anyMatchRequest, new ArrayList<>());
     }
 
-    @Test
-    public void testTlsWithAlias() throws Exception {
-        RestEnrichmentConfig.RestEnrichmentConfigBuilder modelResultPostRequestWithKeyAlias = mockRestServer.configureTLS(mockRestServer.configureModelPostRequest(MockRestServer.ENCRYPTED_PROTOCOL), "client");
-        testSuccessfulModelPosts(modelResultPostRequestWithKeyAlias.build());
-    }
-
-    @Test
-    public void testTlsHandshakeException() throws Exception {
-        RestEnrichmentConfig.RestEnrichmentConfigBuilder modelResultPostRequestWithKeyAlias = mockRestServer.configureTLS(mockRestServer.configureModelPostRequest(MockRestServer.ENCRYPTED_PROTOCOL), "nosuchkeyalias");
-        RestRequest badHandshakePost = new PostRestRequest(modelResultPostRequestWithKeyAlias.build());
-        MockRestServer.ExpectedModelResult expectedResult = MockRestServer.expectedModelResults.get(0);
-        Map<String, String> extensions = new HashMap<String, String>() {{
-            put(MockRestServer.DOMAIN_EXTENSION_NAME, expectedResult.getDomainName());
-        }};
-        RestRequestResult result = badHandshakePost.getResult(true, extensions).get();
-
-        Assert.assertTrue(result.getExtensions().isEmpty());
-        List<String> errors = result.getErrors();
-        Assert.assertEquals(1, errors.size());
-        Matcher<String> expectedString = containsString(String.format("Rest request url='%s://%s/model' entity='{\"accessKey\":\"mup8kz1hsl3erczwepbt8jupamita6y6\",\"request\":{\"domain\":\"google\"}}'", mockRestServer.getMockProtocol(), mockRestServer.getMockHostAndPort()));
-        assertThat(errors.get(0), expectedString);
-    }
-
     @Test
     public void testUndefinedVariableWithSpecificMatchReturnsKeyErrors() throws Exception {
-        RestEnrichmentConfig anyMatchRequest = mockRestServer.configureModelPostRequest(MockRestServer.ENCRYPTED_PROTOCOL).build();
+        RestEnrichmentConfig anyMatchRequest = mockRestServer.configureModelPostRequest().build();
         // accessing an undefined field in a key with a specific source match fails with a key error
         testDomain(anyMatchRequest, Lists.newArrayList("Variable(s) 'domain' required by rest entity are undefined"));
     }
 
-    private void testDomain(RestEnrichmentConfig config, String domainName, Boolean legit) throws Exception {
+    protected void testDomain(RestEnrichmentConfig config, String domainName, Boolean legit) throws Exception {
         Map<String, String> variables = new HashMap<String, String>() {{
             put(MockRestServer.DOMAIN_EXTENSION_NAME, domainName);
         }};
@@ -150,7 +119,7 @@ private void testDomain(RestEnrichmentConfig config, String domainName, Boolean
         Assert.assertTrue(result.getErrors().isEmpty());
     }
 
-    private void testDomain(RestEnrichmentConfig config, List<String> expectedErrors) throws Exception {
+    protected void testDomain(RestEnrichmentConfig config, List<String> expectedErrors) throws Exception {
         Map<String, String> variables = new HashMap<String, String>() {{
             put("wrong field name", "testdomain");
         }};

flink-cyber/flink-enrichment/flink-enrichment-lookup-rest/src/test/java/com/cloudera/cyber/enrichment/rest/RestLookupJobTest.java

@@ -59,9 +59,9 @@ public static void startMockRestServer() throws IOException {
         mockRestServer = new MockRestServer(true);
         File configFile = configTempFolder.newFile("rest-job-test.json");
         List<RestEnrichmentConfig> modelRestConfig = new ArrayList<>();
-        modelRestConfig.add(mockRestServer.configureModelPostRequest(MockRestServer.ENCRYPTED_PROTOCOL).build());
-        modelRestConfig.add(mockRestServer.configureGetAssetRequest(MockRestServer.ENCRYPTED_PROTOCOL).build());
-        modelRestConfig.add(mockRestServer.configureGetUserRequest(MockRestServer.ENCRYPTED_PROTOCOL).build());
+        modelRestConfig.add(mockRestServer.configureModelPostRequest().build());
+        modelRestConfig.add(mockRestServer.configureGetAssetRequest().build());
+        modelRestConfig.add(mockRestServer.configureGetUserRequest().build());
         ObjectWriter ow = getConfigObjectMapper().writerFor(new TypeReference<ArrayList<RestEnrichmentConfig>>() {
         });
         ow.writeValue(configFile, modelRestConfig);

flink-cyber/flink-enrichment/flink-enrichment-lookup-rest/src/test/java/com/cloudera/cyber/enrichment/rest/RestRequestTest.java

@@ -22,14 +22,8 @@ public abstract class RestRequestTest {
     protected static MockRestServer mockRestServer;
 
 
-    protected static void startMockServer() {
-        mockRestServer = new MockRestServer(true);
-    }
-
-    protected static void stopMockServer() {
-        if (mockRestServer != null) {
-            mockRestServer.close();
-        }
+    protected static void startMockServer(boolean enableTlsMutualAuth) {
+        mockRestServer = new MockRestServer(enableTlsMutualAuth);
     }
 
     protected RestRequestResult makeRequest(RestEnrichmentConfig config, Map<String, String> variables) throws Exception {
@@ -38,7 +32,7 @@ protected RestRequestResult makeRequest(RestEnrichmentConfig config, Map<String,
         }
     }
 
-    protected void verifyErrorResult(RestRequestResult result, String expectedErrorFormat, MockRestServer mockRestServer) {
+    protected void verifyErrorResult(RestRequestResult result, String expectedErrorFormat) {
         Assert.assertTrue(result.getExtensions().isEmpty());
         List<String> errors = result.getErrors();
         Assert.assertEquals(1, errors.size());