d0s3t3ch
/
wownero-python
mirror of https://git.wownero.com/lza_menace/wownero-python.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge branch 'txfrmgr'

This commit is contained in:
Michał Sałaban 2018-01-30 12:47:25 +01:00
parent 63d47f200b ed383d6040
commit bf7c369d48
14 changed files with 445 additions and 147 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
monero/account.py
View File

@ -1,5 +1,6 @@
from . import address
from . import prio
from .transaction import PaymentManager
class Account(object):
@ -8,6 +9,8 @@ class Account(object):
def __init__(self, backend, index):
self.index = index
self._backend = backend
self.incoming = PaymentManager(index, backend, 'in')
self.outgoing = PaymentManager(index, backend, 'out')
def balances(self):
return self._backend.balances(account=self.index)
@ -27,17 +30,6 @@ class Account(object):
def new_address(self, label=None):
return self._backend.new_address(account=self.index, label=label)
def payments(self, payment_id=None):
return self._backend.payments(account=self.index, payment_id=payment_id)
def transactions_in(self, confirmed=True, unconfirmed=False):
return self._backend.transactions_in(
account=self.index, confirmed=confirmed, unconfirmed=unconfirmed)
def transactions_out(self, confirmed=True, unconfirmed=True):
return self._backend.transactions_out(
account=self.index, confirmed=confirmed, unconfirmed=unconfirmed)
def transfer(self, address, amount,
priority=prio.NORMAL, ringsize=5, payment_id=None, unlock_time=0,
relay=True):

18
monero/address.py
View File

@ -23,12 +23,12 @@ class Address(object):
_valid_netbytes = (18, 53)
# NOTE: _valid_netbytes order is (real, testnet)
def __init__(self, address, label=None):
address = str(address)
if not _ADDR_REGEX.match(address):
def __init__(self, addr, label=None):
addr = str(addr)
if not _ADDR_REGEX.match(addr):
raise ValueError("Address must be 95 characters long base58-encoded string, "
"is {addr} ({len} chars length)".format(addr=address, len=len(address)))
self._decode(address)
"is {addr} ({len} chars length)".format(addr=addr, len=len(addr)))
self._decode(addr)
self.label = label or self.label
def _decode(self, address):
@ -86,7 +86,7 @@ class Address(object):
return str(self) == str(other)
if isinstance(other, str):
return str(self) == other
return super()
return super(Address, self).__eq__(other)
class SubAddress(Address):
@ -148,12 +148,12 @@ def address(addr, label=None):
return Address(addr, label=label)
elif netbyte in SubAddress._valid_netbytes:
return SubAddress(addr, label=label)
raise ValueError("Invalid address netbyte {nb}. Allowed values are: {allowed}".format(
nb=hexlify(chr(netbyte)),
raise ValueError("Invalid address netbyte {nb:x}. Allowed values are: {allowed}".format(
nb=netbyte,
allowed=", ".join(map(
lambda b: '%02x' % b,
sorted(Address._valid_netbytes + SubAddress._valid_netbytes)))))
elif _IADDR_REGEX.match(addr):
return IntegratedAddress(addr)
raise ValueError("Address must be either 95 or 106 characters long base58-encoded string, "
"is {addr} ({len} chars length)".format(addr=address, len=len(address)))
"is {addr} ({len} chars length)".format(addr=addr, len=len(addr)))

94
monero/backends/jsonrpc.py
View File

@ -169,64 +169,68 @@ class JSONRPCWallet(object):
_balance = self.raw_request('getbalance', {'account_index': account})
return (from_atomic(_balance['balance']), from_atomic(_balance['unlocked_balance']))
def payments(self, account=0, payment_id=0):
payment_id = PaymentID(payment_id)
_log.debug("Getting payments for account {acc}, payment_id {pid}".format(
acc=account, pid=payment_id))
_payments = self.raw_request('get_payments', {
def transfers_in(self, account, pmtfilter):
params = {'account_index': account}
method = 'get_transfers'
if pmtfilter.unconfirmed:
params['in'] = pmtfilter.confirmed
params['out'] = False
params['pool'] = True
else:
if pmtfilter.payment_ids:
method = 'get_bulk_payments'
params['payment_id'] = pmtfilter.payment_ids
else:
params['in'] = pmtfilter.confirmed
params['out'] = False
params['pool'] = False
if method == 'get_transfers':
arg = 'in'
if pmtfilter.min_height:
params['min_height'] = pmtfilter.min_height
params['filter_by_height'] = True
if pmtfilter.max_height:
params['max_height'] = pmtfilter.max_height
params['filter_by_height'] = True
elif pmtfilter.min_height:
arg = 'payments'
params['min_block_height'] = pmtfilter.min_height
_pmts = self.raw_request(method, params)
pmts = _pmts.get(arg, [])
if pmtfilter.unconfirmed:
pmts.extend(_pmts.get('pool', []))
return list(pmtfilter.filter(map(self._inpayment, pmts)))
def transfers_out(self, account, pmtfilter):
_pmts = self.raw_request('get_transfers', {
'account_index': account,
'payment_id': str(payment_id)})
pmts = []
for data in _payments['payments']:
# Monero <= 0.11 : no address is passed because there's only one
data['address'] = data['address'] or self._master_address
pmts.append(self._inpayment(data))
return pmts
def transactions_in(self, account=0, confirmed=True, unconfirmed=False):
_txns = self.raw_request('get_transfers',
{'account_index': account, 'in': confirmed, 'out': False, 'pool': unconfirmed})
txns = _txns.get('in', [])
if unconfirmed:
txns.extend(_txns.get('pool', []))
return [self._inpayment(tx) for tx in sorted(txns, key=operator.itemgetter('timestamp'))]
def transactions_out(self, account=0, confirmed=True, unconfirmed=True):
_txns = self.raw_request('get_transfers',
{'account_index': account, 'in': False, 'out': confirmed, 'pool': unconfirmed})
txns = _txns.get('out', [])
if unconfirmed:
txns.extend(_txns.get('pool', []))
return [self._outpayment(tx) for tx in sorted(txns, key=operator.itemgetter('timestamp'))]
def get_transaction(self, txhash):
_tx = self.raw_request('get_transfer_by_txid', {'txid': str(txhash)})['transfer']
if _tx['type'] == 'in':
return self._inpayment(tx)
elif _tx['type'] == 'out':
return self._outpayment(tx)
return Payment(**self._paymentdict(tx))
'in': False,
'out': pmtfilter.confirmed,
'pool': pmtfilter.unconfirmed})
pmts = _pmts.get('out', [])
if pmtfilter.unconfirmed:
pmts.extend(_pmts.get('pool', []))
return list(pmtfilter.filter(map(self._outpayment, pmts)))
def _paymentdict(self, data):
pid = data.get('payment_id', None)
laddr = data.get('address', None)
if laddr:
laddr = address(laddr)
return {
'txhash': data.get('txid', data.get('tx_hash')),
'payment_id': None if pid is None else PaymentID(pid),
'amount': from_atomic(data['amount']),
'timestamp': datetime.fromtimestamp(data['timestamp']) if 'timestamp' in data else None,
'note': data.get('note'),
'transaction': self._tx(data)
'note': data.get('note', None),
'transaction': self._tx(data),
'local_address': laddr,
}
def _inpayment(self, data):
p = self._paymentdict(data)
p.update({'received_by': address(data['address']) if 'address' in data else None})
return IncomingPayment(**p)
return IncomingPayment(**self._paymentdict(data))
def _outpayment(self, data):
p = self._paymentdict(data)
p.update({'sent_from': address(data['address']) if 'address' in data else None})
return OutgoingPayment(**p)
return OutgoingPayment(**self._paymentdict(data))
def _tx(self, data):
return Transaction(**{

2
monero/numbers.py
View File

@ -59,4 +59,4 @@ class PaymentID(object):
return int(self) == other
elif isinstance(other, _str_types):
return str(self) == other
return super()
return super(PaymentID, self).__eq__(other)

133
monero/transaction.py
View File

@ -1,34 +1,37 @@
import sys
from .address import address
from .numbers import PaymentID
class Payment(object):
tx_hash = None
payment_id = None
amount = None
timestamp = None
transaction = None
def __init__(self, **kwargs):
self.tx_hash = kwargs.get('tx_hash', self.tx_hash)
self.amount = kwargs.get('amount', self.amount)
self.timestamp = kwargs.get('timestamp', self.timestamp)
self.payment_id = kwargs.get('payment_id', self.payment_id)
self.transaction = kwargs.get('transaction', self.transaction)
class IncomingPayment(Payment):
received_by = None
def __init__(self, **kwargs):
super(IncomingPayment, self).__init__(**kwargs)
self.received_by = kwargs.get('received_by', self.received_by)
class OutgoingPayment(Payment):
sent_from = None
local_address = None
note = ''
def __init__(self, **kwargs):
super(OutgoingPayment, self).__init__(**kwargs)
self.sent_from = kwargs.get('sent_from', self.sent_from)
self.note = kwargs.get('note', self.sent_from)
self.amount = kwargs.pop('amount', self.amount)
self.timestamp = kwargs.pop('timestamp', self.timestamp)
self.payment_id = kwargs.pop('payment_id', self.payment_id)
self.transaction = kwargs.pop('transaction', self.transaction)
self.local_address = kwargs.pop('local_address', self.local_address)
self.note = kwargs.pop('note', self.note)
if len(kwargs):
raise ValueError("Excessive arguments for {}: {}".format(type(self), kwargs))
def __repr__(self):
return "{} {:.12f} id={}".format(self.transaction.hash, self.amount, self.payment_id)
class IncomingPayment(Payment):
def __repr__(self):
return "in: {} {:.12f} id={}".format(self.transaction.hash, self.amount, self.payment_id)
class OutgoingPayment(Payment):
def __repr__(self):
return "out: {} {:.12f} id={}".format(self.transaction.hash, self.amount, self.payment_id)
class Transaction(object):
@ -46,3 +49,87 @@ class Transaction(object):
self.timestamp = kwargs.get('timestamp', self.timestamp)
self.key = kwargs.get('key', self.key)
self.blob = kwargs.get('blob', self.blob)
def __repr__(self):
return self.hash
if sys.version_info < (3,):
_str_types = (str, bytes, unicode)
else:
_str_types = (str, bytes)
class PaymentManager(object):
account_idx = 0
backend = None
def __init__(self, account_idx, backend, direction):
self.account_idx = account_idx
self.backend = backend
self.direction = direction
def __call__(self, **filterparams):
fetch = self.backend.transfers_in if self.direction == 'in' else self.backend.transfers_out
return fetch(self.account_idx, PaymentFilter(**filterparams))
class PaymentFilter(object):
def __init__(self, **filterparams):
self.min_height = filterparams.pop('min_height', None)
self.max_height = filterparams.pop('max_height', None)
self.unconfirmed = filterparams.pop('unconfirmed', False)
self.confirmed = filterparams.pop('confirmed', True)
_local_address = filterparams.pop('local_address', None)
_payment_id = filterparams.pop('payment_id', None)
if len(filterparams) > 0:
raise ValueError("Excessive arguments for payment query: {}".format(filterparams))
if _local_address is None:
self.local_addresses = []
else:
if isinstance(_local_address, _str_types):
local_addresses = [_local_address]
else:
try:
iter(_local_address)
local_addresses = _local_address
except TypeError:
local_addresses = [_local_address]
self.local_addresses = list(map(address, local_addresses))
if _payment_id is None:
self.payment_ids = []
else:
if isinstance(_payment_id, _str_types):
payment_ids = [_payment_id]
else:
try:
iter(_payment_id)
payment_ids = _payment_id
except TypeError:
payment_ids = [_payment_id]
self.payment_ids = list(map(PaymentID, payment_ids))
def check(self, payment):
ht = payment.transaction.height
if ht is None:
if not self.unconfirmed:
return False
if self.min_height is not None or self.max_height is not None:
# mempool txns are filtered out if any height range check is present
return False
else:
if not self.confirmed:
return False
if self.min_height is not None and ht < self.min_height:
return False
if self.max_height is not None and ht > self.max_height:
return False
if self.payment_ids and payment.payment_id not in self.payment_ids:
return False
if self.local_addresses and payment.local_address not in self.local_addresses:
return False
return True
def filter(self, payments):
return filter(self.check, payments)

28
monero/wallet.py
View File

@ -1,13 +1,15 @@
from . import address
from . import prio
from . import account
from . import transaction
from .transaction import Payment, PaymentManager
class Wallet(object):
accounts = None
def __init__(self, backend):
self._backend = backend
self.incoming = PaymentManager(0, backend, 'in')
self.outgoing = PaymentManager(0, backend, 'out')
self.refresh()
def refresh(self):
@ -52,14 +54,15 @@ class Wallet(object):
self.accounts.append(acc)
return acc
def get_transaction(self, hash):
return self._backend.get_transaction(hash)
def confirmations(self, txn):
txn = self._backend.get_transaction(txn)
if txn.height is None:
def confirmations(self, txn_or_pmt):
if isinstance(txn_or_pmt, Payment):
txn = txn_or_pmt.transaction
else:
txn = txn_or_pmt
try:
return max(0, self.height() - txn.height)
except TypeError:
return 0
return max(0, self.height() - txn.height)
# Following methods operate on default account (index=0)
def balances(self):
@ -77,15 +80,6 @@ class Wallet(object):
def new_address(self, label=None):
return self.accounts[0].new_address(label=label)
def payments(self, payment_id=None):
return self.accounts[0].payments(payment_id=payment_id)
def transactions_in(self, confirmed=True, unconfirmed=False):
return self.accounts[0].transactions_in(confirmed=confirmed, unconfirmed=unconfirmed)
def transactions_out(self, confirmed=True, unconfirmed=True):
return self.accounts[0].transactions_out(confirmed=confirmed, unconfirmed=unconfirmed)
def transfer(self, address, amount,
priority=prio.NORMAL, ringsize=5, payment_id=None, unlock_time=0,
relay=True):

1
setup.py
View File

@ -24,4 +24,5 @@ setup(
'Programming Language :: Python',
],
keywords = 'monero cryptocurrency',
test_suite='tests',
)

8
tests/__init__.py
View File

@ -1,3 +1,5 @@
from .address import AddressTestCase, TestnetAddressTestCase
from .numbers import NumbersTestCase
from .wallet import SubaddrWalletTestCase
from . import test_address
from . import test_numbers
from . import test_transaction
from . import test_wallet
from . import test_jsonrpcwallet

17
tests/address.py → tests/test_address.py
View File

@ -58,6 +58,8 @@ class Tests(object):
self.assertEqual(sa.is_testnet(), self.testnet)
self.assertEqual(sa2.is_testnet(), self.testnet)
self.assertNotEqual(a, 0)
def test_idempotence(self):
a = Address(self.addr)
a_idem = Address(a)
@ -83,6 +85,21 @@ class Tests(object):
self.assertRaises(TypeError, a.with_payment_id, "%x" % (2**64+1))
s = SubAddress(self.subaddr)
self.assertRaises(TypeError, s.with_payment_id, 0)
self.assertRaises(ValueError, address, 'whatever')
self.assertRaises(ValueError, Address, 'whatever')
self.assertRaises(ValueError, SubAddress, 'whatever')
self.assertRaises(ValueError, IntegratedAddress, 'whatever')
# Aeon
self.assertRaises(
ValueError,
address,
'Wmtj8UAJhdrhbKvwyBJmLEUZKHcffv2VHNBaq6oTxJFwJjUj3QwMUSS32mddSX7vchbxXdmb4QuZA9TsN47441f61yAYLQYTo')
# invalid netbyte
self.assertRaises(
ValueError,
address,
'Cf6RinMUztY5otm6NEFjg3UWBBkXK6Lh23wKrLFMEcCY7i3A6aPLH9i4QMCkf6CdWk8Q9N7yoJf7ANKgtQMuPM6JANXgCWs')
def test_type_mismatch(self):
self.assertRaises(ValueError, Address, self.iaddr)

52
tests/wallet.py → tests/test_jsonrpcwallet.py
View File

@ -7,7 +7,9 @@ except ImportError:
from mock import patch, Mock
from monero.wallet import Wallet
from monero.address import Address
from monero.account import Account
from monero.address import Address, address
from monero.numbers import PaymentID
from monero.transaction import IncomingPayment, OutgoingPayment, Transaction
from monero.backends.jsonrpc import JSONRPCWallet
@ -120,7 +122,7 @@ class SubaddrWalletTestCase(unittest.TestCase):
self.assertEqual(len(self.wallet.accounts[0].addresses()), 8)
@patch('monero.backends.jsonrpc.requests.post')
def test_transactions_in(self, mock_post):
def test_incoming(self, mock_post):
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = self.accounts_result
self.wallet = Wallet(JSONRPCWallet())
@ -133,6 +135,7 @@ class SubaddrWalletTestCase(unittest.TestCase):
'height': 1049947,
'note': '',
'payment_id': '0000000000000000',
'address': '9vgV48wWAPTWik5QSUSoGYicdvvsbSNHrT9Arsx1XBTz6VrWPSgfmnUKSPZDMyX4Ms8R9TkhB4uFqK9s5LUBbV6YQN2Q9ag',
'subaddr_index': {'major': 0, 'minor': 0},
'timestamp': 1511926250,
'txid': '0cdde0eb934c44b523f6e966a5e19b131ed68c3c08600bc087f48ae13015b704',
@ -145,6 +148,7 @@ class SubaddrWalletTestCase(unittest.TestCase):
'note': '',
'payment_id': '0000000000000000',
'subaddr_index': {'major': 0, 'minor': 0},
'address': '9vgV48wWAPTWik5QSUSoGYicdvvsbSNHrT9Arsx1XBTz6VrWPSgfmnUKSPZDMyX4Ms8R9TkhB4uFqK9s5LUBbV6YQN2Q9ag',
'timestamp': 1511926250,
'txid': '8b4154681c48a873550818ecaa6408a7c987a882b80917d6c902befd6ee57109',
'type': 'in',
@ -155,24 +159,24 @@ class SubaddrWalletTestCase(unittest.TestCase):
'height': 1049947,
'note': '',
'payment_id': '0000000000000000',
'address': '9vgV48wWAPTWik5QSUSoGYicdvvsbSNHrT9Arsx1XBTz6VrWPSgfmnUKSPZDMyX4Ms8R9TkhB4uFqK9s5LUBbV6YQN2Q9ag',
'subaddr_index': {'major': 0, 'minor': 0},
'timestamp': 1511926250,
'txid': 'd23a7d086e70df7aa0ca002361c4b35e35a272345b0a513ece4f21b773941f5e',
'type': 'in',
'unlock_time': 0}]}}
pay_in = self.wallet.transactions_in()
pay_in = self.wallet.incoming()
self.assertEqual(len(list(pay_in)), 3)
for pmt in pay_in:
self.assertIsInstance(pmt, IncomingPayment)
# Once PR#3010 has been merged to Monero, update the JSON and enable the following:
# self.assertIsInstance(pmt.received_by, Address)
self.assertIsInstance(pmt.local_address, Address)
self.assertIsInstance(pmt.amount, Decimal)
self.assertIsInstance(pmt.transaction, Transaction)
self.assertIsInstance(pmt.transaction.fee, Decimal)
self.assertIsInstance(pmt.transaction.height, int)
@patch('monero.backends.jsonrpc.requests.post')
def test_transactions_out(self, mock_post):
def test_outgoing(self, mock_post):
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = self.accounts_result
self.wallet = Wallet(JSONRPCWallet())
@ -188,6 +192,7 @@ class SubaddrWalletTestCase(unittest.TestCase):
'note': '',
'payment_id': '0000000000000000',
'subaddr_index': {'major': 2, 'minor': 0},
'address': 'BgCseuY3jFJAZS7kt9mrNg7fEG3bo5BV91CTyKbYu9GFiU6hUZhvdNWCTUdQNPNcA4PyFApsFr3EsQDEDfT3tQSY1mVZeP2',
'timestamp': 1512095241,
'txid': 'eadca0f956a2a60cb3497a7dff1bd80153140a111d2f7db257a264bd9b76f0b3',
'type': 'out',
@ -201,6 +206,7 @@ class SubaddrWalletTestCase(unittest.TestCase):
'note': '',
'payment_id': '0000000000000000',
'subaddr_index': {'major': 2, 'minor': 0},
'address': 'BgCseuY3jFJAZS7kt9mrNg7fEG3bo5BV91CTyKbYu9GFiU6hUZhvdNWCTUdQNPNcA4PyFApsFr3EsQDEDfT3tQSY1mVZeP2',
'timestamp': 1511922110,
'txid': '5486ae9e6867ceb6e5aa478b32cba5c11d28e6d905c8479565c78e3933163ab6',
'type': 'out',
@ -214,6 +220,7 @@ class SubaddrWalletTestCase(unittest.TestCase):
'note': '',
'payment_id': '0000000000000000',
'subaddr_index': {'major': 2, 'minor': 0},
'address': 'BgCseuY3jFJAZS7kt9mrNg7fEG3bo5BV91CTyKbYu9GFiU6hUZhvdNWCTUdQNPNcA4PyFApsFr3EsQDEDfT3tQSY1mVZeP2',
'timestamp': 1512098498,
'txid': '9591c8f6832cc3b7908c2447b2feef58c44e7774a5c05cea617ad2f3b3866c18',
'type': 'out',
@ -227,6 +234,7 @@ class SubaddrWalletTestCase(unittest.TestCase):
'note': '',
'payment_id': '0000000000000000',
'subaddr_index': {'major': 2, 'minor': 0},
'address': 'BgCseuY3jFJAZS7kt9mrNg7fEG3bo5BV91CTyKbYu9GFiU6hUZhvdNWCTUdQNPNcA4PyFApsFr3EsQDEDfT3tQSY1mVZeP2',
'timestamp': 1511926250,
'txid': 'af669b99162d9b514a0e8d3bd1d905e3b8778e6fcb88d172e5e049e909c4cc87',
'type': 'out',
@ -240,6 +248,7 @@ class SubaddrWalletTestCase(unittest.TestCase):
'note': '',
'payment_id': '0000000000000000',
'subaddr_index': {'major': 2, 'minor': 0},
'address': 'BgCseuY3jFJAZS7kt9mrNg7fEG3bo5BV91CTyKbYu9GFiU6hUZhvdNWCTUdQNPNcA4PyFApsFr3EsQDEDfT3tQSY1mVZeP2',
'timestamp': 1511914391,
'txid': '2fa2de7fbf009093c5319d0421d3e8c684b8351a066c48d51369aedbbfd1d9af',
'type': 'out',
@ -253,45 +262,22 @@ class SubaddrWalletTestCase(unittest.TestCase):
'note': '',
'payment_id': '0000000000000000',
'subaddr_index': {'major': 2, 'minor': 0},
'address': 'BgCseuY3jFJAZS7kt9mrNg7fEG3bo5BV91CTyKbYu9GFiU6hUZhvdNWCTUdQNPNcA4PyFApsFr3EsQDEDfT3tQSY1mVZeP2',
'timestamp': 1511928624,
'txid': '7e3db6c59c02d870f18b37a37cfc5857eeb5412df4ea00bb1971f3095f72b0d8',
'type': 'out',
'unlock_time': 0}]}}
pay_out = self.wallet.transactions_out()
pay_out = self.wallet.outgoing()
self.assertEqual(len(list(pay_out)), 6)
for pmt in pay_out:
self.assertIsInstance(pmt, OutgoingPayment)
# Once PR#3010 has been merged to Monero, update the JSON and enable the following:
# self.assertIsInstance(pmt.sent_from, Address)
self.assertIsInstance(pmt.local_address, Address)
self.assertIsInstance(pmt.amount, Decimal)
self.assertIsInstance(pmt.timestamp, datetime)
self.assertIsInstance(pmt.transaction, Transaction)
self.assertIsInstance(pmt.transaction.fee, Decimal)
self.assertIsInstance(pmt.transaction.height, int)
@patch('monero.backends.jsonrpc.requests.post')
def test_payments(self, mock_post):
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = self.accounts_result
self.wallet = Wallet(JSONRPCWallet())
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {'id': 0,
'jsonrpc': '2.0',
'result': {'payments': [{'address': 'BZ9V9tfTDgHYsnAxgeMLaGCUb6yMaGNiZJwiBWQrE23MXcRqSde9DKa9LnPw31o2G8QrdKdUNM7VWhd3dr22ivk54QGqZ6u',
'amount': 2313370000000,
'block_height': 1048268,
'payment_id': 'feedbadbeef12345',
'subaddr_index': {'major': 1, 'minor': 1},
'tx_hash': 'e84343c2ebba4d4d94764e0cd275adee07cf7b4718565513be453d3724f6174b',
'unlock_time': 0}]}}
payments = self.wallet.payments(payment_id=0xfeedbadbeef12345)
self.assertEqual(len(list(payments)), 1)
for pmt in payments:
self.assertIsInstance(pmt, IncomingPayment)
self.assertIsInstance(pmt.received_by, Address)
self.assertIsInstance(pmt.amount, Decimal)
self.assertIsInstance(pmt.transaction, Transaction)
self.assertIsInstance(pmt.transaction.height, int)
self.assertEqual(pmt.note, '')
@patch('monero.backends.jsonrpc.requests.post')
def test_send_transfer(self, mock_post):

4
tests/numbers.py → tests/test_numbers.py
View File

@ -1,7 +1,7 @@
from decimal import Decimal
import unittest
from monero.numbers import to_atomic, from_atomic, PaymentID
from monero.numbers import to_atomic, from_atomic, as_monero, PaymentID
class NumbersTestCase(unittest.TestCase):
def test_simple_numbers(self):
@ -14,6 +14,7 @@ class NumbersTestCase(unittest.TestCase):
def test_rounding(self):
self.assertEqual(to_atomic(Decimal('1.0000000000004')), 1000000000000)
self.assertEqual(as_monero(Decimal('1.0000000000014')), Decimal('1.000000000001'))
def test_payment_id(self):
pid = PaymentID('0')
@ -21,6 +22,7 @@ class NumbersTestCase(unittest.TestCase):
self.assertEqual(pid, 0)
self.assertEqual(pid, '0000000000000000')
self.assertEqual(PaymentID(pid), pid)
self.assertNotEqual(pid, None)
pid = PaymentID('abcdef')
self.assertTrue(pid.is_short())
self.assertEqual(pid, 0xabcdef)

28
tests/test_transaction.py Normal file
View File

@ -0,0 +1,28 @@
from datetime import datetime
from decimal import Decimal
import unittest
from monero.address import address
from monero.numbers import PaymentID
from monero.transaction import IncomingPayment, OutgoingPayment, Transaction
class FiltersTestCase(unittest.TestCase):
def setUp(self):
self.tx1 = Transaction(
timestamp=datetime(2018, 1, 29, 15, 0, 25),
height=1087606,
hash='a0b876ebcf7c1d499712d84cedec836f9d50b608bb22d6cb49fd2feae3ffed14',
fee=Decimal('0.00352891'))
self.pm1 = IncomingPayment(
amount=Decimal('1'),
local_address=address('Bf6ngv7q2TBWup13nEm9AjZ36gLE6i4QCaZ7XScZUKDUeGbYEHmPRdegKGwLT8tBBK7P6L32RELNzCR6QzNFkmogDjvypyV'),
payment_id=PaymentID('0166d8da6c0045c51273dd65d6f63734beb8a84e0545a185b2cfd053fced9f5d'),
transaction=self.tx1)
def test_hash(self):
self.assertIn(
'a0b876ebcf7c1d499712d84cedec836f9d50b608bb22d6cb49fd2feae3ffed14',
repr(self.tx1))
self.assertIn(
'a0b876ebcf7c1d499712d84cedec836f9d50b608bb22d6cb49fd2feae3ffed14',
repr(self.pm1))

185
tests/test_wallet.py Normal file
View File

@ -0,0 +1,185 @@
from datetime import datetime
from decimal import Decimal
import unittest
from monero.wallet import Wallet
from monero.account import Account
from monero.address import address
from monero.numbers import PaymentID
from monero.transaction import IncomingPayment, OutgoingPayment, Transaction
class FiltersTestCase(unittest.TestCase):
def setUp(self):
class MockBackend(object):
def __init__(self):
self.transfers = []
tx = Transaction(
timestamp=datetime(2018, 1, 29, 15, 0, 25),
height=1087606,
hash='a0b876ebcf7c1d499712d84cedec836f9d50b608bb22d6cb49fd2feae3ffed14',
fee=Decimal('0.00352891'))
pm = IncomingPayment(
amount=Decimal('1'),
local_address=address('Bf6ngv7q2TBWup13nEm9AjZ36gLE6i4QCaZ7XScZUKDUeGbYEHmPRdegKGwLT8tBBK7P6L32RELNzCR6QzNFkmogDjvypyV'),
payment_id=PaymentID('0166d8da6c0045c51273dd65d6f63734beb8a84e0545a185b2cfd053fced9f5d'),
transaction=tx)
self.transfers.append(pm)
tx = Transaction(
timestamp=datetime(2018, 1, 29, 14, 57, 47),
height=1087601,
hash='f34b495cec77822a70f829ec8a5a7f1e727128d62e6b1438e9cb7799654d610e',
fee=Decimal('0.008661870000'))
pm = IncomingPayment(
amount=Decimal('3.000000000000'),
local_address=address('BhE3cQvB7VF2uuXcpXp28Wbadez6GgjypdRS1F1Mzqn8Advd6q8VfaX8ZoEDobjejrMfpHeNXoX8MjY8q8prW1PEALgr1En'),
payment_id=PaymentID('f75ad90e25d71a12'),
transaction=tx)
self.transfers.append(pm)
tx = Transaction(
timestamp=datetime(2018, 1, 29, 13, 17, 18),
height=1087530,
hash='5c3ab739346e9d98d38dc7b8d36a4b7b1e4b6a16276946485a69797dbf887cd8',
fee=Decimal('0.000962550000'))
pm = IncomingPayment(
amount=Decimal('10.000000000000'),
local_address=address('9tQoHWyZ4yXUgbz9nvMcFZUfDy5hxcdZabQCxmNCUukKYicXegsDL7nQpcUa3A1pF6K3fhq3scsyY88tdB1MqucULcKzWZC'),
payment_id=PaymentID('f75ad90e25d71a12'),
transaction=tx)
self.transfers.append(pm)
tx = Transaction(
timestamp=datetime(2018, 1, 29, 13, 17, 18),
height=1087530,
hash='4ea70add5d0c7db33557551b15cd174972fcfc73bf0f6a6b47b7837564b708d3',
fee=Decimal('0.000962550000'))
pm = IncomingPayment(
amount=Decimal('4.000000000000'),
local_address=address('9tQoHWyZ4yXUgbz9nvMcFZUfDy5hxcdZabQCxmNCUukKYicXegsDL7nQpcUa3A1pF6K3fhq3scsyY88tdB1MqucULcKzWZC'),
payment_id=PaymentID('f75ad90e25d71a12'),
transaction=tx)
self.transfers.append(pm)
tx = Transaction(
timestamp=datetime(2018, 1, 29, 13, 17, 18),
height=1087530,
hash='e9a71c01875bec20812f71d155bfabf42024fde3ec82475562b817dcc8cbf8dc',
fee=Decimal('0.000962550000'))
pm = IncomingPayment(
amount=Decimal('2.120000000000'),
local_address=address('9tQoHWyZ4yXUgbz9nvMcFZUfDy5hxcdZabQCxmNCUukKYicXegsDL7nQpcUa3A1pF6K3fhq3scsyY88tdB1MqucULcKzWZC'),
payment_id=PaymentID('cb248105ea6a9189'),
transaction=tx)
self.transfers.append(pm)
tx = Transaction(
timestamp=datetime(2018, 1, 29, 14, 57, 47),
height=1087601,
hash='5ef7ead6a041101ed326568fbb59c128403cba46076c3f353cd110d969dac808',
fee=Decimal('0.000962430000'))
pm = IncomingPayment(
amount=Decimal('7.000000000000'),
local_address=address('BhE3cQvB7VF2uuXcpXp28Wbadez6GgjypdRS1F1Mzqn8Advd6q8VfaX8ZoEDobjejrMfpHeNXoX8MjY8q8prW1PEALgr1En'),
payment_id=PaymentID('0000000000000000'),
transaction=tx)
self.transfers.append(pm)
tx = Transaction(
timestamp=datetime(2018, 1, 29, 13, 17, 18),
height=1087530,
hash='cc44568337a186c2e1ccc080b43b4ae9db26a07b7afd7edeed60ce2fc4a6477f',
fee=Decimal('0.000962550000'))
pm = IncomingPayment(
amount=Decimal('10.000000000000'),
local_address=address('9tQoHWyZ4yXUgbz9nvMcFZUfDy5hxcdZabQCxmNCUukKYicXegsDL7nQpcUa3A1pF6K3fhq3scsyY88tdB1MqucULcKzWZC'),
payment_id=PaymentID('0000000000000000'),
transaction=tx)
self.transfers.append(pm)
tx = Transaction(
timestamp=datetime(2018, 1, 29, 21, 13, 28),
height=None,
hash='d29264ad317e8fdb55ea04484c00420430c35be7b3fe6dd663f99aebf41a786c',
fee=Decimal('0.000961950000'))
pm = IncomingPayment(
amount=Decimal('3.140000000000'),
local_address=address('9tQoHWyZ4yXUgbz9nvMcFZUfDy5hxcdZabQCxmNCUukKYicXegsDL7nQpcUa3A1pF6K3fhq3scsyY88tdB1MqucULcKzWZC'),
payment_id=PaymentID('03f6649304ea4cb2'),
transaction=tx)
self.transfers.append(pm)
def height(self):
return 1087607
def accounts(self):
return [Account(self, 0)]
def transfers_in(self, account, pmtfilter):
return list(pmtfilter.filter(self.transfers))
self.wallet = Wallet(MockBackend())
def test_filter_none(self):
pmts = self.wallet.incoming()
self.assertEqual(len(pmts), 7)
def test_filter_payment_id(self):
pmts = self.wallet.incoming(payment_id='cb248105ea6a9189')
self.assertEqual(len(pmts), 1)
self.assertEqual(
pmts[0].transaction.hash,
'e9a71c01875bec20812f71d155bfabf42024fde3ec82475562b817dcc8cbf8dc')
pmts = self.wallet.incoming(payment_id='f75ad90e25d71a12')
self.assertEqual(len(pmts), 3)
pmts = self.wallet.incoming(payment_id=('cb248105ea6a9189', 'f75ad90e25d71a12'))
self.assertEqual(len(pmts), 4)
self.assertEqual(
pmts,
self.wallet.incoming(payment_id=(PaymentID('cb248105ea6a9189'), 'f75ad90e25d71a12')))
def test_filter_address(self):
pmts = self.wallet.incoming(
local_address='BhE3cQvB7VF2uuXcpXp28Wbadez6GgjypdRS1F1Mzqn8Advd6q8VfaX8ZoEDobjejrMfpHeNXoX8MjY8q8prW1PEALgr1En')
self.assertEqual(len(pmts), 2)
self.assertEqual(
pmts,
self.wallet.incoming(
local_address=address('BhE3cQvB7VF2uuXcpXp28Wbadez6GgjypdRS1F1Mzqn8Advd6q8VfaX8ZoEDobjejrMfpHeNXoX8MjY8q8prW1PEALgr1En')))
pmts = self.wallet.incoming(
local_address=(
'BhE3cQvB7VF2uuXcpXp28Wbadez6GgjypdRS1F1Mzqn8Advd6q8VfaX8ZoEDobjejrMfpHeNXoX8MjY8q8prW1PEALgr1En',
'Bf6ngv7q2TBWup13nEm9AjZ36gLE6i4QCaZ7XScZUKDUeGbYEHmPRdegKGwLT8tBBK7P6L32RELNzCR6QzNFkmogDjvypyV'))
self.assertEqual(len(pmts), 3)
def test_filter_mempool(self):
pmts = self.wallet.incoming()
self.assertEqual(len(pmts), 7)
for p in pmts:
self.assertGreater(self.wallet.confirmations(p.transaction), 0)
pmts = self.wallet.incoming(unconfirmed=True)
self.assertEqual(len(pmts), 8)
pmts = self.wallet.incoming(unconfirmed=True, confirmed=False)
self.assertEqual(len(pmts), 1)
self.assertEqual(
pmts[0].transaction.hash,
'd29264ad317e8fdb55ea04484c00420430c35be7b3fe6dd663f99aebf41a786c')
self.assertEqual(self.wallet.confirmations(pmts[0]), 0)
self.assertEqual(self.wallet.confirmations(pmts[0].transaction), 0)
pmts = self.wallet.incoming(unconfirmed=True, confirmed=False, min_height=1)
self.assertEqual(len(pmts), 0)
pmts = self.wallet.incoming(unconfirmed=True, confirmed=False, max_height=99999999999999)
self.assertEqual(len(pmts), 0)
pmts = self.wallet.incoming(payment_id='03f6649304ea4cb2')
self.assertEqual(len(pmts), 0)
pmts = self.wallet.incoming(unconfirmed=True, payment_id='03f6649304ea4cb2')
self.assertEqual(len(pmts), 1)
pmts = self.wallet.incoming(
local_address='9tQoHWyZ4yXUgbz9nvMcFZUfDy5hxcdZabQCxmNCUukKYicXegsDL7nQpcUa3A1pF6K3fhq3scsyY88tdB1MqucULcKzWZC')
self.assertEqual(len(pmts), 4)
pmts = self.wallet.incoming(
unconfirmed=True,
local_address='9tQoHWyZ4yXUgbz9nvMcFZUfDy5hxcdZabQCxmNCUukKYicXegsDL7nQpcUa3A1pF6K3fhq3scsyY88tdB1MqucULcKzWZC')
self.assertEqual(len(pmts), 5)
pmts = self.wallet.incoming(
local_address='9tQoHWyZ4yXUgbz9nvMcFZUfDy5hxcdZabQCxmNCUukKYicXegsDL7nQpcUa3A1pF6K3fhq3scsyY88tdB1MqucULcKzWZC',
payment_id='03f6649304ea4cb2')
self.assertEqual(len(pmts), 0)
pmts = self.wallet.incoming(
unconfirmed=True,
local_address='9tQoHWyZ4yXUgbz9nvMcFZUfDy5hxcdZabQCxmNCUukKYicXegsDL7nQpcUa3A1pF6K3fhq3scsyY88tdB1MqucULcKzWZC',
payment_id='03f6649304ea4cb2')
self.assertEqual(len(pmts), 1)

8
utils/walletdump.py
View File

@ -82,26 +82,26 @@ if len(w.accounts) > 1:
addresses = acc.addresses()
print("{num:2d} address(es):".format(num=len(addresses)))
print("\n".join(map(a2str, addresses)))
ins = acc.transactions_in(unconfirmed=True)
ins = acc.incoming(unconfirmed=True)
if ins:
print("\nIncoming transactions:")
print(_TXHDR.format(dir='received by'))
for tx in ins:
print(pmt2str(tx))
outs = acc.transactions_out(unconfirmed=True)
outs = acc.outgoing(unconfirmed=True)
if outs:
print("\nOutgoing transactions:")
print(_TXHDR.format(dir='sent from'))
for tx in outs:
print(pmt2str(tx))
else:
ins = w.transactions_in(unconfirmed=True)
ins = w.incoming(unconfirmed=True)
if ins:
print("\nIncoming transactions:")
print(_TXHDR.format(dir='received by'))
for tx in ins:
print(pmt2str(tx))
outs = w.transactions_out(unconfirmed=True)
outs = w.outgoing(unconfirmed=True)
if outs:
print("\nOutgoing transactions:")
print(_TXHDR.format(dir='sent from'))