[SPT-777] New structure of functional tests

Author: MorozovMisha

.github/workflows/test.yml

@@ -64,4 +64,3 @@ jobs:
           export PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python
           # sh -ex ./snet/cli/test/utils/run_all_functional.sh
           python3 ./snet/cli/test/functional_tests/test_entry_point.py
-          python3 ./snet/cli/test/functional_tests/func_tests.py

snet/cli/test/functional_tests/func_tests.py

@@ -0,0 +1,46 @@
+import unittest
+from func_tests import BaseTest, execute, ADDR, IDENTITY, PRIVATE_KEY, INFURA, INFURA_KEY
+
+
+class TestAAMainPreparations(BaseTest):
+    def setUp(self):
+        super().setUp()
+        self.new_network = "auto_test"
+
+    def test_1_identity_create(self):
+        execute(["identity", "create", IDENTITY, "key", "--private-key", PRIVATE_KEY, "-de"], self.parser, self.conf)
+        result = execute(["session"], self.parser, self.conf)
+        assert f"identity: {IDENTITY}" in result
+
+    def test_2_set_network_mainnet(self):
+        execute(["network", "mainnet"], self.parser, self.conf)
+        result = execute(["session"], self.parser, self.conf)
+        assert "network: mainnet" in result
+
+    def test_3_set_network_sepolia(self):
+        execute(["network", "sepolia"], self.parser, self.conf)
+        result = execute(["session"], self.parser, self.conf)
+        assert "network: sepolia" in result
+
+    def test_41_network_create(self):
+        result = execute(["network", "create", self.new_network, INFURA], self.parser, self.conf)
+        assert f"add network with name='{self.new_network}'" in result
+
+    def test_42_network_create_confirmation(self):
+        execute(["network", self.new_network], self.parser, self.conf)
+        result = execute(["session"], self.parser, self.conf)
+        execute(["network", "sepolia"], self.parser, self.conf)
+        assert "network: ", self.new_network in result
+
+    def test_5_set_infura(self):
+        execute(["set", "default_eth_rpc_endpoint", INFURA], self.parser, self.conf)
+        result = execute(["session"], self.parser, self.conf)
+        assert INFURA_KEY in result
+
+    def test_6_print_account(self):
+        result = execute(["account", "print"], self.parser, self.conf)
+        print(result)
+        assert ADDR in result
+
+if __name__ == "__main__":
+    unittest.main()

snet/cli/test/functional_tests/test_aapreparation.py

@@ -0,0 +1,79 @@
+from func_tests import BaseTest, execute, ADDR
+import unittest
+import re
+
+
+class TestChannels(BaseTest):
+    def setUp(self):
+        super().setUp()
+        self.org_id = "SNet"
+        self.org_2_id = "singularitynet"
+        self.amount = "0.001"
+        self.password = "12345"
+        self.group = "default_group"
+        data = execute(["channel", "print-filter-sender"], self.parser, self.conf)
+        lines = data.split("\n")
+        self.max_id = ""
+        for line in lines:
+            parts = line.split()
+            if len(parts) >= 6 and parts[0].isdigit() and parts[-1].isdigit():
+                self.max_id = parts[0]
+
+    def test_channel_1_extend(self):
+        execute(["account", "deposit", self.amount, "-y", "-q"], self.parser, self.conf)
+        if self.max_id:
+            result1 = execute(["channel", "extend-add", self.max_id, "--amount", self.amount, "-y"], self.parser, self.conf)
+        else:
+            block_number = int(execute(["channel", "block-number"], self.parser, self.conf))
+            channel_open_output = execute(["channel", "open", self.org_id, self.group, self.amount, f"{block_number+10000}", "-y", "--open-new-anyway"], self.parser, self.conf)
+            match = re.search(r"#channel_id\s+(\d+)", channel_open_output)
+            self.max_id = match.group(1)
+            execute(["channel", "extend-add", self.max_id, "--amount", self.amount, "-y"], self.parser, self.conf)
+            result1 = execute(["channel", "extend-add", self.max_id, "--amount", self.amount, "-y"], self.parser,
+                              self.conf)
+        # result2 = execute(["channel", "extend-add-for-org", self.org_id, "default_group", "--channel-id", f"{self.max_id}", "-y"], self.parser, self.conf)
+        print(result1)
+        assert "event: ChannelAddFunds" in result1
+
+    def test_channel_2_print_filter_sender(self):
+        result = execute(["channel", "print-filter-sender"], self.parser, self.conf)
+        print(result)
+        assert "Channels for sender: ", ADDR in result
+
+    def test_channel_3_print_filter_group_sender(self):
+        result = execute(["channel", "print-filter-group-sender", self.org_id, self.group], self.parser, self.conf)
+        assert "Channels for sender: ", ADDR in result
+
+    def test_channel_4_print_filter_group(self):
+        result = execute(["channel", "print-filter-group", self.org_id, self.group], self.parser, self.conf)
+        assert self.max_id in result
+
+    def test_channel_5_print_filter_recipient(self):
+        result = execute(["channel", "print-filter-recipient"], self.parser, self.conf)
+        assert "Channels for recipient:", ADDR in result
+
+    def test_channel_5_claim(self):
+        execute(["account", "deposit", self.amount, "-y", "-q"], self.parser, self.conf)
+        if self.max_id:
+            execute(["channel", "extend-add", self.max_id, "--amount", self.amount, "-y"], self.parser, self.conf)
+            result1 = execute(["channel", "claim-timeout", f"{self.max_id}", "-y"], self.parser, self.conf)
+            execute(["account", "deposit", self.amount, "-y", "-q"], self.parser, self.conf)
+            execute(["channel", "extend-add", self.max_id, "--amount", self.amount, "-y"], self.parser, self.conf)
+            result2 = execute(["channel", "claim-timeout-all", "-y"], self.parser, self.conf)
+        else:
+            block_number = int(execute(["channel", "block-number"], self.parser, self.conf))
+            execute(["channel", "open", self.org_id, self.group, self.amount, f"{block_number-1}", "-y"], self.parser, self.conf)
+            execute(["channel", "extend-add", self.max_id, "--amount", self.amount, "-y"], self.parser, self.conf)
+            result1 = execute(["channel", "claim-timeout", f"{self.max_id}", "-y"], self.parser, self.conf)
+            execute(["account", "deposit", self.amount, "-y", "-q"], self.parser, self.conf)
+            execute(["channel", "extend-add", self.max_id, "--amount", self.amount, "-y"], self.parser, self.conf)
+            result2 = execute(["channel", "claim-timeout-all", "-y"], self.parser, self.conf)
+        print(result1)
+        assert ("event: ChannelSenderClaim" in result1) and ("event: ChannelSenderClaim" in result2)
+
+    def test_channel_6_print_all(self):
+        result = execute(["channel", "print-all", "-ds"], self.parser, self.conf)
+        assert self.max_id in result
+
+if __name__ == "__main__":
+    unittest.main()

snet/cli/test/functional_tests/test_channels.py

@@ -0,0 +1,67 @@
+from func_tests import BaseTest, execute, ADDR, PRIVATE_KEY, INFURA
+import os
+import unittest
+
+
+class TestClient(BaseTest):
+    def setUp(self):
+        super().setUp()
+        self.org_id = "egor-sing-test"
+        self.service_id = "hate-detection"
+        self.group = "default_group"
+        self.identity_name = "some__name"
+        self.method = "runsync"
+        self.params = "./detection.json"
+        self.endpoint = "https://ai-ui-service.singularitynet.io:8001"
+        self.max_id = "357"
+        self.nonce = "1"
+        self.amount_in_cogs = "1"
+
+    def test_0_preparations(self):
+        identity_list=execute(["identity", "list"], self.parser, self.conf)
+        if self.identity_name not in identity_list:
+            execute(["identity", "create", self.identity_name, "key", "--private-key", PRIVATE_KEY, "-de"], self.parser, self.conf)
+        execute(["network", "sepolia"], self.parser, self.conf)
+        result = execute(["session"], self.parser, self.conf)
+        assert "network: sepolia" in result
+
+    def test_1_channel_open(self):
+        execute(["set", "default_eth_rpc_endpoint", INFURA], self.parser, self.conf)
+        execute(["account", "deposit", "0.1", "-y"], self.parser, self.conf)
+        self.block=int(execute(["channel", "block-number"], self.parser, self.conf))
+        print(self.block)
+        result = execute(["channel", "print-filter-group-sender", self.org_id, self.group], self.parser, self.conf)
+        if ADDR not in result[13:]:
+            result=execute(["channel", "open", self.org_id, "default_group", "0.0001", f"{self.block+100}", "-y"], self.parser, self.conf)
+        else:
+            pass
+        print(result)
+        assert "#channel_id" in result
+
+    def test_2_service_call(self):
+        params_file = open("detection.json", "w+")
+        params_file.write("""{
+    "input": {
+        "text": "Hello man answer me soon"
+    }
+}
+        """)
+        params_file.close()
+        result=execute(["client", "call", self.org_id, self.service_id, self.group, self.method, self.params, "-y"], self.parser, self.conf)
+        assert "spam" in result
+
+    def test_3_service_get_channel_state(self):
+        result=execute(["client", "get-channel-state", self.max_id, self.endpoint], self.parser, self.conf)
+        assert "current_unspent_amount_in_cogs = " in result
+
+    def test_4_call_low_level(self):
+        result = execute(["client", "call-lowlevel", self.org_id, self.service_id, self.group, self.max_id, self.nonce, self.amount_in_cogs, self.method, self.params], self.parser, self.conf)
+        assert "spam" in result
+
+    def test_5_get_api_registry(self):
+        execute(["service", "get-api-registry", self.org_id, self.service_id, "./"], self.parser, self.conf)
+        assert os.path.exists("./hate.proto")
+
+
+if __name__ == "__main__":
+    unittest.main()

snet/cli/test/functional_tests/test_client.py

@@ -0,0 +1,29 @@
+import unittest
+from func_tests import BaseTest, execute, ADDR
+import importlib
+
+
+class TestCommands(BaseTest):
+    def setUp(self):
+        super().setUp()
+
+    def test_balance_output(self):
+        result = execute(["account", "balance"], self.parser, self.conf)
+        assert len(result.split("\n")) >= 4
+
+    def test_balance_address(self):
+        result = execute(["account", "balance"], self.parser, self.conf)
+        assert result.split("\n")[0].split()[1] == ADDR
+
+    def test_version(self):
+        file_path = "./version.py"
+        spec = importlib.util.spec_from_file_location("version", file_path)
+        version_module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(version_module)
+        self.version = version_module.__version__
+        result = execute(["version"], self.parser, self.conf)
+        assert f"version: {self.version}" in result
+
+
+if __name__ == "__main__":
+    unittest.main()

snet/cli/test/functional_tests/test_commands.py

@@ -0,0 +1,71 @@
+from func_tests import BaseTest, execute, ADDR
+import unittest
+
+class TestContract(BaseTest):
+    def setUp(self):
+        super().setUp()
+        self.last_channel_id = int(execute(["contract", "MultiPartyEscrow", "nextChannelId"], self.parser, self.conf)) - 1
+        self.amount = "1"
+        self.block_number = int(execute(["channel", "block-number"], self.parser, self.conf))
+        self.channel = execute(["channel", "print-filter-sender"], self.parser, self.conf).split()[11]
+
+    def test_SingularityNetToken_name(self):
+        result = execute(["contract", "SingularityNetToken", "name"], self.parser, self.conf)
+        assert "SingularityNet Token" in result
+
+    def test_SingularityNetToken_decimals(self):
+        result = execute(["contract", "SingularityNetToken", "decimals"], self.parser, self.conf)
+        assert "8" in result
+
+    def test_SingularityNetToken_symbol(self):
+        result = execute(["contract", "SingularityNetToken", "symbol"], self.parser, self.conf)
+        assert "AGIX" in result
+
+    def test_SingularityNetToken_totalSupply(self):
+        result = int(execute(["contract", "SingularityNetToken", "totalSupply"], self.parser, self.conf))
+        assert 1772090920768158 <= result
+
+    def test_SingularityNetToken_paused(self):
+        result=execute(["contract", "SingularityNetToken", "paused"], self.parser, self.conf)
+        assert "False" in result
+
+    def test_MultiPartyEscrow_1_balances(self):
+        result = int(execute(["contract", "MultiPartyEscrow", "balances", ADDR], self.parser, self.conf))
+        assert result > 0
+
+    def test_MultiPartyEscrow_token(self):
+        result = execute(["contract", "MultiPartyEscrow", "token"], self.parser, self.conf)
+        assert "0xf703b9aB8931B6590CFc95183be4fEf278732016" in result
+
+    def test_MultiPartyEscrow_channels(self):
+        result = execute(["contract", "MultiPartyEscrow", "channels", f"{self.last_channel_id}"], self.parser, self.conf)
+        assert len(result) > 150
+
+    def test_MultiPartyEscrow_addFunds(self):
+        print(self.channel)
+        result = execute(["contract", "MultiPartyEscrow", "channelAddFunds", self.channel, self.amount, "-y"], self.parser, self.conf)
+        assert "event: ChannelAddFunds" in result
+
+    def test_MultiPartyEscrow_channelExtend(self):
+        result = execute(["contract", "MultiPartyEscrow", "channelExtend", self.channel, f"{self.block_number+100}", "-y"], self.parser, self.conf).split()[11]
+        assert "event: ChannelExtend" in result
+
+    def test_Registry_getOrganizationById(self):
+        result = execute(["contract", "Registry", "getOrganizationById", "SNet"], self.parser, self.conf)
+        assert len(result) > 400
+
+    def test_Registry_listOrganizations(self):
+        result = execute(["contract", "Registry", "listOrganizations"], self.parser, self.conf)
+        assert len(result) > 12000
+
+    def test_Registry_getServiceRegistrationById(self):
+        result = execute(["contract", "Registry", "getServiceRegistrationById", "SNet", "example-service-constructor"], self.parser, self.conf)
+        assert "True" in result
+
+    def test_Registry_listServicesForOrganization(self):
+        result = execute(["contract", "Registry", "listServicesForOrganization", "SNet"], self.parser, self.conf)
+        assert "True" in result
+
+
+if __name__ == "__main__":
+    unittest.main()

snet/cli/test/functional_tests/test_contract.py

@@ -0,0 +1,35 @@
+import unittest
+from func_tests import BaseTest, execute, ADDR
+import math
+
+
+class TestDepositWithdrawTransfer(BaseTest):
+    def setUp(self):
+        super().setUp()
+        self.balance_1: int
+        self.balance_2: int
+        self.amount = 0.1
+
+    def test_deposit(self):
+        result = execute(["account", "balance"], self.parser, self.conf)
+        self.balance_1 = float(result.split("\n")[3].split()[1])
+        execute(["account", "deposit", f"{self.amount}", "-y", "-q"], self.parser, self.conf)
+        result = execute(["account", "balance"], self.parser, self.conf)
+        self.balance_2 = float(result.split("\n")[3].split()[1])
+        assert math.isclose(self.balance_2, self.balance_1 + self.amount, rel_tol=1e-3)
+
+    def test_withdraw(self):
+        result = execute(["account", "balance"], self.parser, self.conf)
+        self.balance_1 = float(result.split("\n")[3].split()[1])
+        execute(["account", "withdraw", f"{self.amount}", "-y", "-q"], self.parser, self.conf)
+        result = execute(["account", "balance"], self.parser, self.conf)
+        self.balance_2 = float(result.split("\n")[3].split()[1])
+        assert math.isclose(self.balance_2, self.balance_1 - self.amount, rel_tol=1e-3)
+
+    def test_transfer(self):
+        result = execute(["account", "transfer", ADDR, f"{self.amount}", "-y"], self.parser, self.conf)
+        assert "TransferFunds" in result
+
+
+if __name__ == "__main__":
+    unittest.main()

snet/cli/test/functional_tests/test_deposit_withdraw_transfer.py

@@ -0,0 +1,45 @@
+import unittest
+from func_tests import BaseTest, execute, PRIVATE_KEY
+import unittest.mock as mock
+from snet.cli.commands.commands import BlockchainCommand
+
+
+class TestEncryptionKey(BaseTest):
+    def setUp(self):
+        super().setUp()
+        self.key = PRIVATE_KEY
+        self.password = "some_pass"
+        self.name = "some_name"
+        self.default_name = "default_name"
+        result = execute(["identity", "list"], self.parser, self.conf)
+        if self.default_name not in result:
+            execute(["identity", "create", self.default_name, "key", "--private-key", self.key, "-de"],
+                             self.parser,
+                             self.conf)
+
+    def test_1_create_identity_with_encryption_key(self):
+        with mock.patch('getpass.getpass', return_value=self.password):
+            execute(["identity", "create", self.name, "key", "--private-key", self.key],
+                             self.parser,
+                             self.conf)
+            result = execute(["identity", "list"], self.parser, self.conf)
+            assert self.name in result
+
+    def test_2_get_encryption_key(self):
+        with mock.patch('getpass.getpass', return_value=self.password):
+            execute(["identity", self.name], self.parser, self.conf)
+            cmd = BlockchainCommand(self.conf, self.parser.parse_args(['session']))
+            enc_key = cmd.config.get_session_field("private_key")
+            res_key = cmd._get_decrypted_secret(enc_key)
+            assert res_key == self.key
+
+    def test_3_delete_identity(self):
+        with mock.patch('getpass.getpass', return_value=self.password):
+            execute(["identity", self.default_name], self.parser, self.conf)
+            execute(["identity", "delete", self.name], self.parser, self.conf)
+            result = execute(["identity", "list"], self.parser, self.conf)
+            assert self.name not in result
+
+
+if __name__ == "__main__":
+    unittest.main()

snet/cli/test/functional_tests/test_encryption_key.py

@@ -65,4 +65,7 @@ def test_version(self):
 
 
 if __name__ == '__main__':
-    unittest.main()
+    cli_tests = unittest.TestLoader().loadTestsFromTestCase(TestEntryPoint)
+    functional_tests = unittest.TestLoader().discover("./snet/cli/test/functional_tests", pattern="test_*.py")
+    all_tests = unittest.TestSuite([cli_tests, functional_tests])
+    unittest.TextTestRunner().run(all_tests)