forked from TrueCloudLab/frostfs-testlib
Add unit tests for cli helpers
Signed-off-by: Vladimir Avdeev <v.avdeev@yadro.com>
This commit is contained in:
Vladimir Avdeev
parent
841a61fc30
commit
3816a3c7f1
1 changed files with 174 additions and 0 deletions
174
tests/test_cli.py
Normal file
174
tests/test_cli.py
Normal file
|
|
@ -0,0 +1,174 @@
|
|||
from unittest import TestCase
|
||||
from unittest.mock import Mock
|
||||
|
||||
from neofs_testlib.cli import NeofsAdm, NeofsCli, NeoGo
|
||||
from neofs_testlib.cli.cli_command import CliCommand
|
||||
from neofs_testlib.shell.interfaces import CommandOptions, InteractiveInput
|
||||
|
||||
|
||||
class TestCli(TestCase):
|
||||
neofs_adm_exec_path = "neo-adm-exec"
|
||||
neofs_go_exec_path = "neo-go-exec"
|
||||
neofs_cli_exec_path = "neo-cli-exec"
|
||||
|
||||
address = "0x0000000000000000000"
|
||||
addresses = ["0x000000", "0xDEADBEEF", "0xBABECAFE"]
|
||||
amount = 100
|
||||
file1 = "file_1"
|
||||
file2 = "directory/file_2"
|
||||
manifest = "manifest1"
|
||||
token = "GAS"
|
||||
rpc_endpoint = "endpoint-1"
|
||||
sysgas: float = 0.001
|
||||
wallet = "wallet1"
|
||||
wallet_password = "P@$$w0rd"
|
||||
config_file = "config.yml"
|
||||
basic_acl = "1FBFBFFF"
|
||||
policy = "policy1"
|
||||
timeout = 20
|
||||
xhdr = {"param1": "value1", "param2": "value2"}
|
||||
|
||||
def test_container_create(self):
|
||||
shell = Mock()
|
||||
neofs_cli = NeofsCli(
|
||||
config_file=self.config_file,
|
||||
neofs_cli_exec_path=self.neofs_cli_exec_path,
|
||||
shell=shell,
|
||||
)
|
||||
neofs_cli.container.create(
|
||||
rpc_endpoint=self.rpc_endpoint,
|
||||
wallet=self.wallet,
|
||||
basic_acl=self.basic_acl,
|
||||
policy=self.policy,
|
||||
await_mode=True,
|
||||
xhdr=self.xhdr,
|
||||
)
|
||||
|
||||
xhdr = ",".join(f"{param}={value}" for param, value in self.xhdr.items())
|
||||
expected_command = (
|
||||
f"{self.neofs_cli_exec_path} --config {self.config_file} container create "
|
||||
f"--rpc-endpoint '{self.rpc_endpoint}' --wallet '{self.wallet}' "
|
||||
f"--basic-acl '{self.basic_acl}' --await --policy '{self.policy}' "
|
||||
f"--xhdr '{xhdr}'"
|
||||
)
|
||||
|
||||
shell.exec.assert_called_once_with(expected_command)
|
||||
|
||||
def test_bad_wallet_argument(self):
|
||||
shell = Mock()
|
||||
neo_go = NeoGo(
|
||||
shell=shell, config_path=self.config_file, neo_go_exec_path=self.neofs_go_exec_path
|
||||
)
|
||||
with self.assertRaises(Exception) as exc_msg:
|
||||
neo_go.contract.add_group(
|
||||
address=self.address,
|
||||
manifest=self.manifest,
|
||||
wallet_password=self.wallet_password,
|
||||
)
|
||||
self.assertEqual(CliCommand.WALLET_SOURCE_ERROR_MSG, str(exc_msg.exception))
|
||||
|
||||
with self.assertRaises(Exception) as exc_msg:
|
||||
neo_go.contract.add_group(
|
||||
wallet=self.wallet,
|
||||
wallet_password=self.wallet_password,
|
||||
wallet_config=self.config_file,
|
||||
address=self.address,
|
||||
manifest=self.manifest,
|
||||
)
|
||||
self.assertEqual(CliCommand.WALLET_SOURCE_ERROR_MSG, str(exc_msg.exception))
|
||||
|
||||
with self.assertRaises(Exception) as exc_msg:
|
||||
neo_go.contract.add_group(
|
||||
wallet=self.wallet,
|
||||
address=self.address,
|
||||
manifest=self.manifest,
|
||||
)
|
||||
self.assertEqual(CliCommand.WALLET_PASSWD_ERROR_MSG, str(exc_msg.exception))
|
||||
|
||||
def test_wallet_sign(self):
|
||||
shell = Mock()
|
||||
neo_go = NeoGo(
|
||||
shell=shell, config_path=self.config_file, neo_go_exec_path=self.neofs_go_exec_path
|
||||
)
|
||||
neo_go.wallet.sign(
|
||||
input_file=self.file1,
|
||||
out=self.file2,
|
||||
rpc_endpoint=self.rpc_endpoint,
|
||||
address=self.address,
|
||||
wallet=self.wallet,
|
||||
wallet_password=self.wallet_password,
|
||||
timeout=self.timeout,
|
||||
)
|
||||
|
||||
expected_command = (
|
||||
f"{self.neofs_go_exec_path} --config_path {self.config_file} wallet sign "
|
||||
f"--input-file '{self.file1}' --address '{self.address}' "
|
||||
f"--rpc-endpoint '{self.rpc_endpoint}' --wallet '{self.wallet}' "
|
||||
f"--out '{self.file2}' --timeout '{self.timeout}s'"
|
||||
)
|
||||
|
||||
shell.exec.assert_called_once_with(
|
||||
expected_command,
|
||||
options=CommandOptions(
|
||||
interactive_inputs=[
|
||||
InteractiveInput(prompt_pattern="assword", input=self.wallet_password)
|
||||
]
|
||||
),
|
||||
)
|
||||
|
||||
def test_subnet_create(self):
|
||||
shell = Mock()
|
||||
neofs_adm = NeofsAdm(
|
||||
config_file=self.config_file,
|
||||
neofs_adm_exec_path=self.neofs_adm_exec_path,
|
||||
shell=shell,
|
||||
)
|
||||
neofs_adm.subnet.create(
|
||||
address=self.address,
|
||||
rpc_endpoint=self.rpc_endpoint,
|
||||
wallet=self.wallet,
|
||||
notary=True,
|
||||
)
|
||||
|
||||
expected_command = (
|
||||
f"{self.neofs_adm_exec_path} --config {self.config_file} morph subnet create "
|
||||
f"--rpc-endpoint '{self.rpc_endpoint}' --address '{self.address}' "
|
||||
f"--wallet '{self.wallet}' --notary"
|
||||
)
|
||||
|
||||
shell.exec.assert_called_once_with(expected_command)
|
||||
|
||||
def test_wallet_nep17_multitransfer(self):
|
||||
shell = Mock()
|
||||
neo_go = NeoGo(
|
||||
shell=shell, config_path=self.config_file, neo_go_exec_path=self.neofs_go_exec_path
|
||||
)
|
||||
neo_go.nep17.multitransfer(
|
||||
wallet=self.wallet,
|
||||
token=self.token,
|
||||
to_address=self.addresses,
|
||||
sysgas=self.sysgas,
|
||||
rpc_endpoint=self.rpc_endpoint,
|
||||
amount=self.amount,
|
||||
force=True,
|
||||
from_address=self.address,
|
||||
timeout=self.timeout,
|
||||
)
|
||||
|
||||
to_address = "".join(f" --to '{address}'" for address in self.addresses)
|
||||
expected_command = (
|
||||
f"{self.neofs_go_exec_path} --config_path {self.config_file} "
|
||||
f"wallet nep17 multitransfer --token '{self.token}'"
|
||||
f"{to_address} --sysgas '{self.sysgas}' --rpc-endpoint '{self.rpc_endpoint}' "
|
||||
f"--wallet '{self.wallet}' --from '{self.address}' --force --amount {self.amount} "
|
||||
f"--timeout '{self.timeout}s'"
|
||||
)
|
||||
|
||||
shell.exec.assert_called_once_with(expected_command)
|
||||
|
||||
def test_version(self):
|
||||
shell = Mock()
|
||||
neofs_adm = NeofsAdm(shell=shell, neofs_adm_exec_path=self.neofs_adm_exec_path)
|
||||
neofs_adm.version.get()
|
||||
|
||||
shell.exec.assert_called_once_with(f"{self.neofs_adm_exec_path} --version")
|
||||
Loading…
Reference in a new issue