mirror of
https://github.com/bigchaindb/bigchaindb.git
synced 2024-10-13 13:34:05 +00:00
1089 lines
37 KiB
Python
1089 lines
37 KiB
Python
from pytest import raises, mark
|
|
from unittest.mock import patch
|
|
|
|
|
|
def test_fulfillment_serialization(ffill_uri, user_pub):
|
|
from bigchaindb.common.transaction import Fulfillment
|
|
from cryptoconditions import Fulfillment as CCFulfillment
|
|
|
|
expected = {
|
|
'owners_before': [user_pub],
|
|
'fulfillment': ffill_uri,
|
|
'input': None,
|
|
}
|
|
ffill = Fulfillment(CCFulfillment.from_uri(ffill_uri), [user_pub])
|
|
assert ffill.to_dict() == expected
|
|
|
|
|
|
def test_fulfillment_deserialization_with_uri(ffill_uri, user_pub):
|
|
from bigchaindb.common.transaction import Fulfillment
|
|
from cryptoconditions import Fulfillment as CCFulfillment
|
|
|
|
expected = Fulfillment(CCFulfillment.from_uri(ffill_uri), [user_pub])
|
|
ffill = {
|
|
'owners_before': [user_pub],
|
|
'fulfillment': ffill_uri,
|
|
'input': None,
|
|
}
|
|
ffill = Fulfillment.from_dict(ffill)
|
|
|
|
assert ffill == expected
|
|
|
|
|
|
def test_fulfillment_deserialization_with_invalid_fulfillment(user_pub):
|
|
from bigchaindb.common.transaction import Fulfillment
|
|
|
|
ffill = {
|
|
'owners_before': [user_pub],
|
|
'fulfillment': None,
|
|
'input': None,
|
|
}
|
|
with raises(TypeError):
|
|
Fulfillment.from_dict(ffill)
|
|
|
|
|
|
def test_fulfillment_deserialization_with_invalid_fulfillment_uri(user_pub):
|
|
from bigchaindb.common.exceptions import InvalidSignature
|
|
from bigchaindb.common.transaction import Fulfillment
|
|
|
|
ffill = {
|
|
'owners_before': [user_pub],
|
|
'fulfillment': 'an invalid fulfillment',
|
|
'input': None,
|
|
}
|
|
with raises(InvalidSignature):
|
|
Fulfillment.from_dict(ffill)
|
|
|
|
|
|
def test_fulfillment_deserialization_with_unsigned_fulfillment(ffill_uri,
|
|
user_pub):
|
|
from bigchaindb.common.transaction import Fulfillment
|
|
from cryptoconditions import Fulfillment as CCFulfillment
|
|
|
|
expected = Fulfillment(CCFulfillment.from_uri(ffill_uri), [user_pub])
|
|
ffill = {
|
|
'owners_before': [user_pub],
|
|
'fulfillment': CCFulfillment.from_uri(ffill_uri),
|
|
'input': None,
|
|
}
|
|
ffill = Fulfillment.from_dict(ffill)
|
|
|
|
assert ffill == expected
|
|
|
|
|
|
def test_condition_serialization(user_Ed25519, user_pub):
|
|
from bigchaindb.common.transaction import Condition
|
|
|
|
expected = {
|
|
'condition': {
|
|
'uri': user_Ed25519.condition_uri,
|
|
'details': user_Ed25519.to_dict(),
|
|
},
|
|
'owners_after': [user_pub],
|
|
'amount': 1,
|
|
}
|
|
|
|
cond = Condition(user_Ed25519, [user_pub], 1)
|
|
|
|
assert cond.to_dict() == expected
|
|
|
|
|
|
def test_condition_deserialization(user_Ed25519, user_pub):
|
|
from bigchaindb.common.transaction import Condition
|
|
|
|
expected = Condition(user_Ed25519, [user_pub], 1)
|
|
cond = {
|
|
'condition': {
|
|
'uri': user_Ed25519.condition_uri,
|
|
'details': user_Ed25519.to_dict()
|
|
},
|
|
'owners_after': [user_pub],
|
|
'amount': 1,
|
|
}
|
|
cond = Condition.from_dict(cond)
|
|
|
|
assert cond == expected
|
|
|
|
|
|
def test_condition_hashlock_serialization():
|
|
from bigchaindb.common.transaction import Condition
|
|
from cryptoconditions import PreimageSha256Fulfillment
|
|
|
|
secret = b'wow much secret'
|
|
hashlock = PreimageSha256Fulfillment(preimage=secret).condition_uri
|
|
|
|
expected = {
|
|
'condition': {
|
|
'uri': hashlock,
|
|
},
|
|
'owners_after': None,
|
|
'amount': 1,
|
|
}
|
|
cond = Condition(hashlock, amount=1)
|
|
|
|
assert cond.to_dict() == expected
|
|
|
|
|
|
def test_condition_hashlock_deserialization():
|
|
from bigchaindb.common.transaction import Condition
|
|
from cryptoconditions import PreimageSha256Fulfillment
|
|
|
|
secret = b'wow much secret'
|
|
hashlock = PreimageSha256Fulfillment(preimage=secret).condition_uri
|
|
expected = Condition(hashlock, amount=1)
|
|
|
|
cond = {
|
|
'condition': {
|
|
'uri': hashlock
|
|
},
|
|
'owners_after': None,
|
|
'amount': 1,
|
|
}
|
|
cond = Condition.from_dict(cond)
|
|
|
|
assert cond == expected
|
|
|
|
|
|
def test_invalid_condition_initialization(cond_uri, user_pub):
|
|
from bigchaindb.common.transaction import Condition
|
|
|
|
with raises(TypeError):
|
|
Condition(cond_uri, user_pub)
|
|
|
|
|
|
def test_generate_conditions_split_half_recursive(user_pub, user2_pub,
|
|
user3_pub):
|
|
from bigchaindb.common.transaction import Condition
|
|
from cryptoconditions import Ed25519Fulfillment, ThresholdSha256Fulfillment
|
|
|
|
expected_simple1 = Ed25519Fulfillment(public_key=user_pub)
|
|
expected_simple2 = Ed25519Fulfillment(public_key=user2_pub)
|
|
expected_simple3 = Ed25519Fulfillment(public_key=user3_pub)
|
|
|
|
expected = ThresholdSha256Fulfillment(threshold=2)
|
|
expected.add_subfulfillment(expected_simple1)
|
|
expected_threshold = ThresholdSha256Fulfillment(threshold=2)
|
|
expected_threshold.add_subfulfillment(expected_simple2)
|
|
expected_threshold.add_subfulfillment(expected_simple3)
|
|
expected.add_subfulfillment(expected_threshold)
|
|
|
|
cond = Condition.generate([user_pub, [user2_pub, expected_simple3]], 1)
|
|
assert cond.fulfillment.to_dict() == expected.to_dict()
|
|
|
|
|
|
def test_generate_conditions_split_half_single_owner(user_pub, user2_pub,
|
|
user3_pub):
|
|
from bigchaindb.common.transaction import Condition
|
|
from cryptoconditions import Ed25519Fulfillment, ThresholdSha256Fulfillment
|
|
|
|
expected_simple1 = Ed25519Fulfillment(public_key=user_pub)
|
|
expected_simple2 = Ed25519Fulfillment(public_key=user2_pub)
|
|
expected_simple3 = Ed25519Fulfillment(public_key=user3_pub)
|
|
|
|
expected = ThresholdSha256Fulfillment(threshold=2)
|
|
expected_threshold = ThresholdSha256Fulfillment(threshold=2)
|
|
expected_threshold.add_subfulfillment(expected_simple2)
|
|
expected_threshold.add_subfulfillment(expected_simple3)
|
|
expected.add_subfulfillment(expected_threshold)
|
|
expected.add_subfulfillment(expected_simple1)
|
|
|
|
cond = Condition.generate([[expected_simple2, user3_pub], user_pub], 1)
|
|
assert cond.fulfillment.to_dict() == expected.to_dict()
|
|
|
|
|
|
def test_generate_conditions_flat_ownage(user_pub, user2_pub, user3_pub):
|
|
from bigchaindb.common.transaction import Condition
|
|
from cryptoconditions import Ed25519Fulfillment, ThresholdSha256Fulfillment
|
|
|
|
expected_simple1 = Ed25519Fulfillment(public_key=user_pub)
|
|
expected_simple2 = Ed25519Fulfillment(public_key=user2_pub)
|
|
expected_simple3 = Ed25519Fulfillment(public_key=user3_pub)
|
|
|
|
expected = ThresholdSha256Fulfillment(threshold=3)
|
|
expected.add_subfulfillment(expected_simple1)
|
|
expected.add_subfulfillment(expected_simple2)
|
|
expected.add_subfulfillment(expected_simple3)
|
|
|
|
cond = Condition.generate([user_pub, user2_pub, expected_simple3], 1)
|
|
assert cond.fulfillment.to_dict() == expected.to_dict()
|
|
|
|
|
|
def test_generate_conditions_single_owner(user_pub):
|
|
from bigchaindb.common.transaction import Condition
|
|
from cryptoconditions import Ed25519Fulfillment
|
|
|
|
expected = Ed25519Fulfillment(public_key=user_pub)
|
|
cond = Condition.generate([user_pub], 1)
|
|
|
|
assert cond.fulfillment.to_dict() == expected.to_dict()
|
|
|
|
|
|
def test_generate_conditions_single_owner_with_condition(user_pub):
|
|
from bigchaindb.common.transaction import Condition
|
|
from cryptoconditions import Ed25519Fulfillment
|
|
|
|
expected = Ed25519Fulfillment(public_key=user_pub)
|
|
cond = Condition.generate([expected], 1)
|
|
|
|
assert cond.fulfillment.to_dict() == expected.to_dict()
|
|
|
|
|
|
# TODO FOR CC: see skip reason
|
|
@mark.skip(reason='threshold(hashlock).to_dict() exposes secret')
|
|
def test_generate_threshold_condition_with_hashlock(user_pub, user2_pub,
|
|
user3_pub):
|
|
from bigchaindb.common.transaction import Condition
|
|
from cryptoconditions import (PreimageSha256Fulfillment,
|
|
Ed25519Fulfillment,
|
|
ThresholdSha256Fulfillment)
|
|
|
|
secret = b'much secret, wow'
|
|
hashlock = PreimageSha256Fulfillment(preimage=secret)
|
|
|
|
expected_simple1 = Ed25519Fulfillment(public_key=user_pub)
|
|
expected_simple3 = Ed25519Fulfillment(public_key=user3_pub)
|
|
|
|
expected = ThresholdSha256Fulfillment(threshold=2)
|
|
expected_sub = ThresholdSha256Fulfillment(threshold=2)
|
|
expected_sub.add_subfulfillment(expected_simple1)
|
|
expected_sub.add_subfulfillment(hashlock)
|
|
expected.add_subfulfillment(expected_simple3)
|
|
|
|
cond = Condition.generate([[user_pub, hashlock], expected_simple3], 1)
|
|
assert cond.fulfillment.to_dict() == expected.to_dict()
|
|
|
|
|
|
def test_generate_conditions_invalid_parameters(user_pub, user2_pub,
|
|
user3_pub):
|
|
from bigchaindb.common.transaction import Condition
|
|
|
|
with raises(ValueError):
|
|
Condition.generate([], 1)
|
|
with raises(TypeError):
|
|
Condition.generate('not a list', 1)
|
|
with raises(ValueError):
|
|
Condition.generate([[user_pub, [user2_pub, [user3_pub]]]], 1)
|
|
with raises(ValueError):
|
|
Condition.generate([[user_pub]], 1)
|
|
|
|
|
|
def test_invalid_transaction_initialization():
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
with raises(ValueError):
|
|
Transaction(operation='invalid operation', asset=Asset())
|
|
with raises(TypeError):
|
|
Transaction(operation='CREATE', asset='invalid asset')
|
|
with raises(TypeError):
|
|
Transaction(
|
|
operation='CREATE',
|
|
asset=Asset(),
|
|
conditions='invalid conditions'
|
|
)
|
|
with raises(TypeError):
|
|
Transaction(
|
|
operation='CREATE',
|
|
asset=Asset(),
|
|
conditions=[],
|
|
fulfillments='invalid fulfillments'
|
|
)
|
|
with raises(TypeError):
|
|
Transaction(
|
|
operation='CREATE',
|
|
asset=Asset(),
|
|
conditions=[],
|
|
fulfillments=[],
|
|
metadata='invalid metadata'
|
|
)
|
|
|
|
|
|
def test_create_default_asset_on_tx_initialization():
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
with patch.object(Asset, '_validate_asset', return_value=None):
|
|
tx = Transaction(Transaction.CREATE, None)
|
|
expected = Asset()
|
|
asset = tx.asset
|
|
|
|
expected.data_id = None
|
|
asset.data_id = None
|
|
assert asset == expected
|
|
|
|
|
|
def test_transaction_serialization(user_ffill, user_cond, data, data_id):
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
tx_id = 'l0l'
|
|
timestamp = '66666666666'
|
|
|
|
expected = {
|
|
'id': tx_id,
|
|
'version': Transaction.VERSION,
|
|
'transaction': {
|
|
# NOTE: This test assumes that Fulfillments and Conditions can
|
|
# successfully be serialized
|
|
'fulfillments': [user_ffill.to_dict(0)],
|
|
'conditions': [user_cond.to_dict(0)],
|
|
'operation': Transaction.CREATE,
|
|
'timestamp': timestamp,
|
|
'metadata': None,
|
|
'asset': {
|
|
'id': data_id,
|
|
'divisible': False,
|
|
'updatable': False,
|
|
'refillable': False,
|
|
'data': data,
|
|
}
|
|
}
|
|
}
|
|
|
|
tx = Transaction(Transaction.CREATE, Asset(data, data_id), [user_ffill],
|
|
[user_cond])
|
|
tx_dict = tx.to_dict()
|
|
tx_dict['id'] = tx_id
|
|
tx_dict['transaction']['asset']['id'] = data_id
|
|
tx_dict['transaction']['timestamp'] = timestamp
|
|
|
|
assert tx_dict == expected
|
|
|
|
|
|
def test_transaction_deserialization(user_ffill, user_cond, data, data_id):
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
timestamp = '66666666666'
|
|
|
|
expected_asset = Asset(data, data_id)
|
|
expected = Transaction(Transaction.CREATE, expected_asset, [user_ffill],
|
|
[user_cond], None, timestamp, Transaction.VERSION)
|
|
|
|
tx = {
|
|
'version': Transaction.VERSION,
|
|
'transaction': {
|
|
# NOTE: This test assumes that Fulfillments and Conditions can
|
|
# successfully be serialized
|
|
'fulfillments': [user_ffill.to_dict()],
|
|
'conditions': [user_cond.to_dict()],
|
|
'operation': Transaction.CREATE,
|
|
'timestamp': timestamp,
|
|
'metadata': None,
|
|
'asset': {
|
|
'id': data_id,
|
|
'divisible': False,
|
|
'updatable': False,
|
|
'refillable': False,
|
|
'data': data,
|
|
}
|
|
}
|
|
}
|
|
tx_no_signatures = Transaction._remove_signatures(tx)
|
|
tx['id'] = Transaction._to_hash(Transaction._to_str(tx_no_signatures))
|
|
tx = Transaction.from_dict(tx)
|
|
|
|
assert tx == expected
|
|
|
|
|
|
def test_tx_serialization_with_incorrect_hash(utx):
|
|
from bigchaindb.common.transaction import Transaction
|
|
from bigchaindb.common.exceptions import InvalidHash
|
|
|
|
utx_dict = utx.to_dict()
|
|
utx_dict['id'] = 'abc'
|
|
with raises(InvalidHash):
|
|
Transaction.from_dict(utx_dict)
|
|
utx_dict.pop('id')
|
|
with raises(InvalidHash):
|
|
Transaction.from_dict(utx_dict)
|
|
utx_dict['id'] = []
|
|
with raises(InvalidHash):
|
|
Transaction.from_dict(utx_dict)
|
|
|
|
|
|
def test_invalid_fulfillment_initialization(user_ffill, user_pub):
|
|
from bigchaindb.common.transaction import Fulfillment
|
|
|
|
with raises(TypeError):
|
|
Fulfillment(user_ffill, user_pub)
|
|
with raises(TypeError):
|
|
Fulfillment(user_ffill, [], tx_input='somethingthatiswrong')
|
|
|
|
|
|
def test_invalid_metadata_initialization():
|
|
from bigchaindb.common.transaction import Metadata
|
|
|
|
with raises(TypeError):
|
|
Metadata([])
|
|
|
|
|
|
def test_metadata_serialization(data, data_id):
|
|
from bigchaindb.common.transaction import Metadata
|
|
|
|
expected = {
|
|
'data': data,
|
|
'id': data_id,
|
|
}
|
|
metadata = Metadata(data, data_id)
|
|
|
|
assert metadata.to_dict() == expected
|
|
|
|
|
|
def test_metadata_deserialization(data, data_id):
|
|
from bigchaindb.common.transaction import Metadata
|
|
|
|
expected = Metadata(data, data_id)
|
|
metadata = Metadata.from_dict({'data': data, 'id': data_id})
|
|
|
|
assert metadata == expected
|
|
|
|
|
|
def test_transaction_link_serialization():
|
|
from bigchaindb.common.transaction import TransactionLink
|
|
|
|
tx_id = 'a transaction id'
|
|
expected = {
|
|
'txid': tx_id,
|
|
'cid': 0,
|
|
}
|
|
tx_link = TransactionLink(tx_id, 0)
|
|
|
|
assert tx_link.to_dict() == expected
|
|
|
|
|
|
def test_transaction_link_serialization_with_empty_payload():
|
|
from bigchaindb.common.transaction import TransactionLink
|
|
|
|
expected = None
|
|
tx_link = TransactionLink()
|
|
|
|
assert tx_link.to_dict() == expected
|
|
|
|
|
|
def test_transaction_link_deserialization():
|
|
from bigchaindb.common.transaction import TransactionLink
|
|
|
|
tx_id = 'a transaction id'
|
|
expected = TransactionLink(tx_id, 0)
|
|
tx_link = {
|
|
'txid': tx_id,
|
|
'cid': 0,
|
|
}
|
|
tx_link = TransactionLink.from_dict(tx_link)
|
|
|
|
assert tx_link == expected
|
|
|
|
|
|
def test_transaction_link_deserialization_with_empty_payload():
|
|
from bigchaindb.common.transaction import TransactionLink
|
|
|
|
expected = TransactionLink()
|
|
tx_link = TransactionLink.from_dict(None)
|
|
|
|
assert tx_link == expected
|
|
|
|
|
|
def test_cast_transaction_link_to_boolean():
|
|
from bigchaindb.common.transaction import TransactionLink
|
|
|
|
assert bool(TransactionLink()) is False
|
|
assert bool(TransactionLink('a', None)) is False
|
|
assert bool(TransactionLink(None, 'b')) is False
|
|
assert bool(TransactionLink('a', 'b')) is True
|
|
assert bool(TransactionLink(False, False)) is True
|
|
|
|
|
|
def test_add_fulfillment_to_tx(user_ffill):
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
with patch.object(Asset, '_validate_asset', return_value=None):
|
|
tx = Transaction(Transaction.CREATE, Asset(), [], [])
|
|
tx.add_fulfillment(user_ffill)
|
|
|
|
assert len(tx.fulfillments) == 1
|
|
|
|
|
|
def test_add_fulfillment_to_tx_with_invalid_parameters():
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
with patch.object(Asset, '_validate_asset', return_value=None):
|
|
tx = Transaction(Transaction.CREATE, Asset())
|
|
with raises(TypeError):
|
|
tx.add_fulfillment('somewronginput')
|
|
|
|
|
|
def test_add_condition_to_tx(user_cond):
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
with patch.object(Asset, '_validate_asset', return_value=None):
|
|
tx = Transaction(Transaction.CREATE, Asset())
|
|
tx.add_condition(user_cond)
|
|
|
|
assert len(tx.conditions) == 1
|
|
|
|
|
|
def test_add_condition_to_tx_with_invalid_parameters():
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
with patch.object(Asset, '_validate_asset', return_value=None):
|
|
tx = Transaction(Transaction.CREATE, Asset(), [], [])
|
|
with raises(TypeError):
|
|
tx.add_condition('somewronginput')
|
|
|
|
|
|
def test_sign_with_invalid_parameters(utx, user_priv):
|
|
with raises(TypeError):
|
|
utx.sign(None)
|
|
with raises(TypeError):
|
|
utx.sign(user_priv)
|
|
|
|
|
|
def test_validate_tx_simple_create_signature(user_ffill, user_cond, user_priv):
|
|
from copy import deepcopy
|
|
from bigchaindb.common.crypto import SigningKey
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
tx = Transaction(Transaction.CREATE, Asset(), [user_ffill], [user_cond])
|
|
expected = deepcopy(user_cond)
|
|
expected.fulfillment.sign(str(tx).encode(), SigningKey(user_priv))
|
|
tx.sign([user_priv])
|
|
|
|
assert tx.fulfillments[0].to_dict()['fulfillment'] == \
|
|
expected.fulfillment.serialize_uri()
|
|
assert tx.fulfillments_valid() is True
|
|
|
|
|
|
def test_invoke_simple_signature_fulfillment_with_invalid_params(utx,
|
|
user_ffill):
|
|
from bigchaindb.common.exceptions import KeypairMismatchException
|
|
|
|
with raises(KeypairMismatchException):
|
|
invalid_key_pair = {'wrong_pub_key': 'wrong_priv_key'}
|
|
utx._sign_simple_signature_fulfillment(user_ffill,
|
|
0,
|
|
'somemessage',
|
|
invalid_key_pair)
|
|
|
|
|
|
def test_sign_threshold_with_invalid_params(utx, user_user2_threshold_ffill,
|
|
user3_pub, user3_priv):
|
|
from bigchaindb.common.exceptions import KeypairMismatchException
|
|
|
|
with raises(KeypairMismatchException):
|
|
utx._sign_threshold_signature_fulfillment(user_user2_threshold_ffill,
|
|
0,
|
|
'somemessage',
|
|
{user3_pub: user3_priv})
|
|
with raises(KeypairMismatchException):
|
|
user_user2_threshold_ffill.owners_before = ['somewrongvalue']
|
|
utx._sign_threshold_signature_fulfillment(user_user2_threshold_ffill,
|
|
0,
|
|
'somemessage',
|
|
None)
|
|
|
|
|
|
def test_validate_fulfillment_with_invalid_parameters(utx):
|
|
from bigchaindb.common.transaction import Transaction
|
|
|
|
input_conditions = [cond.fulfillment.condition_uri for cond
|
|
in utx.conditions]
|
|
tx_dict = utx.to_dict()
|
|
tx_dict = Transaction._remove_signatures(tx_dict)
|
|
tx_serialized = Transaction._to_str(tx_dict)
|
|
assert utx._fulfillment_valid(utx.fulfillments[0],
|
|
tx_serialized,
|
|
input_conditions) is False
|
|
|
|
|
|
def test_validate_multiple_fulfillments(user_ffill, user_cond, user_priv):
|
|
from copy import deepcopy
|
|
|
|
from bigchaindb.common.crypto import SigningKey
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
# TODO: Why is there a fulfillment in the conditions list
|
|
tx = Transaction(Transaction.CREATE, Asset(divisible=True),
|
|
[user_ffill, deepcopy(user_ffill)],
|
|
[user_cond, deepcopy(user_cond)])
|
|
|
|
expected_first = deepcopy(tx)
|
|
expected_second = deepcopy(tx)
|
|
expected_first.fulfillments = [expected_first.fulfillments[0]]
|
|
expected_first.conditions = expected_first.conditions
|
|
expected_second.fulfillments = [expected_second.fulfillments[1]]
|
|
expected_second.conditions = expected_second.conditions
|
|
|
|
expected_first_bytes = str(expected_first).encode()
|
|
expected_first.fulfillments[0].fulfillment.sign(expected_first_bytes,
|
|
SigningKey(user_priv))
|
|
expected_second_bytes = str(expected_second).encode()
|
|
expected_second.fulfillments[0].fulfillment.sign(expected_second_bytes,
|
|
SigningKey(user_priv))
|
|
tx.sign([user_priv])
|
|
|
|
assert tx.fulfillments[0].to_dict()['fulfillment'] == \
|
|
expected_first.fulfillments[0].fulfillment.serialize_uri()
|
|
assert tx.fulfillments[1].to_dict()['fulfillment'] == \
|
|
expected_second.fulfillments[0].fulfillment.serialize_uri()
|
|
assert tx.fulfillments_valid() is True
|
|
|
|
|
|
def test_validate_tx_threshold_create_signature(user_user2_threshold_ffill,
|
|
user_user2_threshold_cond,
|
|
user_pub,
|
|
user2_pub,
|
|
user_priv,
|
|
user2_priv):
|
|
from copy import deepcopy
|
|
|
|
from bigchaindb.common.crypto import SigningKey
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
tx = Transaction(Transaction.CREATE, Asset(), [user_user2_threshold_ffill],
|
|
[user_user2_threshold_cond])
|
|
expected = deepcopy(user_user2_threshold_cond)
|
|
expected.fulfillment.subconditions[0]['body'].sign(str(tx).encode(),
|
|
SigningKey(user_priv))
|
|
expected.fulfillment.subconditions[1]['body'].sign(str(tx).encode(),
|
|
SigningKey(user2_priv))
|
|
tx.sign([user_priv, user2_priv])
|
|
|
|
assert tx.fulfillments[0].to_dict()['fulfillment'] == \
|
|
expected.fulfillment.serialize_uri()
|
|
assert tx.fulfillments_valid() is True
|
|
|
|
|
|
def test_multiple_fulfillment_validation_of_transfer_tx(user_ffill, user_cond,
|
|
user_priv, user2_pub,
|
|
user2_priv, user3_pub,
|
|
user3_priv):
|
|
from copy import deepcopy
|
|
from bigchaindb.common.transaction import (Transaction, TransactionLink,
|
|
Fulfillment, Condition, Asset)
|
|
from cryptoconditions import Ed25519Fulfillment
|
|
|
|
tx = Transaction(Transaction.CREATE, Asset(divisible=True),
|
|
[user_ffill, deepcopy(user_ffill)],
|
|
[user_cond, deepcopy(user_cond)])
|
|
tx.sign([user_priv])
|
|
|
|
fulfillments = [Fulfillment(cond.fulfillment, cond.owners_after,
|
|
TransactionLink(tx.id, index))
|
|
for index, cond in enumerate(tx.conditions)]
|
|
conditions = [Condition(Ed25519Fulfillment(public_key=user3_pub),
|
|
[user3_pub]),
|
|
Condition(Ed25519Fulfillment(public_key=user3_pub),
|
|
[user3_pub])]
|
|
transfer_tx = Transaction('TRANSFER', tx.asset, fulfillments, conditions)
|
|
transfer_tx = transfer_tx.sign([user_priv])
|
|
|
|
assert transfer_tx.fulfillments_valid(tx.conditions) is True
|
|
|
|
|
|
def test_validate_fulfillments_of_transfer_tx_with_invalid_params(transfer_tx,
|
|
cond_uri,
|
|
utx,
|
|
user2_pub,
|
|
user_priv):
|
|
from bigchaindb.common.transaction import Condition
|
|
from cryptoconditions import Ed25519Fulfillment
|
|
|
|
invalid_cond = Condition(Ed25519Fulfillment.from_uri('cf:0:'), ['invalid'])
|
|
assert transfer_tx.fulfillments_valid([invalid_cond]) is False
|
|
invalid_cond = utx.conditions[0]
|
|
invalid_cond.owners_after = 'invalid'
|
|
assert transfer_tx.fulfillments_valid([invalid_cond]) is True
|
|
|
|
with raises(TypeError):
|
|
assert transfer_tx.fulfillments_valid(None) is False
|
|
with raises(AttributeError):
|
|
transfer_tx.fulfillments_valid('not a list')
|
|
with raises(ValueError):
|
|
transfer_tx.fulfillments_valid([])
|
|
with raises(TypeError):
|
|
transfer_tx.operation = "Operation that doesn't exist"
|
|
transfer_tx.fulfillments_valid([utx.conditions[0]])
|
|
|
|
|
|
def test_create_create_transaction_single_io(user_cond, user_pub, data,
|
|
data_id):
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
expected = {
|
|
'transaction': {
|
|
'conditions': [user_cond.to_dict(0)],
|
|
'metadata': {
|
|
'data': data,
|
|
},
|
|
'asset': {
|
|
'id': data_id,
|
|
'divisible': False,
|
|
'updatable': False,
|
|
'refillable': False,
|
|
'data': data,
|
|
},
|
|
'fulfillments': [
|
|
{
|
|
'owners_before': [
|
|
user_pub
|
|
],
|
|
'fid': 0,
|
|
'fulfillment': None,
|
|
'input': None
|
|
}
|
|
],
|
|
'operation': 'CREATE',
|
|
},
|
|
'version': 1
|
|
}
|
|
|
|
asset = Asset(data, data_id)
|
|
tx = Transaction.create([user_pub], [([user_pub], 1)],
|
|
data, asset).to_dict()
|
|
tx.pop('id')
|
|
tx['transaction']['metadata'].pop('id')
|
|
tx['transaction'].pop('timestamp')
|
|
tx['transaction']['fulfillments'][0]['fulfillment'] = None
|
|
|
|
assert tx == expected
|
|
|
|
|
|
def test_validate_single_io_create_transaction(user_pub, user_priv, data):
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
tx = Transaction.create([user_pub], [([user_pub], 1)], data, Asset())
|
|
tx = tx.sign([user_priv])
|
|
assert tx.fulfillments_valid() is True
|
|
|
|
|
|
def test_create_create_transaction_multiple_io(user_cond, user2_cond, user_pub,
|
|
user2_pub):
|
|
from bigchaindb.common.transaction import Transaction, Asset, Fulfillment
|
|
|
|
# a fulfillment for a create transaction with multiple `owners_before`
|
|
# is a fulfillment for an implicit threshold condition with
|
|
# weight = len(owners_before)
|
|
ffill = Fulfillment.generate([user_pub, user2_pub]).to_dict()
|
|
ffill.update({'fid': 0})
|
|
expected = {
|
|
'transaction': {
|
|
'conditions': [user_cond.to_dict(0), user2_cond.to_dict(1)],
|
|
'metadata': {
|
|
'data': {
|
|
'message': 'hello'
|
|
}
|
|
},
|
|
'fulfillments': [ffill],
|
|
'operation': 'CREATE',
|
|
},
|
|
'version': 1
|
|
}
|
|
asset = Asset(divisible=True)
|
|
tx = Transaction.create([user_pub, user2_pub],
|
|
[([user_pub], 1), ([user2_pub], 1)],
|
|
asset=asset,
|
|
metadata={'message': 'hello'}).to_dict()
|
|
tx.pop('id')
|
|
tx['transaction']['metadata'].pop('id')
|
|
tx['transaction'].pop('timestamp')
|
|
tx['transaction'].pop('asset')
|
|
|
|
assert tx == expected
|
|
|
|
|
|
def test_validate_multiple_io_create_transaction(user_pub, user_priv,
|
|
user2_pub, user2_priv):
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
# TODO: Fix multiple owners_before in create transactions
|
|
tx = Transaction.create([user_pub, user2_pub],
|
|
[([user_pub], 1), ([user2_pub], 1)],
|
|
metadata={'message': 'hello'},
|
|
asset=Asset(divisible=True))
|
|
tx = tx.sign([user_priv, user2_priv])
|
|
assert tx.fulfillments_valid() is True
|
|
|
|
|
|
def test_create_create_transaction_threshold(user_pub, user2_pub, user3_pub,
|
|
user_user2_threshold_cond,
|
|
user_user2_threshold_ffill, data,
|
|
data_id):
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
expected = {
|
|
'transaction': {
|
|
'conditions': [user_user2_threshold_cond.to_dict(0)],
|
|
'metadata': {
|
|
'data': data,
|
|
},
|
|
'asset': {
|
|
'id': data_id,
|
|
'divisible': False,
|
|
'updatable': False,
|
|
'refillable': False,
|
|
'data': data,
|
|
},
|
|
'fulfillments': [
|
|
{
|
|
'owners_before': [
|
|
user_pub,
|
|
],
|
|
'fid': 0,
|
|
'fulfillment': None,
|
|
'input': None
|
|
},
|
|
],
|
|
'operation': 'CREATE',
|
|
},
|
|
'version': 1
|
|
}
|
|
asset = Asset(data, data_id)
|
|
tx = Transaction.create([user_pub], [([user_pub, user2_pub], 1)],
|
|
data, asset)
|
|
tx_dict = tx.to_dict()
|
|
tx_dict.pop('id')
|
|
tx_dict['transaction']['metadata'].pop('id')
|
|
tx_dict['transaction'].pop('timestamp')
|
|
tx_dict['transaction']['fulfillments'][0]['fulfillment'] = None
|
|
|
|
assert tx_dict == expected
|
|
|
|
|
|
def test_validate_threshold_create_transaction(user_pub, user_priv, user2_pub,
|
|
data):
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
tx = Transaction.create([user_pub], [([user_pub, user2_pub], 1)],
|
|
data, Asset())
|
|
tx = tx.sign([user_priv])
|
|
assert tx.fulfillments_valid() is True
|
|
|
|
|
|
@mark.skip(reason='Hashlocks are not implemented')
|
|
def test_create_create_transaction_hashlock(user_pub, data, data_id):
|
|
from cryptoconditions import PreimageSha256Fulfillment
|
|
from bigchaindb.common.transaction import Transaction, Condition, Asset
|
|
|
|
secret = b'much secret, wow'
|
|
hashlock = PreimageSha256Fulfillment(preimage=secret).condition_uri
|
|
cond = Condition(hashlock)
|
|
|
|
expected = {
|
|
'transaction': {
|
|
'conditions': [cond.to_dict(0)],
|
|
'metadata': {
|
|
'data': data,
|
|
},
|
|
'asset': {
|
|
'id': data_id,
|
|
'divisible': False,
|
|
'updatable': False,
|
|
'refillable': False,
|
|
'data': data,
|
|
},
|
|
'fulfillments': [
|
|
{
|
|
'owners_before': [
|
|
user_pub,
|
|
],
|
|
'fid': 0,
|
|
'fulfillment': None,
|
|
'input': None
|
|
},
|
|
],
|
|
'operation': 'CREATE',
|
|
},
|
|
'version': 1
|
|
}
|
|
|
|
asset = Asset(data, data_id)
|
|
tx = Transaction.create([user_pub], [], data, asset, secret).to_dict()
|
|
tx.pop('id')
|
|
tx['transaction']['metadata'].pop('id')
|
|
tx['transaction'].pop('timestamp')
|
|
tx['transaction']['fulfillments'][0]['fulfillment'] = None
|
|
|
|
assert tx == expected
|
|
|
|
|
|
@mark.skip(reson='Hashlocks are not implemented')
|
|
def test_validate_hashlock_create_transaction(user_pub, user_priv, data):
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
tx = Transaction.create([user_pub], [], data, Asset(), b'much secret, wow')
|
|
tx = tx.sign([user_priv])
|
|
assert tx.fulfillments_valid() is True
|
|
|
|
|
|
def test_create_create_transaction_with_invalid_parameters(user_pub):
|
|
from bigchaindb.common.transaction import Transaction
|
|
|
|
with raises(TypeError):
|
|
Transaction.create('not a list')
|
|
with raises(TypeError):
|
|
Transaction.create([], 'not a list')
|
|
with raises(ValueError):
|
|
Transaction.create([], [user_pub])
|
|
with raises(ValueError):
|
|
Transaction.create([user_pub], [])
|
|
with raises(ValueError):
|
|
Transaction.create([user_pub], [user_pub])
|
|
with raises(ValueError):
|
|
Transaction.create([user_pub], [([user_pub],)])
|
|
|
|
|
|
def test_conditions_to_inputs(tx):
|
|
ffills = tx.to_inputs([0])
|
|
assert len(ffills) == 1
|
|
ffill = ffills.pop()
|
|
assert ffill.fulfillment == tx.conditions[0].fulfillment
|
|
assert ffill.owners_before == tx.conditions[0].owners_after
|
|
assert ffill.tx_input.txid == tx.id
|
|
assert ffill.tx_input.cid == 0
|
|
|
|
|
|
def test_create_transfer_transaction_single_io(tx, user_pub, user2_pub,
|
|
user2_cond, user_priv, data_id):
|
|
from copy import deepcopy
|
|
from bigchaindb.common.crypto import SigningKey
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
from bigchaindb.common.util import serialize
|
|
|
|
expected = {
|
|
'transaction': {
|
|
'conditions': [user2_cond.to_dict(0)],
|
|
'metadata': None,
|
|
'asset': {
|
|
'id': data_id,
|
|
},
|
|
'fulfillments': [
|
|
{
|
|
'owners_before': [
|
|
user_pub
|
|
],
|
|
'fid': 0,
|
|
'fulfillment': None,
|
|
'input': {
|
|
'txid': tx.id,
|
|
'cid': 0
|
|
}
|
|
}
|
|
],
|
|
'operation': 'TRANSFER',
|
|
},
|
|
'version': 1
|
|
}
|
|
inputs = tx.to_inputs([0])
|
|
asset = Asset(None, data_id)
|
|
transfer_tx = Transaction.transfer(inputs, [([user2_pub], 1)], asset=asset)
|
|
transfer_tx = transfer_tx.sign([user_priv])
|
|
transfer_tx = transfer_tx.to_dict()
|
|
transfer_tx_body = transfer_tx['transaction']
|
|
|
|
expected_input = deepcopy(inputs[0])
|
|
expected['id'] = transfer_tx['id']
|
|
expected['transaction']['timestamp'] = transfer_tx_body['timestamp']
|
|
expected_input.fulfillment.sign(serialize(expected).encode(),
|
|
SigningKey(user_priv))
|
|
expected_ffill = expected_input.fulfillment.serialize_uri()
|
|
transfer_ffill = transfer_tx_body['fulfillments'][0]['fulfillment']
|
|
|
|
assert transfer_ffill == expected_ffill
|
|
|
|
transfer_tx = Transaction.from_dict(transfer_tx)
|
|
assert transfer_tx.fulfillments_valid([tx.conditions[0]]) is True
|
|
|
|
|
|
def test_create_transfer_transaction_multiple_io(user_pub, user_priv,
|
|
user2_pub, user2_priv,
|
|
user3_pub, user2_cond):
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
asset = Asset(divisible=True)
|
|
tx = Transaction.create([user_pub], [([user_pub], 1), ([user2_pub], 1)],
|
|
asset=asset, metadata={'message': 'hello'})
|
|
tx = tx.sign([user_priv])
|
|
|
|
expected = {
|
|
'transaction': {
|
|
'conditions': [user2_cond.to_dict(0), user2_cond.to_dict(1)],
|
|
'metadata': None,
|
|
'fulfillments': [
|
|
{
|
|
'owners_before': [
|
|
user_pub
|
|
],
|
|
'fid': 0,
|
|
'fulfillment': None,
|
|
'input': {
|
|
'txid': tx.id,
|
|
'cid': 0
|
|
}
|
|
}, {
|
|
'owners_before': [
|
|
user2_pub
|
|
],
|
|
'fid': 1,
|
|
'fulfillment': None,
|
|
'input': {
|
|
'txid': tx.id,
|
|
'cid': 1
|
|
}
|
|
}
|
|
],
|
|
'operation': 'TRANSFER',
|
|
},
|
|
'version': 1
|
|
}
|
|
|
|
transfer_tx = Transaction.transfer(tx.to_inputs(),
|
|
[([user2_pub], 1), ([user2_pub], 1)],
|
|
asset=tx.asset)
|
|
transfer_tx = transfer_tx.sign([user_priv, user2_priv])
|
|
|
|
assert len(transfer_tx.fulfillments) == 2
|
|
assert len(transfer_tx.conditions) == 2
|
|
|
|
assert transfer_tx.fulfillments_valid(tx.conditions) is True
|
|
|
|
transfer_tx = transfer_tx.to_dict()
|
|
transfer_tx['transaction']['fulfillments'][0]['fulfillment'] = None
|
|
transfer_tx['transaction']['fulfillments'][1]['fulfillment'] = None
|
|
transfer_tx['transaction'].pop('timestamp')
|
|
transfer_tx.pop('id')
|
|
transfer_tx['transaction'].pop('asset')
|
|
|
|
assert expected == transfer_tx
|
|
|
|
|
|
def test_create_transfer_with_invalid_parameters(user_pub):
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
with raises(TypeError):
|
|
Transaction.transfer({}, [], Asset())
|
|
with raises(ValueError):
|
|
Transaction.transfer([], [], Asset())
|
|
with raises(TypeError):
|
|
Transaction.transfer(['fulfillment'], {}, Asset())
|
|
with raises(ValueError):
|
|
Transaction.transfer(['fulfillment'], [], Asset())
|
|
with raises(ValueError):
|
|
Transaction.transfer(['fulfillment'], [user_pub], Asset())
|
|
with raises(ValueError):
|
|
Transaction.transfer(['fulfillment'], [([user_pub],)], Asset())
|
|
|
|
|
|
def test_cant_add_empty_condition():
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
with patch.object(Asset, '_validate_asset', return_value=None):
|
|
tx = Transaction(Transaction.CREATE, None)
|
|
with raises(TypeError):
|
|
tx.add_condition(None)
|
|
|
|
|
|
def test_cant_add_empty_fulfillment():
|
|
from bigchaindb.common.transaction import Transaction, Asset
|
|
|
|
with patch.object(Asset, '_validate_asset', return_value=None):
|
|
tx = Transaction(Transaction.CREATE, None)
|
|
with raises(TypeError):
|
|
tx.add_fulfillment(None)
|