0g-storage-node/tests/utility/run_all.py
Bo QIU 6ca05090c3
update config (#56)
* update config

* use evmos testnet binary all the time
2024-04-16 21:24:50 +08:00

194 lines
6.1 KiB
Python

import argparse
import os
import time
import subprocess
import sys
import shutil
import stat
from concurrent.futures import ProcessPoolExecutor
from utility.utils import is_windows_platform
DEFAULT_PORT_MIN = 11000
DEFAULT_PORT_MAX = 65535
DEFAULT_PORT_RANGE = 500
CONFLUX_BINARY = "conflux.exe" if is_windows_platform() else "conflux"
EVMOS_BINARY = "evmosd.exe" if is_windows_platform() else "evmosd"
def print_testcase_result(color, glyph, script, start_time):
print(color[1] + glyph + " Testcase " + script + "\telapsed: " + str(int(time.time() - start_time)) + " seconds" + color[0], flush=True)
def run_single_test(py, script, test_dir, index, port_min, port_max):
try:
# Make sure python thinks it can write unicode to its stdout
"\u2713".encode("utf_8").decode(sys.stdout.encoding)
TICK = ""
CROSS = ""
# CIRCLE = "○ "
except UnicodeDecodeError:
TICK = "P "
CROSS = "x "
# CIRCLE = "o "
# BOLD, BLUE, RED, GREY = ("", ""), ("", ""), ("", ""), ("", "")
BLUE, RED = ("", ""), ("", "")
if os.name == "posix" or os.name == "nt":
# primitive formatting on supported
# terminal via ANSI escape sequences:
# BOLD = ("\033[0m", "\033[1m")
BLUE = ("\033[0m", "\033[0;34m")
RED = ("\033[0m", "\033[0;31m")
# GREY = ("\033[0m", "\033[1;30m")
print("Running " + script, flush=True)
port_min = port_min + (index * DEFAULT_PORT_RANGE) % (port_max - port_min)
start_time = time.time()
try:
subprocess.check_output(
args=[py, script, "--randomseed=1", f"--port-min={port_min}"],
stdin=None,
cwd=test_dir,
)
except subprocess.CalledProcessError as err:
print_testcase_result(RED, CROSS, script, start_time)
print("Output of " + script + "\n" + err.output.decode("utf-8"), flush=True)
raise err
print_testcase_result(BLUE, TICK, script, start_time)
def run_all(test_dir: str, test_subdirs: list[str]=[], slow_tests: set[str]={}, long_manual_tests: set[str]={}):
tmp_dir = os.path.join(test_dir, "tmp")
if not os.path.exists(tmp_dir):
os.makedirs(tmp_dir, exist_ok=True)
# Build conflux binary if absent
build_from_github(
dir=tmp_dir,
binary_name=CONFLUX_BINARY,
github_url="https://github.com/Conflux-Chain/conflux-rust.git",
build_cmd="cargo build --release --bin conflux",
compiled_relative_path=["target", "release"],
)
# Build evmos binary if absent
build_from_github(
dir=tmp_dir,
binary_name=EVMOS_BINARY,
github_url="-b testnet https://github.com/0glabs/0g-evmos.git",
build_cmd="make install; cp $(go env GOPATH)/bin/evmosd .",
compiled_relative_path=[],
)
start_time = time.time()
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=DEFAULT_PORT_MAX,
type=int,
)
parser.add_argument(
"--port-min",
dest="port_min",
default=DEFAULT_PORT_MIN,
type=int,
)
options = parser.parse_args()
TEST_SCRIPTS = []
# include test_dir itself
test_subdirs.insert(0, "")
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 and rel_path not in long_manual_tests:
TEST_SCRIPTS.append(rel_path)
executor = ProcessPoolExecutor(max_workers=options.max_workers)
test_results = []
py = "python3"
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)
print("Elapsed: " + str(int(time.time() - start_time)) + " seconds", flush=True)
if len(failed) > 0:
print("The following test fails: ")
for c in failed:
print(c)
sys.exit(1)
def build_from_github(dir: str, binary_name: str, github_url: str, build_cmd: str, compiled_relative_path: list[str]) -> bool:
binary_path = os.path.join(dir, binary_name)
if os.path.exists(binary_path):
return False
start_time = time.time()
# clone code from github to a temp folder
code_tmp_dir_name = (binary_name[:-4] if is_windows_platform() else binary_name) + "_tmp"
code_tmp_dir = os.path.join(dir, code_tmp_dir_name)
if os.path.exists(code_tmp_dir):
shutil.rmtree(code_tmp_dir)
clone_command = "git clone " + github_url + " " + code_tmp_dir
os.system(clone_command)
# build binary
origin_path = os.getcwd()
os.chdir(code_tmp_dir)
os.system(build_cmd)
# copy compiled binary to right place
compiled_binary = os.path.join(code_tmp_dir, *compiled_relative_path, binary_name)
shutil.copyfile(compiled_binary, binary_path)
if not is_windows_platform():
st = os.stat(binary_path)
os.chmod(binary_path, st.st_mode | stat.S_IEXEC)
os.chdir(origin_path)
shutil.rmtree(code_tmp_dir, ignore_errors=True)
print("Completed to build binary " + binary_name + ", Elapsed: " + str(int(time.time() - start_time)) + " seconds", flush=True)
return True