test: Add E2E test and enhance simulator flexibility (#330)

This PR adds new functionality aimed at making the simulator a better and more flexible test driver for E2E tests. It achieves this by:

- Adding a new StreamStatus data object, which collects and stores information about the recently published/consumed blocks, as well as the latest statuses.
- Exposing the necessary methods to make this usable in other modules, primarily the E2E Tests module.
- Adding a new E2E test that covers the positive scenario where the simulator sends blocks to the Block Node, which responds with the correct response.
- Adds a new CustomThreadPoolExecutor to handle simulator threads, for now I'm just handling thrown errors, but in the future we might extend, depending on our needs.

Signed-off-by: georgi-l95 <[email protected]>

.gitignore

@@ -9,6 +9,7 @@
 
 # Mobile Tools for Java (J2ME)
 .mtj.tmp/
+bin/
 
 # Package Files #
 *.jar

common/src/main/java/com/hedera/block/common/utils/Preconditions.java

@@ -75,6 +75,42 @@ public static int requirePositive(final int toCheck) {
         return requirePositive(toCheck, null);
     }
 
+    /**
+     * This method asserts a given long is a whole number. A long is whole
+     * if it is greater or equal to zero.
+     *
+     * @param toCheck the long to check if it is a whole number
+     * @return the number to check if it is whole number
+     * @throws IllegalArgumentException if the input number to check is not
+     * positive
+     */
+    public static long requireWhole(final long toCheck) {
+        return requireWhole(toCheck, null);
+    }
+
+    /**
+     * This method asserts a given long is a whole number. A long is whole
+     * if it is greater or equal to zero.
+     *
+     * @param toCheck the long to check if it is a whole number
+     * @param errorMessage the error message to be used in the exception if the
+     * input long to check is not a whole number, if null, a default message will
+     * be used
+     * @return the number to check if it is whole number
+     * @throws IllegalArgumentException if the input number to check is not
+     * positive
+     */
+    public static long requireWhole(final long toCheck, final String errorMessage) {
+        if (toCheck >= 0) {
+            return toCheck;
+        }
+
+        final String message = Objects.isNull(errorMessage)
+                ? "The input integer [%d] is required be whole.".formatted(toCheck)
+                : errorMessage;
+        throw new IllegalArgumentException(message);
+    }
+
     /**
      * This method asserts a given integer is a positive. An integer is positive
      * if it is NOT equal to zero and is greater than zero.

common/src/test/java/com/hedera/block/common/CommonsTestUtility.java

@@ -183,11 +183,17 @@ public static Stream<Arguments> positiveIntegers() {
     }
 
     /**
-     * Zero and some negative integers.
+     * Some whole numbers.
      */
-    public static Stream<Arguments> zeroAndNegativeIntegers() {
+    public static Stream<Arguments> wholeNumbers() {
+        return Stream.concat(Stream.of(Arguments.of(0)), positiveIntegers());
+    }
+
+    /**
+     * Some negative integers.
+     */
+    public static Stream<Arguments> negativeIntegers() {
         return Stream.of(
-                Arguments.of(0),
                 Arguments.of(-1),
                 Arguments.of(-2),
                 Arguments.of(-3),
@@ -201,5 +207,12 @@ public static Stream<Arguments> zeroAndNegativeIntegers() {
                 Arguments.of(-10_000_000));
     }
 
+    /**
+     * Zero and some negative integers.
+     */
+    public static Stream<Arguments> zeroAndNegativeIntegers() {
+        return Stream.concat(Stream.of(Arguments.of(0)), negativeIntegers());
+    }
+
     private CommonsTestUtility() {}
 }

common/src/test/java/com/hedera/block/common/utils/PreconditionsTest.java

@@ -67,6 +67,45 @@ void testRequireNotBlankFail(final String toTest) {
                 .withMessage(testErrorMessage);
     }
 
+    /**
+     * This test aims to verify that the
+     * {@link Preconditions#requireWhole(long)} will return the input 'toTest'
+     * parameter if the positive check passes. Test includes overloads.
+     *
+     * @param toTest parameterized, the number to test
+     */
+    @ParameterizedTest
+    @MethodSource("com.hedera.block.common.CommonsTestUtility#wholeNumbers")
+    void testRequireWholePass(final int toTest) {
+        final Consumer<Integer> asserts =
+                actual -> assertThat(actual).isGreaterThanOrEqualTo(0).isEqualTo(toTest);
+
+        final int actual = (int) Preconditions.requireWhole(toTest);
+        assertThat(actual).satisfies(asserts);
+
+        final int actualOverload = (int) Preconditions.requireWhole(toTest, "test error message");
+        assertThat(actualOverload).satisfies(asserts);
+    }
+
+    /**
+     * This test aims to verify that the
+     * {@link Preconditions#requireWhole(long)} will throw an
+     * {@link IllegalArgumentException} if the positive check fails. Test
+     * includes overloads.
+     *
+     * @param toTest parameterized, the number to test
+     */
+    @ParameterizedTest
+    @MethodSource("com.hedera.block.common.CommonsTestUtility#negativeIntegers")
+    void testRequireWholeFail(final int toTest) {
+        assertThatIllegalArgumentException().isThrownBy(() -> Preconditions.requireWhole(toTest));
+
+        final String testErrorMessage = "test error message";
+        assertThatIllegalArgumentException()
+                .isThrownBy(() -> Preconditions.requireWhole(toTest, testErrorMessage))
+                .withMessage(testErrorMessage);
+    }
+
     /**
      * This test aims to verify that the
      * {@link Preconditions#requirePositive(int)} will return the input 'toTest'

simulator/src/main/java/com/hedera/block/simulator/BlockStreamSimulatorApp.java

@@ -20,6 +20,7 @@
 import static java.util.Objects.requireNonNull;
 
 import com.hedera.block.simulator.config.data.BlockStreamConfig;
+import com.hedera.block.simulator.config.data.StreamStatus;
 import com.hedera.block.simulator.config.types.SimulatorMode;
 import com.hedera.block.simulator.exception.BlockSimulatorParsingException;
 import com.hedera.block.simulator.generator.BlockStreamManager;
@@ -101,10 +102,30 @@ public boolean isRunning() {
         return isRunning.get();
     }
 
-    /** Stops the Block Stream Simulator and closes off all grpc channels. */
-    public void stop() {
+    /**
+     * Stops the Block Stream Simulator and closes off all grpc channels.
+     *
+     * @throws InterruptedException if the thread is interrupted
+     */
+    public void stop() throws InterruptedException {
+        simulatorModeHandler.stop();
+        publishStreamGrpcClient.completeStreaming();
+
         publishStreamGrpcClient.shutdown();
         isRunning.set(false);
+
         LOGGER.log(INFO, "Block Stream Simulator has stopped");
     }
+
+    /**
+     * Gets the stream status from both the publisher and the consumer.
+     *
+     * @return the stream status
+     */
+    public StreamStatus getStreamStatus() {
+        return StreamStatus.builder()
+                .publishedBlocks(publishStreamGrpcClient.getPublishedBlocks())
+                .lastKnownPublisherStatuses(publishStreamGrpcClient.getLastKnownStatuses())
+                .build();
+    }
 }

simulator/src/main/java/com/hedera/block/simulator/config/data/StreamStatus.java

@@ -0,0 +1,122 @@
+/*
+ * Copyright (C) 2024 Hedera Hashgraph, LLC
+ *
+ * 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
+ *
+ * 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 com.hedera.block.simulator.config.data;
+
+import static com.hedera.block.common.utils.Preconditions.requireWhole;
+import static java.util.Objects.requireNonNull;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Represents the status of the stream.
+ *
+ * @param publishedBlocks the number of published blocks
+ * @param consumedBlocks the number of consumed blocks
+ * @param lastKnownPublisherStatuses the last known publisher statuses
+ * @param lastKnownConsumersStatuses the last known consumers statuses
+ */
+public record StreamStatus(
+        long publishedBlocks,
+        long consumedBlocks,
+        List<String> lastKnownPublisherStatuses,
+        List<String> lastKnownConsumersStatuses) {
+
+    /**
+     * Creates a new {@link Builder} instance for constructing a {@code StreamStatus}.
+     *
+     * @return a new {@code Builder}
+     */
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    /**
+     * A builder for creating instances of {@link StreamStatus}.
+     */
+    public static class Builder {
+        private long publishedBlocks = 0;
+        private long consumedBlocks = 0;
+        private List<String> lastKnownPublisherStatuses = new ArrayList<>();
+        private List<String> lastKnownConsumersStatuses = new ArrayList<>();
+
+        /**
+         * Creates a new instance of the {@code Builder} class with default configuration values.
+         */
+        public Builder() {
+            // Default constructor
+        }
+
+        /**
+         * Sets the number of published blocks.
+         *
+         * @param publishedBlocks the number of published blocks
+         * @return the builder instance
+         */
+        public Builder publishedBlocks(long publishedBlocks) {
+            requireWhole(publishedBlocks);
+            this.publishedBlocks = publishedBlocks;
+            return this;
+        }
+
+        /**
+         * Sets the number of consumed blocks.
+         *
+         * @param consumedBlocks the number of consumed blocks
+         * @return the builder instance
+         */
+        public Builder consumedBlocks(long consumedBlocks) {
+            requireWhole(consumedBlocks);
+            this.consumedBlocks = consumedBlocks;
+            return this;
+        }
+
+        /**
+         * Sets the last known publisher statuses.
+         *
+         * @param lastKnownPublisherStatuses the last known publisher statuses
+         * @return the builder instance
+         */
+        public Builder lastKnownPublisherStatuses(List<String> lastKnownPublisherStatuses) {
+            requireNonNull(lastKnownPublisherStatuses);
+            this.lastKnownPublisherStatuses = new ArrayList<>(lastKnownPublisherStatuses);
+            return this;
+        }
+
+        /**
+         * Sets the last known consumers statuses.
+         *
+         * @param lastKnownConsumersStatuses the last known consumers statuses
+         * @return the builder instance
+         */
+        public Builder lastKnownConsumersStatuses(List<String> lastKnownConsumersStatuses) {
+            requireNonNull(lastKnownConsumersStatuses);
+            this.lastKnownConsumersStatuses = new ArrayList<>(lastKnownConsumersStatuses);
+            return this;
+        }
+
+        /**
+         * Builds a new {@link StreamStatus} instance.
+         *
+         * @return a new {@link StreamStatus} instance
+         */
+        public StreamStatus build() {
+            return new StreamStatus(
+                    publishedBlocks, consumedBlocks, lastKnownPublisherStatuses, lastKnownConsumersStatuses);
+        }
+    }
+}

simulator/src/main/java/com/hedera/block/simulator/grpc/PublishStreamGrpcClient.java

@@ -45,6 +45,27 @@ public interface PublishStreamGrpcClient {
      */
     boolean streamBlock(Block block);
 
+    /**
+     * Sends a onCompleted message to the server and waits for a short period of time to ensure the message is sent.
+     *
+     * @throws InterruptedException if the thread is interrupted
+     */
+    void completeStreaming() throws InterruptedException;
+
+    /**
+     * Gets the number of published blocks.
+     *
+     * @return the number of published blocks
+     */
+    long getPublishedBlocks();
+
+    /**
+     * Gets the last known statuses.
+     *
+     * @return the last known statuses
+     */
+    List<String> getLastKnownStatuses();
+
     /**
      * Shutdowns the channel.
      */