zebra.lucky
/
joinmarket-clientserver
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

fix test/unified/test_segwit.py

add_frost_channel_encryption
zebra-lucky 4 months ago
parent
d512c58543
commit
d5213ec45a
2 changed files with 146 additions and 132 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 7
      test/unified/common.py
  2. 271
      test/unified/test_segwit.py

7
test/unified/common.py
Unescape View File

@ -50,7 +50,7 @@ async def make_sign_and_push(ins_full,
output_addr = await wallet_service.get_new_addr(
1,
BaseWallet.ADDRESS_TYPE_INTERNAL) if not output_addr else output_addr
change_addr = wallet_service.get_new_addr(
change_addr = await wallet_service.get_new_addr(
0,
BaseWallet.ADDRESS_TYPE_INTERNAL) if not change_addr else change_addr
fee_est = estimate_tx_fee(len(ins), 2) if estimate_fee else 10000
@ -135,8 +135,9 @@ async def make_wallets(n,
amt = mean_amt - sdev_amt / 2.0 + deviation
if amt < 0: amt = 0.001
amt = float(Decimal(amt).quantize(Decimal(10)**-8))
jm_single().bc_interface.grab_coins(wallet_service.get_new_addr(
j, BaseWallet.ADDRESS_TYPE_INTERNAL), amt)
dest_addr = await wallet_service.get_new_addr(
j, BaseWallet.ADDRESS_TYPE_INTERNAL)
jm_single().bc_interface.grab_coins(dest_addr , amt)
return wallets

271
test/unified/test_segwit.py
Unescape View File

@ -6,140 +6,153 @@ from common import make_wallets
from pprint import pformat
import jmbitcoin as btc
import pytest
from unittest_parametrize import parametrize, ParametrizedTestCase
from jmbase import get_log, hextobin
from jmclient import load_test_config, jm_single, LegacyWallet, BaseWallet
from common import TrialAsyncioTestCase
pytestmark = pytest.mark.usefixtures("setup_regtest_bitcoind")
log = get_log()
def test_segwit_valid_txs(setup_segwit):
with open("test/tx_segwit_valid.json", "r") as f:
json_data = f.read()
valid_txs = json.loads(json_data)
for j in valid_txs:
if len(j) < 2:
continue
deserialized_tx = btc.CMutableTransaction.deserialize(hextobin(j[1]))
print(pformat(deserialized_tx))
assert deserialized_tx.serialize() == hextobin(j[1])
#TODO use bcinterface to decoderawtransaction
#and compare the json values
@pytest.mark.parametrize(
"wallet_structure, in_amt, amount, segwit_amt, segwit_ins, o_ins", [
([[1, 0, 0, 0, 0]], 1, 1000000, 1, [0, 1, 2], []),
([[4, 0, 0, 0, 1]], 3, 100000000, 1, [0, 2], [1, 3]),
([[4, 0, 0, 0, 1]], 3, 100000000, 1, [0, 5], [1, 2, 3, 4]),
([[4, 0, 0, 0, 0]], 2, 200000007, 0.3, [0, 1, 4, 5], [2, 3, 6]),
])
def test_spend_p2sh_p2wpkh_multi(setup_segwit, wallet_structure, in_amt, amount,
segwit_amt, segwit_ins, o_ins):
"""Creates a wallet from which non-segwit inputs/
outputs can be created, constructs one or more
p2wpkh in p2sh spendable utxos (by paying into the
corresponding address) and tests spending them
in combination.
wallet_structure is in accordance with commontest.make_wallets, see docs there
in_amt is the amount to pay into each address into the wallet (non-segwit adds)
amount (in satoshis) is how much we will pay to the output address
segwit_amt in BTC is the amount we will fund each new segwit address with
segwit_ins is a list of input indices (where to place the funding segwit utxos)
other_ins is a list of input indices (where to place the funding non-sw utxos)
"""
MIXDEPTH = 0
# set up wallets and inputs
nsw_wallet_service = make_wallets(1, wallet_structure, in_amt,
walletclass=LegacyWallet)[0]['wallet']
nsw_wallet_service.sync_wallet(fast=True)
sw_wallet_service = make_wallets(1, [[len(segwit_ins), 0, 0, 0, 0]], segwit_amt)[0]['wallet']
sw_wallet_service.sync_wallet(fast=True)
nsw_utxos = nsw_wallet_service.get_utxos_by_mixdepth()[MIXDEPTH]
sw_utxos = sw_wallet_service.get_utxos_by_mixdepth()[MIXDEPTH]
assert len(o_ins) <= len(nsw_utxos), "sync failed"
assert len(segwit_ins) <= len(sw_utxos), "sync failed"
total_amt_in_sat = 0
nsw_ins = {}
for nsw_in_index in o_ins:
total_amt_in_sat += in_amt * 10**8
nsw_ins[nsw_in_index] = nsw_utxos.popitem()
sw_ins = {}
for sw_in_index in segwit_ins:
total_amt_in_sat += int(segwit_amt * 10**8)
sw_ins[sw_in_index] = sw_utxos.popitem()
all_ins = {}
all_ins.update(nsw_ins)
all_ins.update(sw_ins)
# sanity checks
assert len(all_ins) == len(nsw_ins) + len(sw_ins), \
"test broken, duplicate index"
for k in all_ins:
assert 0 <= k < len(all_ins), "test broken, missing input index"
# FIXME: encoding mess, mktx should accept binary input formats
tx_ins = []
for i, (txin, data) in sorted(all_ins.items(), key=lambda x: x[0]):
tx_ins.append(txin)
# create outputs
FEE = 50000
assert FEE < total_amt_in_sat - amount, "test broken, not enough funds"
cj_script = nsw_wallet_service.get_new_script(MIXDEPTH + 1,
BaseWallet.ADDRESS_TYPE_INTERNAL)
change_script = nsw_wallet_service.get_new_script(MIXDEPTH,
BaseWallet.ADDRESS_TYPE_INTERNAL)
change_amt = total_amt_in_sat - amount - FEE
tx_outs = [
{'address': nsw_wallet_service.script_to_addr(cj_script),
'value': amount},
{'address': nsw_wallet_service.script_to_addr(change_script),
'value': change_amt}]
tx = btc.mktx(tx_ins, tx_outs)
# import new addresses to bitcoind
jm_single().bc_interface.import_addresses(
[nsw_wallet_service.script_to_addr(x)
for x in [cj_script, change_script]], nsw_wallet_service.get_wallet_name())
# sign tx
scripts = {}
for nsw_in_index in o_ins:
inp = nsw_ins[nsw_in_index][1]
scripts[nsw_in_index] = (inp['script'], inp['value'])
success, msg = nsw_wallet_service.sign_tx(tx, scripts)
assert success, msg
scripts = {}
for sw_in_index in segwit_ins:
inp = sw_ins[sw_in_index][1]
scripts[sw_in_index] = (inp['script'], inp['value'])
success, msg = sw_wallet_service.sign_tx(tx, scripts)
assert success, msg
print(tx)
# push and verify
txid = jm_single().bc_interface.pushtx(tx.serialize())
assert txid
balances = jm_single().bc_interface.get_received_by_addr(
[nsw_wallet_service.script_to_addr(cj_script),
nsw_wallet_service.script_to_addr(change_script)])['data']
assert balances[0]['balance'] == amount
assert balances[1]['balance'] == change_amt
@pytest.fixture(scope="module")
def setup_segwit():
load_test_config()
jm_single().bc_interface.tick_forward_chain_interval = 1
class SegWitTests(TrialAsyncioTestCase, ParametrizedTestCase):
async def asyncSetUp(self):
load_test_config()
jm_single().bc_interface.tick_forward_chain_interval = 1
async def test_segwit_valid_txs(self):
with open("test/tx_segwit_valid.json", "r") as f:
json_data = f.read()
valid_txs = json.loads(json_data)
for j in valid_txs:
if len(j) < 2:
continue
deserialized_tx = btc.CMutableTransaction.deserialize(hextobin(j[1]))
print(pformat(deserialized_tx))
assert deserialized_tx.serialize() == hextobin(j[1])
#TODO use bcinterface to decoderawtransaction
#and compare the json values
@parametrize(
"wallet_structure, in_amt, amount, segwit_amt, segwit_ins, o_ins", [
([[1, 0, 0, 0, 0]], 1, 1000000, 1, [0, 1, 2], []),
([[4, 0, 0, 0, 1]], 3, 100000000, 1, [0, 2], [1, 3]),
([[4, 0, 0, 0, 1]], 3, 100000000, 1, [0, 5], [1, 2, 3, 4]),
([[4, 0, 0, 0, 0]], 2, 200000007, 0.3, [0, 1, 4, 5], [2, 3, 6]),
])
async def test_spend_p2sh_p2wpkh_multi(self, wallet_structure, in_amt, amount,
segwit_amt, segwit_ins, o_ins):
"""Creates a wallet from which non-segwit inputs/
outputs can be created, constructs one or more
p2wpkh in p2sh spendable utxos (by paying into the
corresponding address) and tests spending them
in combination.
wallet_structure is in accordance with commontest.make_wallets, see docs there
in_amt is the amount to pay into each address into the wallet (non-segwit adds)
amount (in satoshis) is how much we will pay to the output address
segwit_amt in BTC is the amount we will fund each new segwit address with
segwit_ins is a list of input indices (where to place the funding segwit utxos)
other_ins is a list of input indices (where to place the funding non-sw utxos)
"""
MIXDEPTH = 0
# set up wallets and inputs
wallet_services = await make_wallets(
1, wallet_structure, in_amt, walletclass=LegacyWallet)
nsw_wallet_service = wallet_services[0]['wallet']
await nsw_wallet_service.sync_wallet(fast=True)
sw_wallet_services = await make_wallets(
1, [[len(segwit_ins), 0, 0, 0, 0]], segwit_amt)
sw_wallet_service = sw_wallet_services[0]['wallet']
await sw_wallet_service.sync_wallet(fast=True)
nsw_utxos_by_md = await nsw_wallet_service.get_utxos_by_mixdepth()
nsw_utxos = nsw_utxos_by_md[MIXDEPTH]
sw_utxos_by_md = await sw_wallet_service.get_utxos_by_mixdepth()
sw_utxos = sw_utxos_by_md[MIXDEPTH]
assert len(o_ins) <= len(nsw_utxos), "sync failed"
assert len(segwit_ins) <= len(sw_utxos), "sync failed"
total_amt_in_sat = 0
nsw_ins = {}
for nsw_in_index in o_ins:
total_amt_in_sat += in_amt * 10**8
nsw_ins[nsw_in_index] = nsw_utxos.popitem()
sw_ins = {}
for sw_in_index in segwit_ins:
total_amt_in_sat += int(segwit_amt * 10**8)
sw_ins[sw_in_index] = sw_utxos.popitem()
all_ins = {}
all_ins.update(nsw_ins)
all_ins.update(sw_ins)
# sanity checks
assert len(all_ins) == len(nsw_ins) + len(sw_ins), \
"test broken, duplicate index"
for k in all_ins:
assert 0 <= k < len(all_ins), "test broken, missing input index"
# FIXME: encoding mess, mktx should accept binary input formats
tx_ins = []
for i, (txin, data) in sorted(all_ins.items(), key=lambda x: x[0]):
tx_ins.append(txin)
# create outputs
FEE = 50000
assert FEE < total_amt_in_sat - amount, "test broken, not enough funds"
cj_script = await nsw_wallet_service.get_new_script(
MIXDEPTH + 1, BaseWallet.ADDRESS_TYPE_INTERNAL)
change_script = await nsw_wallet_service.get_new_script(
MIXDEPTH, BaseWallet.ADDRESS_TYPE_INTERNAL)
change_amt = total_amt_in_sat - amount - FEE
tx_outs = [
{'address': await nsw_wallet_service.script_to_addr(cj_script),
'value': amount},
{'address': await nsw_wallet_service.script_to_addr(change_script),
'value': change_amt}]
tx = btc.mktx(tx_ins, tx_outs)
# import new addresses to bitcoind
jm_single().bc_interface.import_addresses(
[(await nsw_wallet_service.script_to_addr(x))
for x in [cj_script, change_script]],
nsw_wallet_service.get_wallet_name())
# sign tx
scripts = {}
for nsw_in_index in o_ins:
inp = nsw_ins[nsw_in_index][1]
scripts[nsw_in_index] = (inp['script'], inp['value'])
success, msg = await nsw_wallet_service.sign_tx(tx, scripts)
assert success, msg
scripts = {}
for sw_in_index in segwit_ins:
inp = sw_ins[sw_in_index][1]
scripts[sw_in_index] = (inp['script'], inp['value'])
success, msg = await sw_wallet_service.sign_tx(tx, scripts)
assert success, msg
print(tx)
# push and verify
txid = jm_single().bc_interface.pushtx(tx.serialize())
assert txid
balances = jm_single().bc_interface.get_received_by_addr(
[(await nsw_wallet_service.script_to_addr(cj_script)),
(await nsw_wallet_service.script_to_addr(change_script))])['data']
assert balances[0]['balance'] == amount
assert balances[1]['balance'] == change_amt
'''

Write Preview
Loading…
Cancel
Save