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