plm-electrumx/tests/integration/test_protocol.py

'''Electrum protocol methods per protocol version, daemon limits and reorgs, against the regtest node (see conftest.py).'''
import pytest

import ssl

from tests.integration.conftest import SSL_SERVICE, ElectrumClient, new_address, scripthash, wait_for


@pytest.fixture
def client(electrumx):
    client = ElectrumClient(electrumx)
    yield client
    client.close()


def negotiate(client, version):
    return client.call('server.version', 'plm-integration-test', version)


def spk_of(node, address):
    return node.call('getaddressinfo', address)['scriptPubKey']


@pytest.mark.parametrize('version', ['1.4', '1.4.2', '1.5', '1.6', '1.7'])
def test_protocol_versions(client, version):
    assert negotiate(client, version)[1] == version


def test_unsupported_protocol_versions(electrumx):
    for version in ('1.3', '2.0'):
        client = ElectrumClient(electrumx)
        try:
            with pytest.raises(RuntimeError, match='unsupported protocol version'):
                negotiate(client, version)
        finally:
            client.close()


def test_server_features(node, client):
    negotiate(client, '1.4')
    features = client.call('server.features')
    assert features['genesis_hash'] == node.call('getblockhash', 0)
    assert features['hash_function'] == 'sha256'
    assert features['protocol_min'] == '1.4'


def test_relayfee_and_estimatefee(client):
    negotiate(client, '1.5')  # blockchain.relayfee was removed in 1.6
    assert client.call('blockchain.relayfee') > 0
    assert isinstance(client.call('blockchain.estimatefee', 2), (int, float))


def test_mempool_info(client):
    '''The daemon reports 'incrementalfee' in getnetworkinfo, not 'incrementalrelayfee' in getmempoolinfo'''
    negotiate(client, '1.6')
    info = client.call('mempool.get_info')
    assert set(info) == {'mempoolminfee', 'minrelaytxfee', 'incrementalrelayfee'}
    assert all(fee > 0 for fee in info.values())


def test_daemon_limits_are_reported_as_errors(client):
    '''The Palladium daemon has no submitpackage and no gettxspendingprevout: clients get an error.
    If a newer daemon gains them, update CLAUDE.md, README.md and DEPLOY.md and this test.'''
    negotiate(client, '1.7')
    with pytest.raises(RuntimeError, match='does not support package submission'):
        client.call('blockchain.transaction.broadcast_package', ['00'])
    # a subscription would be accepted for an outpoint in the mempool and then close the session at the next block
    with pytest.raises(RuntimeError, match='outpoint subscriptions are not supported'):
        client.call('blockchain.outpoint.subscribe', '00' * 32, 0, '')


def test_protocol_1_7_scriptpubkey_methods(node, wallet, client):
    negotiate(client, '1.7')
    target = node.call('getnewaddress', '', 'bech32m', wallet=wallet)
    spk = spk_of(node, target)
    txid = node.call('sendtoaddress', target, 2, wallet=wallet)
    wait_for(lambda: client.call('blockchain.scriptpubkey.get_mempool', spk)['history'], 'the tx in the mempool')
    assert txid in [h['tx_hash'] for h in client.call('blockchain.scriptpubkey.get_history', spk)['history']]
    assert client.call('blockchain.scriptpubkey.get_balance', spk)['unconfirmed'] == 200_000_000
    assert client.call('blockchain.scriptpubkey.subscribe', spk) is not None
    # mempool.recent lists it
    assert txid in [t['txid'] for t in client.call('mempool.recent')]
    node.mine(1, new_address(node, wallet))
    wait_for(lambda: client.call('blockchain.scriptpubkey.get_balance', spk)['confirmed'] == 200_000_000, 'confirmation')
    unspent = client.call('blockchain.scriptpubkey.listunspent', spk)['utxos']
    assert [(u['tx_hash'], u['value']) for u in unspent] == [(txid, 200_000_000)]
    assert client.call('blockchain.scriptpubkey.unsubscribe', spk) is True


def test_testmempoolaccept(node, wallet, client):
    negotiate(client, '1.7')
    target = new_address(node, wallet)
    funded = node.call('fundrawtransaction', node.call('createrawtransaction', [], {target: 0.1}), wallet=wallet)
    signed = node.call('signrawtransactionwithwallet', funded['hex'], wallet=wallet)['hex']
    result, = client.call('blockchain.transaction.testmempoolaccept', [signed])
    assert result['allowed'] is True


def test_reorg_replaces_blocks(node, wallet, electrumx, client):
    '''The server follows a reorg: the block is replaced by a longer chain that confirms the tx again.
    Like upstream ElectrumX it reorganises when the daemon's chain gets longer than its own, not before.'''
    negotiate(client, '1.4.2')
    miner = new_address(node, wallet)
    target = node.call('getnewaddress', '', 'bech32', wallet=wallet)
    sh = scripthash(node, target)
    txid = node.call('sendtoaddress', target, 1, wallet=wallet)
    block = node.mine(1, miner)[0]
    height = node.call('getblockcount')
    wait_for(lambda: client.call('blockchain.scripthash.get_balance', sh)['confirmed'] == 100_000_000, 'confirmation')
    old_header = client.call('blockchain.block.header', height)

    node.call('invalidateblock', block)
    new_blocks = node.mine(2, new_address(node, wallet))  # other coinbase, else the same block; the tx goes back to the mempool and is mined again in the first one
    assert block not in new_blocks
    wait_for(lambda: client.call('blockchain.headers.subscribe')['height'] == height + 1, 'the server to follow the new chain')
    assert client.call('blockchain.block.header', height) != old_header
    wait_for(lambda: client.call('blockchain.scripthash.get_balance', sh) == {'confirmed': 100_000_000, 'unconfirmed': 0},
             'the balance after the reorg')
    history = client.call('blockchain.scripthash.get_history', sh)
    assert [(h['tx_hash'], h['height']) for h in history] == [(txid, height)]


def test_ssl_service(node, electrumx):
    '''The SSL port serves the same protocol, with the certificate given by SSL_CERTFILE/SSL_KEYFILE'''
    context = ssl.create_default_context(cafile=SSL_SERVICE['certfile'])
    client = ElectrumClient(SSL_SERVICE['port'], context)
    try:
        assert negotiate(client, '1.4')[1] == '1.4'
        assert client.call('blockchain.block.header', 0)
    finally:
        client.close()
    # a client that does not trust the (self-signed) certificate refuses it
    with pytest.raises(ssl.SSLCertVerificationError):
        ElectrumClient(SSL_SERVICE['port'], ssl.create_default_context())