Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_transactionbuilder(self, node_param):
if node_param == "instance":
set_shared_steem_instance(self.bts)
o = TransactionBuilder()
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
o = TransactionBuilder(steem_instance=Steem(node="https://abc.d", autoconnect=False, num_retries=1))
o.steem.get_config()
else:
set_shared_steem_instance(Steem(node="https://abc.d", autoconnect=False, num_retries=1))
stm = self.bts
o = TransactionBuilder(steem_instance=stm)
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
o = TransactionBuilder()
o.steem.get_config()
tx = TransactionBuilder(use_condenser_api=True, steem_instance=steem)
tx.appendOps(Transfer(**{"from": 'beem5',
"to": 'beem1',
"amount": Amount("0.01 STEEM", steem_instance=steem),
"memo": '2 of 2 serialized/deserialized transaction'}))
tx.appendSigner("beem5", "active")
tx.addSigningInformation("beem5", "active")
tx.sign()
tx.clearWifs()
self.assertEqual(len(tx['signatures']), 1)
# steem.wallet.removeAccount("beem5")
steem.wallet.removePrivateKeyFromPublicKey(str(PublicKey(self.active_private_key_of_beem5, prefix=core_unit)))
tx_json = tx.json()
del tx
new_tx = TransactionBuilder(tx=tx_json, steem_instance=steem)
self.assertEqual(len(new_tx['signatures']), 1)
steem.wallet.addPrivateKey(self.active_private_key_of_beem4)
new_tx.appendMissingSignatures()
new_tx.sign(reconstruct_tx=False)
self.assertEqual(len(new_tx['signatures']), 2)
new_tx.broadcast()
steem.nobroadcast = True
def test_TransactionConstructor(self):
stm = self.bts
opTransfer = Transfer(**{"from": "beem",
"to": "beem1",
"amount": Amount("1 STEEM", steem_instance=stm),
"memo": ""})
tx1 = TransactionBuilder(use_condenser_api=True, steem_instance=stm)
tx1.appendOps(opTransfer)
tx = TransactionBuilder(tx1, steem_instance=stm)
self.assertFalse(tx.is_empty())
self.assertTrue(len(tx.list_operations()) == 1)
self.assertTrue(repr(tx) is not None)
self.assertTrue(str(tx) is not None)
account = Account("beem", steem_instance=stm)
tx.appendSigner(account, "active")
self.assertTrue(len(tx.wifs) > 0)
tx.sign()
self.assertTrue(len(tx["signatures"]) > 0)
def test_transfer_2of2_wallet(self):
# Send a 2 of 2 transaction from beem5 which needs beem4's cosign to send
# priv key of beem5 and beem4 are stored in the wallet
# appendSigner fetches both keys and signs automatically with both keys.
steem = self.bts
steem.nobroadcast = False
steem.wallet.unlock("123")
tx = TransactionBuilder(use_condenser_api=True, steem_instance=steem)
tx.appendOps(Transfer(**{"from": 'beem5',
"to": 'beem1',
"amount": Amount("0.01 STEEM", steem_instance=steem),
"memo": '2 of 2 serialized/deserialized transaction'}))
tx.appendSigner("beem5", "active")
tx.sign()
self.assertEqual(len(tx['signatures']), 2)
tx.broadcast()
steem.nobroadcast = True
def test_appendSigner(self):
nodelist = NodeList()
stm = Steem(node=self.nodes,
keys=[self.active_key],
nobroadcast=True,
expiration=120,
num_retries=10)
tx = TransactionBuilder(use_condenser_api=True, steem_instance=stm)
tx.appendOps(Transfer(**{"from": "beem",
"to": "beem1",
"amount": Amount("1 STEEM", steem_instance=stm),
"memo": ""}))
account = Account("beem", steem_instance=stm)
with self.assertRaises(
AssertionError
):
tx.appendSigner(account, "abcdefg")
tx.appendSigner(account, "active")
self.assertTrue(len(tx.wifs) > 0)
tx.sign()
self.assertTrue(len(tx["signatures"]) > 0)
def test_transfer_2of2_offline(self):
# Send a 2 of 2 transaction from beem5 which needs beem4's cosign to send
# funds but sign the transaction with beem5's key and then serialize the transaction
# and deserialize the transaction. After that, sign with beem4's key.
steem = self.bts
steem.nobroadcast = False
steem.wallet.unlock("123")
# steem.wallet.removeAccount("beem4")
steem.wallet.removePrivateKeyFromPublicKey(str(PublicKey(self.active_private_key_of_beem4, prefix=core_unit)))
tx = TransactionBuilder(use_condenser_api=True, steem_instance=steem)
tx.appendOps(Transfer(**{"from": 'beem5',
"to": 'beem',
"amount": Amount("0.01 STEEM", steem_instance=steem),
"memo": '2 of 2 serialized/deserialized transaction'}))
tx.appendSigner("beem5", "active")
tx.addSigningInformation("beem5", "active")
tx.sign()
tx.clearWifs()
self.assertEqual(len(tx['signatures']), 1)
# steem.wallet.removeAccount("beem5")
steem.wallet.removePrivateKeyFromPublicKey(str(PublicKey(self.active_private_key_of_beem5, prefix=core_unit)))
steem.wallet.addPrivateKey(self.active_private_key_of_beem4)
tx.appendMissingSignatures()
tx.sign(reconstruct_tx=False)
self.assertEqual(len(tx['signatures']), 2)
def test_transfer_2of2_wif(self):
nodelist = NodeList()
# Send a 2 of 2 transaction from elf which needs beem4's cosign to send
# funds but sign the transaction with elf's key and then serialize the transaction
# and deserialize the transaction. After that, sign with beem4's key.
steem = Steem(
node=self.nodes,
num_retries=10,
keys=[self.active_private_key_of_beem5],
expiration=360,
)
tx = TransactionBuilder(use_condenser_api=True, steem_instance=steem)
tx.appendOps(Transfer(**{"from": 'beem5',
"to": 'beem',
"amount": Amount("0.01 STEEM", steem_instance=steem),
"memo": '2 of 2 serialized/deserialized transaction'}))
tx.appendSigner("beem5", "active")
tx.addSigningInformation("beem5", "active")
tx.sign()
tx.clearWifs()
self.assertEqual(len(tx['signatures']), 1)
tx_json = tx.json()
del steem
del tx
steem = Steem(
node=self.nodes,
def test_transfer_2of2_serialized_deserialized(self):
# Send a 2 of 2 transaction from beem5 which needs beem4's cosign to send
# funds but sign the transaction with beem5's key and then serialize the transaction
# and deserialize the transaction. After that, sign with beem4's key.
steem = self.bts
steem.nobroadcast = False
steem.wallet.unlock("123")
# steem.wallet.removeAccount("beem4")
steem.wallet.removePrivateKeyFromPublicKey(str(PublicKey(self.active_private_key_of_beem4, prefix=core_unit)))
tx = TransactionBuilder(use_condenser_api=True, steem_instance=steem)
tx.appendOps(Transfer(**{"from": 'beem5',
"to": 'beem1',
"amount": Amount("0.01 STEEM", steem_instance=steem),
"memo": '2 of 2 serialized/deserialized transaction'}))
tx.appendSigner("beem5", "active")
tx.addSigningInformation("beem5", "active")
tx.sign()
tx.clearWifs()
self.assertEqual(len(tx['signatures']), 1)
# steem.wallet.removeAccount("beem5")
steem.wallet.removePrivateKeyFromPublicKey(str(PublicKey(self.active_private_key_of_beem5, prefix=core_unit)))
tx_json = tx.json()
del tx
new_tx = TransactionBuilder(tx=tx_json, steem_instance=steem)
self.assertEqual(len(new_tx['signatures']), 1)
if not use_api:
from beembase.signedtransactions import Signed_Transaction
for trx in trxs:
if not use_api:
# trx is now identical to the output of get_transaction
# This is just for testing porpuse
if True:
signed_tx = Signed_Transaction(trx.copy())
else:
tx = b.get_transaction(trx["transaction_id"])
signed_tx = Signed_Transaction(tx)
public_keys = []
for key in signed_tx.verify(chain=stm.chain_params, recover_parameter=True):
public_keys.append(format(Base58(key, prefix=stm.prefix), stm.prefix))
else:
tx = TransactionBuilder(tx=trx, steem_instance=stm)
public_keys = tx.get_potential_signatures()
accounts = []
empty_public_keys = []
for key in public_keys:
account = wallet.getAccountFromPublicKey(key)
if account is None:
empty_public_keys.append(key)
else:
accounts.append(account)
new_public_keys = []
for key in public_keys:
if key not in empty_public_keys or use_api:
new_public_keys.append(key)
if isinstance(new_public_keys, list) and len(new_public_keys) == 1:
new_public_keys = new_public_keys[0]
else:
def clear(self):
""" Clear the transaction builder and start from scratch
"""
self.ops = []
self.wifs = set()
self.signing_accounts = []
self.ref_block_num = None
self.ref_block_prefix = None
# This makes sure that _is_constructed will return False afterwards
self["expiration"] = None
super(TransactionBuilder, self).__init__({})