ttest_wallet.py - electrum - Electrum Bitcoin wallet
HTML git clone https://git.parazyd.org/electrum
DIR Log
DIR Files
DIR Refs
DIR Submodules
---
ttest_wallet.py (17462B)
---
1 import shutil
2 import tempfile
3 import sys
4 import os
5 import json
6 from decimal import Decimal
7 import time
8 from io import StringIO
9 import asyncio
10
11 from electrum.storage import WalletStorage
12 from electrum.wallet_db import FINAL_SEED_VERSION
13 from electrum.wallet import (Abstract_Wallet, Standard_Wallet, create_new_wallet,
14 restore_wallet_from_text, Imported_Wallet, Wallet)
15 from electrum.exchange_rate import ExchangeBase, FxThread
16 from electrum.util import TxMinedInfo, InvalidPassword
17 from electrum.bitcoin import COIN
18 from electrum.wallet_db import WalletDB
19 from electrum.simple_config import SimpleConfig
20 from electrum import util
21
22 from . import ElectrumTestCase
23
24
25 class FakeSynchronizer(object):
26
27 def __init__(self):
28 self.store = []
29
30 def add(self, address):
31 self.store.append(address)
32
33
34 class WalletTestCase(ElectrumTestCase):
35
36 def setUp(self):
37 super(WalletTestCase, self).setUp()
38 self.user_dir = tempfile.mkdtemp()
39 self.config = SimpleConfig({'electrum_path': self.user_dir})
40
41 self.wallet_path = os.path.join(self.user_dir, "somewallet")
42
43 self._saved_stdout = sys.stdout
44 self._stdout_buffer = StringIO()
45 sys.stdout = self._stdout_buffer
46
47 def tearDown(self):
48 super(WalletTestCase, self).tearDown()
49 shutil.rmtree(self.user_dir)
50 # Restore the "real" stdout
51 sys.stdout = self._saved_stdout
52
53
54 class TestWalletStorage(WalletTestCase):
55
56 def test_read_dictionary_from_file(self):
57
58 some_dict = {"a":"b", "c":"d"}
59 contents = json.dumps(some_dict)
60 with open(self.wallet_path, "w") as f:
61 contents = f.write(contents)
62
63 storage = WalletStorage(self.wallet_path)
64 db = WalletDB(storage.read(), manual_upgrades=True)
65 self.assertEqual("b", db.get("a"))
66 self.assertEqual("d", db.get("c"))
67
68 def test_write_dictionary_to_file(self):
69
70 storage = WalletStorage(self.wallet_path)
71 db = WalletDB('', manual_upgrades=True)
72
73 some_dict = {
74 u"a": u"b",
75 u"c": u"d",
76 u"seed_version": FINAL_SEED_VERSION}
77
78 for key, value in some_dict.items():
79 db.put(key, value)
80 db.write(storage)
81
82 with open(self.wallet_path, "r") as f:
83 contents = f.read()
84 d = json.loads(contents)
85 for key, value in some_dict.items():
86 self.assertEqual(d[key], value)
87
88 class FakeExchange(ExchangeBase):
89 def __init__(self, rate):
90 super().__init__(lambda self: None, lambda self: None)
91 self.quotes = {'TEST': rate}
92
93 class FakeFxThread:
94 def __init__(self, exchange):
95 self.exchange = exchange
96 self.ccy = 'TEST'
97
98 remove_thousands_separator = staticmethod(FxThread.remove_thousands_separator)
99 timestamp_rate = FxThread.timestamp_rate
100 ccy_amount_str = FxThread.ccy_amount_str
101 history_rate = FxThread.history_rate
102
103 class FakeWallet:
104 def __init__(self, fiat_value):
105 super().__init__()
106 self.fiat_value = fiat_value
107 self.db = WalletDB("{}", manual_upgrades=True)
108 self.db.transactions = self.db.verified_tx = {'abc':'Tx'}
109
110 def get_tx_height(self, txid):
111 # because we use a current timestamp, and history is empty,
112 # FxThread.history_rate will use spot prices
113 return TxMinedInfo(height=10, conf=10, timestamp=int(time.time()), header_hash='def')
114
115 default_fiat_value = Abstract_Wallet.default_fiat_value
116 price_at_timestamp = Abstract_Wallet.price_at_timestamp
117 class storage:
118 put = lambda self, x: None
119
120 txid = 'abc'
121 ccy = 'TEST'
122
123 class TestFiat(ElectrumTestCase):
124 def setUp(self):
125 super().setUp()
126 self.value_sat = COIN
127 self.fiat_value = {}
128 self.wallet = FakeWallet(fiat_value=self.fiat_value)
129 self.fx = FakeFxThread(FakeExchange(Decimal('1000.001')))
130 default_fiat = Abstract_Wallet.default_fiat_value(self.wallet, txid, self.fx, self.value_sat)
131 self.assertEqual(Decimal('1000.001'), default_fiat)
132 self.assertEqual('1,000.00', self.fx.ccy_amount_str(default_fiat, commas=True))
133
134 def test_save_fiat_and_reset(self):
135 self.assertEqual(False, Abstract_Wallet.set_fiat_value(self.wallet, txid, ccy, '1000.01', self.fx, self.value_sat))
136 saved = self.fiat_value[ccy][txid]
137 self.assertEqual('1,000.01', self.fx.ccy_amount_str(Decimal(saved), commas=True))
138 self.assertEqual(True, Abstract_Wallet.set_fiat_value(self.wallet, txid, ccy, '', self.fx, self.value_sat))
139 self.assertNotIn(txid, self.fiat_value[ccy])
140 # even though we are not setting it to the exact fiat value according to the exchange rate, precision is truncated away
141 self.assertEqual(True, Abstract_Wallet.set_fiat_value(self.wallet, txid, ccy, '1,000.002', self.fx, self.value_sat))
142
143 def test_too_high_precision_value_resets_with_no_saved_value(self):
144 self.assertEqual(True, Abstract_Wallet.set_fiat_value(self.wallet, txid, ccy, '1,000.001', self.fx, self.value_sat))
145
146 def test_empty_resets(self):
147 self.assertEqual(True, Abstract_Wallet.set_fiat_value(self.wallet, txid, ccy, '', self.fx, self.value_sat))
148 self.assertNotIn(ccy, self.fiat_value)
149
150 def test_save_garbage(self):
151 self.assertEqual(False, Abstract_Wallet.set_fiat_value(self.wallet, txid, ccy, 'garbage', self.fx, self.value_sat))
152 self.assertNotIn(ccy, self.fiat_value)
153
154
155 class TestCreateRestoreWallet(WalletTestCase):
156
157 def test_create_new_wallet(self):
158 passphrase = 'mypassphrase'
159 password = 'mypassword'
160 encrypt_file = True
161 d = create_new_wallet(path=self.wallet_path,
162 passphrase=passphrase,
163 password=password,
164 encrypt_file=encrypt_file,
165 gap_limit=1,
166 config=self.config)
167 wallet = d['wallet'] # type: Standard_Wallet
168
169 # lightning initialization
170 self.assertTrue(wallet.db.get('lightning_privkey2').startswith('xprv'))
171
172 wallet.check_password(password)
173 self.assertEqual(passphrase, wallet.keystore.get_passphrase(password))
174 self.assertEqual(d['seed'], wallet.keystore.get_seed(password))
175 self.assertEqual(encrypt_file, wallet.storage.is_encrypted())
176
177 def test_restore_wallet_from_text_mnemonic(self):
178 text = 'bitter grass shiver impose acquire brush forget axis eager alone wine silver'
179 passphrase = 'mypassphrase'
180 password = 'mypassword'
181 encrypt_file = True
182 d = restore_wallet_from_text(text,
183 path=self.wallet_path,
184 passphrase=passphrase,
185 password=password,
186 encrypt_file=encrypt_file,
187 gap_limit=1,
188 config=self.config)
189 wallet = d['wallet'] # type: Standard_Wallet
190 self.assertEqual(passphrase, wallet.keystore.get_passphrase(password))
191 self.assertEqual(text, wallet.keystore.get_seed(password))
192 self.assertEqual(encrypt_file, wallet.storage.is_encrypted())
193 self.assertEqual('bc1q2ccr34wzep58d4239tl3x3734ttle92a8srmuw', wallet.get_receiving_addresses()[0])
194
195 def test_restore_wallet_from_text_xpub(self):
196 text = 'zpub6nydoME6CFdJtMpzHW5BNoPz6i6XbeT9qfz72wsRqGdgGEYeivso6xjfw8cGcCyHwF7BNW4LDuHF35XrZsovBLWMF4qXSjmhTXYiHbWqGLt'
197 d = restore_wallet_from_text(text, path=self.wallet_path, gap_limit=1, config=self.config)
198 wallet = d['wallet'] # type: Standard_Wallet
199 self.assertEqual(text, wallet.keystore.get_master_public_key())
200 self.assertEqual('bc1q2ccr34wzep58d4239tl3x3734ttle92a8srmuw', wallet.get_receiving_addresses()[0])
201
202 def test_restore_wallet_from_text_xkey_that_is_also_a_valid_electrum_seed_by_chance(self):
203 text = 'yprvAJBpuoF4FKpK92ofzQ7ge6VJMtorow3maAGPvPGj38ggr2xd1xCrC9ojUVEf9jhW5L9SPu6fU2U3o64cLrRQ83zaQGNa6YP3ajZS6hHNPXj'
204 d = restore_wallet_from_text(text, path=self.wallet_path, gap_limit=1, config=self.config)
205 wallet = d['wallet'] # type: Standard_Wallet
206 self.assertEqual(text, wallet.keystore.get_master_private_key(password=None))
207 self.assertEqual('3Pa4hfP3LFWqa2nfphYaF7PZfdJYNusAnp', wallet.get_receiving_addresses()[0])
208
209 def test_restore_wallet_from_text_xprv(self):
210 text = 'zprvAZzHPqhCMt51fskXBUYB1fTFYgG3CBjJUT4WEZTpGw6hPSDWBPZYZARC5sE9xAcX8NeWvvucFws8vZxEa65RosKAhy7r5MsmKTxr3hmNmea'
211 d = restore_wallet_from_text(text, path=self.wallet_path, gap_limit=1, config=self.config)
212 wallet = d['wallet'] # type: Standard_Wallet
213 self.assertEqual(text, wallet.keystore.get_master_private_key(password=None))
214 self.assertEqual('bc1q2ccr34wzep58d4239tl3x3734ttle92a8srmuw', wallet.get_receiving_addresses()[0])
215
216 def test_restore_wallet_from_text_addresses(self):
217 text = 'bc1q2ccr34wzep58d4239tl3x3734ttle92a8srmuw bc1qnp78h78vp92pwdwq5xvh8eprlga5q8gu66960c'
218 d = restore_wallet_from_text(text, path=self.wallet_path, config=self.config)
219 wallet = d['wallet'] # type: Imported_Wallet
220 self.assertEqual('bc1q2ccr34wzep58d4239tl3x3734ttle92a8srmuw', wallet.get_receiving_addresses()[0])
221 self.assertEqual(2, len(wallet.get_receiving_addresses()))
222 # also test addr deletion
223 wallet.delete_address('bc1qnp78h78vp92pwdwq5xvh8eprlga5q8gu66960c')
224 self.assertEqual(1, len(wallet.get_receiving_addresses()))
225
226 def test_restore_wallet_from_text_privkeys(self):
227 text = 'p2wpkh:L4jkdiXszG26SUYvwwJhzGwg37H2nLhrbip7u6crmgNeJysv5FHL p2wpkh:L24GxnN7NNUAfCXA6hFzB1jt59fYAAiFZMcLaJ2ZSawGpM3uqhb1'
228 d = restore_wallet_from_text(text, path=self.wallet_path, config=self.config)
229 wallet = d['wallet'] # type: Imported_Wallet
230 addr0 = wallet.get_receiving_addresses()[0]
231 self.assertEqual('bc1q2ccr34wzep58d4239tl3x3734ttle92a8srmuw', addr0)
232 self.assertEqual('p2wpkh:L4jkdiXszG26SUYvwwJhzGwg37H2nLhrbip7u6crmgNeJysv5FHL',
233 wallet.export_private_key(addr0, password=None))
234 self.assertEqual(2, len(wallet.get_receiving_addresses()))
235 # also test addr deletion
236 wallet.delete_address('bc1qnp78h78vp92pwdwq5xvh8eprlga5q8gu66960c')
237 self.assertEqual(1, len(wallet.get_receiving_addresses()))
238
239
240 class TestWalletPassword(WalletTestCase):
241
242 def setUp(self):
243 super().setUp()
244 self.asyncio_loop, self._stop_loop, self._loop_thread = util.create_and_start_event_loop()
245
246 def tearDown(self):
247 super().tearDown()
248 self.asyncio_loop.call_soon_threadsafe(self._stop_loop.set_result, 1)
249 self._loop_thread.join(timeout=1)
250
251 def test_update_password_of_imported_wallet(self):
252 wallet_str = '{"addr_history":{"1364Js2VG66BwRdkaoxAaFtdPb1eQgn8Dr":[],"15CyDgLffJsJgQrhcyooFH4gnVDG82pUrA":[],"1Exet2BhHsFxKTwhnfdsBMkPYLGvobxuW6":[]},"addresses":{"change":[],"receiving":["1364Js2VG66BwRdkaoxAaFtdPb1eQgn8Dr","1Exet2BhHsFxKTwhnfdsBMkPYLGvobxuW6","15CyDgLffJsJgQrhcyooFH4gnVDG82pUrA"]},"keystore":{"keypairs":{"0344b1588589958b0bcab03435061539e9bcf54677c104904044e4f8901f4ebdf5":"L2sED74axVXC4H8szBJ4rQJrkfem7UMc6usLCPUoEWxDCFGUaGUM","0389508c13999d08ffae0f434a085f4185922d64765c0bff2f66e36ad7f745cc5f":"L3Gi6EQLvYw8gEEUckmqawkevfj9s8hxoQDFveQJGZHTfyWnbk1U","04575f52b82f159fa649d2a4c353eb7435f30206f0a6cb9674fbd659f45082c37d559ffd19bea9c0d3b7dcc07a7b79f4cffb76026d5d4dff35341efe99056e22d2":"5JyVyXU1LiRXATvRTQvR9Kp8Rx1X84j2x49iGkjSsXipydtByUq"},"type":"imported"},"pruned_txo":{},"seed_version":13,"stored_height":-1,"transactions":{},"tx_fees":{},"txi":{},"txo":{},"use_encryption":false,"verified_tx3":{},"wallet_type":"standard","winpos-qt":[100,100,840,405]}'
253 db = WalletDB(wallet_str, manual_upgrades=False)
254 storage = WalletStorage(self.wallet_path)
255 wallet = Wallet(db, storage, config=self.config)
256
257 wallet.check_password(None)
258
259 wallet.update_password(None, "1234")
260
261 with self.assertRaises(InvalidPassword):
262 wallet.check_password(None)
263 with self.assertRaises(InvalidPassword):
264 wallet.check_password("wrong password")
265 wallet.check_password("1234")
266
267 def test_update_password_of_standard_wallet(self):
parazyd.org:70 /git/electrum/file/electrum/tests/test_wallet.py.gph:277: line too long