test_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): 268 wallet_str = '''{"addr_history":{"12ECgkzK6gHouKAZ7QiooYBuk1CgJLJxes":[],"12iR43FPb5M7sw4Mcrr5y1nHKepg9EtZP1":[],"13HT1pfWctsSXVFzF76uYuVdQvcAQ2MAgB":[],"13kG9WH9JqS7hyCcVL1ssLdNv4aXocQY9c":[],"14Tf3qiiHJXStSU4KmienAhHfHq7FHpBpz":[],"14gmBxYV97mzYwWdJSJ3MTLbTHVegaKrcA":[],"15FGuHvRssu1r8fCw98vrbpfc3M4xs5FAV":[],"17oJzweA2gn6SDjsKgA9vUD5ocT1sSnr2Z":[],"18hNcSjZzRcRP6J2bfFRxp9UfpMoC4hGTv":[],"18n9PFxBjmKCGhd4PCDEEqYsi2CsnEfn2B":[],"19a98ZfEezDNbCwidVigV5PAJwrR2kw4Jz":[],"19z3j2ELqbg2pR87byCCt3BCyKR7rc3q8G":[],"1A3XSmvLQvePmvm7yctsGkBMX9ZKKXLrVq":[],"1CmhFe2BN1h9jheFpJf4v39XNPj8F9U6d":[],"1DuphhHUayKzbkdvjVjf5dtjn2ACkz4zEs":[],"1E4ygSNJpWL2uPXZHBptmU2LqwZTqb1Ado":[],"1GTDSjkVc9vaaBBBGNVqTANHJBcoT5VW9z":[],"1GWqgpThAuSq3tDg6uCoLQxPXQNnU8jZ52":[],"1GhmpwqSF5cqNgdr9oJMZx8dKxPRo4pYPP":[],"1J5TTUQKhwehEACw6Jjte1E22FVrbeDmpv":[],"1JWySzjzJhsETUUcqVZHuvQLA7pfFfmesb":[],"1KQHxcy3QUHAWMHKUtJjqD9cMKXcY2RTwZ":[],"1KoxZfc2KsgovjGDxwqanbFEA76uxgYH4G":[],"1KqVEPXdpbYvEbwsZcEKkrA4A2jsgj9hYN":[],"1N16yDSYe76c5A3CoVoWAKxHeAUc8Jhf9J":[],"1Pm8JBhzUJDqeQQKrmnop1Frr4phe1jbTt":[]},"addresses":{"change":["1GhmpwqSF5cqNgdr9oJMZx8dKxPRo4pYPP","1GTDSjkVc9vaaBBBGNVqTANHJBcoT5VW9z","15FGuHvRssu1r8fCw98vrbpfc3M4xs5FAV","1A3XSmvLQvePmvm7yctsGkBMX9ZKKXLrVq","19z3j2ELqbg2pR87byCCt3BCyKR7rc3q8G","1JWySzjzJhsETUUcqVZHuvQLA7pfFfmesb"],"receiving":["14gmBxYV97mzYwWdJSJ3MTLbTHVegaKrcA","13HT1pfWctsSXVFzF76uYuVdQvcAQ2MAgB","19a98ZfEezDNbCwidVigV5PAJwrR2kw4Jz","1J5TTUQKhwehEACw6Jjte1E22FVrbeDmpv","1Pm8JBhzUJDqeQQKrmnop1Frr4phe1jbTt","13kG9WH9JqS7hyCcVL1ssLdNv4aXocQY9c","1KQHxcy3QUHAWMHKUtJjqD9cMKXcY2RTwZ","12ECgkzK6gHouKAZ7QiooYBuk1CgJLJxes","12iR43FPb5M7sw4Mcrr5y1nHKepg9EtZP1","14Tf3qiiHJXStSU4KmienAhHfHq7FHpBpz","1KqVEPXdpbYvEbwsZcEKkrA4A2jsgj9hYN","17oJzweA2gn6SDjsKgA9vUD5ocT1sSnr2Z","1E4ygSNJpWL2uPXZHBptmU2LqwZTqb1Ado","18hNcSjZzRcRP6J2bfFRxp9UfpMoC4hGTv","1KoxZfc2KsgovjGDxwqanbFEA76uxgYH4G","18n9PFxBjmKCGhd4PCDEEqYsi2CsnEfn2B","1CmhFe2BN1h9jheFpJf4v39XNPj8F9U6d","1DuphhHUayKzbkdvjVjf5dtjn2ACkz4zEs","1GWqgpThAuSq3tDg6uCoLQxPXQNnU8jZ52","1N16yDSYe76c5A3CoVoWAKxHeAUc8Jhf9J"]},"keystore":{"seed":"cereal wise two govern top pet frog nut rule sketch bundle logic","type":"bip32","xprv":"xprv9s21ZrQH143K29XjRjUs6MnDB9wXjXbJP2kG1fnRk8zjdDYWqVkQYUqaDtgZp5zPSrH5PZQJs8sU25HrUgT1WdgsPU8GbifKurtMYg37d4v","xpub":"xpub661MyMwAqRbcEdcCXm1sTViwjBn28zK9kFfrp4C3JUXiW1sfP34f6HA45B9yr7EH5XGzWuTfMTdqpt9XPrVQVUdgiYb5NW9m8ij1FSZgGBF"},"pruned_txo":{},"seed_type":"standard","seed_version":13,"stored_height":-1,"transactions":{},"tx_fees":{},"txi":{},"txo":{},"use_encryption":false,"verified_tx3":{},"wallet_type":"standard","winpos-qt":[619,310,840,405]}''' 269 db = WalletDB(wallet_str, manual_upgrades=False) 270 storage = WalletStorage(self.wallet_path) 271 wallet = Wallet(db, storage, config=self.config) 272 273 wallet.check_password(None) 274 275 wallet.update_password(None, "1234") 276 with self.assertRaises(InvalidPassword): 277 wallet.check_password(None) 278 with self.assertRaises(InvalidPassword): 279 wallet.check_password("wrong password") 280 wallet.check_password("1234") 281 282 def test_update_password_with_app_restarts(self): 283 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]}' 284 db = WalletDB(wallet_str, manual_upgrades=False) 285 storage = WalletStorage(self.wallet_path) 286 wallet = Wallet(db, storage, config=self.config) 287 asyncio.run_coroutine_threadsafe(wallet.stop(), self.asyncio_loop).result() 288 289 storage = WalletStorage(self.wallet_path) 290 # if storage.is_encrypted(): 291 # storage.decrypt(password) 292 db = WalletDB(storage.read(), manual_upgrades=False) 293 wallet = Wallet(db, storage, config=self.config) 294 295 wallet.check_password(None) 296 297 wallet.update_password(None, "1234") 298 with self.assertRaises(InvalidPassword): 299 wallet.check_password(None) 300 with self.assertRaises(InvalidPassword): 301 wallet.check_password("wrong password") 302 wallet.check_password("1234")