diff --git a/jmclient/test/test_tx_notify.py b/jmclient/test/test_tx_notify.py deleted file mode 100644 index a202f73..0000000 --- a/jmclient/test/test_tx_notify.py +++ /dev/null @@ -1,130 +0,0 @@ -#! /usr/bin/env python -from __future__ import absolute_import -'''test of the add_tx_notify() timeouts and related callbacks''' - -import sys - -import time -from commontest import make_wallets -import subprocess -import jmbitcoin as bitcoin -import pytest -from jmclient import (load_program_config, jm_single, get_log, Wallet, sync_wallet) -from twisted.trial import unittest -from twisted.internet import reactor, task -log = get_log() - -unconfirm_called = [False] -confirm_called = [False] -timeout_unconfirm_called = [False] -timeout_confirm_called = [False] - -def unconfirm_callback(txd, txid): - unconfirm_called[0] = True - log.debug('unconfirm callback()') - -def confirm_callback(txd, txid, confirmations): - confirm_called[0] = True - log.debug('confirm callback()') - -def timeout_callback(confirmed): - if not confirmed: - timeout_unconfirm_called[0] = True - log.debug('timeout unconfirm callback()') - else: - timeout_confirm_called[0] = True - log.debug('timeout confirm callback()') - -def run_tests(): - #TODO: rewrite with deferreds not callLater() - no_timeout() - -def notify_handler_errors(): - #TODO this has hardcoded 62612 which is from regtest_joinmarket.cfg - #default testing config - txhex = make_tx_add_notify() - with pytest.raises(subprocess.CalledProcessError) as e_info: - res = subprocess.check_output(["curl", "-I", "--connect-timeout", "1", - "http://localhost:62612/walletnotify?abcd"]) - with pytest.raises(subprocess.CalledProcessError) as e_info: - res = subprocess.check_output(["curl", "-I", "--connect-timeout", "1", - "http://localhost:62612/walletnotify?xxXX"]) - #unrecognised calls are returned an error message, not a curl failure - res = subprocess.check_output(["curl", "-I", "--connect-timeout", "1", - "http://localhost:62612/dummycommand"]) - #alertnotifys with an alert message should be accepted (todo, deprecate) - res = subprocess.check_output(["curl", "-I", "--connect-timeout", "1", - "http://localhost:62612/alertnotify?dummyalertmessage"]) - jm_single().bc_interface.tick_forward_chain_interval = 2 - -def no_timeout(): - txhex = make_tx_add_notify() - jm_single().bc_interface.pushtx(txhex) - reactor.callLater(6, checkstate, True, True, False, False, unconfirm_timeout) - -def checkstate(uc, c, tu, tc, f=None): - assert unconfirm_called[0] == uc - assert confirm_called[0] == c - assert timeout_unconfirm_called[0] == tu - assert timeout_confirm_called[0] == tc - if f: - f() - -def unconfirm_timeout(): - txhex = make_tx_add_notify() - #dont pushtx - reactor.callLater(6, checkstate, False, False, True, False) - -def confirm_timeout(): - txhex = make_tx_add_notify() - jm_single().bc_interface.tick_forward_chain_interval = -1 - jm_single().bc_interface.pushtx(txhex) - reactor.callLater(True, False, False, True, notify_handler_errors) - -def make_tx_add_notify(): - wallet_dict = make_wallets(1, [[1, 0, 0, 0, 0]], mean_amt=4, sdev_amt=0)[0] - amount = 250000000 - txfee = 10000 - wallet = wallet_dict['wallet'] - sync_wallet(wallet) - inputs = wallet.select_utxos(0, amount) - ins = inputs.keys() - input_value = sum([i['value'] for i in inputs.values()]) - output_addr = wallet.get_new_addr(1, 0) - change_addr = wallet.get_new_addr(0, 1) - outs = [{'value': amount, 'address': output_addr}, - {'value': input_value - amount - txfee, 'address': change_addr}] - tx = bitcoin.mktx(ins, outs) - de_tx = bitcoin.deserialize(tx) - for index, ins in enumerate(de_tx['ins']): - utxo = ins['outpoint']['hash'] + ':' + str(ins['outpoint']['index']) - addr = inputs[utxo]['address'] - priv = wallet.get_key_from_addr(addr) - tx = bitcoin.sign(tx, index, priv) - - unconfirm_called[0] = confirm_called[0] = False - timeout_unconfirm_called[0] = timeout_confirm_called[0] = False - jm_single().bc_interface.add_tx_notify( - bitcoin.deserialize(tx), unconfirm_callback, - confirm_callback, output_addr, timeout_callback) - return tx - -class TxNotifies(unittest.TestCase): - - def setUp(self): - setup_tx_notify() - reactor.callLater(0.0, run_tests) - - def test_waiter(self): - #TODO fixed timeout, horrible.. - return task.deferLater(reactor, 30, self._called_by_deferred) - - def _called_by_deferred(self): - pass - -def setup_tx_notify(): - load_program_config() - jm_single().config.set('TIMEOUT', 'unconfirm_timeout_sec', '3') - jm_single().config.set('TIMEOUT', 'confirm_timeout_hours', str(6.0 / 60 / 60)) - jm_single().bc_interface.tick_forward_chain_interval = 2 -