Skip to content

Commit

Permalink
use assert_in in wallet functional tests for better error message
Browse files Browse the repository at this point in the history
  • Loading branch information
@OBorce
OBorce committed Sep 29, 2023
1 parent 9b15ffb commit 2bfed59
Show file tree
Hide file tree
Showing 9 changed files with 134 additions and 134 deletions.
3 changes: 3 additions & 0 deletions test/functional/test_framework/util.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,9 @@ def assert_fee_amount(fee, tx_size, feerate_BTC_kvB):
if fee > high_fee:
raise AssertionError("Fee of %s BTC too high! (Should be %s BTC)" % (str(fee), str(target_fee)))

def assert_in(thing1: str, thing2: str):
if thing1 not in thing2:
raise AssertionError(f"{thing1} is not in {thing2}")

def assert_equal(thing1, thing2, *args):
if thing1 != thing2 or any(thing1 != arg for arg in args):
Expand Down
94 changes: 47 additions & 47 deletions test/functional/wallet_delegations.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,16 +42,13 @@
from scalecodec.base import ScaleBytes, RuntimeConfiguration, ScaleDecoder
from test_framework.test_framework import BitcoinTestFramework
from test_framework.mintlayer import (make_tx, reward_input, tx_input)
from test_framework.util import assert_raises_rpc_error
from test_framework.util import assert_equal, assert_greater_than, assert_in
from test_framework.mintlayer import mintlayer_hash, block_input_data_obj
from test_framework.wallet_cli_controller import DEFAULT_ACCOUNT_INDEX, WalletCliController
from test_framework.util import (
assert_equal,
)

import asyncio
import sys
import random, time, re
import time, re

GENESIS_POOL_ID = "123c4c600097c513e088b9be62069f0c74c7671c523c8e3469a1c3f14b7ea2c4"
GENESIS_STAKE_PRIVATE_KEY = "8717e6946febd3a33ccdc3f3a27629ec80c33461c33a0fc56b4836fcedd26638"
Expand Down Expand Up @@ -294,11 +291,11 @@ async def async_test(self):
# check it is on genesis
best_block_height = await wallet.get_best_block_height()
self.log.info(f"best block height = {best_block_height}")
assert best_block_height == '0'
assert_equal(best_block_height, '0')

# new address
pub_key_bytes = await wallet.new_public_key()
assert len(pub_key_bytes) == 33
assert_equal(len(pub_key_bytes), 33)

# Get chain tip
tip_id = node.chainstate_best_block_id()
Expand All @@ -314,118 +311,121 @@ async def async_test(self):
self.setup_pool_and_transfer([encoded_tx])

# sync the wallet
assert "Success" in await wallet.sync()
assert_in("Success", await wallet.sync())

# check wallet best block if it is synced
best_block_height = await wallet.get_best_block_height()
assert best_block_height == '1'
assert_in(best_block_height, '1')

balance = await wallet.get_balance()
assert "Coins amount: 50000" in balance
assert_in("Coins amount: 50000", balance)

assert "Success" in await wallet.create_new_account()
assert "Success" in await wallet.select_account(1)
assert_in("Success", await wallet.create_new_account())
assert_in("Success", await wallet.select_account(1))
acc1_address = await wallet.new_address()
assert "Success" in await wallet.select_account(DEFAULT_ACCOUNT_INDEX)
assert "The transaction was submitted successfully" in await wallet.send_to_address(acc1_address, 100)
assert "Success" in await wallet.select_account(1)
assert_in("Success", await wallet.select_account(DEFAULT_ACCOUNT_INDEX))
assert_in("The transaction was submitted successfully", await wallet.send_to_address(acc1_address, 100))
assert_in("Success", await wallet.select_account(1))
transactions = node.mempool_transactions()

assert "Success" in await wallet.select_account(DEFAULT_ACCOUNT_INDEX)
assert "The transaction was submitted successfully" in await wallet.create_stake_pool(40000, 0, 0.5)
assert_in("Success", await wallet.select_account(DEFAULT_ACCOUNT_INDEX))
assert_in("The transaction was submitted successfully", await wallet.create_stake_pool(40000, 0, 0.5))
transactions2 = node.mempool_transactions()
for tx in transactions2:
if tx not in transactions:
transactions.append(tx)

self.gen_pos_block(transactions, 2)
assert "Success" in await wallet.sync()
assert_in("Success", await wallet.sync())

pools = await wallet.list_pool_ids()
assert len(pools) == 1
assert pools[0].balance == 40000
assert_equal(len(pools), 1)
assert_equal(pools[0].balance, 40000)

assert "Success" in await wallet.select_account(1)
assert_in("Success", await wallet.select_account(1))
delegation_id = await wallet.create_delegation(acc1_address, pools[0].pool_id)
assert delegation_id is not None
transactions = node.mempool_transactions()

# still not in a block
delegations = await wallet.list_delegation_ids()
assert len(delegations) == 0
assert_equal(len(delegations), 0)

assert "Success" in await wallet.stake_delegation(10, delegation_id)
assert_in("Success", await wallet.stake_delegation(10, delegation_id))
transactions2 = node.mempool_transactions()
for tx in transactions2:
if tx not in transactions:
transactions.append(tx)

self.gen_pos_block(transactions, 3)
assert "Success" in await wallet.sync()
assert_in("Success", await wallet.sync())
last_block_id = self.previous_block_id()

delegations = await wallet.list_delegation_ids()
assert len(delegations) == 1
assert delegations[0].balance == 10
assert_equal(len(delegations), 1)
assert_equal(delegations[0].balance, 10)

assert "Success" in await wallet.select_account(DEFAULT_ACCOUNT_INDEX)
assert "Staking started successfully" in await wallet.start_staking()
assert "Success" in await wallet.select_account(1)
assert_in("Success", await wallet.select_account(DEFAULT_ACCOUNT_INDEX))
assert_in("Staking started successfully", await wallet.start_staking())
assert_in("Success", await wallet.select_account(1))

last_delegation_balance = delegations[0].balance
for _ in range(4, 10):
tip_id = node.chainstate_best_block_id()
assert "The transaction was submitted successfully" in await wallet.send_to_address(acc1_address, 1)
assert_in("The transaction was submitted successfully", await wallet.send_to_address(acc1_address, 1))
transactions = node.mempool_transactions()
self.wait_until(lambda: node.chainstate_best_block_id() != tip_id, timeout = 5)
assert "Success" in await wallet.sync()
assert_in("Success", await wallet.sync())

delegations = await wallet.list_delegation_ids()
assert len(delegations) == 1
assert delegations[0].balance > last_delegation_balance
assert_equal(len(delegations), 1)
assert_greater_than(delegations[0].balance, last_delegation_balance)
last_delegation_balance = delegations[0].balance

# stake to acc1 delegation from acc 0
assert "Success" in await wallet.select_account(DEFAULT_ACCOUNT_INDEX)
assert "Success" in await wallet.stake_delegation(10, delegation_id)
assert_in("Success", await wallet.select_account(DEFAULT_ACCOUNT_INDEX))
assert_in("Success", await wallet.stake_delegation(10, delegation_id))
self.wait_until(lambda: node.chainstate_best_block_id() != tip_id, timeout = 5)
assert "Success" in await wallet.sync()
assert_in("Success", await wallet.sync())

# check that we still don't have any delagations for this account
delegations = await wallet.list_delegation_ids()
assert len(delegations) == 0
assert_equal(len(delegations), 0)

# create a delegation from acc 0 but with destination address for acc1
delegation_id = await wallet.create_delegation(acc1_address, pools[0].pool_id)
tip_id = node.chainstate_best_block_id()
self.wait_until(lambda: node.chainstate_best_block_id() != tip_id, timeout = 5)
assert "Success" in await wallet.sync()
assert_in("Success", await wallet.sync())

# check that we still don't have any delagations for this account
delegations = await wallet.list_delegation_ids()
assert len(delegations) == 0
assert_equal(len(delegations), 0)

assert "Success" in await wallet.select_account(1)
assert_in("Success", await wallet.select_account(1))
delegations = await wallet.list_delegation_ids()
assert len(delegations) == 2
assert_equal(len(delegations), 2)
assert delegation_id in [delegation.delegation_id for delegation in delegations]

assert "Success" in await wallet.select_account(DEFAULT_ACCOUNT_INDEX)
assert "Success" in await wallet.stop_staking()
assert "The transaction was submitted successfully" in await wallet.decommission_stake_pool(pools[0].pool_id)
assert_in("Success", await wallet.select_account(DEFAULT_ACCOUNT_INDEX))
assert_in("Success", await wallet.stop_staking())
assert_in("The transaction was submitted successfully", await wallet.decommission_stake_pool(pools[0].pool_id))

transactions = node.mempool_transactions()
block_height = await wallet.get_best_block_height()
self.gen_pos_block(transactions, int(block_height)+1, last_block_id)
assert "Success" in await wallet.sync()
assert_in("Success", await wallet.sync())

pools = await wallet.list_pool_ids()
assert len(pools) == 0
assert_equal(len(pools), 0)

balance = await wallet.get_balance("locked")
pattern = r"Coins amount: 4\d{4}"
pattern = r"Coins amount: (\d{5})"
result = re.search(pattern, balance)
assert(result)
g = result.group(1)
self.log.info(f"extracted group {g}")
assert_greater_than(int(g), 40000)


if __name__ == '__main__':
Expand Down
14 changes: 7 additions & 7 deletions test/functional/wallet_get_address_usage.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@

from test_framework.test_framework import BitcoinTestFramework
from test_framework.mintlayer import (MLT_COIN, make_tx, reward_input, tx_input)
from test_framework.util import assert_raises_rpc_error
from test_framework.util import assert_in, assert_equal
from test_framework.mintlayer import mintlayer_hash, block_input_data_obj
from test_framework.wallet_cli_controller import WalletCliController

Expand Down Expand Up @@ -74,19 +74,19 @@ async def async_test(self):
# new wallet
async with WalletCliController(node, self.config, self.log) as wallet:
mnemonic = "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"
assert "New wallet created successfully" in await wallet.recover_wallet(mnemonic)
assert_in("New wallet created successfully", await wallet.recover_wallet(mnemonic))

# check it is on genesis
best_block_height = await wallet.get_best_block_height()
assert best_block_height == '0'
assert_equal(best_block_height, '0')

# new address
for _ in range(4):
pub_key_bytes = await wallet.new_public_key()
assert len(pub_key_bytes) == 33
assert_equal(len(pub_key_bytes), 33)

pub_key_bytes = await wallet.new_public_key()
assert len(pub_key_bytes) == 33
assert_equal(len(pub_key_bytes), 33)

# 2 more unused addresses
await wallet.new_address()
Expand All @@ -110,7 +110,7 @@ async def async_test(self):

# sync the wallet
output = await wallet.sync()
assert "Success" in output
assert_in("Success", output)

expected_output = """+-------+----------------------------------------------+--------------------------------+
| Index | Address | Is used in transaction history |
Expand All @@ -131,7 +131,7 @@ async def async_test(self):
+-------+----------------------------------------------+--------------------------------+"""
output = await wallet.get_addresses_usage()
for (line, expected_line) in zip(output.split(), expected_output.split()):
assert line == expected_line
assert_equal(line, expected_line)


if __name__ == '__main__':
Expand Down
14 changes: 7 additions & 7 deletions test/functional/wallet_high_fee.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
import scalecodec
from test_framework.test_framework import BitcoinTestFramework
from test_framework.mintlayer import (calc_tx_id, make_tx_dict, reward_input, tx_input, MLT_COIN, tx_output)
from test_framework.util import assert_raises_rpc_error
from test_framework.util import assert_in, assert_equal
from test_framework.mintlayer import mintlayer_hash, block_input_data_obj
from test_framework.wallet_cli_controller import WalletCliController

Expand Down Expand Up @@ -92,11 +92,11 @@ async def async_test(self):
# check it is on genesis
best_block_height = await wallet.get_best_block_height()
self.log.info(f"best block height = {best_block_height}")
assert best_block_height == '0'
assert_equal(best_block_height, '0')

# new address
pub_key_bytes = await wallet.new_public_key()
assert len(pub_key_bytes) == 33
assert_equal(len(pub_key_bytes), 33)

# Get chain tip
tip_id = node.chainstate_best_block_id()
Expand All @@ -122,17 +122,17 @@ async def async_test(self):

# sync the wallet
output = await wallet.sync()
assert "Success" in output
assert_in("Success", output)

# check wallet best block if it is synced
best_block_height = await wallet.get_best_block_height()
assert best_block_height == '1'
assert_equal(best_block_height, '1')

best_block_id = await wallet.get_best_block()
assert best_block_id == block_id
assert_equal(best_block_id, block_id)

balance = await wallet.get_balance()
assert "Coins amount: 10" in balance
assert_in("Coins amount: 10", balance)

transactions = node.test_functions_generate_transactions(tx_id, 25, total - 300, 300)
for idx, encoded_tx in enumerate(transactions):
Expand Down
24 changes: 12 additions & 12 deletions test/functional/wallet_nfts.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@

from test_framework.test_framework import BitcoinTestFramework
from test_framework.mintlayer import (make_tx, reward_input, tx_input, MLT_COIN)
from test_framework.util import assert_raises_rpc_error
from test_framework.util import assert_in, assert_equal
from test_framework.mintlayer import mintlayer_hash, block_input_data_obj
from test_framework.wallet_cli_controller import WalletCliController

Expand Down Expand Up @@ -78,11 +78,11 @@ async def async_test(self):
await wallet.create_wallet()

# check it is on genesis
assert '0' == await wallet.get_best_block_height()
assert_equal('0', await wallet.get_best_block_height())

# new address
pub_key_bytes = await wallet.new_public_key()
assert len(pub_key_bytes) == 33
assert_equal(len(pub_key_bytes), 33)

# Get chain tip
tip_id = node.chainstate_best_block_id()
Expand All @@ -102,38 +102,38 @@ async def async_test(self):


# sync the wallet
assert "Success" in await wallet.sync()
assert_in("Success", await wallet.sync())

# check wallet best block if it is synced
assert await wallet.get_best_block_height() == '1'
assert await wallet.get_best_block() == block_id
assert_equal(await wallet.get_best_block_height(), '1')
assert_equal(await wallet.get_best_block(), block_id)

assert "Coins amount: 1000" in await wallet.get_balance()
assert_in("Coins amount: 1000", await wallet.get_balance())

address = await wallet.new_address()
nft_id = await wallet.issue_new_nft(address,"123456", "Name", "SomeNFT", "XXX")
assert nft_id is not None
self.log.info(f"new nft id: '{nft_id}'")

self.generate_block()
assert "Success" in await wallet.sync()
assert_in("Success", await wallet.sync())

self.log.info(await wallet.get_balance())
assert f"{nft_id} amount: 1" in await wallet.get_balance()
assert_in(f"{nft_id} amount: 1", await wallet.get_balance())

# create a new account and send some tokens to it
await wallet.create_new_account()
await wallet.select_account(1)
address = await wallet.new_address()

await wallet.select_account(0)
assert f"{nft_id} amount: 1" in await wallet.get_balance()
assert_in(f"{nft_id} amount: 1", await wallet.get_balance())
output = await wallet.send_tokens_to_address(nft_id, address, 1)
self.log.info(output)
assert "The transaction was submitted successfully" in output
assert_in("The transaction was submitted successfully", output)

self.generate_block()
assert "Success" in await wallet.sync()
assert_in("Success", await wallet.sync())

# check the new balance nft is not present
assert nft_id not in await wallet.get_balance()
Expand Down
Loading

0 comments on commit 2bfed59

Please sign in to comment.