frostfs-testcases/robot/resources/lib/payment_neogo.py
anatoly@nspcc.ru 61358128bd update readme
2020-11-29 01:16:52 +03:00

288 lines
9.1 KiB
Python

#!/usr/bin/python3
import subprocess
import pexpect
import re
import uuid
from robot.api.deco import keyword
from robot.api import logger
import logging
import robot.errors
import requests
import json
from robot.libraries.BuiltIn import BuiltIn
ROBOT_AUTO_KEYWORDS = False
NEOFS_CONTRACT = "5f490fbd8010fd716754073ee960067d28549b7d"
NEOGO_CLI_PREFIX = "docker exec -it main_chain neo-go"
NEO_MAINNET_ENDPOINT = "main_chain.neofs.devenv:30333"
@keyword('Init wallet')
def init_wallet():
filename = "wallets/" + str(uuid.uuid4()) + ".json"
cmd = ( f"{NEOGO_CLI_PREFIX} wallet init -w {filename}" )
logger.info(f"Executing shell command: {cmd}")
out = _run_sh(cmd)
logger.info(f"Command completed with output: {out}")
return filename
@keyword('Generate wallet')
def generate_wallet(wallet: str):
cmd = ( f"{NEOGO_CLI_PREFIX} wallet create -w {wallet}" )
logger.info(f"Executing command: {cmd}")
p = pexpect.spawn(cmd)
p.expect(".*")
p.sendline('\n')
p.sendline('\n')
p.sendline('\n')
p.wait()
out = p.read()
logger.info(f"Command completed with output: {out}")
@keyword('Dump Address')
def dump_address(wallet: str):
address = ""
cmd = ( f"{NEOGO_CLI_PREFIX} wallet dump -w {wallet}" )
logger.info(f"Executing command: {cmd}")
out = _run_sh(cmd)
logger.info(f"Command completed with output: {out}")
m = re.search(r'"address": "(\w+)"', out)
if m.start() != m.end():
address = m.group(1)
else:
raise Exception("Can not get address.")
return address
@keyword('Dump PrivKey')
def dump_privkey(wallet: str, address: str):
cmd = ( f"{NEOGO_CLI_PREFIX} wallet export -w {wallet} --decrypt {address}" )
logger.info(f"Executing command: {cmd}")
out = _run_sh_with_passwd('', cmd)
logger.info(f"Command completed with output: {out}")
return out
@keyword('Transfer Mainnet Gas')
def transfer_mainnet_gas(wallet: str, address: str, address_to: str, amount: int):
cmd = ( f"{NEOGO_CLI_PREFIX} wallet nep5 transfer -w {wallet} -r http://main_chain.neofs.devenv:30333 --from {address} "
f"--to {address_to} --token gas --amount {amount}" )
logger.info(f"Executing command: {cmd}")
out = _run_sh_with_passwd('', cmd)
logger.info(f"Command completed with output: {out}")
if not re.match(r'^(\w{64})$', out):
raise Exception("Can not get Tx.")
return out
@keyword('Withdraw Mainnet Gas')
def withdraw_mainnet_gas(wallet: str, address: str, scripthash: str, amount: int):
cmd = ( f"{NEOGO_CLI_PREFIX} contract invokefunction -w {wallet} -a {address} -r http://main_chain.neofs.devenv:30333 "
f"{NEOFS_CONTRACT} withdraw {scripthash} int:{amount} -- {scripthash}" )
logger.info(f"Executing command: {cmd}")
out = _run_sh_with_passwd('', cmd)
logger.info(f"Command completed with output: {out}")
#if not re.match(r'^(\w{64})$', out):
# raise Exception("Can not get Tx.")
return out
@keyword('Mainnet Balance')
def mainnet_balance(address: str):
request = 'curl -X POST '+NEO_MAINNET_ENDPOINT+' --cacert ca/nspcc-ca.pem -H \'Content-Type: application/json\' -d \'{ "jsonrpc": "2.0", "id": 5, "method": "getnep5balances", "params": [\"'+address+'\"] }\''
logger.info(f"Executing request: {request}")
complProc = subprocess.run(request, check=True, universal_newlines=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=15, shell=True)
out = complProc.stdout
logger.info(out)
m = re.search(r'"668e0c1f9d7b70a99dd9e06eadd4c784d641afbc","amount":"([\d\.]+)"', out)
if not m.start() != m.end():
raise Exception("Can not get mainnet gas balance.")
amount = m.group(1)
return amount
@keyword('Expexted Mainnet Balance')
def expected_mainnet_balance(address: str, expected: int):
amount = mainnet_balance(address)
if float(amount) != float(expected):
raise Exception(f"Expected amount ({expected}) of GAS has not been found. Found {amount}.")
return True
@keyword('NeoFS Deposit')
def neofs_deposit(wallet: str, address: str, scripthash: str, amount: int):
cmd = ( f"{NEOGO_CLI_PREFIX} contract invokefunction -w {wallet} -a {address} "
f"-r http://main_chain.neofs.devenv:30333 {NEOFS_CONTRACT} "
f"deposit {scripthash} int:{amount} bytes: -- {scripthash}")
logger.info(f"Executing command: {cmd}")
out = _run_sh_with_passwd('', cmd)
logger.info(f"Command completed with output: {out}")
m = re.match(r'^Sent invocation transaction (\w{64})$', out)
if m is None:
raise Exception("Can not get Tx.")
tx = m.group(1)
return tx
@keyword('Transaction accepted in block')
def transaction_accepted_in_block(tx_id):
"""
This function return True in case of accepted TX.
Parameters:
:param tx_id: transaction is
:rtype: block number or Exception
"""
logger.info("Transaction id: %s" % tx_id)
TX_request = 'curl -X POST '+NEO_MAINNET_ENDPOINT+' --cacert ca/nspcc-ca.pem -H \'Content-Type: application/json\' -d \'{ "jsonrpc": "2.0", "id": 5, "method": "gettransactionheight", "params": [\"'+ tx_id +'\"] }\''
logger.info(f"Executing command: {TX_request}")
complProc = subprocess.run(TX_request, check=True, universal_newlines=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=15, shell=True)
logger.info(complProc.stdout)
response = json.loads(complProc.stdout)
if (response['result'] == 0):
raise Exception( "Transaction is not found in the blocks." )
logger.info("Transaction has been found in the block %s." % response['result'] )
return response['result']
@keyword('Get Transaction')
def get_transaction(tx_id: str):
"""
This function return information about TX.
Parameters:
:param tx_id: transaction id
"""
TX_request = 'curl -X POST '+NEO_MAINNET_ENDPOINT+' --cacert ca/nspcc-ca.pem -H \'Content-Type: application/json\' -d \'{ "jsonrpc": "2.0", "id": 5, "method": "getapplicationlog", "params": [\"'+tx_id+'\"] }\''
complProc = subprocess.run(TX_request, check=True, universal_newlines=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=15, shell=True)
logger.info(complProc.stdout)
@keyword('Request NeoFS Deposit')
def request_neofs_deposit(public_key: str):
"""
This function requests Deposit to the selected public key.
:param public_key: neo public key
"""
response = requests.get('https://fs.localtest.nspcc.ru/api/deposit/'+str(public_key), verify='ca/nspcc-ca.pem')
if response.status_code != 200:
BuiltIn().fatal_error('Can not run Deposit to {} with error: {}'.format(public_key, response.text))
else:
logger.info("Deposit has been completed for '%s'; tx: '%s'" % (public_key, response.text) )
return response.text
@keyword('Get Balance')
def get_balance(privkey: str):
"""
This function returns NeoFS balance for selected public key.
:param public_key: neo public key
"""
balance = _get_balance_request(privkey)
return balance
@keyword('Expected Balance')
def expected_balance(privkey: str, init_amount: float, deposit_size: float):
"""
This function returns NeoFS balance for selected public key.
:param public_key: neo public key
:param init_amount: initial number of tokens in the account
:param deposit_size: expected amount of the balance increasing
"""
balance = _get_balance_request(privkey)
deposit_change = round((float(balance) - init_amount),8)
if deposit_change != deposit_size:
raise Exception('Expected deposit increase: {}. This does not correspond to the actual change in account: {}'.format(deposit_size, deposit_change))
logger.info('Expected deposit increase: {}. This correspond to the actual change in account: {}'.format(deposit_size, deposit_change))
return deposit_change
def _get_balance_request(privkey: str):
'''
Internal method.
'''
Cmd = f'neofs-cli --key {privkey} --rpc-endpoint s01.neofs.devenv:8080 accounting balance'
logger.info("Cmd: %s" % Cmd)
complProc = subprocess.run(Cmd, check=True, universal_newlines=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=150, shell=True)
output = complProc.stdout
logger.info("Output: %s" % output)
m = re.match(r'(-?[\d.\.?\d*]+)', output )
if m is None:
BuiltIn().fatal_error('Can not parse balance: "%s"' % output)
balance = m.group(1)
logger.info("Balance for '%s' is '%s'" % (privkey, balance) )
return balance
def _run_sh(args):
complProc = subprocess.run(args, check=True, universal_newlines=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
timeout=150, shell=True)
output, errors = complProc.stdout, complProc.stderr
if errors:
return errors
return output
def _run_sh_with_passwd(passwd, cmd):
p = pexpect.spawn(cmd)
p.expect(".*")
p.sendline(passwd)
p.wait()
# throw a string with password prompt
# take a string with tx hash
tx_hash = p.read().splitlines()[-1]
return tx_hash.decode()