specter-desktop/tests/conftest.py

import atexit
import code
import json
import logging
import os
import shutil
import signal
import sys
import tempfile
import traceback

import pytest

from cryptoadvance.specter.config import TestConfig
from cryptoadvance.specter.managers.device_manager import DeviceManager
from cryptoadvance.specter.managers.node_manager import NodeManager
from cryptoadvance.specter.managers.user_manager import UserManager
from cryptoadvance.specter.node import Node
from cryptoadvance.specter.process_controller.bitcoind_controller import (
    BitcoindPlainController,
)
from cryptoadvance.specter.process_controller.elementsd_controller import (
    ElementsPlainController,
)
from cryptoadvance.specter.server import SpecterFlask, create_app, init_app
from cryptoadvance.specter.specter import Specter
from cryptoadvance.specter.specter_error import (
    BrokenCoreConnectionException,
    SpecterError,
    handle_exception,
)
from cryptoadvance.specter.user import User, hash_password
from cryptoadvance.specter.util.common import str2bool
from cryptoadvance.specter.util.shell import which
from cryptoadvance.specter.util.wallet_importer import WalletImporter

logger = logging.getLogger(__name__)

pytest_plugins = [
    "conftest_visibility",
    "fix_ghost_machine",
    "fix_keys_and_seeds",
    "fix_devices_and_wallets",
    "fix_testnet",
]

# This is from https://stackoverflow.com/questions/132058/showing-the-stack-trace-from-a-running-python-application
# it enables stopping a hanging test via sending the pytest-process a SIGUSR2 (12)
# kill 12 pid-of-pytest
# In the article they claim to open a debug-console which didn't work for me but at least
# you get a stacktrace in the output.
def debug(sig, frame):
    """Interrupt running process, and provide a python prompt for
    interactive debugging."""
    d = {"_frame": frame}  # Allow access to frame object.
    d.update(frame.f_globals)  # Unless shadowed by global
    d.update(frame.f_locals)

    i = code.InteractiveConsole(d)
    message = "Signal received : entering python shell.\nTraceback:\n"
    message += "".join(traceback.format_stack(frame))
    i.interact(message)


def listen():
    signal.signal(signal.SIGUSR2, debug)  # Register handler


def pytest_addoption(parser):
    """Internally called to add options to pytest
    see pytest_generate_tests(metafunc) on how to check that
    Also used to register the SIGUSR2 (12) as decribed in conftest.py
    """
    parser.addoption(
        "--bitcoind-version",
        action="store",
        default="v0.20.1",
        help="Version of bitcoind (something which works with git checkout ...)",
    )
    parser.addoption(
        "--bitcoind-log-stdout",
        action="store",
        default=False,
        help="Whether bitcoind should log to stdout (default:False)",
    )
    parser.addoption(
        "--elementsd-version",
        action="store",
        default="master",
        help="Version of elementsd (something which works with git checkout ...)",
    )
    parser.addoption(
        "--run-jade-hardware",
        action="store_true",
        default=False,
        help="Run tests marked jade_hardware (real Jade attached + operator).",
    )
    listen()


def pytest_collection_modifyitems(config, items):
    if config.getoption("--run-jade-hardware"):
        return
    skip = pytest.mark.skip(reason="opt-in via --run-jade-hardware")
    for item in items:
        if item.get_closest_marker("jade_hardware") is not None:
            item.add_marker(skip)


def pytest_generate_tests(metafunc):
    # ToDo: use custom compiled version of bitcoind
    # E.g. test again bitcoind version [currentRelease] + master-branch
    if "docker" in metafunc.fixturenames:
        if metafunc.config.getoption("docker"):
            # That's a list because we could do both (see above) but currently that doesn't make sense in that context
            metafunc.parametrize("docker", [True], scope="session")
        else:
            metafunc.parametrize("docker", [False], scope="session")


def instantiate_bitcoind_controller(
    request, rpcport=18543, extra_args=[]
) -> BitcoindPlainController:
    # logging.getLogger().setLevel(logging.DEBUG)
    requested_version = request.config.getoption("--bitcoind-version")
    log_stdout = str2bool(request.config.getoption("--bitcoind-log-stdout"))
    if os.path.isfile("tests/bitcoin/src/bitcoind"):
        bitcoind_controller = BitcoindPlainController(
            bitcoind_path="tests/bitcoin/src/bitcoind", rpcport=rpcport
        )  # always prefer the self-compiled bitcoind if existing
    elif os.path.isfile("tests/bitcoin/bin/bitcoind"):
        bitcoind_controller = BitcoindPlainController(
            bitcoind_path="tests/bitcoin/bin/bitcoind", rpcport=rpcport
        )  # next take the self-installed binary if existing
    else:
        bitcoind_controller = BitcoindPlainController(
            rpcport=rpcport
        )  # Alternatively take the one on the path for now
    bitcoind_controller.start_bitcoind(
        cleanup_at_exit=True,
        cleanup_hard=True,
        extra_args=extra_args,
        log_stdout=log_stdout,
    )
    assert not bitcoind_controller.datadir is None
    running_version = bitcoind_controller.version()
    requested_version = request.config.getoption("--bitcoind-version")
    assert running_version == requested_version, (
        "Please make sure that the Bitcoind-version (%s) matches with the version in pyproject.toml (%s)"
        % (running_version, requested_version)
    )
    return bitcoind_controller


def instantiate_elementsd_controller(request, rpcport=18643, extra_args=[]):
    if os.path.isfile("tests/elements/src/elementsd"):
        elementsd_controller = ElementsPlainController(
            elementsd_path="tests/elements/src/elementsd", rpcport=rpcport
        )  # always prefer the self-compiled bitcoind if existing
    elif os.path.isfile("tests/elements/bin/elementsd"):
        elementsd_controller = ElementsPlainController(
            elementsd_path="tests/elements/bin/elementsd", rpcport=rpcport
        )  # next take the self-installed binary if existing
    else:
        elementsd_controller = ElementsPlainController(
            rpcport=rpcport
        )  # Alternatively take the one on the path for now
    elementsd_controller.start_elementsd(
        cleanup_at_exit=True, cleanup_hard=True, extra_args=extra_args
    )
    assert not elementsd_controller.datadir is None
    running_version = elementsd_controller.version()
    requested_version = request.config.getoption("--elementsd-version")
    assert running_version == requested_version, (
        "Please make sure that the elementsd-version (%s) matches with the version in pyproject.toml (%s)"
        % (running_version, requested_version)
    )
    return elementsd_controller


# Below this point are fixtures. Fixtures have a scope. Check about scopes here:
# https://docs.pytest.org/en/6.2.x/fixture.html#scope-sharing-fixtures-across-classes-modules-packages-or-session
# possible values: function, class, module, package or session.
# The nodes are of scope session. All else is the default (function)


@pytest.fixture(scope="session")
def bitcoind_path():
    if os.path.isfile("tests/bitcoin/src/bitcoind"):
        return "tests/bitcoin/src/bitcoind"
    elif os.path.isfile("tests/bitcoin/bin/bitcoind"):
        return "tests/bitcoin/bin/bitcoind"
    else:
        return which("bitcoind")


@pytest.fixture(scope="session")
def bitcoin_regtest(request) -> BitcoindPlainController:
    bitcoind_regtest = instantiate_bitcoind_controller(request, extra_args=None)
    try:
        assert bitcoind_regtest.get_rpc().test_connection()
        assert not bitcoind_regtest.datadir is None
        assert bitcoind_regtest.datadir is not ""
        yield bitcoind_regtest
    finally:
        bitcoind_regtest.stop_bitcoind()


@pytest.fixture(scope="session")
def bitcoin_regtest2(request) -> BitcoindPlainController:
    """If a test needs two nodes ..."""
    bitcoind_regtest = instantiate_bitcoind_controller(
        request, rpcport=18545, extra_args=None
    )
    try:
        assert bitcoind_regtest.get_rpc().test_connection()
        assert not bitcoind_regtest.datadir is None
        yield bitcoind_regtest
    finally:
        bitcoind_regtest.stop_bitcoind()


@pytest.fixture
def node(empty_data_folder, bitcoin_regtest):
    nodes_folder = empty_data_folder + "/nodes"
    if not os.path.isdir(nodes_folder):
        os.makedirs(nodes_folder)
    nm = NodeManager(data_folder=nodes_folder)
    node: Node = nm.add_external_node(
        "BTC",
        "Standard node",
        False,
        bitcoin_regtest.datadir,
        bitcoin_regtest.rpcconn.rpcuser,
        bitcoin_regtest.rpcconn.rpcpassword,
        bitcoin_regtest.rpcconn.rpcport,
        bitcoin_regtest.rpcconn._ipaddress,
        "http",
    )
    assert node.rpc.test_connection()
    return node


@pytest.fixture
def node_with_different_port(empty_data_folder, bitcoin_regtest):
    nodes_folder = empty_data_folder + "/nodes"
    if not os.path.isdir(nodes_folder):
        os.makedirs(nodes_folder)
    nm = NodeManager(data_folder=nodes_folder)
    node = nm.add_external_node(
        "BTC",
        "Node with a different port",
        False,
        "",
        bitcoin_regtest.rpcconn.rpcuser,
        bitcoin_regtest.rpcconn.rpcpassword,
        18333,
        bitcoin_regtest.rpcconn._ipaddress,
        "http",
    )
    return node


@pytest.fixture
def node_with_empty_datadir(empty_data_folder, bitcoin_regtest):
    nodes_folder = empty_data_folder + "/nodes"
    if not os.path.isdir(nodes_folder):
        os.makedirs(nodes_folder)
    node = Node.from_json(
        {
            "autodetect": False,
            "datadir": "",
            "user": bitcoin_regtest.rpcconn.rpcuser,
            "password": bitcoin_regtest.rpcconn.rpcpassword,
            "port": bitcoin_regtest.rpcconn.rpcport,
            "host": bitcoin_regtest.rpcconn.ipaddress,
            "protocol": "http",
        },
        manager=NodeManager(data_folder=nodes_folder),
        default_fullpath=os.path.join(nodes_folder, "node_with_empty_datadir.json"),
    )
    return node


@pytest.fixture(scope="session")
def elements_elreg(request):
    elements_elreg = instantiate_elementsd_controller(request, extra_args=None)
    try:
        yield elements_elreg
        assert not elements_elreg.datadir is None
    finally:
        elements_elreg.stop_elementsd()


@pytest.fixture
def empty_data_folder():
    # Make sure that this folder never ever gets a reasonable non-testing use-case
    with tempfile.TemporaryDirectory(
        prefix="specter_home_tmp_", ignore_cleanup_errors=False
    ) as data_folder:
        yield data_folder


@pytest.fixture
def devices_filled_data_folder(empty_data_folder):
    devices_folder = empty_data_folder + "/devices"
    if not os.path.isdir(devices_folder):
        os.makedirs(devices_folder)

    shutil.copy2(
        "./tests/misc_testdata/trezor_device.json",
        empty_data_folder + "/devices/trezor.json",
    )
    shutil.copy2(
        "./tests/misc_testdata/specter_device.json",
        empty_data_folder + "/devices/specter.json",
    )
    return empty_data_folder  # no longer empty, though


@pytest.fixture
def wallets_filled_data_folder(devices_filled_data_folder):
    simple_json = """
{
    "alias": "simple",
    "fullpath": "/home/kim/.specter/wallets/regtest/simple.json",
    "name": "Simple",
    "address_index": 0,
    "keypool": 5,
    "address": "bcrt1qcatuhg0gll3h7py4cmn53rjjn9xlsqfwj3zcej",
    "change_index": 0,
    "change_address": "bcrt1qt28v03278lmmxllys89acddp2p5y4zds94944n",
    "change_keypool": 5,
    "type": "simple",
    "description": "Single (Segwit)",
    "keys": [{
        "derivation": "m/84h/1h/0h",
        "original": "vpub5Y35MNUT8sUR2SnRCU9A9S6z1JDACMTuNnM8WHXvuS7hCwuVuoRAWJGpi66Yo8evGPiecN26oLqx19xf57mqVQjiYb9hbb4QzbNmFfsS9ko",
        "fingerprint": "1ef4e492",
        "type": "wpkh",
        "xpub": "tpubDC5EUwdy9WWpzqMWKNhVmXdMgMbi4ywxkdysRdNr1MdM4SCfVLbNtsFvzY6WKSuzsaVAitj6FmP6TugPuNT6yKZDLsHrSwMd816TnqX7kuc"
    }],
    "recv_descriptor": "wpkh([1ef4e492/84h/1h/0h]tpubDC5EUwdy9WWpzqMWKNhVmXdMgMbi4ywxkdysRdNr1MdM4SCfVLbNtsFvzY6WKSuzsaVAitj6FmP6TugPuNT6yKZDLsHrSwMd816TnqX7kuc/0/*)#xp8lv5nr",
    "change_descriptor": "wpkh([1ef4e492/84h/1h/0h]tpubDC5EUwdy9WWpzqMWKNhVmXdMgMbi4ywxkdysRdNr1MdM4SCfVLbNtsFvzY6WKSuzsaVAitj6FmP6TugPuNT6yKZDLsHrSwMd816TnqX7kuc/1/*)#h4z73prm",
    "device": "Trezor",
    "device_type": "trezor",
    "address_type": "bech32"
}
"""
    another_wallet_json = """
{
    "name": "sdsd",
    "alias": "sdsd",
    "description": "Single (Segwit)",
    "address_type": "bech32",
    "address": "bcrt1q4h86vfanswhsle63hw2muv9h5a45cg2878uez5",
    "address_index": 0,
    "change_address": "bcrt1qxsj28ddr95xvp7xjyzkkfq6qknrn4kap30zkut",
    "change_index": 0,
    "keypool": 60,
    "change_keypool": 20,
    "recv_descriptor": "wpkh([41490ec7/84h/1h/0h]tpubDCTPz7KwyetfhQNMSWiK34pPR2zSTsTybrMPgRVAzouNLqtgsv51o81KjccmTbjkWJ8mVhRJM1LxZD6AfRH2635tHpHeCAKW446iwADNv7C/0/*)#rn833s5g",
    "change_descriptor": "wpkh([41490ec7/84h/1h/0h]tpubDCTPz7KwyetfhQNMSWiK34pPR2zSTsTybrMPgRVAzouNLqtgsv51o81KjccmTbjkWJ8mVhRJM1LxZD6AfRH2635tHpHeCAKW446iwADNv7C/1/*)#j8zsv9ys",
    "keys": [
        {
            "original": "vpub5YRErYARy1rFj1oGKc9yQyJ1jybtbEyvDziem5eFttPiVMbXJNtoQZ2DTAcowHUfu7NFPAiJtaop6TNRqAbkc8GPVY9VLp2HveP2PygjuYh",
            "fingerprint": "41490ec7",
            "derivation": "m/84h/1h/0h",
            "type": "wpkh",
            "purpose": "#0 Single Sig (Segwit)",
            "xpub": "tpubDCTPz7KwyetfhQNMSWiK34pPR2zSTsTybrMPgRVAzouNLqtgsv51o81KjccmTbjkWJ8mVhRJM1LxZD6AfRH2635tHpHeCAKW446iwADNv7C"
        }
    ],
    "devices": [
        "dsds"
    ],
    "sigs_required": 1,
    "blockheight": 0,
    "pending_psbts": {},
    "frozen_utxo": [],
    "last_block": "187e2db380eb6d901efd87188f00c7074506c9c3813b8ecec7300ecc4e55eb46"
}
"""

    os.makedirs(os.path.join(devices_filled_data_folder, "wallets", "regtest"))
    with open(
        os.path.join(devices_filled_data_folder, "wallets", "regtest", "simple.json"),
        "w",
    ) as json_file:
        json_file.write(simple_json)
    os.makedirs(os.path.join(devices_filled_data_folder, "wallets_someuser", "regtest"))
    with open(
        os.path.join(
            devices_filled_data_folder, "wallets_someuser", "regtest", "simple.json"
        ),
        "w",
    ) as json_file:
        json_file.write(another_wallet_json)
    return devices_filled_data_folder  # and with wallets obviously


@pytest.fixture
def device_manager(devices_filled_data_folder):
    return DeviceManager(os.path.join(devices_filled_data_folder, "devices"))


# @pytest.fixture
# def user_manager(empty_data_folder) -> UserManager:
#     """A UserManager having users alice, bob and eve"""
#     specter = Specter(data_folder=empty_data_folder)
#     user_manager = UserManager(specter=specter)
#     config = {}
#     user_manager.get_user("admin").decrypt_user_secret("admin")
#     user_manager.create_user(
#         user_id="alice",
#         username="alice",
#         plaintext_password="plain_pass_alice",
#         config=config,
#     )
#     user_manager.create_user(
#         user_id="bob",
#         username="bob",
#         plaintext_password="plain_pass_bob",
#         config=config,
#     )
#     user_manager.create_user(
#         user_id="eve",
#         username="eve",
#         plaintext_password="plain_pass_eve",
#         config=config,
#     )
#     return user_manager


@pytest.fixture
def specter_regtest_configured(bitcoin_regtest, devices_filled_data_folder, node):
    assert bitcoin_regtest.get_rpc().test_connection()
    config = {
        "rpc": {
            "autodetect": False,
            "datadir": bitcoin_regtest.datadir,
            "user": bitcoin_regtest.rpcconn.rpcuser,
            "password": bitcoin_regtest.rpcconn.rpcpassword,
            "port": bitcoin_regtest.rpcconn.rpcport,
            "host": bitcoin_regtest.rpcconn.ipaddress,
            "protocol": "http",
        },
        "auth": {
            "method": "rpcpasswordaspin",
        },
        "testing": {
            "allow_threading_for_testing": False,
        },
    }
    specter: Specter = Specter(
        data_folder=devices_filled_data_folder, config=config, checker_threads=False
    )
    assert specter.active_node_alias == "bitcoin_core"
    assert specter.node_manager.active_node.alias == "bitcoin_core"
    assert specter.chain == "regtest"
    # Create a User
    someuser = specter.user_manager.add_user(
        User.from_json(
            user_dict={
                "id": "someuser",
                "username": "someuser",
                "password": hash_password("somepassword"),
                "config": {},
                "is_admin": False,
                "services": None,
            },
            specter=specter,
        )
    )
    specter.user_manager.save()
    specter.check()

    assert not specter.wallet_manager.working_folder is None
    try:
        yield specter
    finally:
        # End all threads
        # Deleting all Wallets (this will also purge them on core)
        for user in specter.user_manager.users:
            for wallet in list(user.wallet_manager.wallets.values()):
                user.wallet_manager.delete_wallet(wallet, node)


@pytest.fixture
def specter_regtest_configured_with_threading(
    bitcoin_regtest, devices_filled_data_folder, node
):
    assert bitcoin_regtest.get_rpc().test_connection()
    config = {
        "rpc": {
            "autodetect": False,
            "datadir": bitcoin_regtest.datadir,
            "user": bitcoin_regtest.rpcconn.rpcuser,
            "password": bitcoin_regtest.rpcconn.rpcpassword,
            "port": bitcoin_regtest.rpcconn.rpcport,
            "host": bitcoin_regtest.rpcconn.ipaddress,
            "protocol": "http",
        },
        "auth": {
            "method": "rpcpasswordaspin",
        },
        "testing": {
            "allow_threading_for_testing": True,
        },
    }
    specter = Specter(data_folder=devices_filled_data_folder, config=config)
    assert specter.chain == "regtest"
    # Create a User
    someuser = specter.user_manager.add_user(
        User.from_json(
            user_dict={
                "id": "someuser",
                "username": "someuser",
                "password": hash_password("somepassword"),
                "config": {},
                "is_admin": False,
                "services": None,
            },
            specter=specter,
        )
    )
    specter.user_manager.save()
    specter.check()

    assert not specter.wallet_manager.working_folder is None
    try:
        yield specter
    finally:
        # Deleting all Wallets (this will also purge them on core)
        for user in specter.user_manager.users:
            for wallet in list(user.wallet_manager.wallets.values()):
                user.wallet_manager.delete_wallet(wallet, node)


def specter_app_with_config(config={}, specter=None):
    """helper-function to create SpecterFlasks"""
    if isinstance(config, dict):
        tempClass = type("tempClass", (TestConfig,), {})
        for key, value in config.items():
            setattr(tempClass, key, value)
        # service_manager will expect the class to be defined as a direct property of the module:
        if hasattr(sys.modules[__name__], "tempClass"):
            delattr(sys.modules[__name__], "tempClass")
        assert not hasattr(sys.modules[__name__], "tempClass")
        setattr(sys.modules[__name__], "tempClass", tempClass)
        assert hasattr(sys.modules[__name__], "tempClass")
        assert getattr(sys.modules[__name__], "tempClass") == tempClass
        config = tempClass
    app = create_app(config=config)
    app.app_context().push()
    app.config["TESTING"] = True
    app.testing = True
    app.tor_service_id = None
    app.tor_enabled = False
    init_app(app, specter=specter)
    return app


@pytest.fixture
def app(specter_regtest_configured) -> SpecterFlask:
    """the Flask-App, but uninitialized"""
    return specter_app_with_config(
        config="cryptoadvance.specter.config.TestConfig",
        specter=specter_regtest_configured,
    )


@pytest.fixture
def app_no_node(empty_data_folder) -> SpecterFlask:
    specter = Specter(data_folder=empty_data_folder, checker_threads=False)
    app = create_app(config="cryptoadvance.specter.config.TestConfig")
    app.app_context().push()
    app.config["TESTING"] = True
    app.testing = True
    app.tor_service_id = None
    app.tor_enabled = False
    init_app(app, specter=specter)
    return app


@pytest.fixture
def client(app):
    """a test_client from an initialized Flask-App"""
    return app.test_client()