Fix integration tests (#375)

thjarvin · web-flow · commit e7ede7a4c441 · 2025-07-03T15:53:53.000+03:00
* Revert "Disable integration tests" This reverts commit a8c5be8. * AB#45574: Update Pulsar version to 3.0.7 in integration tests * AB#45574: Restore deleted integration test code * Fix compile errors * Fix HTTP response status codes in integration tests
diff --git a/.github/workflows/test-and-build.yml b/.github/workflows/test-and-build.yml
@@ -15,6 +15,8 @@ jobs:
           cache: 'maven'
       - name: Build with Maven
         run: mvn clean package -Dmaven.javadoc.skip=true -B -V
+      - name: Run integration tests
+        run: mvn failsafe:integration-test && mvn failsafe:verify
   publish:
     needs: test
     runs-on: ubuntu-latest
diff --git a/src/integration-test/java/fi/hsl/common/pulsar/ITBaseTestSuite.java b/src/integration-test/java/fi/hsl/common/pulsar/ITBaseTestSuite.java
@@ -4,6 +4,7 @@
 import fi.hsl.common.config.ConfigUtils;
 import fi.hsl.common.transitdata.TransitdataProperties;
 import org.apache.pulsar.client.api.*;
+import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -40,6 +41,15 @@ public class ITBaseTestSuite {
     @ClassRule
     public static PulsarContainer pulsar = MockContainers.newPulsarContainer();
 
+    @BeforeClass
+    public static void setUp() throws Exception {
+        MockContainers.configurePulsarContainer(pulsar, TENANT, NAMESPACE);
+
+        if (PRINT_PULSAR_LOG) {
+            MockContainers.tail(pulsar, logger);
+        }
+    }
+
     protected static PulsarApplication createPulsarApp(String config, String testId) throws Exception {
         logger.info("Creating Pulsar Application for config " + config);
         Config configObj = PulsarMockApplication.readConfig(config);
diff --git a/src/integration-test/java/fi/hsl/common/pulsar/ITPulsarApplication.java b/src/integration-test/java/fi/hsl/common/pulsar/ITPulsarApplication.java
@@ -3,9 +3,11 @@
 import com.typesafe.config.Config;
 import fi.hsl.common.config.ConfigParser;
 import fi.hsl.common.config.ConfigUtils;
+import fi.hsl.common.health.HealthServer;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.Producer;
+import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -14,9 +16,15 @@
 import org.testcontainers.containers.PulsarContainer;
 import redis.clients.jedis.Jedis;
 
+import java.net.HttpURLConnection;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
 import java.nio.charset.Charset;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
 
 import static org.junit.Assert.*;
 
@@ -36,13 +44,22 @@ public class ITPulsarApplication {
 
     @ClassRule
     public static PulsarContainer pulsar = MockContainers.newPulsarContainer();
-    
+
+    @BeforeClass
+    public static void setUp() throws Exception {
+        MockContainers.configurePulsarContainer(pulsar, TENANT, NAMESPACE);
+
+        if (PRINT_PULSAR_LOG) {
+            MockContainers.tail(pulsar, logger);
+        }
+    }
+
     @Test
     public void testRedisContainer() {
         Jedis jedis = MockContainers.newMockJedisConnection(redis);
         jedis.set("key", "value");
         String value = jedis.get("key");
-        assertEquals(value, "value");
+        assertEquals("value", value);
     }
 
     @Test
@@ -51,10 +68,11 @@ public void testPulsarApplicationRedis() throws Exception {
         assertNotNull(config);
         PulsarApplication app = PulsarMockApplication.newInstance(config, redis, pulsar);
         assertNotNull(app);
-
+        
+        assertNotNull(app.getContext().getJedis());
         app.getContext().getJedis().set("pulsar-application-jedis", "should work");
         String value = app.getContext().getJedis().get("pulsar-application-jedis");
-        assertEquals(value, "should work");
+        assertEquals("should work", value);
     }
 
 
@@ -76,17 +94,20 @@ public void testPulsar() throws Exception {
         final String payload = "Test-message";
 
         Producer<byte[]> producer = app.getContext().getSingleProducer();
+        assertNotNull(producer);
         producer.send(payload.getBytes());
 
         logger.info("Message sent, reading it back");
 
         Consumer<byte[]> consumer = app.getContext().getConsumer();
-        readAndValidateMsg(consumer, new HashSet<>(Arrays.asList(payload)));
+        readAndValidateMsg(consumer, new HashSet<>(List.of(payload)));
 
         Jedis jedis = app.getContext().getJedis();
-
+        
+        assertNotNull(consumer);
         assertTrue(consumer.isConnected());
         assertTrue(producer.isConnected());
+        assertNotNull(jedis);
         assertTrue(jedis.isConnected());
 
         app.close();
@@ -115,7 +136,8 @@ public void testPulsarWithMultipleTopics() throws Exception {
 
         PulsarApplication app = PulsarMockApplication.newInstance(producer1Config, redis, pulsar);
         assertNotNull(app);
-
+        
+        assertNotNull(app.getContext().getProducers());
         Producer<byte[]> producer = app.getContext().getProducers().get("test-1");
 
         //Create a second producer but bind into different topic
@@ -173,12 +195,15 @@ public void testPulsarAutoClose() throws Exception {
             assertNotNull(app);
 
             producer = app.getContext().getSingleProducer();
+            assertNotNull(producer);
             assertTrue(producer.isConnected());
 
             consumer = app.getContext().getConsumer();
+            assertNotNull(consumer);
             assertTrue(consumer.isConnected());
 
             jedis = app.getContext().getJedis();
+            assertNotNull(jedis);
             assertTrue(jedis.isConnected());
         }
 
@@ -218,10 +243,120 @@ public void testInitFailureOnInvalidTopicsPattern() {
     public void testInitFailure(Config config) {
         try(PulsarApplication app = PulsarMockApplication.newInstance(config, redis, pulsar)) {
             logger.info("You should never see this message, init should throw an exception");
-            assertTrue(false);
+            fail();
         }
         catch (Exception e) {
             logger.debug("Exception as expected");
         }
     }
+
+    @Test
+    public void testHttpServer() throws Exception {
+        Config base = PulsarMockApplication.readConfig(CONFIG_FILE);
+
+        PulsarApplication app = PulsarMockApplication.newInstance(base, redis, pulsar);
+        assertNotNull(app);
+
+        logger.info("Pulsar Application created, testing HealthServer");
+
+        final Producer<byte[]> producer = app.getContext().getSingleProducer();
+        final Consumer<byte[]> consumer = app.getContext().getConsumer();
+        final Jedis jedis = app.getContext().getJedis();
+        final HealthServer healthServer = app.getContext().getHealthServer();
+        
+        assertNotNull(consumer);
+        assertTrue(consumer.isConnected());
+        assertNotNull(producer);
+        assertTrue(producer.isConnected());
+        assertNotNull(jedis);
+        assertTrue(jedis.isConnected());
+
+        logger.info("Creating health check function");
+        final BooleanSupplier healthCheck = () -> {
+            boolean status = true;
+            status &= producer.isConnected();
+            status &= consumer.isConnected();
+            status &= jedis.isConnected();
+            return status;
+        };
+        assertNotNull(healthServer);
+        healthServer.addCheck(healthCheck);
+
+        String url = "http://localhost:" + healthServer.port + healthServer.endpoint;
+
+        logger.info("Checking health");
+        HttpResponse<String> response = makeGetRequest(url);
+        assertEquals(HttpURLConnection.HTTP_OK, response.statusCode());
+        assertEquals("OK", getContent(response));
+
+        logger.info("Disconnecting Jedis and checking health");
+        jedis.disconnect();
+        assertFalse(jedis.isConnected());
+
+        response = makeGetRequest(url);
+        assertEquals(HttpURLConnection.HTTP_UNAVAILABLE, response.statusCode());
+        assertEquals("FAIL", getContent(response));
+
+        logger.info("Reconnecting Jedis and checking health");
+        jedis.connect();
+        assertTrue(jedis.isConnected());
+
+        response = makeGetRequest(url);
+        assertEquals(HttpURLConnection.HTTP_OK, response.statusCode());
+        assertEquals("OK", getContent(response));
+
+        logger.info("Closing Pulsar consumer and checking health");
+        consumer.close();
+        assertFalse(consumer.isConnected());
+
+        response = makeGetRequest(url);
+        assertEquals(HttpURLConnection.HTTP_UNAVAILABLE, response.statusCode());
+        assertEquals("FAIL", getContent(response));
+
+        response = makePostRequest(url);
+        assertEquals(HttpURLConnection.HTTP_BAD_METHOD, response.statusCode());
+        assertEquals("Method Not Allowed", getContent(response));
+
+        url = "http://localhost:" + healthServer.port + "/foo";
+        response = makeGetRequest(url);
+        assertEquals(HttpURLConnection.HTTP_NOT_FOUND, response.statusCode());
+        assertEquals("Not Found", getContent(response));
+
+        url = "http://localhost:" + healthServer.port + healthServer.endpoint + "foo";
+        response = makeGetRequest(url);
+        assertEquals(HttpURLConnection.HTTP_NOT_FOUND, response.statusCode());
+        assertEquals("Not Found", getContent(response));
+
+        app.close();
+        assertFalse(consumer.isConnected());
+        assertFalse(producer.isConnected());
+        assertFalse(jedis.isConnected());
+    }
+
+    private HttpResponse<String> makeGetRequest(final String url) {
+        return makeRequest("GET", url);
+    }
+
+    private HttpResponse<String> makePostRequest(final String url) {
+        return makeRequest("POST", url);
+    }
+
+    private HttpResponse<String> makeRequest(final String method, final String url) {
+        HttpClient client = HttpClient.newHttpClient();
+        HttpRequest request;
+        if (method.equalsIgnoreCase("post")) {
+            request = HttpRequest.newBuilder(URI.create(url)).POST(HttpRequest.BodyPublishers.noBody()).build();
+        } else {
+            request = HttpRequest.newBuilder(URI.create(url)).GET().build();
+        }
+        try {
+            return client.send(request, HttpResponse.BodyHandlers.ofString());
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private String getContent(final HttpResponse<String> response) {
+        return response.body();
+    }
 }
diff --git a/src/integration-test/java/fi/hsl/common/pulsar/MockContainers.java b/src/integration-test/java/fi/hsl/common/pulsar/MockContainers.java
@@ -1,13 +1,18 @@
 package fi.hsl.common.pulsar;
 
-import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.common.policies.data.TenantInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.PulsarContainer;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
 import redis.clients.jedis.Jedis;
 
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
 public class MockContainers {
 
     static final Logger logger = LoggerFactory.getLogger(MockContainers.class);
@@ -36,13 +41,24 @@ public static Jedis newMockJedisConnection(GenericContainer redis) {
      * For using the "real" tenant/namespace/topicname pattern @see MockContainers#configurePulsarContainer(PulsarContainer, String, String)
      */
     public static PulsarContainer newPulsarContainer() {
-        return new PulsarContainer("2.3.1");
+        return new PulsarContainer("3.0.7");
     }
-    
-    public static PulsarClient newMockPulsarClient(PulsarContainer pulsar) throws Exception {
-        return PulsarClient.builder()
-                .serviceUrl(pulsar.getPulsarBrokerUrl())
+
+    public static PulsarContainer configurePulsarContainer(PulsarContainer pulsar, final String tenant, final String namespace) throws Exception {
+        PulsarAdmin admin = PulsarAdmin.builder()
+                .serviceHttpUrl(pulsar.getHttpServiceUrl())
                 .build();
+
+        Set<String> clusters = new HashSet<>(List.of("standalone"));
+        TenantInfo info = TenantInfo.builder()
+                .allowedClusters(clusters)
+                .adminRoles(new HashSet<>(List.of("all")))
+                .build();
+        admin.tenants().createTenant(tenant, info);
+
+        admin.namespaces().createNamespace(tenant + "/" + namespace, clusters);
+        logger.info("Pulsar setup done");
+        return pulsar;
     }
 
     public static void tail(GenericContainer container, Logger logger) {
