test_network.py (6873B)
1 import asyncio 2 import tempfile 3 import unittest 4 5 from electrum import constants 6 from electrum.simple_config import SimpleConfig 7 from electrum import blockchain 8 from electrum.interface import Interface, ServerAddr 9 from electrum.crypto import sha256 10 from electrum.util import bh2u 11 12 from . import ElectrumTestCase 13 14 15 class MockTaskGroup: 16 async def spawn(self, x): return 17 18 class MockNetwork: 19 taskgroup = MockTaskGroup() 20 asyncio_loop = asyncio.get_event_loop() 21 22 class MockInterface(Interface): 23 def __init__(self, config): 24 self.config = config 25 network = MockNetwork() 26 network.config = config 27 super().__init__(network=network, server=ServerAddr.from_str('mock-server:50000:t'), proxy=None) 28 self.q = asyncio.Queue() 29 self.blockchain = blockchain.Blockchain(config=self.config, forkpoint=0, 30 parent=None, forkpoint_hash=constants.net.GENESIS, prev_hash=None) 31 self.tip = 12 32 self.blockchain._size = self.tip + 1 33 async def get_block_header(self, height, assert_mode): 34 assert self.q.qsize() > 0, (height, assert_mode) 35 item = await self.q.get() 36 print("step with height", height, item) 37 assert item['block_height'] == height, (item['block_height'], height) 38 assert assert_mode in item['mock'], (assert_mode, item) 39 return item 40 41 class TestNetwork(ElectrumTestCase): 42 43 @classmethod 44 def setUpClass(cls): 45 super().setUpClass() 46 constants.set_regtest() 47 48 @classmethod 49 def tearDownClass(cls): 50 super().tearDownClass() 51 constants.set_mainnet() 52 53 def setUp(self): 54 super().setUp() 55 self.config = SimpleConfig({'electrum_path': self.electrum_path}) 56 self.interface = MockInterface(self.config) 57 58 def test_fork_noconflict(self): 59 blockchain.blockchains = {} 60 self.interface.q.put_nowait({'block_height': 8, 'mock': {'catchup':1, 'check': lambda x: False, 'connect': lambda x: False}}) 61 def mock_connect(height): 62 return height == 6 63 self.interface.q.put_nowait({'block_height': 7, 'mock': {'backward':1,'check': lambda x: False, 'connect': mock_connect, 'fork': self.mock_fork}}) 64 self.interface.q.put_nowait({'block_height': 2, 'mock': {'backward':1,'check':lambda x: True, 'connect': lambda x: False}}) 65 self.interface.q.put_nowait({'block_height': 4, 'mock': {'binary':1,'check':lambda x: True, 'connect': lambda x: True}}) 66 self.interface.q.put_nowait({'block_height': 5, 'mock': {'binary':1,'check':lambda x: True, 'connect': lambda x: True}}) 67 self.interface.q.put_nowait({'block_height': 6, 'mock': {'binary':1,'check':lambda x: True, 'connect': lambda x: True}}) 68 ifa = self.interface 69 self.assertEqual(('fork', 8), asyncio.get_event_loop().run_until_complete(ifa.sync_until(8, next_height=7))) 70 self.assertEqual(self.interface.q.qsize(), 0) 71 72 def test_fork_conflict(self): 73 blockchain.blockchains = {7: {'check': lambda bad_header: False}} 74 self.interface.q.put_nowait({'block_height': 8, 'mock': {'catchup':1, 'check': lambda x: False, 'connect': lambda x: False}}) 75 def mock_connect(height): 76 return height == 6 77 self.interface.q.put_nowait({'block_height': 7, 'mock': {'backward':1,'check': lambda x: False, 'connect': mock_connect, 'fork': self.mock_fork}}) 78 self.interface.q.put_nowait({'block_height': 2, 'mock': {'backward':1,'check':lambda x: True, 'connect': lambda x: False}}) 79 self.interface.q.put_nowait({'block_height': 4, 'mock': {'binary':1,'check':lambda x: True, 'connect': lambda x: True}}) 80 self.interface.q.put_nowait({'block_height': 5, 'mock': {'binary':1,'check':lambda x: True, 'connect': lambda x: True}}) 81 self.interface.q.put_nowait({'block_height': 6, 'mock': {'binary':1,'check':lambda x: True, 'connect': lambda x: True}}) 82 ifa = self.interface 83 self.assertEqual(('fork', 8), asyncio.get_event_loop().run_until_complete(ifa.sync_until(8, next_height=7))) 84 self.assertEqual(self.interface.q.qsize(), 0) 85 86 def test_can_connect_during_backward(self): 87 blockchain.blockchains = {} 88 self.interface.q.put_nowait({'block_height': 8, 'mock': {'catchup':1, 'check': lambda x: False, 'connect': lambda x: False}}) 89 def mock_connect(height): 90 return height == 2 91 self.interface.q.put_nowait({'block_height': 7, 'mock': {'backward':1, 'check': lambda x: False, 'connect': mock_connect, 'fork': self.mock_fork}}) 92 self.interface.q.put_nowait({'block_height': 2, 'mock': {'backward':1, 'check': lambda x: False, 'connect': mock_connect, 'fork': self.mock_fork}}) 93 self.interface.q.put_nowait({'block_height': 3, 'mock': {'catchup':1, 'check': lambda x: False, 'connect': lambda x: True}}) 94 self.interface.q.put_nowait({'block_height': 4, 'mock': {'catchup':1, 'check': lambda x: False, 'connect': lambda x: True}}) 95 ifa = self.interface 96 self.assertEqual(('catchup', 5), asyncio.get_event_loop().run_until_complete(ifa.sync_until(8, next_height=4))) 97 self.assertEqual(self.interface.q.qsize(), 0) 98 99 def mock_fork(self, bad_header): 100 forkpoint = bad_header['block_height'] 101 b = blockchain.Blockchain(config=self.config, forkpoint=forkpoint, parent=None, 102 forkpoint_hash=bh2u(sha256(str(forkpoint))), prev_hash=bh2u(sha256(str(forkpoint-1)))) 103 return b 104 105 def test_chain_false_during_binary(self): 106 blockchain.blockchains = {} 107 self.interface.q.put_nowait({'block_height': 8, 'mock': {'catchup':1, 'check': lambda x: False, 'connect': lambda x: False}}) 108 mock_connect = lambda height: height == 3 109 self.interface.q.put_nowait({'block_height': 7, 'mock': {'backward':1, 'check': lambda x: False, 'connect': mock_connect}}) 110 self.interface.q.put_nowait({'block_height': 2, 'mock': {'backward':1, 'check': lambda x: True, 'connect': mock_connect}}) 111 self.interface.q.put_nowait({'block_height': 4, 'mock': {'binary':1, 'check': lambda x: False, 'fork': self.mock_fork, 'connect': mock_connect}}) 112 self.interface.q.put_nowait({'block_height': 3, 'mock': {'binary':1, 'check': lambda x: True, 'connect': lambda x: True}}) 113 self.interface.q.put_nowait({'block_height': 5, 'mock': {'catchup':1, 'check': lambda x: False, 'connect': lambda x: True}}) 114 self.interface.q.put_nowait({'block_height': 6, 'mock': {'catchup':1, 'check': lambda x: False, 'connect': lambda x: True}}) 115 ifa = self.interface 116 self.assertEqual(('catchup', 7), asyncio.get_event_loop().run_until_complete(ifa.sync_until(8, next_height=6))) 117 self.assertEqual(self.interface.q.qsize(), 0) 118 119 120 if __name__=="__main__": 121 constants.set_regtest() 122 unittest.main()