frostfs-testcases/robot/resources/lib/payment_neogo.py

#!/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()