diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..f607121 --- /dev/null +++ b/tests/test_cli.py @@ -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")