test_scripts.py

"""Tests for config.py, node.sh and setup.sh. XDC is a stand-in script, so no node runs and nothing is installed."""
import copy
import os
import shutil
import signal
import stat
import subprocess
import sys
import tempfile
import unittest

import yaml

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)

import config  # noqa: E402

KEY = "11" * 32
BOOTNODE = "enode://abc@10.0.0.1:30301"
FAKE_XDC = """#!/bin/sh
echo "$*" >> "$FAKE_LOG"
env >> "$FAKE_ENV"
case "$1" in
  account)
    if [ "$2" = import ]; then
      for last; do :; done
      cat "$last" >> "$FAKE_KEYS"
      [ -n "$FAKE_BAD_IMPORT" ] || echo "Address: {abc123}"
    else
      echo "Account #0: {abc123} keystore://x"
    fi
    ;;
  --datadir)
    if [ "$3" = init ]; then mkdir -p "$2/XDC/chaindata"; exit 0; fi
    [ -z "$FAKE_CRASH" ] || exit 1
    exec sleep 30
    ;;
esac
"""


def read(path):
    with open(path) as source:
        return source.read()


def shipped_config():
    return config.load(os.path.join(ROOT, "config.yml"))


def valid_config_text():
    """config.yml with the node addresses filled in, as a person would have it."""
    return read(os.path.join(ROOT, "config.yml")).replace("0x" + "0" * 40, "0x" + "ab" * 20)


class ConfigTest(unittest.TestCase):
    def errors_with(self, **changes):
        settings = copy.deepcopy(yaml.safe_load(valid_config_text()))
        for key, value in changes.items():
            settings[key.replace("__", "-")] = value
        return " ".join(config.validate(settings))

    def test_the_shipped_config_needs_real_node_addresses(self):
        errors = config.validate(shipped_config())
        self.assertEqual(len(errors), 1)
        self.assertIn("Nodes", errors[0])
        self.assertEqual(config.validate(yaml.safe_load(valid_config_text())), [])

    def test_the_zero_address_is_refused(self):
        for zero in ("0x" + "0" * 40, "xdc" + "0" * 40):
            with self.subTest(zero=zero):
                self.assertFalse(config.is_address(zero))
                self.assertIn("MasterNode", self.errors_with(MasterNode=zero))
                self.assertIn("signers", self.errors_with(signers=[zero]))

    def test_bad_values_are_named(self):
        address = "0x" + "ab" * 20
        cases = [
            ({"Chain_Name": "my-chain"}, "Chain_Name"),
            ({"Chain_Name": "my chain"}, "Chain_Name"),
            ({"Consensus": 4}, "Consensus"),
            ({"Consensus": True}, "Consensus"),
            ({"BlockTime": 0}, "BlockTime"),
            ({"BlockTime": "2"}, "BlockTime"),
            ({"EpochReward": -1}, "EpochReward"),
            ({"MasterNode": "0x123"}, "MasterNode"),
            ({"foundationAddress": None}, "foundationAddress"),
            ({"swapWallet": "z" * 42}, "swapWallet"),
            ({"signers": []}, "signers"),
            ({"signers": [address, "nope"]}, "signers"),
            ({"preFund": address}, "preFund"),
            ({"networkID": 50}, "XDC mainnet"),
            ({"networkID": 51}, "Apothem"),
            ({"networkID": 0}, "networkID"),
            ({"setGap": 900}, "setGap must be smaller"),
            ({"walletConfirmations": 4}, "walletConfirmations cannot"),
            ({"Total_Nodes": 3}, "Total_Nodes is 3 but Nodes lists 4"),
            ({"MultiSig__Address": [address]}, "walletConfirmations cannot"),
        ]
        for changes, expected in cases:
            with self.subTest(changes=changes):
                self.assertIn(expected, self.errors_with(**changes))

    def test_all_problems_are_reported_together(self):
        self.assertEqual(len(config.validate({})), 16)

    def test_both_address_prefixes_are_accepted(self):
        self.assertTrue(config.is_address("xdc" + "ab" * 20))
        self.assertTrue(config.is_address("0x" + "AB" * 20))
        self.assertFalse(config.is_address("0x" + "ab" * 19))
        self.assertFalse(config.is_address("0x" + "ab" * 20 + "\n"))


class Workspace(unittest.TestCase):
    """A copy of the scripts, so tests never write keys or node data into the repository."""

    def setUp(self):
        self.dir = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, self.dir)
        for name in ("config.py", "config.yml", "node.sh", "setup.sh"):
            shutil.copy(os.path.join(ROOT, name), self.dir)
        self.path("config.yml", valid_config_text())
        self.project = os.path.join(self.dir, "XDPoSChain")
        binary = os.path.join(self.project, "build/bin/XDC")
        os.makedirs(os.path.dirname(binary))
        with open(binary, "w") as fake:
            fake.write(FAKE_XDC)
        os.chmod(binary, 0o755)
        os.makedirs(os.path.join(self.dir, "genesis"))
        self.path("genesis/genesis.json", "{}")
        self.log = os.path.join(self.dir, "calls.log")
        self.keys = os.path.join(self.dir, "keys.log")
        self.node_env = os.path.join(self.dir, "env.log")
        self.addCleanup(self.stop_nodes)

    def path(self, name, content):
        with open(os.path.join(self.dir, name), "w") as out:
            out.write(content)

    def env(self, **extra):
        env = {
            "PATH": os.path.dirname(sys.executable) + os.pathsep + os.environ["PATH"],
            "HOME": self.dir,
            "PROJECT_DIR": self.project,
            "FAKE_LOG": self.log,
            "FAKE_KEYS": self.keys,
            "FAKE_ENV": self.node_env,
            "START_WAIT": "0.3",
            "BOOTNODES": BOOTNODE,
            "PRIVATE_KEY_1": KEY,
        }
        env.update(extra)
        return {key: value for key, value in env.items() if value is not None}

    def script(self, name, *args, **extra):
        return subprocess.run(["bash", name, *args], cwd=self.dir, env=self.env(**extra), capture_output=True, text=True, timeout=30)

    def node(self, *args, **extra):
        return self.script("node.sh", *(args or ("1",)), **extra)

    def calls(self):
        with open(self.log) as log:
            return log.read().splitlines()

    def start_args(self):
        return [line for line in self.calls() if "--mine" in line][-1]

    def stop_nodes(self):
        nodes = os.path.join(self.dir, "nodes")
        if not os.path.isdir(nodes):
            return
        for node in os.listdir(nodes):
            pid = os.path.join(nodes, node, "XDC.pid")
            if os.path.exists(pid):
                try:
                    os.kill(int(read(pid)), signal.SIGKILL)
                except (OSError, ValueError):
                    pass


class NodeScriptTest(Workspace):
    def test_rpc_stays_private_by_default(self):
        result = self.node()
        self.assertEqual(result.returncode, 0, result.stderr)
        args = self.start_args()
        self.assertIn("--http-addr 127.0.0.1", args)
        self.assertIn("--ws-addr 127.0.0.1", args)
        self.assertIn("--http-api eth,net,web3,XDPoS", args)
        for unsafe in ("0.0.0.0", "personal", "admin", "corsdomain", "origins", "vhosts"):
            self.assertNotIn(unsafe, args)

    def test_settings_come_from_config_and_env(self):
        self.path("config.yml", valid_config_text().replace("networkID: 72", "networkID: 99"))
        self.node()
        args = self.start_args()
        self.assertIn("--networkid 99", args)
        self.assertIn("--bootnodes %s" % BOOTNODE, args)

    def test_env_file_is_read(self):
        self.path(".env", 'BOOTNODES="enode://fromfile@10.0.0.2:30301"\nPRIVATE_KEY_1=%s\n' % KEY)
        os.chmod(os.path.join(self.dir, ".env"), 0o600)
        result = self.node(BOOTNODES=None, PRIVATE_KEY_1=None)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("enode://fromfile@10.0.0.2:30301", self.start_args())

    def test_an_env_file_others_can_read_is_refused(self):
        self.path(".env", "PRIVATE_KEY_1=%s\n" % KEY)
        for mode in (0o644, 0o640, 0o604, 0o620, 0o602):
            with self.subTest(mode=mode):
                os.chmod(os.path.join(self.dir, ".env"), mode)
                result = self.node()
                self.assertNotEqual(result.returncode, 0)
                self.assertIn("chmod 600 .env", result.stderr)
                self.assertFalse(os.path.exists(self.log))

    def test_the_node_does_not_inherit_secrets(self):
        self.path(".env", "PRIVATE_KEY_1=%s\nPRIVATE_KEY_2=%s\nETHSTATS=name:secret@stats.example:3000\n" % (KEY, "22" * 32))
        os.chmod(os.path.join(self.dir, ".env"), 0o600)
        result = self.node(PRIVATE_KEY_1=None, PRIVATE_KEY_3="33" * 32, ETHSTATS="name:secret@stats.example:3000")
        self.assertEqual(result.returncode, 0, result.stderr)
        seen = read(self.node_env)
        self.assertIn("FAKE_LOG", seen)
        for secret in ("PRIVATE_KEY", KEY, "22" * 32, "33" * 32, "ETHSTATS", "name:secret"):
            self.assertNotIn(secret, seen)

    def test_the_key_is_imported_but_never_on_a_command_line(self):
        self.node(PRIVATE_KEY_1="0x" + KEY)
        with open(self.keys) as keys:
            self.assertEqual(keys.read().strip(), KEY)
        for line in self.calls():
            self.assertNotIn(KEY, line)
        with open(os.path.join(self.dir, "nodes/1/XDC.log")) as log:
            self.assertNotIn(KEY, log.read())

    def test_the_password_is_generated_and_private(self):
        self.node()
        path = os.path.join(self.dir, ".pwd")
        self.assertEqual(len(read(path).strip()), 64)
        self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), 0o600)

    def test_an_existing_password_is_kept(self):
        self.path(".pwd", "mine\n")
        self.node()
        self.assertEqual(read(os.path.join(self.dir, ".pwd")), "mine\n")

    def test_an_existing_password_is_made_private(self):
        self.path(".pwd", "mine\n")
        os.chmod(os.path.join(self.dir, ".pwd"), 0o644)
        self.node()
        self.assertEqual(stat.S_IMODE(os.stat(os.path.join(self.dir, ".pwd")).st_mode), 0o600)

    def test_each_node_gets_its_own_ports_and_folder(self):
        self.node("1")
        first = self.start_args()
        self.node("2", PRIVATE_KEY_2=KEY)
        second = self.start_args()
        for expected in ("--port 30303", "--http-port 8545", "--ws-port 8555", "--datadir ./nodes/1"):
            self.assertIn(expected, first)
        for expected in ("--port 30304", "--http-port 8546", "--ws-port 8556", "--datadir ./nodes/2"):
            self.assertIn(expected, second)

    def test_a_running_node_is_not_started_twice(self):
        self.node()
        result = self.node()
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("already running", result.stderr)

    def test_a_restart_reuses_the_wallet_and_needs_no_key(self):
        self.node()
        self.stop_nodes()
        result = self.node(PRIVATE_KEY_1=None)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("--unlock abc123", self.start_args())
        self.assertEqual(len([line for line in self.calls() if " init " in line]), 1)

    def test_refusals(self):
        cases = [
            (("0",), {}, "usage"),
            (("one",), {}, "usage"),
            (("5",), {}, "outside Total_Nodes"),
            (("1",), {"BOOTNODES": None}, "BOOTNODES"),
            (("1",), {"PRIVATE_KEY_1": None}, "PRIVATE_KEY_1"),
            (("1",), {"PROJECT_DIR": "/nonexistent"}, "Run setup.sh"),
            (("1",), {"GENESIS_FILE": "./missing.json"}, "missing.json"),
            (("1",), {"RPC_ADDR": "0.0.0.0", "RPC_API": "eth,personal"}, "refusing"),
            (("1",), {"RPC_ADDR": "0.0.0.0", "RPC_API": "admin"}, "refusing"),
            (("1",), {"RPC_ADDR": "0.0.0.0"}, "refusing to offer the eth API"),
            (("1",), {"RPC_ADDR": "0.0.0.0", "RPC_API": " "}, "RPC_API is blank"),
            (("1",), {"RPC_API": " "}, "RPC_API is blank"),
            (("1",), {"RPC_ADDR": "10.0.0.5", "RPC_API": "net, eth"}, "refusing to offer the eth API"),
            (("1",), {"RPC_ADDR": "10.0.0.5", "RPC_API": "net,txpool"}, "refusing to offer the txpool API"),
            (("1",), {"FAKE_BAD_IMPORT": "1"}, "could not import"),
        ]
        for args, extra, expected in cases:
            with self.subTest(args=args, extra=extra):
                result = self.node(*args, **extra)
                self.assertNotEqual(result.returncode, 0)
                self.assertIn(expected, result.stderr)
                self.assertFalse(os.path.exists(self.log) and any("--mine" in line for line in self.calls()))
            shutil.rmtree(os.path.join(self.dir, "nodes"), ignore_errors=True)

    def test_a_public_address_with_safe_apis_is_allowed(self):
        result = self.node(RPC_ADDR="10.0.0.5", RPC_API="net, web3,XDPoS")
        self.assertEqual(result.returncode, 0, result.stderr)
        args = self.start_args()
        self.assertIn("--http-addr 10.0.0.5", args)
        self.assertIn("--http-api net,web3,XDPoS", args)

    def test_a_local_address_may_offer_eth(self):
        for address in ("127.0.0.1", "localhost", "::1"):
            with self.subTest(address=address):
                result = self.node(RPC_ADDR=address)
                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertIn("--http-api eth,net,web3,XDPoS", self.start_args())
                self.stop_nodes()

    def test_a_node_that_dies_at_startup_is_reported(self):
        result = self.node(FAKE_CRASH="1")
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("stopped straight away", result.stderr)

    def test_a_bad_config_stops_the_node(self):
        self.path("config.yml", valid_config_text().replace("networkID: 72", "networkID: 50"))
        result = self.node()
        self.assertNotEqual(result.returncode, 0)
        self.assertFalse(os.path.exists(self.log))

    def test_stats_are_only_sent_when_asked_for(self):
        self.node()
        self.assertNotIn("--ethstats", self.start_args())
        self.stop_nodes()
        self.node(ETHSTATS="name:secret@stats.example:3000")
        self.assertIn("--ethstats name:secret@stats.example:3000", self.start_args())


class ConfigCommandTest(Workspace):
    def run_config(self, *args):
        return subprocess.run([sys.executable, "config.py", *args], cwd=self.dir, capture_output=True, text=True, timeout=30)

    def test_get_prints_one_setting(self):
        self.assertEqual(self.run_config("get", "networkID").stdout.strip(), "72")
        self.assertEqual(len(self.run_config("get", "Nodes").stdout.split()), 4)

    def test_get_refuses_an_unknown_setting(self):
        result = self.run_config("get", "Nothing")
        self.assertEqual(result.returncode, 1)
        self.assertIn("Nothing", result.stderr)

    def test_summary_lists_the_puppeth_values(self):
        output = self.run_config("summary").stdout
        self.assertIn("Chain_Name: Leo", output)
        self.assertIn("networkID: 72", output)

    def test_problems_exit_non_zero_and_are_listed(self):
        self.path("config.yml", yaml.safe_dump({"Chain_Name": "bad name"}))
        result = self.run_config("check")
        self.assertEqual(result.returncode, 1)
        self.assertIn("Chain_Name", result.stderr)

    def test_unreadable_files_are_reported(self):
        for content in ("just text", "a: [", ""):
            with self.subTest(content=content):
                self.path("config.yml", content)
                result = self.run_config("check")
                self.assertEqual(result.returncode, 1)
                self.assertIn("Could not read", result.stderr)

    def test_usage(self):
        self.assertEqual(self.run_config().returncode, 2)
        self.assertEqual(self.run_config("get").returncode, 2)


class SetupScriptTest(Workspace):
    def dry_run(self, **extra):
        return self.script("setup.sh", DRY_RUN="1", INSTALL_DIR=self.dir, PROJECT_DIR=None, **extra)

    def test_dry_run_changes_nothing_and_lists_the_steps(self):
        before = sorted(os.listdir(self.dir))
        result = self.dry_run(SETUP_MACHINE="x86_64")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(sorted(os.listdir(self.dir)), before)
        for expected in ("go1.23.12.linux-amd64.tar.gz", "checkout v2.7.1", "make -C %s/XDPoSChain all" % self.dir, "GOTOOLCHAIN=local"):
            self.assertIn(expected, result.stdout)

    def test_it_never_upgrades_removes_or_writes_outside_its_folder(self):
        output = self.dry_run(SETUP_MACHINE="x86_64").stdout
        for forbidden in ("upgrade", "apt-get remove", "add-apt-repository", "rm -rf", "/home/ubuntu", "chmod 777", "npm"):
            self.assertNotIn(forbidden, output)

    def test_the_config_is_checked_before_the_slow_steps(self):
        output = self.dry_run(SETUP_MACHINE="x86_64").stdout
        self.assertLess(output.index("config.py check"), output.index("curl -fsSL"))

    def test_arm_gets_the_arm_download(self):
        self.assertIn("linux-arm64", self.dry_run(SETUP_MACHINE="aarch64").stdout)

    def test_unknown_machines_are_refused(self):
        result = self.dry_run(SETUP_MACHINE="sparc")
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("sparc", result.stderr)

    def test_version_can_be_chosen(self):
        self.assertIn("checkout v9.9.9", self.dry_run(SETUP_MACHINE="x86_64", XDPOS_VERSION="v9.9.9").stdout)

    def fake_tools(self, **tools):
        folder = os.path.join(self.dir, "tools")
        os.makedirs(folder, exist_ok=True)
        for name, body in {"apt-get": "exit 0", **tools}.items():
            path = os.path.join(folder, name)
            with open(path, "w") as tool:
                tool.write("#!/bin/sh\necho \"%s $*\" >> \"$FAKE_LOG\"\n%s\n" % (name, body))
            os.chmod(path, 0o755)
        return folder + os.pathsep + os.path.dirname(sys.executable) + os.pathsep + os.environ["PATH"]

    def real_run(self, path):
        return subprocess.run(["bash", "setup.sh"], cwd=self.dir, capture_output=True, text=True, timeout=30,
                              env={"PATH": path, "HOME": self.dir, "INSTALL_DIR": self.dir, "SETUP_MACHINE": "x86_64", "FAKE_LOG": self.log})

    def test_a_failed_step_stops_the_script(self):
        result = self.real_run(self.fake_tools(sudo="exit 1", curl="exit 0"))
        self.assertNotEqual(result.returncode, 0)
        self.assertNotIn("curl", read(self.log))

    def test_a_bad_download_is_refused_before_it_is_unpacked(self):
        if shutil.which("sha256sum") is None:
            self.skipTest("sha256sum is not installed")
        result = self.real_run(self.fake_tools(sudo="exit 0", curl='echo tampered > "$2"', tar="exit 0"))
        self.assertNotEqual(result.returncode, 0)
        calls = read(self.log)
        self.assertIn("curl -fsSL", calls)
        self.assertNotIn("tar ", calls)
        self.assertFalse(os.path.exists(os.path.join(self.dir, "go-1.23.12")))


class RepositoryTest(unittest.TestCase):
    def test_secrets_and_node_data_are_ignored(self):
        with open(os.path.join(ROOT, ".gitignore")) as ignore:
            lines = ignore.read().splitlines()
        for name in (".env", ".pwd", "*.key", "nodes/", "genesis/", "*.log"):
            self.assertIn(name, lines)


if __name__ == "__main__":
    unittest.main()