path: root/tests/phase0/monzero_multinode.py

#!/usr/bin/env python3

"""Isolated Monzero three-node propagation and restart smoke test.

This script uses temporary regtest data and wallet directories. It never reads
or writes the user's normal Monzero data directory or wallet files.
"""

from __future__ import annotations

import argparse
import json
import pathlib
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request


HOST = "127.0.0.1"
P2P_PORTS = (26174, 26184, 26194)
RPC_PORTS = (26175, 26185, 26195)
ZMQ_PORTS = (26176, 26186, 26196)
WALLET_RPC_PORT = 26205


class TestFailure(RuntimeError):
    pass


def request(port: int, path: str, body: dict, timeout: float = 5.0) -> dict:
    payload = json.dumps(body).encode("utf-8")
    req = urllib.request.Request(
        f"http://{HOST}:{port}{path}",
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=timeout) as response:
        result = json.loads(response.read().decode("utf-8"))
    if "error" in result:
        raise TestFailure(f"RPC error from port {port}: {result['error']}")
    return result


def json_rpc(port: int, method: str, params: dict | None = None) -> dict:
    response = request(
        port,
        "/json_rpc",
        {"jsonrpc": "2.0", "id": "0", "method": method, "params": params or {}},
    )
    return response.get("result", {})


def get_info(port: int) -> dict:
    return request(port, "/get_info", {})


def transaction_pool_hashes(port: int) -> set[str]:
    response = request(port, "/get_transaction_pool", {})
    return {
        transaction.get("id_hash", "")
        for transaction in response.get("transactions", [])
        if transaction.get("id_hash")
    }


def wait_until(description: str, predicate, timeout: float = 45.0) -> None:
    deadline = time.monotonic() + timeout
    last_error = None
    while time.monotonic() < deadline:
        try:
            if predicate():
                return
        except (OSError, TestFailure, urllib.error.URLError, ValueError) as error:
            last_error = error
        time.sleep(0.25)
    suffix = f"; last error: {last_error}" if last_error else ""
    raise TestFailure(f"Timed out waiting for {description}{suffix}")


def stop_process(process: subprocess.Popen | None) -> None:
    if process is None or process.poll() is not None:
        return
    process.send_signal(signal.SIGTERM)
    try:
        process.wait(timeout=10)
    except subprocess.TimeoutExpired:
        process.kill()
        process.wait(timeout=5)


def daemon_command(binary: pathlib.Path, root: pathlib.Path, index: int) -> list[str]:
    peers = []
    if index > 0:
        peers.extend(["--add-exclusive-node", f"{HOST}:{P2P_PORTS[index - 1]}"])
    if index + 1 < len(P2P_PORTS):
        peers.extend(["--add-exclusive-node", f"{HOST}:{P2P_PORTS[index + 1]}"])
    return [
        str(binary),
        "--regtest",
        "--fixed-difficulty", "1",
        "--data-dir", str(root / f"node-{index}"),
        "--p2p-bind-ip", HOST,
        "--p2p-bind-port", str(P2P_PORTS[index]),
        "--rpc-bind-ip", HOST,
        "--rpc-bind-port", str(RPC_PORTS[index]),
        "--zmq-rpc-bind-ip", HOST,
        "--zmq-rpc-bind-port", str(ZMQ_PORTS[index]),
        "--rpc-ssl", "disabled",
        "--disable-dns-checkpoints",
        "--check-updates", "disabled",
        "--no-igd",
        "--non-interactive",
        "--log-level", "0",
        *peers,
    ]


def start_daemon(binary: pathlib.Path, root: pathlib.Path, index: int, logs) -> subprocess.Popen:
    log = open(root / f"node-{index}.log", "ab", buffering=0)
    logs.append(log)
    return subprocess.Popen(
        daemon_command(binary, root, index),
        stdout=log,
        stderr=subprocess.STDOUT,
        start_new_session=True,
    )


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--build-dir", default="build", help="CMake build directory")
    parser.add_argument("--keep-data", action="store_true", help="keep temporary logs and databases")
    args = parser.parse_args()

    build_dir = pathlib.Path(args.build_dir).resolve()
    daemon = build_dir / "bin" / "monzerod"
    wallet_rpc = build_dir / "bin" / "monzero-wallet-rpc"
    for binary in (daemon, wallet_rpc):
        if not binary.is_file():
            raise TestFailure(f"Required binary not found: {binary}")

    root = pathlib.Path(tempfile.mkdtemp(prefix="monzero-phase0-"))
    processes: list[subprocess.Popen | None] = [None, None, None]
    wallet_process = None
    logs = []

    try:
        print(f"Temporary test network: {root}")
        for index in range(3):
            processes[index] = start_daemon(daemon, root, index, logs)

        for index, port in enumerate(RPC_PORTS):
            wait_until(f"node {index} RPC", lambda port=port: get_info(port).get("status") == "OK")

        wait_until(
            "the three-node P2P topology",
            lambda: all(
                get_info(port).get("incoming_connections_count", 0)
                + get_info(port).get("outgoing_connections_count", 0) >= 1
                for port in RPC_PORTS
            ),
            timeout=60,
        )

        wallet_log = open(root / "wallet-rpc.log", "ab", buffering=0)
        logs.append(wallet_log)
        (root / "wallets").mkdir()
        wallet_process = subprocess.Popen(
            [
                str(wallet_rpc),
                "--wallet-dir", str(root / "wallets"),
                "--rpc-bind-ip", HOST,
                "--rpc-bind-port", str(WALLET_RPC_PORT),
                "--disable-rpc-login",
                "--rpc-ssl", "disabled",
                "--daemon-address", f"{HOST}:{RPC_PORTS[0]}",
                "--daemon-ssl", "disabled",
                # Regtest intentionally activates the latest hard fork at
                # height 1, which cannot match the public-network schedule.
                "--allow-mismatched-daemon-version",
                "--log-level", "0",
            ],
            stdout=wallet_log,
            stderr=subprocess.STDOUT,
            start_new_session=True,
        )
        wait_until("wallet RPC", lambda: json_rpc(WALLET_RPC_PORT, "get_version").get("version", 0) > 0)
        json_rpc(WALLET_RPC_PORT, "create_wallet", {"filename": "miner", "password": "", "language": "English"})
        address = json_rpc(WALLET_RPC_PORT, "get_address").get("address")
        if not address:
            raise TestFailure("Temporary wallet did not return an address")

        initial_height = get_info(RPC_PORTS[0])["height"]
        json_rpc(RPC_PORTS[0], "generateblocks", {"wallet_address": address, "amount_of_blocks": 12})
        target_height = initial_height + 12
        wait_until(
            f"all nodes to reach height {target_height}",
            lambda: all(get_info(port)["height"] >= target_height for port in RPC_PORTS),
            timeout=90,
        )
        hashes = [get_info(port)["top_block_hash"] for port in RPC_PORTS]
        if len(set(hashes)) != 1:
            raise TestFailure(f"Nodes disagree on the propagated tip: {hashes}")

        print("Stopping node 2 and extending the chain by five blocks...")
        stop_process(processes[2])
        processes[2] = None
        json_rpc(RPC_PORTS[0], "generateblocks", {"wallet_address": address, "amount_of_blocks": 5})
        restart_target = target_height + 5
        wait_until("connected nodes to accept the extension", lambda: get_info(RPC_PORTS[1])["height"] >= restart_target)

        processes[2] = start_daemon(daemon, root, 2, logs)
        wait_until("restarted node 2 RPC", lambda: get_info(RPC_PORTS[2]).get("status") == "OK")
        wait_until(
            f"restarted node 2 to catch up to height {restart_target}",
            lambda: get_info(RPC_PORTS[2])["height"] >= restart_target,
            timeout=90,
        )
        final_info = [get_info(port) for port in RPC_PORTS]
        final_hashes = [info["top_block_hash"] for info in final_info]
        if len(set(final_hashes)) != 1:
            raise TestFailure(f"Nodes disagree after restart: {final_hashes}")

        print("Testing competing forks and longest-chain convergence...")
        stop_process(processes[1])
        processes[1] = None
        wait_until(
            "edge nodes to become isolated",
            lambda: all(
                get_info(port).get("incoming_connections_count", 0)
                + get_info(port).get("outgoing_connections_count", 0) == 0
                for port in (RPC_PORTS[0], RPC_PORTS[2])
            ),
            timeout=30,
        )
        json_rpc(RPC_PORTS[0], "generateblocks", {"wallet_address": address, "amount_of_blocks": 3})
        json_rpc(RPC_PORTS[2], "generateblocks", {"wallet_address": address, "amount_of_blocks": 5})
        longer_fork_height = restart_target + 5
        longer_fork_tip = get_info(RPC_PORTS[2])["top_block_hash"]

        processes[1] = start_daemon(daemon, root, 1, logs)
        wait_until("restarted node 1 RPC", lambda: get_info(RPC_PORTS[1]).get("status") == "OK")
        wait_until(
            "all nodes to adopt the longer competing fork",
            lambda: all(
                get_info(port)["height"] >= longer_fork_height
                and get_info(port)["top_block_hash"] == longer_fork_tip
                for port in RPC_PORTS
            ),
            timeout=120,
        )

        print("Testing wallet-to-wallet transaction relay and confirmation...")
        json_rpc(RPC_PORTS[0], "generateblocks", {"wallet_address": address, "amount_of_blocks": 60})
        spendable_height = longer_fork_height + 60
        wait_until(
            f"all nodes to reach spendable height {spendable_height}",
            lambda: all(get_info(port)["height"] >= spendable_height for port in RPC_PORTS),
            timeout=120,
        )
        json_rpc(WALLET_RPC_PORT, "refresh")

        json_rpc(WALLET_RPC_PORT, "close_wallet")
        json_rpc(WALLET_RPC_PORT, "create_wallet", {"filename": "receiver", "password": "", "language": "English"})
        receiver_address = json_rpc(WALLET_RPC_PORT, "get_address").get("address")
        if not receiver_address:
            raise TestFailure("Receiver wallet did not return an address")
        json_rpc(WALLET_RPC_PORT, "close_wallet")
        json_rpc(WALLET_RPC_PORT, "open_wallet", {"filename": "miner", "password": ""})
        json_rpc(WALLET_RPC_PORT, "refresh")

        transfer = json_rpc(
            WALLET_RPC_PORT,
            "transfer",
            {
                "destinations": [{"amount": 100_000_000_000, "address": receiver_address}],
                "get_tx_key": True,
            },
        )
        transaction_hash = transfer.get("tx_hash")
        if not transaction_hash:
            raise TestFailure(f"Transfer did not return a transaction hash: {transfer}")
        wait_until(
            "transaction relay to every node",
            lambda: all(transaction_hash in transaction_pool_hashes(port) for port in RPC_PORTS),
            timeout=60,
        )

        json_rpc(RPC_PORTS[1], "generateblocks", {"wallet_address": address, "amount_of_blocks": 1})
        confirmed_height = spendable_height + 1
        wait_until(
            f"transaction confirmation at height {confirmed_height}",
            lambda: all(
                get_info(port)["height"] >= confirmed_height
                and transaction_hash not in transaction_pool_hashes(port)
                for port in RPC_PORTS
            ),
            timeout=90,
        )
        json_rpc(WALLET_RPC_PORT, "close_wallet")
        json_rpc(WALLET_RPC_PORT, "open_wallet", {"filename": "receiver", "password": ""})
        json_rpc(WALLET_RPC_PORT, "refresh")
        receiver_balance = json_rpc(WALLET_RPC_PORT, "get_balance").get("balance", 0)
        if receiver_balance != 100_000_000_000:
            raise TestFailure(f"Receiver balance is {receiver_balance}, expected 100000000000")

        receiver_seed = json_rpc(WALLET_RPC_PORT, "query_key", {"key_type": "mnemonic"}).get("key")
        if not receiver_seed:
            raise TestFailure("Receiver wallet did not return its temporary recovery seed")
        json_rpc(WALLET_RPC_PORT, "close_wallet")
        restored = json_rpc(
            WALLET_RPC_PORT,
            "restore_deterministic_wallet",
            {
                "filename": "receiver-restored",
                "password": "",
                "seed": receiver_seed,
                "restore_height": 0,
            },
        )
        if restored.get("address") != receiver_address:
            raise TestFailure(
                f"Restored address {restored.get('address')} does not match {receiver_address}"
            )
        json_rpc(WALLET_RPC_PORT, "refresh")
        restored_balance = json_rpc(WALLET_RPC_PORT, "get_balance").get("balance", 0)
        if restored_balance != receiver_balance:
            raise TestFailure(
                f"Restored balance is {restored_balance}, expected {receiver_balance}"
            )

        final_info = [get_info(port) for port in RPC_PORTS]
        final_hashes = [info["top_block_hash"] for info in final_info]
        if len(set(final_hashes)) != 1:
            raise TestFailure(f"Nodes disagree after transaction confirmation: {final_hashes}")

        print("PASS: connectivity, propagation, catch-up, reorg, transfer, and wallet restoration")
        print("Final height:", final_info[0]["height"])
        print("Final tip:", final_hashes[0])
        print("Confirmed transaction:", transaction_hash)
        return 0
    finally:
        stop_process(wallet_process)
        for process in reversed(processes):
            stop_process(process)
        for log in logs:
            log.close()
        if args.keep_data:
            kept = pathlib.Path.cwd() / f"monzero-phase0-{int(time.time())}"
            shutil.move(str(root), str(kept))
            print(f"Kept test data at {kept}")
        else:
            shutil.rmtree(root, ignore_errors=True)


if __name__ == "__main__":
    try:
        sys.exit(main())
    except (TestFailure, OSError, urllib.error.URLError) as error:
        print(f"FAIL: {error}", file=sys.stderr)
        sys.exit(1)