ttest_network.py - electrum - Electrum Bitcoin wallet
HTML git clone https://git.parazyd.org/electrum
DIR Log
DIR Files
DIR Refs
DIR Submodules
---
ttest_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()