rostrum 14.0.1

from enum import Enum
import functools
import subprocess
import os


class Network(Enum):
    BCH = 0
    NEX = 1


class Node(Enum):
    NEXA = 1
    BCHN = 3


class NodeFeature(Enum):
    SPAWN_ROSTRUM = (0,)
    TOKENS = (1,)
    RPC_LOGLINE = (3,)


@functools.cache
def rostrum_network() -> Network:
    """
    Determine what network rostrum is built for from version string
    """
    output: str = ""
    try:
        args = [rostrum_path(), "--version"]
        wrapper = process_wrapper()
        if wrapper is not None:
            args = [wrapper] + args

        subprocess.run(
            args,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=True,
        )
    except subprocess.CalledProcessError as e:
        # With --version parameter, the process is expected to return with returncode 1 and output version info to stderr
        if e.returncode != 1:
            raise

        output = e.stderr.decode("utf-8")

    if "nexa" in output:
        return Network.NEX
    if "bitcoin" in output:
        return Network.BCH
    raise Exception(f"Cannot tell network from following version info: '{output}'")


def node_network() -> Network:
    """
    Determine what network the full node is built for
    """
    if node() == Node.BCHN:
        return Network.BCH
    if node() == Node.NEXA:
        return Network.NEX
    raise NotImplementedError()


@functools.cache
def node() -> Node:
    output: str = ""

    args = [full_node_path(), "--version"]
    wrapper = process_wrapper()
    if wrapper is not None:
        args = [wrapper] + args

    result = subprocess.run(
        args,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=True,
    )

    output = result.stdout.decode("utf-8")
    if "Nexa" in output:
        return Node.NEXA

    if "Bitcoin Cash Node Daemon" in output:
        return Node.BCHN

    raise Exception(f"Cannot tell network from following version info: '{output}'")


@functools.cache
def network() -> Network:
    if rostrum_network() != node_network():
        raise Exception(
            f"Rostrum is built for {rostrum_network()}, while node software is built for {node_network()}"
        )
    return rostrum_network()


def on_bch() -> bool:
    return network() == Network.BCH


def on_nex() -> bool:
    return network() == Network.NEX


def full_node_path() -> str:
    if not "NODE_PATH" in os.environ:
        raise Exception(
            "Environment variable NODE_PATH is not set. Set NODE_PATH to the full path of the node software executable."
        )
    path = os.environ["NODE_PATH"]
    if not os.path.exists(path):
        raise Exception(
            f"Path '{path}' in environment variable NODE_PATH does not exist"
        )
    return path


def rostrum_path() -> str:
    if "ROSTRUM_PATH" in os.environ:
        path = os.environ["ROSTRUM_PATH"]
        if not os.path.exists(path):
            raise Exception(
                f"Path '{path}' set in environment variable ROSTRUM_PATH does not exist"
            )
        return path

    # When ROSTRUM_PATH is not defined, then assume rostrum is in the same directory as the node binary.

    path = os.path.join(os.path.dirname(full_node_path()), "rostrum")
    if not os.path.exists(path):
        raise Exception(
            f"Path '{path}' does not exist. Set path with environment variable ROSTRUM_PATH"
        )

    return path


def process_wrapper() -> str | None:
    """
    Processes wrapper, such as qemu and wine64 for testing builds for other platforms.
    """
    if not "PROCESS_WRAPPER" in os.environ:
        return None
    path = os.environ["PROCESS_WRAPPER"]

    if not os.path.exists(path):
        raise Exception(
            f"Path '{path}' set in environment variable PROCESS_WRAPPER does not exist"
        )
    return path


def node_supports(n: Node, feature: NodeFeature) -> bool:
    """
    If node supports given feature
    """
    if feature == NodeFeature.SPAWN_ROSTRUM and "NO_SPAWN_ROSTRUM" in os.environ:
        # Act as if this feature does not exist.
        # This makes the test framework spawn rostrum by itself.
        return False

    support_map = {
        Node.NEXA: [
            NodeFeature.SPAWN_ROSTRUM,
            NodeFeature.TOKENS,
            NodeFeature.RPC_LOGLINE,
        ],
        Node.BCHN: [NodeFeature.TOKENS],
    }

    if not n in support_map:
        raise NotImplementedError()

    return feature in support_map[n]


def testing_websocket():
    """
    If we are running tests via websocket.

    By default tests are run via TCP interface.
    """
    return "ROSTRUM_TEST_WEBSOCKET" in os.environ


def testing_http():
    """
    If we are running tests via HTTP interface.
    """
    return "ROSTRUM_TEST_HTTP" in os.environ


def announce_enabled():
    """
    If we want to enable discovery announcement on regtest
    """
    return "ROSTRUM_ANNOUNCE" in os.environ


def testing_tcp_ssl():
    """
    If we are running tests via TCP SSL interface.
    """
    return "ROSTRUM_TEST_TCP_SSL" in os.environ


def testing_websocket_ssl():
    """
    If we are running tests via WebSocket SSL interface.
    """
    return "ROSTRUM_TEST_WEBSOCKET_SSL" in os.environ


def testing_https():
    """
    If we are running tests via HTTPS interface.
    """
    return "ROSTRUM_TEST_HTTPS" in os.environ