Oren
130 lines
5.8 KiB
Python
130 lines
5.8 KiB
Python
from io import StringIO
|
|
import json
|
|
import os, sys
|
|
from electrum.bitcoin import address_to_script
|
|
from electrum.fee_policy import FixedFeePolicy
|
|
from electrum.plugins.timelock_recovery.timelock_recovery import TimelockRecoveryContext
|
|
from electrum.simple_config import SimpleConfig
|
|
from electrum.storage import WalletStorage
|
|
from electrum.transaction import PartialTxOutput
|
|
from electrum.wallet import Wallet
|
|
from electrum.wallet_db import WalletDB
|
|
from electrum.plugins.timelock_recovery.qt import Plugin as TimelockRecoveryQtPlugin
|
|
|
|
from . import ElectrumTestCase
|
|
|
|
|
|
class TestTimelockRecovery(ElectrumTestCase):
|
|
TESTNET = True
|
|
|
|
def setUp(self):
|
|
super(TestTimelockRecovery, self).setUp()
|
|
self.config = SimpleConfig({'electrum_path': self.electrum_path})
|
|
|
|
self.wallet_path = os.path.join(self.electrum_path, "timelock_recovery_wallet")
|
|
|
|
self._saved_stdout = sys.stdout
|
|
self._stdout_buffer = StringIO()
|
|
sys.stdout = self._stdout_buffer
|
|
|
|
def tearDown(self):
|
|
super(TestTimelockRecovery, self).tearDown()
|
|
# Restore the "real" stdout
|
|
sys.stdout = self._saved_stdout
|
|
|
|
def _create_default_wallet(self):
|
|
with open(os.path.join(os.path.dirname(__file__), "test_timelock_recovery", "default_wallet"), "r") as f:
|
|
wallet_str = f.read()
|
|
storage = WalletStorage(self.wallet_path)
|
|
db = WalletDB(wallet_str, storage=storage, upgrade=True)
|
|
wallet = Wallet(db, config=self.config)
|
|
return wallet
|
|
|
|
async def test_get_alert_address(self):
|
|
wallet = self._create_default_wallet()
|
|
|
|
context = TimelockRecoveryContext(wallet)
|
|
alert_address = context.get_alert_address()
|
|
self.assertEqual(alert_address, 'tb1qchyc02y9mv4xths4je9puc4yzuxt8rfm26ef07')
|
|
|
|
async def test_get_cancellation_address(self):
|
|
wallet = self._create_default_wallet()
|
|
|
|
context = TimelockRecoveryContext(wallet)
|
|
context.get_alert_address()
|
|
cancellation_address = context.get_cancellation_address()
|
|
self.assertEqual(cancellation_address, 'tb1q6k5h4cz6ra8nzhg90xm9wldvadgh0fpttfthcg')
|
|
|
|
async def test_make_unsigned_alert_tx(self):
|
|
wallet = self._create_default_wallet()
|
|
|
|
context = TimelockRecoveryContext(wallet)
|
|
context.outputs = [
|
|
PartialTxOutput(scriptpubkey=address_to_script('tb1q4s8z6g5jqzllkgt8a4har94wl8tg0k9m8kv5zd'), value='!'),
|
|
]
|
|
|
|
alert_tx = context.make_unsigned_alert_tx(fee_policy=FixedFeePolicy(5000))
|
|
self.assertEqual(alert_tx.version, 2)
|
|
alert_tx_inputs = [tx_input.prevout.to_str() for tx_input in alert_tx.inputs()]
|
|
self.assertEqual(alert_tx_inputs, [
|
|
'59a9ff5fa62586f102b92504584f52e47f4ca0d5af061e99a0a3023fa70a70e2:1',
|
|
'778b01899d5ed48df03e406bc5babd1fdc8f1be4b7e5b9d20dd8caf24dd66ff4:1',
|
|
])
|
|
alert_tx_outputs = [(tx_output.address, tx_output.value) for tx_output in alert_tx.outputs()]
|
|
self.assertEqual(alert_tx_outputs, [
|
|
('tb1q4s8z6g5jqzllkgt8a4har94wl8tg0k9m8kv5zd', 600),
|
|
('tb1qchyc02y9mv4xths4je9puc4yzuxt8rfm26ef07', 743065),
|
|
])
|
|
self.assertEqual(alert_tx.txid(), '01c227f136c4490ec7cb0fe2ba5e44c436f58906b7fc29a83cb865d7e3bfaa60')
|
|
|
|
async def test_make_unsigned_recovery_tx(self):
|
|
wallet = self._create_default_wallet()
|
|
|
|
context = TimelockRecoveryContext(wallet)
|
|
context.outputs = [
|
|
PartialTxOutput(scriptpubkey=address_to_script('tb1q4s8z6g5jqzllkgt8a4har94wl8tg0k9m8kv5zd'), value='!'),
|
|
]
|
|
context.alert_tx = context.make_unsigned_alert_tx(fee_policy=FixedFeePolicy(5000))
|
|
context.timelock_days = 90
|
|
|
|
recovery_tx = context.make_unsigned_recovery_tx(fee_policy=FixedFeePolicy(5000))
|
|
self.assertEqual(recovery_tx.version, 2)
|
|
recovery_tx_inputs = [tx_input.prevout.to_str() for tx_input in recovery_tx.inputs()]
|
|
self.assertEqual(recovery_tx_inputs, [
|
|
'01c227f136c4490ec7cb0fe2ba5e44c436f58906b7fc29a83cb865d7e3bfaa60:1',
|
|
])
|
|
self.assertEqual(recovery_tx.inputs()[0].nsequence, 0x00403b54)
|
|
|
|
recovery_tx_outputs = [(tx_output.address, tx_output.value) for tx_output in recovery_tx.outputs()]
|
|
self.assertEqual(recovery_tx_outputs, [
|
|
('tb1q4s8z6g5jqzllkgt8a4har94wl8tg0k9m8kv5zd', 738065),
|
|
])
|
|
|
|
async def test_make_unsigned_cancellation_tx(self):
|
|
wallet = self._create_default_wallet()
|
|
|
|
context = TimelockRecoveryContext(wallet)
|
|
context.outputs = [
|
|
PartialTxOutput(scriptpubkey=address_to_script('tb1q4s8z6g5jqzllkgt8a4har94wl8tg0k9m8kv5zd'), value='!'),
|
|
]
|
|
context.alert_tx = context.make_unsigned_alert_tx(fee_policy=FixedFeePolicy(5000))
|
|
|
|
cancellation_tx = context.make_unsigned_cancellation_tx(fee_policy=FixedFeePolicy(6000))
|
|
self.assertEqual(cancellation_tx.version, 2)
|
|
cancellation_tx_inputs = [tx_input.prevout.to_str() for tx_input in cancellation_tx.inputs()]
|
|
self.assertEqual(cancellation_tx_inputs, [
|
|
'01c227f136c4490ec7cb0fe2ba5e44c436f58906b7fc29a83cb865d7e3bfaa60:1',
|
|
])
|
|
self.assertEqual(cancellation_tx.inputs()[0].nsequence, 0xfffffffd)
|
|
cancellation_tx_outputs = [(tx_output.address, tx_output.value) for tx_output in cancellation_tx.outputs()]
|
|
self.assertEqual(cancellation_tx_outputs, [
|
|
('tb1q6k5h4cz6ra8nzhg90xm9wldvadgh0fpttfthcg', 737065),
|
|
])
|
|
|
|
def test_checksum_non_ascii(self):
|
|
# Non-ASCII characters must be serialized as-is (ensure_ascii=False),
|
|
# not escaped as \uXXXX sequences, before hashing.
|
|
json_data = {"wallet_name": "Ωmega Wörld Ñoño 日本語 中文 עברית العربية", "id": "abc-123"}
|
|
result = TimelockRecoveryQtPlugin._checksum(json_data)
|
|
self.assertEqual(result, "74674eca")
|