davide/purple-electrumwallet
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
af1997438103f3a09a0c1ccb3c681d5cfa76520e
purple-electrumwallet/tests/test_bitcoinpurple.py
T
davide af19974381 fix: restore BIP44_COIN_TYPE=13496 for BitcoinPurple and fix LN stresstest race 
BIP44_COIN_TYPE was accidentally set to 0 (Bitcoin mainnet), which would
cause BTCP wallets to derive keys identical to Bitcoin mainnet wallets from
the same seed. Restored to 13496 (provisional private constant matching the
BTCP P2P port) per tecnichal-data.md spec; updated table entry and test.

Also fixes a race condition in TestPeerDirectAnchors::test_payments_stresstest:
gath.cancel() was called immediately after OldTaskGroup exited, without any
await, so the event loop never ran the message-loop tasks to drain the final
revoke_and_ack for the last batch of 5 concurrent HTLCs. Added
asyncio.sleep(0) to yield one event-loop iteration before cancelling.
2026-05-05 19:26:26 +02:00

486 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Tests for BitcoinPurple (BTCP) network support.
Covers:
- Network constants against the technical specification
- 120-block difficulty adjustment logic in blockchain.py
- Address encoding under the BTCP network parameters
"""
import os
from unittest.mock import patch
from electrum import bitcoin, blockchain, constants, segwit_addr
from electrum.bitcoin import (
address_to_script,
is_address,
is_b58_address,
is_segwit_address,
public_key_to_p2pkh,
serialize_privkey,
deserialize_privkey,
)
from electrum.blockchain import Blockchain, InvalidHeader
from electrum.constants import BitcoinPurple, BitcoinPurpleTestnet
from electrum.simple_config import SimpleConfig
from electrum.util import make_dir
from . import ElectrumTestCase
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
class TestBitcoinPurpleConstants(ElectrumTestCase):
"""Verify all BTCP network constants against the technical specification."""
# --- identity ---
def test_net_names(self):
self.assertEqual("bitcoinpurple", BitcoinPurple.NET_NAME)
self.assertEqual("bitcoinpurple_testnet", BitcoinPurpleTestnet.NET_NAME)
def test_testnet_flags(self):
self.assertFalse(BitcoinPurple.TESTNET)
self.assertTrue(BitcoinPurpleTestnet.TESTNET)
def test_cli_flags_and_datadir(self):
self.assertEqual("bitcoinpurple", BitcoinPurple.cli_flag())
self.assertEqual("bitcoinpurple", BitcoinPurple.datadir_subdir())
self.assertEqual("bitcoinpurple_testnet", BitcoinPurpleTestnet.cli_flag())
self.assertEqual("bitcoinpurple_testnet", BitcoinPurpleTestnet.datadir_subdir())
# --- address encoding ---
def test_address_prefixes_mainnet(self):
self.assertEqual(56, BitcoinPurple.ADDRTYPE_P2PKH) # 0x38
self.assertEqual(55, BitcoinPurple.ADDRTYPE_P2SH) # 0x37
self.assertEqual(0xb7, BitcoinPurple.WIF_PREFIX) # 183
def test_address_prefixes_testnet(self):
# BTCP testnet keeps the same prefixes as mainnet
self.assertEqual(56, BitcoinPurpleTestnet.ADDRTYPE_P2PKH)
self.assertEqual(55, BitcoinPurpleTestnet.ADDRTYPE_P2SH)
self.assertEqual(0xb7, BitcoinPurpleTestnet.WIF_PREFIX)
def test_segwit_hrp(self):
self.assertEqual("btcp", BitcoinPurple.SEGWIT_HRP)
self.assertEqual("tbtcp", BitcoinPurpleTestnet.SEGWIT_HRP)
def test_bolt11_hrp(self):
self.assertEqual("btcp", BitcoinPurple.BOLT11_HRP)
self.assertEqual("tbtcp", BitcoinPurpleTestnet.BOLT11_HRP)
# --- genesis ---
def test_genesis_mainnet(self):
self.assertEqual(
"000003823fbf82ea4906cbe214617ce7a70a5da29c19ecb1d65618bcf04ec015",
BitcoinPurple.GENESIS,
)
def test_genesis_testnet(self):
self.assertEqual(
"000002fdc3921c1ad368816fcc587f499698d42b42ab5a5d94ee67882ef9d998",
BitcoinPurpleTestnet.GENESIS,
)
def test_rev_genesis_bytes_mainnet(self):
# Wire-order (reversed) bytes used in LN chain_hash
expected_wire = "15c04ef0bc1856d6b1ec199ca25d0aa7e77c6114e2cb0649ea82bf3f82030000"
self.assertEqual(expected_wire, BitcoinPurple.rev_genesis_bytes().hex())
def test_rev_genesis_bytes_testnet(self):
expected_wire = "98d9f92e8867ee945d5aab422bd49896497f58cc6f8168d31a1c92c3fd020000"
self.assertEqual(expected_wire, BitcoinPurpleTestnet.rev_genesis_bytes().hex())
# --- ports ---
def test_default_ports_mainnet(self):
self.assertEqual({'t': '50001', 's': '50002'}, BitcoinPurple.DEFAULT_PORTS)
def test_default_ports_testnet(self):
self.assertEqual({'t': '60001', 's': '60002'}, BitcoinPurpleTestnet.DEFAULT_PORTS)
# --- PoW constants ---
def test_pow_adjustment_interval(self):
self.assertEqual(120, BitcoinPurple.DIFFICULTY_ADJUSTMENT_INTERVAL)
# testnet inherits the same value
self.assertEqual(120, BitcoinPurpleTestnet.DIFFICULTY_ADJUSTMENT_INTERVAL)
def test_pow_target_timespan(self):
self.assertEqual(7200, BitcoinPurple.POW_TARGET_TIMESPAN) # 120 * 60 s
def test_pow_max_adjustment_factor(self):
self.assertEqual(4, BitcoinPurple.MAX_ADJUSTMENT_FACTOR)
def test_pow_max_target_exceeds_bitcoin(self):
# BTCP starts with easier proof-of-work than Bitcoin
self.assertGreater(BitcoinPurple.MAX_TARGET, constants.BitcoinMainnet.MAX_TARGET)
def test_pow_max_target_value(self):
# compact 0x1e0ffff0 maps to a target just under BTCP MAX_TARGET
genesis_bits = 0x1e0ffff0
genesis_target = Blockchain.bits_to_target(genesis_bits)
self.assertLessEqual(genesis_target, BitcoinPurple.MAX_TARGET)
# --- HD key headers ---
def test_xpub_headers_mainnet_match_bitcoin(self):
# BTCP mainnet reuses Bitcoin's BIP32 root bytes (0488B21E / 0488ADE4)
for script_type in ('standard', 'p2wpkh-p2sh', 'p2wpkh', 'p2wsh-p2sh', 'p2wsh'):
with self.subTest(script_type=script_type):
self.assertEqual(
constants.BitcoinMainnet.XPUB_HEADERS[script_type],
BitcoinPurple.XPUB_HEADERS[script_type],
)
self.assertEqual(
constants.BitcoinMainnet.XPRV_HEADERS[script_type],
BitcoinPurple.XPRV_HEADERS[script_type],
)
def test_xpub_headers_testnet_match_bitcoin_testnet(self):
for script_type in ('standard', 'p2wpkh-p2sh', 'p2wpkh', 'p2wsh-p2sh', 'p2wsh'):
with self.subTest(script_type=script_type):
self.assertEqual(
constants.BitcoinTestnet.XPUB_HEADERS[script_type],
BitcoinPurpleTestnet.XPUB_HEADERS[script_type],
)
self.assertEqual(
constants.BitcoinTestnet.XPRV_HEADERS[script_type],
BitcoinPurpleTestnet.XPRV_HEADERS[script_type],
)
def test_xpub_inv_headers_are_inverses(self):
for hdr, hdr_inv in (
(BitcoinPurple.XPUB_HEADERS, BitcoinPurple.XPUB_HEADERS_INV),
(BitcoinPurple.XPRV_HEADERS, BitcoinPurple.XPRV_HEADERS_INV),
):
for k, v in hdr.items():
self.assertEqual(k, hdr_inv[v])
# --- Lightning ---
def test_ln_realm_byte(self):
self.assertEqual(0, BitcoinPurple.LN_REALM_BYTE)
self.assertEqual(1, BitcoinPurpleTestnet.LN_REALM_BYTE)
def test_ln_dns_seeds_empty(self):
self.assertEqual([], BitcoinPurple.LN_DNS_SEEDS)
self.assertEqual([], BitcoinPurpleTestnet.LN_DNS_SEEDS)
def test_bip44_coin_type(self):
self.assertEqual(13496, BitcoinPurple.BIP44_COIN_TYPE)
self.assertEqual(1, BitcoinPurpleTestnet.BIP44_COIN_TYPE)
# --- NETS_LIST integrity ---
def test_nets_list_contains_btcp(self):
net_names = [c.NET_NAME for c in constants.NETS_LIST]
self.assertIn("bitcoinpurple", net_names)
self.assertIn("bitcoinpurple_testnet", net_names)
def test_nets_list_unique(self):
net_names = [c.NET_NAME for c in constants.NETS_LIST]
self.assertEqual(len(net_names), len(set(net_names)))
# --- inheritance: BitcoinPurple must not inherit from any Bitcoin class ---
def test_btcp_independent_from_bitcoin_mainnet(self):
self.assertFalse(issubclass(BitcoinPurple, constants.BitcoinMainnet))
def test_btcp_independent_from_bitcoin_testnet(self):
self.assertFalse(issubclass(BitcoinPurple, constants.BitcoinTestnet))
def test_btcp_testnet_inherits_from_btcp(self):
self.assertTrue(issubclass(BitcoinPurpleTestnet, BitcoinPurple))
# ---------------------------------------------------------------------------
# Difficulty adjustment (120-block retarget)
# ---------------------------------------------------------------------------
class TestBitcoinPurpleDifficultyAdjustment(ElectrumTestCase):
"""
Test the 120-block BTCP difficulty retarget logic in blockchain.get_target().
Uses BitcoinPurple (mainnet) so that get_target() executes the real retarget
formula. (Testnet always returns 0, which prevents testing the formula.)
For can_connect() tests we either rely on the ordering of checks inside
can_connect() or patch the parts that are not under test.
"""
GENESIS_BITS = 0x1e0ffff0 # nBits from BTCP genesis block
@classmethod
def setUpClass(cls):
super().setUpClass()
constants.BitcoinPurple.set_as_network()
@classmethod
def tearDownClass(cls):
super().tearDownClass()
constants.BitcoinMainnet.set_as_network()
def setUp(self):
super().setUp()
make_dir(os.path.join(self.electrum_path, 'forks'))
self.config = SimpleConfig({'electrum_path': self.electrum_path})
blockchain.blockchains = {}
def _make_chain(self) -> Blockchain:
chain = Blockchain(
config=self.config, forkpoint=0, parent=None,
forkpoint_hash=constants.net.GENESIS, prev_hash=None)
open(chain.path(), 'w+').close()
blockchain.blockchains[constants.net.GENESIS] = chain
return chain
def _fake_headers(self, first_ts: int, last_ts: int, bits: int = GENESIS_BITS):
"""Return a read_header side_effect that yields controlled timestamps/bits."""
adj = constants.net.DIFFICULTY_ADJUSTMENT_INTERVAL
def read_header(height: int):
if height % adj == 0:
return {'bits': bits, 'timestamp': first_ts}
if height % adj == adj - 1:
return {'bits': bits, 'timestamp': last_ts}
return {'bits': bits, 'timestamp': first_ts + (last_ts - first_ts) * (height % adj) // (adj - 1)}
return read_header
# --- index -1 ---
def test_get_target_genesis_index_returns_genesis_target(self):
# Index -1 must return the genesis-era target (derived from POW_GENESIS_BITS
# 0x1e0ffff0), which is slightly harder than MAX_TARGET (0x1e0fffff).
chain = self._make_chain()
got = chain.get_target(-1)
expected = Blockchain.bits_to_target(BitcoinPurple.POW_GENESIS_BITS)
self.assertEqual(expected, got)
# Genesis target must be at or below powLimit
self.assertLessEqual(got, BitcoinPurple.MAX_TARGET)
# Differs from Bitcoin mainnet's MAX_TARGET
self.assertNotEqual(constants.BitcoinMainnet.MAX_TARGET, got)
# --- retarget formula ---
def test_get_target_on_time(self):
"""actual == target_timespan → target unchanged."""
ts = constants.net.POW_TARGET_TIMESPAN # 7200
initial_target = Blockchain.bits_to_target(self.GENESIS_BITS)
chain = self._make_chain()
with patch.object(chain, 'read_header', side_effect=self._fake_headers(0, ts)):
new_target = chain.get_target(0)
expected = Blockchain.bits_to_target(Blockchain.target_to_bits(initial_target))
self.assertEqual(expected, new_target)
def test_get_target_fast_blocks_increases_difficulty(self):
"""Blocks mined twice as fast → target halves (difficulty doubles)."""
ts = constants.net.POW_TARGET_TIMESPAN # 7200
actual = ts // 2 # 3600, above the 1800-floor clamp
initial_target = Blockchain.bits_to_target(self.GENESIS_BITS)
expected_target = Blockchain.bits_to_target(
Blockchain.target_to_bits(initial_target * actual // ts)
)
chain = self._make_chain()
with patch.object(chain, 'read_header', side_effect=self._fake_headers(0, actual)):
new_target = chain.get_target(0)
self.assertEqual(expected_target, new_target)
self.assertLess(new_target, initial_target) # harder
def test_get_target_slow_blocks_decreases_difficulty(self):
"""Blocks mined twice as slow → target doubles (difficulty halves)."""
ts = constants.net.POW_TARGET_TIMESPAN # 7200
actual = ts * 2 # 14400, below the 28800-ceiling clamp
initial_target = Blockchain.bits_to_target(self.GENESIS_BITS)
expected_target = Blockchain.bits_to_target(
Blockchain.target_to_bits(min(BitcoinPurple.MAX_TARGET, initial_target * actual // ts))
)
chain = self._make_chain()
with patch.object(chain, 'read_header', side_effect=self._fake_headers(0, actual)):
new_target = chain.get_target(0)
self.assertEqual(expected_target, new_target)
self.assertGreater(new_target, initial_target) # easier
def test_get_target_clamp_lower(self):
"""actual < target/4 → clamped to target/4 (max 4× harder per retarget)."""
ts = constants.net.POW_TARGET_TIMESPAN # 7200
floor = ts // constants.net.MAX_ADJUSTMENT_FACTOR # 1800
# Use actual timespan far below the floor
actual = 100
initial_target = Blockchain.bits_to_target(self.GENESIS_BITS)
expected_target = Blockchain.bits_to_target(
Blockchain.target_to_bits(initial_target * floor // ts)
)
chain = self._make_chain()
with patch.object(chain, 'read_header', side_effect=self._fake_headers(0, actual)):
new_target = chain.get_target(0)
self.assertEqual(expected_target, new_target)
def test_get_target_clamp_upper(self):
"""actual > target*4 → clamped to target*4 (max 4× easier per retarget)."""
ts = constants.net.POW_TARGET_TIMESPAN # 7200
ceiling = ts * constants.net.MAX_ADJUSTMENT_FACTOR # 28800
# Use actual timespan far above the ceiling
actual = 999_999
initial_target = Blockchain.bits_to_target(self.GENESIS_BITS)
raw = initial_target * ceiling // ts
expected_target = Blockchain.bits_to_target(
Blockchain.target_to_bits(min(BitcoinPurple.MAX_TARGET, raw))
)
chain = self._make_chain()
with patch.object(chain, 'read_header', side_effect=self._fake_headers(0, actual)):
new_target = chain.get_target(0)
self.assertEqual(expected_target, new_target)
# --- period index computation ---
def test_adj_interval_is_120_not_2016(self):
"""adj_interval from constants must equal 120 when BTCP network is active."""
self.assertEqual(120, constants.net.DIFFICULTY_ADJUSTMENT_INTERVAL)
def test_get_target_uses_120_block_window(self):
"""get_target(1) reads headers at heights 120 and 239, not 2016 and 4031."""
ts = constants.net.POW_TARGET_TIMESPAN
read_calls = []
def tracking_read_header(height):
read_calls.append(height)
return {'bits': self.GENESIS_BITS, 'timestamp': height * 60}
chain = self._make_chain()
with patch.object(chain, 'read_header', side_effect=tracking_read_header):
chain.get_target(1)
# Must have read exactly headers 120 and 239 (period 1 = [120, 239])
self.assertIn(120, read_calls)
self.assertIn(239, read_calls)
# Must NOT have touched 2016 or 4031 (Bitcoin-style chunk boundary)
self.assertNotIn(2016, read_calls)
self.assertNotIn(4031, read_calls)
def test_can_connect_uses_120_block_period(self):
"""
can_connect should look up target at height // 120 - 1, not height // 2016 - 1.
Verify by checking that get_target is called with the correct period index.
"""
chain = self._make_chain()
get_target_calls = []
original_get_target = chain.get_target
def tracking_get_target(index):
get_target_calls.append(index)
return original_get_target(index)
# Height 1 (period 0): can_connect calls get_target(1 // 120 - 1) = get_target(-1)
# before verify_header. Even though verify_header rejects the PoW (nonce=0),
# get_target is called first so the index is recorded.
dummy_header = {
'block_height': 1,
'prev_block_hash': constants.net.GENESIS,
'version': 1,
'merkle_root': '00' * 32,
'timestamp': 1691126892,
'bits': self.GENESIS_BITS,
'nonce': 0,
}
with patch.object(chain, 'get_target', side_effect=tracking_get_target):
chain.can_connect(dummy_header, check_height=False)
self.assertIn(-1, get_target_calls)
# Height 121 (first block of period 1): get_target(121 // 120 - 1) = get_target(0).
# get_hash(120) would raise MissingHeader (header not on disk) before get_target
# is reached, so we patch get_hash to return a fake prev hash.
get_target_calls.clear()
fake_prev = 'ab' * 32
dummy_header['block_height'] = 121
dummy_header['prev_block_hash'] = fake_prev
with patch.object(chain, 'get_hash', return_value=fake_prev):
with patch.object(chain, 'get_target', side_effect=tracking_get_target):
chain.can_connect(dummy_header, check_height=False)
self.assertIn(0, get_target_calls)
# ---------------------------------------------------------------------------
# Address encoding under BTCP network parameters
# ---------------------------------------------------------------------------
class TestBitcoinPurpleAddress(ElectrumTestCase):
"""P2PKH, P2SH, Bech32, and WIF encoding under BitcoinPurple network."""
# compressed pubkey for a known private key (k=1, secp256k1)
PUBKEY_HEX = "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798"
@classmethod
def setUpClass(cls):
super().setUpClass()
constants.BitcoinPurple.set_as_network()
@classmethod
def tearDownClass(cls):
super().tearDownClass()
constants.BitcoinMainnet.set_as_network()
def test_p2pkh_address_starts_with_P(self):
# P2PKH prefix 56 (0x38) encodes to Base58 addresses starting with 'P'
pubkey = bytes.fromhex(self.PUBKEY_HEX)
addr = public_key_to_p2pkh(pubkey)
self.assertTrue(addr.startswith('P'), f"Expected 'P...' address, got: {addr}")
def test_p2pkh_address_is_valid(self):
pubkey = bytes.fromhex(self.PUBKEY_HEX)
addr = public_key_to_p2pkh(pubkey)
self.assertTrue(is_address(addr))
self.assertTrue(is_b58_address(addr))
def test_p2pkh_not_valid_on_bitcoin_mainnet(self):
# BTCP address must be rejected on Bitcoin mainnet (different prefix)
pubkey = bytes.fromhex(self.PUBKEY_HEX)
addr = public_key_to_p2pkh(pubkey)
self.assertFalse(is_address(addr, net=constants.BitcoinMainnet))
def test_bech32_hrp_is_btcp(self):
# A native SegWit witness-v0 address encoded with HRP 'btcp' must be
# accepted as valid by the BTCP network and rejected by Bitcoin mainnet.
witness_program = bytes(20) # all-zero 20-byte hash (P2WPKH)
addr = segwit_addr.encode_segwit_address(BitcoinPurple.SEGWIT_HRP, 0, witness_program)
self.assertIsNotNone(addr)
self.assertTrue(is_segwit_address(addr))
self.assertFalse(is_segwit_address(addr, net=constants.BitcoinMainnet))
def test_bech32_address_to_script(self):
# address_to_script must produce a valid P2WPKH scriptPubKey for a BTCP bech32 addr
witness_program = bytes(20)
addr = segwit_addr.encode_segwit_address(BitcoinPurple.SEGWIT_HRP, 0, witness_program)
script = address_to_script(addr)
# P2WPKH scriptPubKey: OP_0 <20-byte-hash> = 0x0014 + 20 zero bytes
self.assertEqual('0014' + '00' * 20, script.hex())
def test_wif_roundtrip(self):
# serialize_privkey / deserialize_privkey must round-trip under BTCP prefix 0xb7
raw_key = bytes(31) + bytes([1]) # 32-byte privkey with value 1
wif = serialize_privkey(raw_key, True, 'p2pkh')
txin_type, got_key, compressed = deserialize_privkey(wif)
self.assertEqual(raw_key, got_key)
self.assertTrue(compressed)
def test_bitcoin_address_not_valid_on_btcp(self):
# A Bitcoin mainnet P2PKH address ('1...') must not pass is_address on BTCP
btc_addr = "1A1zP1eP5QGefi2DMPTfTL5SLmv7Divf" # Bitcoin genesis coinbase
self.assertFalse(is_address(btc_addr))
def test_bitcoin_bech32_not_valid_on_btcp(self):
# A Bitcoin mainnet bech32 address (HRP 'bc') must be rejected on BTCP
btc_bech32 = "bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4"
self.assertFalse(is_segwit_address(btc_bech32))
Reference in New Issue View Git Blame Copy Permalink