UPDATE: test framework

.gitmodules

@@ -13,3 +13,6 @@
 [submodule "contracts/lib/eigenlayer-middleware"]
 	path = contracts/lib/eigenlayer-middleware
 	url = https://github.com/Layr-Labs/eigenlayer-middleware
+[submodule "zerog_storage_kv"]
+	path = zerog_storage_kv
+	url = git@github.com:zero-gravity-labs/zerog-storage-kv.git

tests/da_hello_test.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+import sys
+sys.path.append("../zerog_storage_kv/tests")
+import time
+
+from da_test_framework.da_test_framework import DATestFramework
+
+
+class DAHelloTest(DATestFramework):
+    def setup_params(self):
+        self.num_blockchain_nodes = 1
+        self.num_nodes = 2
+
+    def run_test(self):
+        self.log.debug("===============================================")
+        self.log.debug("root dir:" + self.root_dir)
+        self.log.debug("blockchain binary:" + self.blockchain_binary)
+        self.log.debug("zgs binary:" + self.zgs_binary)
+        self.log.debug("cli binary:" + self.cli_binary)
+        self.log.debug("kv binary:" + self.kv_binary)
+        self.log.debug("contract path:" + self.contract_path)
+        self.log.debug("token contract path:" + self.token_contract_path)
+        self.log.debug("mine contract path:" + self.mine_contract_path)
+        self.log.debug("localstack binary:" + self.localstack_binary)
+        self.log.debug("da encoder binary:" + self.da_encoder_binary)
+        self.log.debug("da batcher binary:" + self.da_batcher_binary)
+        self.log.debug("da server binary:" + self.da_server_binary)
+        self.log.debug("===============================================")
+
+        self.log.info("hello")
+        time.sleep(3)
+
+
+if __name__ == "__main__":
+    DAHelloTest().main()

tests/da_test_all.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+import argparse
+import os
+import subprocess
+import sys
+
+sys.path.append("../zerog_storage_kv/tests")
+
+from concurrent.futures import ProcessPoolExecutor
+from test_all import run_single_test
+
+PORT_MIN = 11000
+PORT_MAX = 65535
+PORT_RANGE = 600
+
+__file_path__ = os.path.dirname(os.path.realpath(__file__))
+
+
+def run():
+    parser = argparse.ArgumentParser(usage="%(prog)s [options]")
+    parser.add_argument(
+        "--max-workers",
+        dest="max_workers",
+        default=5,
+        type=int,
+    )
+    parser.add_argument(
+        "--port-max",
+        dest="port_max",
+        default=PORT_MAX,
+        type=int,
+    )
+    parser.add_argument(
+        "--port-min",
+        dest="port_min",
+        default=PORT_MIN,
+        type=int,
+    )
+
+    options = parser.parse_args()
+
+    TEST_SCRIPTS = []
+
+    test_dir = os.path.dirname(os.path.realpath(__file__))
+    test_subdirs = [
+        "",  # include test_dir itself
+    ]
+
+    slow_tests = {}
+
+    for subdir in test_subdirs:
+        subdir_path = os.path.join(test_dir, subdir)
+        for file in os.listdir(subdir_path):
+            if file.endswith("_test.py"):
+                rel_path = os.path.join(subdir, file)
+                if rel_path not in slow_tests:
+                    TEST_SCRIPTS.append(rel_path)
+
+    executor = ProcessPoolExecutor(max_workers=options.max_workers)
+    test_results = []
+
+    py = "python"
+    if hasattr(sys, "getwindowsversion"):
+        py = "python"
+
+    i = 0
+    # Start slow tests first to avoid waiting for long-tail jobs
+    for script in slow_tests:
+        f = executor.submit(
+            run_single_test, py, script, test_dir, i, options.port_min, options.port_max
+        )
+        test_results.append((script, f))
+        i += 1
+    for script in TEST_SCRIPTS:
+        f = executor.submit(
+            run_single_test, py, script, test_dir, i, options.port_min, options.port_max
+        )
+        test_results.append((script, f))
+        i += 1
+
+    failed = set()
+    for script, f in test_results:
+        try:
+            f.result()
+        except subprocess.CalledProcessError as err:
+            print("CalledProcessError " + repr(err))
+            failed.add(script)
+
+    if len(failed) > 0:
+        print("The following test fails: ")
+        for c in failed:
+            print(c)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    run()

tests/da_test_framework/da_batcher.py

@@ -0,0 +1,86 @@
+import os
+import sys
+import time
+
+sys.path.append("../../zerog_storage_kv/tests")
+
+from test_framework.blockchain_node import TestNode
+from utility.utils import blockchain_rpc_port
+from config.node_config import GENESIS_PRIV_KEY
+from da_test_framework.da_node_type import DANodeType
+
+__file_path__ = os.path.dirname(os.path.realpath(__file__))
+
+
+class DABatcher(TestNode):
+    def __init__(
+            self,
+            root_dir,
+            binary,
+            updated_config,
+            log_contract_address,
+            log,
+    ):
+        local_conf = {
+            "log_config_file": "log_config",
+            "log_contract_address": log_contract_address,
+            "blockchain_rpc_endpoint": f"http://127.0.0.1:{blockchain_rpc_port(0)}",
+        }
+
+        local_conf.update(updated_config)
+        data_dir = os.path.join(root_dir, "da_batcher")
+        # rpc_url = "http://" + local_conf["rpc_listen_address"]
+        super().__init__(
+            DANodeType.DA_BATCHER,
+            0,
+            data_dir,
+            None,
+            binary,
+            local_conf,
+            log,
+            None,
+        )
+        self.args = [binary, "--batcher.pull-interval", "10s",
+                     "--chain.rpc", local_conf['blockchain_rpc_endpoint'],
+                     "--chain.private-key", GENESIS_PRIV_KEY,
+                     "--batcher.finalizer-interval", "20s",
+                     "--batcher.aws.region", "us-east-1",
+                     "--batcher.aws.access-key-id", "localstack",
+                     "--batcher.aws.secret-access-key", "localstack",
+                     "--batcher.aws.endpoint-url", "http://0.0.0.0:4566",
+                     "--batcher.s3-bucket-name", "test-zgda-blobstore",
+                     "--batcher.dynamodb-table-name", "test-BlobMetadata",
+                     "--encoder-socket", "0.0.0.0:34000",
+                     "--batcher.batch-size-limit", "10000",
+                     "--batcher.srs-order", "300000",
+                     "--encoding-timeout", "10s",
+                     "--chain-read-timeout", "12s",
+                     "--chain-write-timeout", "13s",
+                     "--batcher.storage.node-url", "http://0.0.0.0:5678",
+                     "--batcher.storage.node-url", "http://0.0.0.0:6789",
+                     "--batcher.storage.kv-url", "http://0.0.0.0:7890",
+                     "--batcher.storage.kv-stream-id",
+                     "000000000000000000000000000000000000000000000000000000000000f2bd",
+                     "--batcher.storage.flow-contract", local_conf['log_contract_address']]
+
+    def start(self):
+        self.log.info("Start DA batcher")
+        super().start()
+
+    def wait_for_rpc_connection(self):
+        time.sleep(1)
+
+    def stop(self):
+        self.log.info("Stop DA batcher")
+        try:
+            super().stop(kill=True, wait=False)
+        except AssertionError as e:
+            err = repr(e)
+            # The batcher will check return_code via rpc when error log exists
+            # that is written when the batcher starts normally.
+            # The exception handling can be removed when rpc is added or the error
+            # is not written when the batcher starts normally.
+            if "no RPC connection" in err:
+                self.log.debug(f"Stop DA encoder: no RPC connection")
+            else:
+                raise e

tests/da_test_framework/da_encoder.py

@@ -0,0 +1,66 @@
+import os
+import sys
+import time
+
+sys.path.append("../../zerog_storage_kv/tests")
+
+from test_framework.blockchain_node import TestNode
+from da_test_framework.da_node_type import DANodeType
+
+__file_path__ = os.path.dirname(os.path.realpath(__file__))
+
+
+class DAEncoder(TestNode):
+    def __init__(
+            self,
+            root_dir,
+            binary,
+            updated_config,
+            log,
+    ):
+        local_conf = dict(log_config_file="log_config")
+
+        local_conf.update(updated_config)
+        data_dir = os.path.join(root_dir, "da_encoder")
+        # rpc_url = "http://" + local_conf["rpc_listen_address"]
+        super().__init__(
+            DANodeType.DA_ENCODER,
+            0,
+            data_dir,
+            None,
+            binary,
+            local_conf,
+            log,
+            None,
+        )
+        self.args = [binary, "--disperser-encoder.grpc-port", "34000",
+                     "--disperser-encoder.metrics-http-port", "9109",
+                     "--kzg.g1-path", f"{__file_path__}/../../inabox/resources/kzg/g1.point.300000",
+                     "--kzg.g2-path", f"{__file_path__}/../../inabox/resources/kzg/g2.point.300000",
+                     "--kzg.cache-path", f"{__file_path__}/../../inabox/resources/kzg/SRSTables",
+                     "--kzg.srs-order", "300000",
+                     "--kzg.num-workers", "12",
+                     "--disperser-encoder.log.level-std", "trace",
+                     "--disperser-encoder.log.level-file", "trace"]
+
+    def start(self):
+        self.log.info("Start DA encoder")
+        super().start()
+
+    def wait_for_rpc_connection(self):
+        time.sleep(1)
+
+    def stop(self):
+        self.log.info("Stop DA encoder")
+        # The encoder will check return_code via rpc when error log exists
+        # that is written when the encoder starts normally.
+        # The exception handling can be removed when rpc is added or the error
+        # is not written when the encoder starts normally.
+        try:
+            super().stop(kill=True, wait=False)
+        except AssertionError as e:
+            err = repr(e)
+            if "no RPC connection" in err:
+                self.log.debug(f"Stop DA encoder: no RPC connection")
+            else:
+                raise e

tests/da_test_framework/da_node_type.py

@@ -0,0 +1,9 @@
+from enum import Enum, unique
+
+
+@unique
+class DANodeType(Enum):
+    DA_LOCAL_STACK = 3
+    DA_ENCODER = 4
+    DA_BATCHER = 5
+    DA_SERVER = 6

tests/da_test_framework/da_server.py

@@ -0,0 +1,53 @@
+import os
+import sys
+import time
+
+sys.path.append("../../zerog_storage_kv/tests")
+
+from test_framework.blockchain_node import TestNode
+from da_test_framework.da_node_type import DANodeType
+
+__file_path__ = os.path.dirname(os.path.realpath(__file__))
+
+
+class DAServer(TestNode):
+    def __init__(
+            self,
+            root_dir,
+            binary,
+            updated_config,
+            log,
+    ):
+        local_conf = dict(log_config_file="log_config")
+
+        local_conf.update(updated_config)
+        data_dir = os.path.join(root_dir, "da_server")
+        # rpc_url = "http://" + local_conf["rpc_listen_address"]
+        super().__init__(
+            DANodeType.DA_SERVER,
+            0,
+            data_dir,
+            None,
+            binary,
+            local_conf,
+            log,
+            None,
+        )
+        self.args = [binary, "--disperser-server.grpc-port", "51001",
+                     "--disperser-server.s3-bucket-name", "test-zgda-blobstore",
+                     "--disperser-server.dynamodb-table-name", "test-BlobMetadata",
+                     "--disperser-server.aws.region", "us-east-1",
+                     "--disperser-server.aws.access-key-id", "localstack",
+                     "--disperser-server.aws.secret-access-key", "localstack",
+                     "--disperser-server.aws.endpoint-url", "http://0.0.0.0:4566"]
+
+    def start(self):
+        self.log.info("Start DA server")
+        super().start()
+
+    def wait_for_rpc_connection(self):
+        time.sleep(1)
+
+    def stop(self):
+        self.log.info("Stop DA server")
+        super().stop(kill=True, wait=False)