forked from LBRYCommunity/lbry-sdk
165 lines
6.4 KiB
Python
165 lines
6.4 KiB
Python
from twisted.trial import unittest
|
|
from twisted.internet import defer
|
|
|
|
from torba.wallet import Wallet
|
|
from torba.constants import COIN
|
|
from torba.coin.bitcoinsegwit import MainNetLedger as ledger_class
|
|
from torba.basedatabase import constraints_to_sql
|
|
|
|
from .test_transaction import get_output, NULL_HASH
|
|
|
|
|
|
class TestConstraintBuilder(unittest.TestCase):
|
|
|
|
def test_dot(self):
|
|
constraints = {'txo.position': 18}
|
|
self.assertEqual(
|
|
constraints_to_sql(constraints, prepend_sql=''),
|
|
'txo.position = :txo_position'
|
|
)
|
|
self.assertEqual(constraints, {'txo_position': 18})
|
|
|
|
def test_any(self):
|
|
constraints = {
|
|
'ages__any': {
|
|
'txo.age__gt': 18,
|
|
'txo.age__lt': 38
|
|
}
|
|
}
|
|
self.assertEqual(
|
|
constraints_to_sql(constraints, prepend_sql=''),
|
|
'(txo.age > :ages__any_txo_age__gt OR txo.age < :ages__any_txo_age__lt)'
|
|
)
|
|
self.assertEqual(
|
|
constraints, {
|
|
'ages__any_txo_age__gt': 18,
|
|
'ages__any_txo_age__lt': 38
|
|
}
|
|
)
|
|
|
|
def test_in_list(self):
|
|
constraints = {'txo.age__in': [18, 38]}
|
|
self.assertEqual(
|
|
constraints_to_sql(constraints, prepend_sql=''),
|
|
'txo.age IN (:txo_age_1, :txo_age_2)'
|
|
)
|
|
self.assertEqual(
|
|
constraints, {
|
|
'txo_age_1': 18,
|
|
'txo_age_2': 38
|
|
}
|
|
)
|
|
|
|
def test_in_query(self):
|
|
constraints = {'txo.age__in': 'SELECT age from ages_table'}
|
|
self.assertEqual(
|
|
constraints_to_sql(constraints, prepend_sql=''),
|
|
'txo.age IN (SELECT age from ages_table)'
|
|
)
|
|
self.assertEqual(constraints, {})
|
|
|
|
def test_not_in_query(self):
|
|
constraints = {'txo.age__not_in': 'SELECT age from ages_table'}
|
|
self.assertEqual(
|
|
constraints_to_sql(constraints, prepend_sql=''),
|
|
'txo.age NOT IN (SELECT age from ages_table)'
|
|
)
|
|
self.assertEqual(constraints, {})
|
|
|
|
def test_in_invalid(self):
|
|
constraints = {'ages__in': 9}
|
|
with self.assertRaisesRegex(ValueError, 'list or string'):
|
|
constraints_to_sql(constraints, prepend_sql='')
|
|
|
|
|
|
class TestQueries(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.ledger = ledger_class({
|
|
'db': ledger_class.database_class(':memory:'),
|
|
'headers': ledger_class.headers_class(':memory:'),
|
|
})
|
|
return self.ledger.db.open()
|
|
|
|
@defer.inlineCallbacks
|
|
def create_account(self):
|
|
account = self.ledger.account_class.generate(self.ledger, Wallet())
|
|
yield account.ensure_address_gap()
|
|
return account
|
|
|
|
@defer.inlineCallbacks
|
|
def create_tx_from_nothing(self, my_account, height):
|
|
to_address = yield my_account.receiving.get_or_create_usable_address()
|
|
to_hash = ledger_class.address_to_hash160(to_address)
|
|
tx = ledger_class.transaction_class(height=height, is_verified=True) \
|
|
.add_inputs([self.txi(self.txo(1, NULL_HASH))]) \
|
|
.add_outputs([self.txo(1, to_hash)])
|
|
yield self.ledger.db.save_transaction_io('insert', tx, to_address, to_hash, '')
|
|
return tx
|
|
|
|
@defer.inlineCallbacks
|
|
def create_tx_from_txo(self, txo, to_account, height):
|
|
from_hash = txo.script.values['pubkey_hash']
|
|
from_address = self.ledger.hash160_to_address(from_hash)
|
|
to_address = yield to_account.receiving.get_or_create_usable_address()
|
|
to_hash = ledger_class.address_to_hash160(to_address)
|
|
tx = ledger_class.transaction_class(height=height, is_verified=True) \
|
|
.add_inputs([self.txi(txo)]) \
|
|
.add_outputs([self.txo(1, to_hash)])
|
|
yield self.ledger.db.save_transaction_io('insert', tx, from_address, from_hash, '')
|
|
yield self.ledger.db.save_transaction_io('', tx, to_address, to_hash, '')
|
|
return tx
|
|
|
|
@defer.inlineCallbacks
|
|
def create_tx_to_nowhere(self, txo, height):
|
|
from_hash = txo.script.values['pubkey_hash']
|
|
from_address = self.ledger.hash160_to_address(from_hash)
|
|
to_hash = NULL_HASH
|
|
tx = ledger_class.transaction_class(height=height, is_verified=True) \
|
|
.add_inputs([self.txi(txo)]) \
|
|
.add_outputs([self.txo(1, to_hash)])
|
|
yield self.ledger.db.save_transaction_io('insert', tx, from_address, from_hash, '')
|
|
return tx
|
|
|
|
def txo(self, amount, address):
|
|
return get_output(int(amount*COIN), address)
|
|
|
|
def txi(self, txo):
|
|
return ledger_class.transaction_class.input_class.spend(txo)
|
|
|
|
@defer.inlineCallbacks
|
|
def test_get_transactions(self):
|
|
account1 = yield self.create_account()
|
|
account2 = yield self.create_account()
|
|
tx1 = yield self.create_tx_from_nothing(account1, 1)
|
|
tx2 = yield self.create_tx_from_txo(tx1.outputs[0], account2, 2)
|
|
tx3 = yield self.create_tx_to_nowhere(tx2.outputs[0], 3)
|
|
|
|
txs = yield self.ledger.db.get_transactions()
|
|
self.assertEqual([tx3.id, tx2.id, tx1.id], [tx.id for tx in txs])
|
|
self.assertEqual([3, 2, 1], [tx.height for tx in txs])
|
|
|
|
txs = yield self.ledger.db.get_transactions(account1)
|
|
self.assertEqual([tx2.id, tx1.id], [tx.id for tx in txs])
|
|
self.assertEqual(txs[0].inputs[0].is_my_account, True)
|
|
self.assertEqual(txs[0].outputs[0].is_my_account, False)
|
|
self.assertEqual(txs[1].inputs[0].is_my_account, False)
|
|
self.assertEqual(txs[1].outputs[0].is_my_account, True)
|
|
|
|
txs = yield self.ledger.db.get_transactions(account2)
|
|
self.assertEqual([tx3.id, tx2.id], [tx.id for tx in txs])
|
|
self.assertEqual(txs[0].inputs[0].is_my_account, True)
|
|
self.assertEqual(txs[0].outputs[0].is_my_account, False)
|
|
self.assertEqual(txs[1].inputs[0].is_my_account, False)
|
|
self.assertEqual(txs[1].outputs[0].is_my_account, True)
|
|
|
|
tx = yield self.ledger.db.get_transaction(tx2.id)
|
|
self.assertEqual(tx.id, tx2.id)
|
|
self.assertEqual(tx.inputs[0].is_my_account, False)
|
|
self.assertEqual(tx.outputs[0].is_my_account, False)
|
|
tx = yield self.ledger.db.get_transaction(tx2.id, account1)
|
|
self.assertEqual(tx.inputs[0].is_my_account, True)
|
|
self.assertEqual(tx.outputs[0].is_my_account, False)
|
|
tx = yield self.ledger.db.get_transaction(tx2.id, account2)
|
|
self.assertEqual(tx.inputs[0].is_my_account, False)
|
|
self.assertEqual(tx.outputs[0].is_my_account, True)
|