199 lines
8.1 KiB
Python
199 lines
8.1 KiB
Python
import logging
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from aucoin import dsa, util
|
|
from aucoin.block import Block
|
|
from aucoin.blockchain import Blockchain
|
|
from aucoin.database import session_scope
|
|
from aucoin.exceptions import InvalidTransactionException, OrphanException
|
|
from aucoin.mempool import Mempool
|
|
from aucoin.network import Network
|
|
from aucoin.transactions import Transaction, Input, Output, CoinbaseTransaction
|
|
from aucoin.validation import Validator
|
|
from aucoin.wallet import Wallet, public_bytes
|
|
from tests import helpers
|
|
from tests.helpers import mine
|
|
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
|
|
|
|
class TestAddTransaction(unittest.TestCase):
|
|
def setUp(self):
|
|
self.blockchain = Blockchain(reset=True)
|
|
self.mempool = Mempool()
|
|
self.wallet = Wallet(self.blockchain, self.mempool)
|
|
self.network = Network(self.blockchain, self.mempool, max_peers=0)
|
|
self.validator = Validator(helpers.Core(), self.blockchain, self.mempool, self.network)
|
|
|
|
# add a block with an unspent transaction output we can reference
|
|
with session_scope() as session:
|
|
genesis = self.blockchain.genesis_block(session)
|
|
address = self.wallet.new_address()
|
|
self.private_key, self.public_key = self.wallet.keys[address]
|
|
self.block = Block(
|
|
hash_prev_block=genesis.hash,
|
|
target=genesis.target,
|
|
public_key=public_bytes(self.public_key),
|
|
transactions=[
|
|
CoinbaseTransaction(
|
|
address=address,
|
|
value=100,
|
|
block_height=1
|
|
)
|
|
]
|
|
)
|
|
self.validator.add_block(mine(self.block, self.private_key))
|
|
|
|
# the transaction which we will be modifying to provoke validation exceptions
|
|
self.transaction = Transaction(
|
|
inputs=[
|
|
Input(
|
|
prev_tx_hash=self.block.transactions[0].hash,
|
|
txout_index=0
|
|
)
|
|
],
|
|
outputs=[
|
|
Output(
|
|
value=50,
|
|
address=b"some_address"
|
|
)
|
|
]
|
|
)
|
|
self.wallet.sign(self.transaction, session)
|
|
|
|
def test_valid(self):
|
|
self.validator.add_transaction(self.transaction)
|
|
|
|
def test_reject_invalid_syntax(self):
|
|
self.transaction.inputs = []
|
|
self.assertRaisesRegex(InvalidTransactionException, "Input list is empty",
|
|
self.validator.add_transaction, self.transaction)
|
|
|
|
def test_reject_coinbase_transaction(self):
|
|
coinbase_transaction = CoinbaseTransaction(b"some_address")
|
|
self.assertRaisesRegex(InvalidTransactionException, "Standalone transaction cannot be coinbase",
|
|
self.validator.add_transaction, coinbase_transaction)
|
|
|
|
def test_reject_duplicate_in_blockchain(self):
|
|
with session_scope() as session:
|
|
block = Block(
|
|
hash_prev_block=self.block.hash,
|
|
target=self.block.target,
|
|
transactions=[
|
|
CoinbaseTransaction(
|
|
address=self.wallet.new_address(),
|
|
value=100
|
|
),
|
|
self.transaction
|
|
]
|
|
)
|
|
self.blockchain.add(block, session, main_branch=True)
|
|
self.assertRaisesRegex(InvalidTransactionException, "Transaction already exists in blockchain's main branch",
|
|
self.validator.add_transaction, self.transaction)
|
|
|
|
def test_reject_duplicate_mempool(self):
|
|
self.mempool[self.transaction.hash] = self.transaction
|
|
self.assertRaisesRegex(InvalidTransactionException, "Transaction already exists in mempool",
|
|
self.validator.add_transaction, self.transaction)
|
|
|
|
def test_reject_mempool_conflict(self):
|
|
# a transaction that spends the same input as self.transaction:
|
|
transaction = Transaction(
|
|
inputs=[
|
|
Input(
|
|
prev_tx_hash=self.block.transactions[0].hash,
|
|
txout_index=0
|
|
)
|
|
],
|
|
outputs=[
|
|
Output(
|
|
value=100,
|
|
address=b"another_address"
|
|
)
|
|
]
|
|
)
|
|
self.assertNotEqual(transaction.hash, self.transaction.hash)
|
|
self.assertEqual((self.transaction.inputs[0].prev_tx_hash, self.transaction.inputs[0].txout_index),
|
|
(transaction.inputs[0].prev_tx_hash, transaction.inputs[0].txout_index))
|
|
self.mempool[transaction.hash] = transaction
|
|
self.assertRaisesRegex(InvalidTransactionException, "Transaction conflicts with mempool: one or more input's referenced output is spent by another transaction in the mempool",
|
|
self.validator.add_transaction, self.transaction)
|
|
|
|
def test_reject_orphan(self):
|
|
input = Input(
|
|
prev_tx_hash=b"nonexistent_tx",
|
|
txout_index=123
|
|
)
|
|
self.transaction.inputs.append(input)
|
|
self.assertRaises(OrphanException, self.validator.add_transaction, self.transaction)
|
|
|
|
def test_reject_spent(self):
|
|
address = self.wallet.new_address()
|
|
private_key, public_key = self.wallet.keys[address]
|
|
block = Block(
|
|
hash_prev_block=self.block.hash,
|
|
target=self.block.target,
|
|
public_key=public_bytes(public_key),
|
|
transactions=[
|
|
CoinbaseTransaction(
|
|
address=address,
|
|
value=100,
|
|
block_height=2
|
|
)
|
|
] + [self.transaction]
|
|
)
|
|
self.validator.add_block(mine(block, private_key))
|
|
|
|
# a transaction that spends the same input as self.transaction:
|
|
transaction = Transaction(
|
|
inputs=[
|
|
Input(
|
|
prev_tx_hash=self.block.transactions[0].hash,
|
|
txout_index=0
|
|
)
|
|
],
|
|
outputs=[
|
|
Output(
|
|
value=20,
|
|
address=b"other_address"
|
|
)
|
|
]
|
|
)
|
|
|
|
self.assertRaisesRegex(InvalidTransactionException, "Referenced output already spent \(not in blockchain/mempool UTXO\) for one or more inputs",
|
|
self.validator.add_transaction, transaction)
|
|
|
|
def test_reject_negative_fee(self):
|
|
output = Output(
|
|
value=10000,
|
|
address=b"doesnt_matter"
|
|
)
|
|
self.transaction.outputs.append(output)
|
|
self.assertRaisesRegex(InvalidTransactionException, "Sum of input values < sum of output values \(negative fee\)",
|
|
self.validator.add_transaction, self.transaction)
|
|
|
|
def test_reject_too_low_fee(self):
|
|
with patch("aucoin.consensus.tx_min_fee", 51): # fee of self.transaction is 50
|
|
self.assertRaisesRegex(InvalidTransactionException, "Transaction fee too low",
|
|
self.validator.add_transaction, self.transaction)
|
|
|
|
def test_reject_invalid_signature(self):
|
|
self.transaction.inputs[0].signature = b"Wrong"
|
|
self.assertRaisesRegex(InvalidTransactionException, "Invalid signature for one or more inputs",
|
|
self.validator.add_transaction, self.transaction)
|
|
|
|
def test_reject_wrong_public_key(self):
|
|
with session_scope() as session:
|
|
for input, copy_hash in self.transaction.truncated_copies(self.blockchain, self.mempool, session):
|
|
keypair = dsa.generate_keypair() # generate new, wrong, keypair
|
|
input.public_key = public_bytes(keypair.public)
|
|
input.signature = dsa.sign(keypair.private, copy_hash)
|
|
|
|
self.assertRaisesRegex(InvalidTransactionException, "Public key doesn't match address of the output it is spending for one or more inputs",
|
|
self.validator.add_transaction, self.transaction)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|