Mirroristas/bigchaindb
2
0
Fork 0
mirror of https://github.com/bigchaindb/bigchaindb.git synced 2024-10-13 13:34:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use pytest fixtures instead of imports

Browse Source
This commit is contained in:
Sylvain Bellemare 2016-02-13 02:35:09 +01:00
parent 2db5197619
commit df242a35c6
3 changed files with 61 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tests/db/conftest.py
View File

@ -12,8 +12,6 @@ import rethinkdb as r
from bigchaindb import Bigchain
from bigchaindb.db import get_conn
from ..conftest import config, USER_PRIVATE_KEY, USER_PUBLIC_KEY
NOOP = None

118
tests/db/test_bigchain_api.py
View File

@ -13,7 +13,7 @@ from bigchaindb.voter import Voter
from bigchaindb.block import Block
import bigchaindb.config_utils
from .conftest import USER_PUBLIC_KEY, USER_PRIVATE_KEY
from ..conftest import USER_PUBLIC_KEY
def create_inputs(amount=1, b=None):
@ -86,11 +86,11 @@ class TestBigchainApi(object):
tx = b.create_transaction('a', 'b', 'c', 'd')
assert b.deserialize(b.serialize(tx)) == tx
def test_write_transaction(self, b):
def test_write_transaction(self, b, user_public_key, user_private_key):
create_inputs()
input_tx = b.get_owned_ids(USER_PUBLIC_KEY).pop()
tx = b.create_transaction(USER_PUBLIC_KEY, 'b', input_tx, 'd')
tx_signed = b.sign_transaction(tx, USER_PRIVATE_KEY)
input_tx = b.get_owned_ids(user_public_key).pop()
tx = b.create_transaction(user_public_key, 'b', input_tx, 'd')
tx_signed = b.sign_transaction(tx, user_private_key)
response = b.write_transaction(tx_signed)
assert response['skipped'] == 0
@ -100,11 +100,11 @@ class TestBigchainApi(object):
assert response['replaced'] == 0
assert response['inserted'] == 1
def test_read_transaction(self, b):
def test_read_transaction(self, b, user_public_key, user_private_key):
create_inputs()
input_tx = b.get_owned_ids(USER_PUBLIC_KEY).pop()
tx = b.create_transaction(USER_PUBLIC_KEY, 'b', input_tx, 'd')
tx_signed = b.sign_transaction(tx, USER_PRIVATE_KEY)
input_tx = b.get_owned_ids(user_public_key).pop()
tx = b.create_transaction(user_public_key, 'b', input_tx, 'd')
tx_signed = b.sign_transaction(tx, user_private_key)
b.write_transaction(tx_signed)
# create block and write it to the bighcain before retrieving the transaction
@ -114,11 +114,11 @@ class TestBigchainApi(object):
response = b.get_transaction(tx_signed["id"])
assert b.serialize(tx_signed) == b.serialize(response)
def test_assign_transaction_one_node(self, b):
def test_assign_transaction_one_node(self, b, user_public_key, user_private_key):
create_inputs()
input_tx = b.get_owned_ids(USER_PUBLIC_KEY).pop()
tx = b.create_transaction(USER_PUBLIC_KEY, 'b', input_tx, 'd')
tx_signed = b.sign_transaction(tx, USER_PRIVATE_KEY)
input_tx = b.get_owned_ids(user_public_key).pop()
tx = b.create_transaction(user_public_key, 'b', input_tx, 'd')
tx_signed = b.sign_transaction(tx, user_private_key)
b.write_transaction(tx_signed)
# retrieve the transaction
@ -127,7 +127,7 @@ class TestBigchainApi(object):
# check if the assignee is the current node
assert response['assignee'] == b.me
def test_assign_transaction_multiple_nodes(self, b):
def test_assign_transaction_multiple_nodes(self, b, user_public_key, user_private_key):
# create 5 federation nodes
for _ in range(5):
b.federation_nodes.append(b.generate_keys()[1])
@ -135,9 +135,9 @@ class TestBigchainApi(object):
# test assignee for several transactions
for _ in range(20):
input_tx = b.get_owned_ids(USER_PUBLIC_KEY).pop()
tx = b.create_transaction(USER_PUBLIC_KEY, 'b', input_tx, 'd')
tx_signed = b.sign_transaction(tx, USER_PRIVATE_KEY)
input_tx = b.get_owned_ids(user_public_key).pop()
tx = b.create_transaction(user_public_key, 'b', input_tx, 'd')
tx_signed = b.sign_transaction(tx, user_private_key)
b.write_transaction(tx_signed)
# retrieve the transaction
@ -280,9 +280,9 @@ class TestTransactionValidation(object):
assert excinfo.value.args[0] == 'input `c` does not exist in the bigchain'
assert b.is_valid_transaction(tx) == False
def test_non_create_valid_input_wrong_owner(self, b):
def test_non_create_valid_input_wrong_owner(self, b, user_public_key):
create_inputs()
valid_input = b.get_owned_ids(USER_PUBLIC_KEY).pop()
valid_input = b.get_owned_ids(user_public_key).pop()
tx = b.create_transaction('a', 'b', valid_input, 'c')
with pytest.raises(exceptions.TransactionOwnerError) as excinfo:
b.validate_transaction(tx)
@ -290,11 +290,11 @@ class TestTransactionValidation(object):
assert excinfo.value.args[0] == 'current_owner `a` does not own the input `{}`'.format(valid_input)
assert b.is_valid_transaction(tx) == False
def test_non_create_double_spend(self, b):
def test_non_create_double_spend(self, b, user_public_key, user_private_key):
create_inputs()
input_valid = b.get_owned_ids(USER_PUBLIC_KEY).pop()
tx_valid = b.create_transaction(USER_PUBLIC_KEY, 'b', input_valid, 'd')
tx_valid_signed = b.sign_transaction(tx_valid, USER_PRIVATE_KEY)
input_valid = b.get_owned_ids(user_public_key).pop()
tx_valid = b.create_transaction(user_public_key, 'b', input_valid, 'd')
tx_valid_signed = b.sign_transaction(tx_valid, user_private_key)
b.write_transaction(tx_valid_signed)
# create and write block to bigchain
@ -302,17 +302,17 @@ class TestTransactionValidation(object):
b.write_block(block, durability='hard')
# create another transaction with the same input
tx_double_spend = b.create_transaction(USER_PUBLIC_KEY, 'd', input_valid, 'd')
tx_double_spend = b.create_transaction(user_public_key, 'd', input_valid, 'd')
with pytest.raises(exceptions.DoubleSpend) as excinfo:
b.validate_transaction(tx_double_spend)
assert excinfo.value.args[0] == 'input `{}` was already spent'.format(input_valid)
assert b.is_valid_transaction(tx_double_spend) == False
def test_wrong_transaction_hash(self, b):
def test_wrong_transaction_hash(self, b, user_public_key):
create_inputs()
input_valid = b.get_owned_ids(USER_PUBLIC_KEY).pop()
tx_valid = b.create_transaction(USER_PUBLIC_KEY, 'b', input_valid, 'd')
input_valid = b.get_owned_ids(user_public_key).pop()
tx_valid = b.create_transaction(user_public_key, 'b', input_valid, 'd')
# change the transaction hash
tx_valid.update({'id': 'abcd'})
@ -320,10 +320,10 @@ class TestTransactionValidation(object):
b.validate_transaction(tx_valid)
assert b.is_valid_transaction(tx_valid) == False
def test_wrong_signature(self, b):
def test_wrong_signature(self, b, user_public_key):
create_inputs()
input_valid = b.get_owned_ids(USER_PUBLIC_KEY).pop()
tx_valid = b.create_transaction(USER_PUBLIC_KEY, 'b', input_valid, 'd')
input_valid = b.get_owned_ids(user_public_key).pop()
tx_valid = b.create_transaction(user_public_key, 'b', input_valid, 'd')
wrong_private_key = '4fyvJe1aw2qHZ4UNRYftXK7JU7zy9bCqoU5ps6Ne3xrY'
@ -332,18 +332,18 @@ class TestTransactionValidation(object):
b.validate_transaction(tx_invalid_signed)
assert b.is_valid_transaction(tx_invalid_signed) == False
def test_valid_create_transaction(self, b):
tx = b.create_transaction(b.me, USER_PUBLIC_KEY, None, 'CREATE')
def test_valid_create_transaction(self, b, user_public_key):
tx = b.create_transaction(b.me, user_public_key, None, 'CREATE')
tx_signed = b.sign_transaction(tx, b.me_private)
assert tx_signed == b.validate_transaction(tx_signed)
assert tx_signed == b.is_valid_transaction(tx_signed)
def test_valid_non_create_transaction(self, b):
def test_valid_non_create_transaction(self, b, user_public_key, user_private_key):
create_inputs()
input_valid = b.get_owned_ids(USER_PUBLIC_KEY).pop()
tx_valid = b.create_transaction(USER_PUBLIC_KEY, 'b', input_valid, 'd')
input_valid = b.get_owned_ids(user_public_key).pop()
tx_valid = b.create_transaction(user_public_key, 'b', input_valid, 'd')
tx_valid_signed = b.sign_transaction(tx_valid, USER_PRIVATE_KEY)
tx_valid_signed = b.sign_transaction(tx_valid, user_private_key)
assert tx_valid_signed == b.validate_transaction(tx_valid_signed)
assert tx_valid_signed == b.is_valid_transaction(tx_valid_signed)
@ -359,10 +359,10 @@ class TestBlockValidation(object):
b.validate_block(block)
@pytest.mark.skipif(reason='Separated tx validation from block creation.')
def test_invalid_transactions_in_block(self, b):
def test_invalid_transactions_in_block(self, b, user_public_key, ):
# invalid transaction
create_inputs()
valid_input = b.get_owned_ids(USER_PUBLIC_KEY).pop()
valid_input = b.get_owned_ids(user_public_key).pop()
tx_invalid = b.create_transaction('a', 'b', valid_input, 'c')
block = b.create_block([tx_invalid])
@ -400,12 +400,12 @@ class TestBlockValidation(object):
with pytest.raises(exceptions.InvalidHash):
b.validate_block(block)
def test_valid_block(self, b):
def test_valid_block(self, b, user_public_key, user_private_key):
create_inputs()
# create valid transaction
input_valid = b.get_owned_ids(USER_PUBLIC_KEY).pop()
tx_valid = b.create_transaction(USER_PUBLIC_KEY, 'b', input_valid, 'd')
tx_valid_signed = b.sign_transaction(tx_valid, USER_PRIVATE_KEY)
input_valid = b.get_owned_ids(user_public_key).pop()
tx_valid = b.create_transaction(user_public_key, 'b', input_valid, 'd')
tx_valid_signed = b.sign_transaction(tx_valid, user_private_key)
# create valid block
block = b.create_block([tx_valid_signed])
@ -491,13 +491,13 @@ class TestBigchainVoter(object):
assert vote['node_pubkey'] == b.me
assert PublicKey(b.me).verify(b.serialize(vote['vote']), vote['signature']) == True
def test_invalid_block_voting(self, b):
def test_invalid_block_voting(self, b, user_public_key):
# create queue and voter
q_new_block = mp.Queue()
voter = Voter(q_new_block)
# create transaction
transaction = b.create_transaction(b.me, USER_PUBLIC_KEY, None, 'CREATE')
transaction = b.create_transaction(b.me, user_public_key, None, 'CREATE')
transaction_signed = b.sign_transaction(transaction, b.me_private)
genesis = b.create_genesis_block()
@ -564,12 +564,12 @@ class TestBigchainVoter(object):
class TestBigchainBlock(object):
def test_by_assignee(self, b):
def test_by_assignee(self, b, user_public_key):
# create transactions and randomly assigne them
transactions = mp.Queue()
count_assigned_to_me = 0
for i in range(100):
tx = b.create_transaction(b.me, USER_PUBLIC_KEY, None, 'CREATE')
tx = b.create_transaction(b.me, user_public_key, None, 'CREATE')
assignee = random.choice([b.me, 'aaa', 'bbb', 'ccc'])
if assignee == b.me:
count_assigned_to_me += 1
@ -588,13 +588,13 @@ class TestBigchainBlock(object):
# the queue minus 'stop'
assert block.q_tx_to_validate.qsize() - 1 == count_assigned_to_me
def test_validate_transactions(self, b):
def test_validate_transactions(self, b, user_public_key):
# create transactions and randomly invalidate some of them by changing the hash
transactions = mp.Queue()
count_valid = 0
for i in range(100):
valid = random.choice([True, False])
tx = b.create_transaction(b.me, USER_PUBLIC_KEY, None, 'CREATE')
tx = b.create_transaction(b.me, user_public_key, None, 'CREATE')
tx = b.sign_transaction(tx, b.me_private)
if not valid:
tx['id'] = 'a' * 64
@ -613,11 +613,11 @@ class TestBigchainBlock(object):
assert block.q_tx_validated.qsize() - 1 == count_valid
assert block.q_tx_delete.qsize() - 1 == 100
def test_create_block(self, b):
def test_create_block(self, b, user_public_key):
# create transactions
transactions = mp.Queue()
for i in range(100):
tx = b.create_transaction(b.me, USER_PUBLIC_KEY, None, 'CREATE')
tx = b.create_transaction(b.me, user_public_key, None, 'CREATE')
tx = b.sign_transaction(tx, b.me_private)
transactions.put(tx)
transactions.put('stop')
@ -631,12 +631,12 @@ class TestBigchainBlock(object):
# check if the number of valid transactions
assert block.q_block.qsize() - 1 == 1
def test_write_block(self, b):
def test_write_block(self, b, user_public_key):
# create transactions
transactions = []
blocks = mp.Queue()
for i in range(100):
tx = b.create_transaction(b.me, USER_PUBLIC_KEY, None, 'CREATE')
tx = b.create_transaction(b.me, user_public_key, None, 'CREATE')
tx = b.sign_transaction(tx, b.me_private)
transactions.append(tx)
@ -661,14 +661,14 @@ class TestBigchainBlock(object):
# check if the number of blocks in bigchain increased
assert r.table('bigchain').count() == 2
def test_delete_transactions(self, b):
def test_delete_transactions(self, b, user_public_key):
# make sure that there are no transactions in the backlog
r.table('backlog').delete().run(b.conn)
# create and write transactions to the backlog
transactions = mp.Queue()
for i in range(100):
tx = b.create_transaction(b.me, USER_PUBLIC_KEY, None, 'CREATE')
tx = b.create_transaction(b.me, user_public_key, None, 'CREATE')
tx = b.sign_transaction(tx, b.me_private)
b.write_transaction(tx)
transactions.put(tx['id'])
@ -689,13 +689,13 @@ class TestBigchainBlock(object):
# check if all transactions were deleted from the backlog
assert r.table('backlog').count() == 0
def test_bootstrap(self, b):
def test_bootstrap(self, b, user_public_key):
# make sure that there are no transactions in the backlog
r.table('backlog').delete().run(b.conn)
# create and write transactions to the backlog
for i in range(100):
tx = b.create_transaction(b.me, USER_PUBLIC_KEY, None, 'CREATE')
tx = b.create_transaction(b.me, user_public_key, None, 'CREATE')
tx = b.sign_transaction(tx, b.me_private)
b.write_transaction(tx)
@ -708,7 +708,7 @@ class TestBigchainBlock(object):
# we should have gotten a queue with 100 results
assert initial_results.qsize() - 1 == 100
def test_start(self, b):
def test_start(self, b, user_public_key):
# start with 100 transactions in the backlog and 100 in the changefeed
# make sure that there are no transactions in the backlog
@ -716,14 +716,14 @@ class TestBigchainBlock(object):
# create and write transactions to the backlog
for i in range(100):
tx = b.create_transaction(b.me, USER_PUBLIC_KEY, None, 'CREATE')
tx = b.create_transaction(b.me, user_public_key, None, 'CREATE')
tx = b.sign_transaction(tx, b.me_private)
b.write_transaction(tx)
# create 100 more transactions to emulate the changefeed
new_transactions = mp.Queue()
for i in range(100):
tx = b.create_transaction(b.me, USER_PUBLIC_KEY, None, 'CREATE')
tx = b.create_transaction(b.me, user_public_key, None, 'CREATE')
tx = b.sign_transaction(tx, b.me_private)
b.write_transaction(tx)
new_transactions.put(tx)

6
tests/db/test_voter.py
View File

@ -7,8 +7,6 @@ from bigchaindb import Bigchain
from bigchaindb.voter import Voter, BlockStream
from bigchaindb.crypto import PublicKey
from .conftest import USER_PUBLIC_KEY
class TestBigchainVoter(object):
@ -50,13 +48,13 @@ class TestBigchainVoter(object):
assert PublicKey(b.me).verify(b.serialize(vote['vote']), vote['signature']) == True
def test_invalid_block_voting(self, b):
def test_invalid_block_voting(self, b, user_public_key):
# create queue and voter
q_new_block = mp.Queue()
voter = Voter(q_new_block)
# create transaction
transaction = b.create_transaction(b.me, USER_PUBLIC_KEY, None, 'CREATE')
transaction = b.create_transaction(b.me, user_public_key, None, 'CREATE')
transaction_signed = b.sign_transaction(transaction, b.me_private)
genesis = b.create_genesis_block()