From 5047b2bae6efe824c1666a2807d8d4ee195ec89b Mon Sep 17 00:00:00 2001 From: ychang Date: Tue, 12 Dec 2023 18:40:35 +0800 Subject: [PATCH 1/5] update testcase --- .github/workflows/unittest.yml | 6 +- .gitignore | 3 + cobo_custody/config/__init__.py | 35 +- test.py | 374 ------------------ tests/__init__.py | 0 tests/run_test.py | 27 ++ tests/test_requirements.txt | 1 + tests/testcase/__init__.py | 0 tests/testcase/test_client.py | 279 +++++++++++++ .../testcase/test_mpc_client.py | 109 ++--- .../testcase/test_web3_client.py | 51 +-- tox.ini | 8 + 12 files changed, 383 insertions(+), 510 deletions(-) delete mode 100644 test.py create mode 100644 tests/__init__.py create mode 100644 tests/run_test.py create mode 100644 tests/test_requirements.txt create mode 100644 tests/testcase/__init__.py create mode 100644 tests/testcase/test_client.py rename test_mpc.py => tests/testcase/test_mpc_client.py (61%) rename test_web3.py => tests/testcase/test_web3_client.py (79%) create mode 100644 tox.ini diff --git a/.github/workflows/unittest.yml b/.github/workflows/unittest.yml index 8a6fba3..9eedc72 100644 --- a/.github/workflows/unittest.yml +++ b/.github/workflows/unittest.yml @@ -16,7 +16,9 @@ on: push: branches: [ main ] pull_request: - branches: [ main ] + branches: + - main + - 'main/**' jobs: test: @@ -40,5 +42,5 @@ jobs: SECRET: "${{ github.event.inputs.environment == 'prod' && secrets.PRODSECRET || secrets.DEVELOPSECRET}}" Web3SECRET: "${{secrets.DEVELOPWEB3SECRET}}" MPCSECRET: "${{secrets.DEVELOPMPCSECRET}}" - run: python test.py --env $ENV --secret $SECRET --web3ApiSecret $Web3SECRET --mpcApiSecret $MPCSECRET + run: python tests/run_test.py --api_secret $SECRET --mpc_api_secret $MPCSECRET diff --git a/.gitignore b/.gitignore index 9702015..1f9f6ec 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,6 @@ venv /cobo_python_api.egg-info cobo_custody.egg-info .DS_Store +__pycache__/ +.python-version +.tox \ No newline at end of file diff --git a/cobo_custody/config/__init__.py b/cobo_custody/config/__init__.py index 386b938..ba61781 100644 --- a/cobo_custody/config/__init__.py +++ b/cobo_custody/config/__init__.py @@ -1,37 +1,6 @@ from cobo_custody.config.env import Env DEV_ENV = Env(host="https://api.dev.cobo.com", - coboPub="03596da539963fb1dd29d5859e25903eb76b9f7ed2d58516e29c9f80c201ff2c1b") + coboPub="03596da539963fb1dd29d5859e25903eb76b9f7ed2d58516e29c9f80c201ff2c1b") PROD_ENV = Env(host="https://api.custody.cobo.com", - coboPub="02c3e5bacf436fbf4da78597e791579f022a2e85073ae36c54a361ff97f2811376") - -DEV_TEST_DATA = { - "cobo_id": "20220314181458000331767000003732", - "tx_id": "0x1c4d137bc2a2ee8f22cbdf9e90405974e72e65d922f42eb81d9f7a05d0f64fc6", - "withdraw_id": "web_send_by_user_915_1647252768642", - "deposit_address": {"BTC": "3JBYNrbB4bHtGWHTEa3ZPuRK9kwTiEUo4D", - "XRP": "rfKyCMyoV6Ln2GZ7YDbrBrnXCbAyBbxRqB|2047482901"}, - "deposit_addresses": {"BTC": "3JBYNrbB4bHtGWHTEa3ZPuRK9kwTiEUo4D,bc1qf22hpu33u2tkyy528mdvpnre45n8lu5s3ycatu", - "XRP": "rfKyCMyoV6Ln2GZ7YDbrBrnXCbAyBbxRqB|2047482901," - "rfKyCMyoV6Ln2GZ7YDbrBrnXCbAyBbxRqB|3752417374"}, - "loop_address": {"BTC": "35eXJPLRTSp4Wn8n2f6pkQF4t3KdU2cuhz", - "XRP": "rfKyCMyoV6Ln2GZ7YDbrBrnXCbAyBbxRqB|477817505"}, - "loop_addresses": {"BTC": "35eXJPLRTSp4Wn8n2f6pkQF4t3KdU2cuhz,34R4JHecUwGNEFVGKz1vR8R6BHGi5FUqPt", - "XRP": "rfKyCMyoV6Ln2GZ7YDbrBrnXCbAyBbxRqB|477817505," - "rfKyCMyoV6Ln2GZ7YDbrBrnXCbAyBbxRqB|2874421071"} - } - -PROD_TEST_DATA = { - "cobo_id": "20220311154108000184408000002833", - "tx_id": "4041A888C9966BE8916FE65F2FEE7AE9A9DC3F49D0F1643A768C842CA95FA736", - "withdraw_id": "sdk_request_id_fe80cc5f_1647068483396", - "deposit_address": {"BTC": "36xYx7vf7DUKpJDixpY3EoV2jchFwYSNCb", - "XRP": "rBWpYJhuJWBPAkzJ4kYQqHShSkkF3rgeD|3992922539"}, - "deposit_addresses": {"BTC": "36xYx7vf7DUKpJDixpY3EoV2jchFwYSNCb,bc1q0l24tf5sjdu9t7l6hrlhxz9aq9yeej9h2sc7tk", - "XRP": "rBWpYJhuJWBPAkzJ4kYQqHShSkkF3rgeD|3992922539," - "rBWpYJhuJWBPAkzJ4kYQqHShSkkF3rgeD|1492154866"}, - "loop_address": {"BTC": "34WLjtk9ta96BVxc1jRF7j5eVvehoftsVV", - "XRP": "rBWpYJhuJWBPAkzJ4kYQqHShSkkF3rgeD|633829231"}, - "loop_addresses": {"BTC": "34WLjtk9ta96BVxc1jRF7j5eVvehoftsVV,33P1kjMfDCKipR58S7XbsCqbmPT5YGrhUo", - "XRP": "rBWpYJhuJWBPAkzJ4kYQqHShSkkF3rgeD|633829231,rBWpYJhuJWBPAkzJ4kYQqHShSkkF3rgeD|935940214"} - } + coboPub="02c3e5bacf436fbf4da78597e791579f022a2e85073ae36c54a361ff97f2811376") \ No newline at end of file diff --git a/test.py b/test.py deleted file mode 100644 index 61d1ae1..0000000 --- a/test.py +++ /dev/null @@ -1,374 +0,0 @@ -from cobo_custody.config import DEV_ENV -from cobo_custody.config import DEV_TEST_DATA -from cobo_custody.config import PROD_ENV -from cobo_custody.config import PROD_TEST_DATA - -import unittest -from cobo_custody.client import Client -from cobo_custody.model.enums import SortFlagEnum -from cobo_custody.signer.local_signer import LocalSigner -from parameterized import param, parameterized -import sys -import argparse -import time -from hashlib import sha256 - - -class ClientTest(unittest.TestCase): - api_secret = "api_secret" - ENV = DEV_ENV - TEST_DATA = DEV_TEST_DATA - - def setUp(self): - self.client = Client(signer=LocalSigner(self.api_secret), - env=self.ENV, - debug=False) - - # account and address - def test_get_account_info(self): - response = self.client.get_account_info() - self.assertTrue(response.success) - - @parameterized.expand( - [ - param(coin="BTC"), - param(coin="ETH"), - param(coin="ETH_USDT"), - param(coin="XRP"), - ] - ) - def test_get_valid_coin_info(self, coin): - response = self.client.get_coin_info(coin) - self.assertTrue(response.success) - - @parameterized.expand( - [ - param(coin="BTTB") - ] - ) - def test_get_invalid_coin_info(self, coin): - response = self.client.get_coin_info(coin) - self.assertFalse(response.success) - self.assertEqual(response.exception.errorCode, 12002) - - @parameterized.expand( - [ - param(coin="BTC", native_segwit=True), - param(coin="BTC", native_segwit=False), - param(coin="ETH", native_segwit=False), - param(coin="ETH_USDT", native_segwit=False), - param(coin="XRP", native_segwit=False), - ] - ) - def test_new_valid_deposit_address(self, coin, native_segwit): - response = self.client.new_deposit_address(coin=coin, native_segwit=native_segwit) - self.assertTrue(response.success) - - @parameterized.expand( - [ - param(coin="BTTB", native_segwit=True), - param(coin="ETTE", native_segwit=False) - ] - ) - def test_new_invalid_deposit_address(self, coin, native_segwit): - response = self.client.new_deposit_address(coin=coin, native_segwit=native_segwit) - self.assertFalse(response.success) - self.assertEqual(response.exception.errorCode, 12002) - - - @parameterized.expand( - [ - param(coin="BTC", native_segwit=True, count=2), - param(coin="BTC", native_segwit=False, count=2), - param(coin="ETH", native_segwit=False, count=2), - param(coin="ETH_USDT", native_segwit=False, count=2), - param(coin="XRP", native_segwit=False, count=2), - ] - ) - def test_batch_new_valid_deposit_address(self, coin, native_segwit, count): - response = self.client.batch_new_deposit_address(coin=coin, native_segwit=native_segwit, count=count) - self.assertTrue(response.success) - - @parameterized.expand( - [ - param(coin="BTTB", native_segwit=True, count=2), - param(coin="ETTE", native_segwit=False, count=2) - ] - ) - def test_batch_new_invalid_deposit_address(self, coin, native_segwit, count): - response = self.client.batch_new_deposit_address(coin=coin,native_segwit=native_segwit, count=count) - self.assertFalse(response.success) - self.assertEqual(response.exception.errorCode, 12002) - - @parameterized.expand( - [ - param(coin="BTC"), - param(coin="XRP") - ] - ) - def test_verify_valid_deposit_address(self, coin): - response = self.client.verify_deposit_address(coin=coin, address=self.TEST_DATA["deposit_address"][coin]) - self.assertTrue(response.success) - - @parameterized.expand( - [ - param(coin="BTC", address="3Kd5rjiLtvpHv5nhYQNTTeRLgrz4om32PJ"), - param(coin="XRP", address="rndm7RphBZG6CpZvKcG9AjoFbSvcKhwLCx") - ] - ) - def test_verify_invalid_deposit_address(self, coin, address): - response = self.client.verify_deposit_address(coin=coin, address=address) - self.assertFalse(response.success) - self.assertEqual(response.exception.errorCode, 12015) - - @parameterized.expand( - [ - param(coin="BTC", count=2), - param(coin="XRP", count=2) - ] - ) - def test_batch_verify_valid_deposit_address(self, coin, count): - response = self.client.batch_verify_deposit_address(coin=coin, addresses= self.TEST_DATA["deposit_addresses"][coin]) - self.assertTrue(response.success) - self.assertEqual(len(response.result["addresses"].split(",")), count) - - @parameterized.expand( - [ - param(coin="BTC", addresses="3Kd5rjiLtvpHv5nhYQNTTeRLgrz4om32PJ,bc1q9unqc738dxjg5mk8zqtz33zg59cahrj29s24lp", count=0), - param(coin="XRP", addresses="rndm7RphBZG6CpZvKcG9AjoFbSvcKhwLCx,rrBD4sBsxrpzbohAEYWH4moPSsoxupWLA|00000000", count=0) - ] - ) - def test_batch_verify_invalid_deposit_address(self, coin, addresses, count): - response = self.client.batch_verify_deposit_address(coin=coin, addresses=addresses) - self.assertTrue(response.success) - self.assertEqual(len(response.result["addresses"]), count) - - @parameterized.expand( - [ - param(coin="BTC", address="3Kd5rjiLtvpHv5nhYQNTTeRLgrz4om32PJ"), - param(coin="BTC", address="bc1q9unqc738dxjg5mk8zqtz33zg59cahrj29s24lp"), - param(coin="ETH", address="0xE410157345be56688F43FF0D9e4B2B38Ea8F7828"), - param(coin="ETH_USDT", address="0xEEACb7a5e53600c144C0b9839A834bb4b39E540c"), - param(coin="XRP", address="rndm7RphBZG6CpZvKcG9AjoFbSvcKhwLCx"), - param(coin="XRP", address="rGNXLMNHkUEtoo7qkCSHEm2sfMo8F969oZ|2200701580") - ] - ) - def test_verify_valid_address(self, coin, address): - response = self.client.verify_valid_address(coin=coin, address=address) - self.assertTrue(response.success) - - @parameterized.expand( - [ - param(coin="BTC", address="0xE410157345be56688F43FF0D9e4B2B38Ea8F7828"), - param(coin="XRP", address="rBWpYJhuJWBPAkzJ4kYQqHShSkkF3rgeDE"), - ] - ) - def test_verify_invalid_address(self, coin, address): - response = self.client.verify_valid_address(coin, address) - self.assertTrue(response.success) - self.assertFalse(response.result) - - @parameterized.expand( - [ - param(coin="BTC"), - param(coin="ETH"), - param(coin="ETH_USDT"), - param(coin="XRP"), - ] - ) - def test_get_valid_address_history(self, coin): - response = self.client.get_address_history(coin=coin) - self.assertTrue(response.success) - self.assertTrue(len(response.result) > 0) - - @parameterized.expand( - [ - param(coin="BTTB"), - ] - ) - def test_get_invalid_address_history(self, coin): - response = self.client.get_address_history(coin=coin) - self.assertFalse(response.success) - self.assertEqual(response.exception.errorCode, 12002) - - @parameterized.expand( - [ - param(coin="BTC", page_index=0, page_length=10), - ] - ) - def test_get_invalid_address_history_with_valid_page(self, coin, page_index, page_length): - response = self.client.get_address_history(coin=coin, page_index=page_index, page_length=page_length) - self.assertTrue(response.success) - self.assertTrue(len(response.result) > 0) - - @parameterized.expand( - [ - param(coin="BTC", page_index=0, page_length=51), - param(coin="BTC", page_index=0, page_length=0), - ] - ) - def test_get_invalid_address_history_with_invalid_page(self, coin, page_index, page_length): - response = self.client.get_address_history(coin=coin, page_index=page_index, page_length=page_length) - self.assertFalse(response.success) - self.assertEqual(response.exception.errorCode, 1011) - - @parameterized.expand( - [ - param(coin="BTC", page_index=0, page_length=10, sort_flag=SortFlagEnum.ASCENDING), - ] - ) - def test_get_address_history_with_sort(self, coin, page_index, page_length, sort_flag): - response = self.client.get_address_history( - coin=coin, page_index=page_index, page_length=page_length, sort_flag=sort_flag) - self.assertTrue(response.success) - self.assertTrue(len(response.result) > 0) - - @parameterized.expand( - [ - param(coin="BTC", memo=False), - param(coin="XRP", memo=True), - ] - ) - def test_check_loop_address_details(self, coin, memo=None): - if memo: - address = self.TEST_DATA["loop_address"][coin].split("|")[0] - mome_info = self.TEST_DATA["loop_address"][coin].split("|")[1] - else: - address = self.TEST_DATA["loop_address"][coin] - mome_info =None - - response = self.client.check_loop_address_details(coin=coin, address=address, memo=mome_info) - self.assertTrue(response.success) - self.assertTrue(response.result["is_internal_address"]) - - @parameterized.expand( - [ - param(coin="BTC"), - param(coin="XRP"), - ] - ) - def test_verify_loop_address_list(self, coin): - response = self.client.verify_loop_address_list(coin=coin, addresses=self.TEST_DATA["loop_addresses"][coin]) - self.assertTrue(response.success) - for address_info in response.result: - self.assertTrue(address_info["is_internal_address"]) - - def test_get_transaction_details(self): - response = self.client.get_transaction_details(self.TEST_DATA["cobo_id"]) - self.assertTrue(response.success) - - def test_get_transaction_by_txid(self): - response = self.client.get_transaction_by_txid(self.TEST_DATA["tx_id"]) - self.assertTrue(response.success) - - def test_get_transactions_by_id(self): - response = self.client.get_transactions_by_id() - self.assertTrue(response.success) - - def test_get_transactions_by_time(self): - response = self.client.get_transactions_by_time() - self.assertTrue(response.success) - - def test_get_transactions_by_time_ex(self): - response = self.client.get_transactions_by_time_ex() - self.assertTrue(response.success) - - def test_get_pending_transactions(self): - response = self.client.get_pending_transactions() - self.assertTrue(response.success) - - def test_get_transaction_history(self): - response = self.client.get_transaction_history() - self.assertTrue(response.success) - - def test_get_pending_transaction(self): - response = self.client.get_pending_transactions() - if len(response.result): - pending_id = response.result[0]["id"] - response = self.client.get_pending_transaction(id=pending_id) - self.assertTrue(response.success) - else: - self.skipTest("no pending transactions") - - def test_get_transactions_by_request_ids(self): - request_ids = "web_send_by_user_1049_1677469364250,slp_uc_1559_1675822822477,web_send_by_user_1212_1678330749364" - response = self.client.get_transactions_by_request_ids(request_ids=request_ids) - print(response) - self.assertTrue(response.success) - - @parameterized.expand( - [ - param(coin="COBO_ETH", address="0xE410157345be56688F43FF0D9e4B2B38Ea8F7828", memo=None, amount=1), - param(coin="XLM", address="GBJDU6TPWHKGV7HRLNTIBA46MG3MB5DUG6BISHX3BF7I75H2HLPV6RJX", memo="4e73f03b", amount=1) - ] - ) - def test_withdraw(self, coin, address, memo, amount): - request_id = f"request_id_{sha256(address.encode()).digest().hex()[:8]}_{str(int(time.time() * 1000))}" - response = self.client.withdraw(coin=coin, address=address, memo=memo, amount=amount, request_id=request_id) - self.assertTrue(response.success) - - def test_query_withdraw_info(self): - response = self.client.query_withdraw_info(self.TEST_DATA["withdraw_id"]) - self.assertTrue(response.success) - - def test_get_staking_product_list(self): - response = self.client.get_staking_product_list() - self.assertTrue(response.success) - - def test_get_staking_product_details(self): - response = self.client.get_staking_product_list() - product_id = response.result[0]['product_id'] - response = self.client.get_staking_product_details(product_id) - self.assertTrue(response.success) - - def test_stake(self): - response = self.client.get_staking_product_list(coin="TETH") - if len(response.result)>0: - product_id = response.result[0]['product_id'] - response = self.client.stake(product_id, 1000000) - self.assertTrue(response.success) - else: - self.skipTest("no TETH staking product") - - def test_unstake(self): - response = self.client.get_staking_product_list(coin="TETH") - if len(response.result)>0: - product_id = response.result[0]['product_id'] - response = self.client.unstake(product_id, 1000000) - self.assertTrue(response.success) - else: - self.skipTest("no TETH staking product") - - def test_get_stakings(self): - response = self.client.get_stakings() - self.assertTrue(response.success) - - def test_get_unstakings(self): - response = self.client.get_unstakings() - self.assertTrue(response.success) - - def test_get_staking_history(self): - response = self.client.get_staking_history() - self.assertTrue(response.success) - - -if __name__ == '__main__': - - if len(sys.argv) > 1: - parser = argparse.ArgumentParser() - parser.add_argument("--env", nargs='?', default="develop") - parser.add_argument("--secret", type=str, required=True) - args = parser.parse_args() - env = args.env if args.env else "develop" - api_secret = args.secret - - ClientTest.api_secret = api_secret - ClientTest.ENV = DEV_ENV if env == "develop" else PROD_ENV - ClientTest.TEST_DATA = DEV_TEST_DATA if env == "develop" else PROD_TEST_DATA - - # unittest.main() - runner = unittest.TextTestRunner() - suite = unittest.TestLoader().loadTestsFromTestCase(ClientTest) - runner.run(suite) - - diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/run_test.py b/tests/run_test.py new file mode 100644 index 0000000..5707892 --- /dev/null +++ b/tests/run_test.py @@ -0,0 +1,27 @@ +import unittest +import argparse +from cobo_custody.config import DEV_ENV +from testcase.test_client import ClientTest +from testcase.test_mpc_client import MPCClientTest + + +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument("--api_secret", type=str, required=True) + parser.add_argument("--mpc_api_secret", type=str, required=True) + + args = parser.parse_args() + + ClientTest.api_secret = args.api_secret + ClientTest.env = DEV_ENV + + MPCClientTest.mpc_api_secret = args.mpc_api_secret + MPCClientTest.env = DEV_ENV + + loader = unittest.TestLoader() + suite = unittest.TestSuite() + + for testcase in (ClientTest, MPCClientTest): + suite.addTests(loader.loadTestsFromTestCase(testcase)) + runner = unittest.TextTestRunner(verbosity=3) + runner.run(suite) diff --git a/tests/test_requirements.txt b/tests/test_requirements.txt new file mode 100644 index 0000000..ec6ecae --- /dev/null +++ b/tests/test_requirements.txt @@ -0,0 +1 @@ +parameterized===0.8.1 diff --git a/tests/testcase/__init__.py b/tests/testcase/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/testcase/test_client.py b/tests/testcase/test_client.py new file mode 100644 index 0000000..1a6dd7a --- /dev/null +++ b/tests/testcase/test_client.py @@ -0,0 +1,279 @@ +import unittest +import time +from cobo_custody.client import Client +from cobo_custody.model.enums import SortFlagEnum +from cobo_custody.signer.local_signer import LocalSigner + +from parameterized import param, parameterized +from hashlib import sha256 + + +class ClientTest(unittest.TestCase): + api_secret = "" + env = None + + @classmethod + def setUpClass(cls): + cls.client = Client(signer=LocalSigner(cls.api_secret), + env=cls.env, + debug=False) + + # account and address + def test_get_account_info(self): + response = self.client.get_account_info() + self.assertTrue(response.success) + + @parameterized.expand( + [ + param(coin="XRP"), + ] + ) + def test_get_valid_coin_info(self, coin): + response = self.client.get_coin_info(coin) + self.assertTrue(response.success) + + @parameterized.expand( + [ + param(coin="BTC", native_segwit=True, start="bc1"), + param(coin="BTC", native_segwit=False, start="3"), + param(coin="BTC", native_segwit=None, start="3"), + ] + ) + def test_new_valid_deposit_address(self, coin, native_segwit, start): + response = self.client.new_deposit_address(coin=coin, native_segwit=native_segwit) + self.assertTrue(response.success) + self.assertTrue(response.result["address"].startswith(start)) + + @parameterized.expand( + [ + param(coin="BTC", native_segwit=None, count=2), + ] + ) + def test_batch_new_valid_deposit_address(self, coin, native_segwit, count): + response = self.client.batch_new_deposit_address(coin=coin, native_segwit=native_segwit, count=count) + self.assertTrue(response.success) + + @parameterized.expand( + [ + param(coin="BTC", address="38kcymiNQXk8WTWX9tPLRZP9wxvXPXcsFy"), + ] + ) + def test_verify_valid_deposit_address(self, coin, address): + response = self.client.verify_deposit_address(coin, address) + self.assertTrue(response.success) + + @parameterized.expand( + [ + param(coin="BTC", addresses="38kcymiNQXk8WTWX9tPLRZP9wxvXPXcsFy,3ApTsekq5XpUtM5CzAKqntHkvoSpYdCDHw"), + ] + ) + def test_batch_verify_valid_deposit_address(self, coin, addresses): + response = self.client.batch_verify_deposit_address(coin, addresses) + self.assertTrue(response.success) + + @parameterized.expand( + [ + param(coin="BTC", address="38kcymiNQXk8WTWX9tPLRZP9wxvXPXcsFy") + ] + ) + def test_verify_valid_address(self, coin, address): + response = self.client.verify_valid_address(coin=coin, address=address) + self.assertTrue(response.success) + + def test_get_valid_address_history(self): + coin = "BTC" + response = self.client.get_address_history(coin) + self.assertTrue(response.success) + + coin = "BTC" + page_index = "1" + page_length = "1" + sort_flag = SortFlagEnum.ASCENDING + response = self.client.get_address_history(coin, page_index, page_length, sort_flag) + self.assertTrue(response.success) + + def test_check_loop_address_details(self): + coin = "XRP" + address = "rBphERztHKga1cyMgWiDen7WDkbkfn1iPE" + memo = "3414236551" + response = self.client.check_loop_address_details(coin=coin, address=address, memo=memo) + self.assertTrue(response.success) + self.assertFalse(response.result["is_internal_address"]) + + coin = "BTC" + address = "38kcymiNQXk8WTWX9tPLRZP9wxvXPXcsFy" + memo = None + response = self.client.check_loop_address_details(coin=coin, address=address, memo=memo) + self.assertTrue(response.success) + self.assertFalse(response.result["is_internal_address"]) + + def test_verify_loop_address_list(self): + coin = "BTC" + address = "38kcymiNQXk8WTWX9tPLRZP9wxvXPXcsFy,32WT5Thw3o2awNX113EbX9kNHgn7VcuRsi" + + response = self.client.verify_loop_address_list(coin=coin, addresses=address) + self.assertTrue(response.success) + + def test_get_transaction_details_by_coboID(self): + cobo_id = "20231206130534000116223000007648" + response = self.client.get_transaction_details(cobo_id) + self.assertTrue(response.success) + + def test_get_transaction_by_txid(self): + tx_id = "0x36cec6c56310172325d0c4ac342447b3b93aa248289fea2e2a6740857a6a59a0" + response = self.client.get_transaction_by_txid(tx_id) + self.assertTrue(response.success) + + def test_get_transactions_by_id(self): + response = self.client.get_transactions_by_id() + self.assertTrue(response.success) + + coin = "COBO_ETH" + side = "deposit" + address = "0xefeff29688deaa32c20cdb4fc54d8285a31ada4e" + max_id = "20231206130534000116223000007648" + response = self.client.get_transactions_by_id(coin=coin, side=side, address=address, max_id=max_id) + self.assertTrue(response.success) + + coin = "COBO_ETH" + min_id = "20231206130534000116223000007648" + response = self.client.get_transactions_by_id(coin=coin, min_id=min_id) + self.assertTrue(response.success) + + def test_get_transactions_by_time(self): + response = self.client.get_transactions_by_time() + self.assertTrue(response.success) + + coin = "COBO_ETH" + side = "deposit" + address = "0xefeff29688deaa32c20cdb4fc54d8285a31ada4e" + begin_time = "1701839134" + response = self.client.get_transactions_by_time(coin=coin, side=side, address=address, + begin_time=begin_time) + self.assertTrue(response.success) + + end_time = "1701839134" + response = self.client.get_transactions_by_time(coin=coin, side=side, address=address, end_time=end_time) + self.assertTrue(response.success) + + def test_get_transactions_by_time_ex(self): + response = self.client.get_transactions_by_time_ex() + self.assertTrue(response.success) + + coins = "COBO_ETH" + side = 2 + address = "0xefeff29688deaa32c20cdb4fc54d8285a31ada4e" + status = 900 + begin_time = 1701839134 + end_time = 1701839134 + limit = 10 + offset = 0 + order_by = "created_time" + order = "ASC" + + response = self.client.get_transactions_by_time_ex(coins=coins, + side=side, + address=address, + status=status, + begin_time=begin_time, + end_time=end_time, + limit=limit, + offset=offset, + order=order, + order_by=order_by) + self.assertTrue(response.success) + + def test_get_pending_transactions(self): + response = self.client.get_pending_transactions() + self.assertTrue(response.success) + + coin = "COBO_ETH" + side = "deposit" + max_id = "20231206130534000116223000007648" + min_id = "20231206130534000116223000007648" + limit = "1" + + response = self.client.get_pending_transactions(coin=coin, + side=side, + max_id=max_id, + min_id=min_id, + limit=limit) + self.assertTrue(response.success) + + def test_get_transaction_history(self): + response = self.client.get_transaction_history() + self.assertTrue(response.success) + + def test_get_pending_transaction(self): + response = self.client.get_pending_transactions() + if len(response.result): + pending_id = response.result[0]["id"] + response = self.client.get_pending_transaction(id=pending_id) + self.assertTrue(response.success) + else: + self.skipTest("no pending transactions") + + def test_get_transactions_by_request_ids(self): + request_ids = "IntegrationTest-848427217479639409,IntegrationTest-848427217358004633" + response = self.client.get_transactions_by_request_ids(request_ids=request_ids) + self.assertTrue(response.success) + + @parameterized.expand( + [ + param(coin="COBO_ETH", address="0xE410157345be56688F43FF0D9e4B2B38Ea8F7828", memo=None, amount=1), + param(coin="XLM", address="GBJDU6TPWHKGV7HRLNTIBA46MG3MB5DUG6BISHX3BF7I75H2HLPV6RJX", memo="4e73f03b", + amount=1) + ] + ) + def test_withdraw(self, coin, address, memo, amount): + request_id = f"request_id_{sha256(address.encode()).digest().hex()[:8]}_{str(int(time.time() * 1000))}" + response = self.client.withdraw(coin=coin, address=address, memo=memo, amount=amount, request_id=request_id) + self.assertTrue(response.success) + + def test_query_withdraw_info(self): + request_id = "IntegrationTest-848427217479639409" + response = self.client.query_withdraw_info(request_id) + self.assertTrue(response.success) + + def test_get_staking_product_list(self): + response = self.client.get_staking_product_list() + self.assertTrue(response.success) + + def test_get_staking_product_details(self): + response = self.client.get_staking_product_list() + if len(response.result): + product_id = response.result[0]['product_id'] + response = self.client.get_staking_product_details(product_id) + self.assertTrue(response.success) + else: + self.skipTest("no staking product") + + def test_stake(self): + response = self.client.get_staking_product_list(coin="TETH") + if len(response.result) > 0: + product_id = response.result[0]['product_id'] + response = self.client.stake(product_id, 1000000) + self.assertTrue(response.success) + else: + self.skipTest("no TETH staking product") + + def test_unstake(self): + response = self.client.get_staking_product_list(coin="TETH") + if len(response.result) > 0: + product_id = response.result[0]['product_id'] + response = self.client.unstake(product_id, 1000000) + self.assertTrue(response.success) + else: + self.skipTest("no TETH staking product") + + def test_get_stakings(self): + response = self.client.get_stakings() + self.assertTrue(response.success) + + def test_get_unstakings(self): + response = self.client.get_unstakings() + self.assertTrue(response.success) + + def test_get_staking_history(self): + response = self.client.get_staking_history() + self.assertTrue(response.success) diff --git a/test_mpc.py b/tests/testcase/test_mpc_client.py similarity index 61% rename from test_mpc.py rename to tests/testcase/test_mpc_client.py index b9c91f3..c334156 100644 --- a/test_mpc.py +++ b/tests/testcase/test_mpc_client.py @@ -1,73 +1,68 @@ -from cobo_custody.client.mpc_client import MPCClient -from cobo_custody.config import DEV_ENV -from cobo_custody.config import DEV_TEST_DATA - import unittest -from cobo_custody.signer.local_signer import LocalSigner -import sys -import argparse import time +from cobo_custody.signer.local_signer import LocalSigner +from cobo_custody.client.mpc_client import MPCClient class MPCClientTest(unittest.TestCase): mpc_api_secret = "" - ENV = DEV_ENV - TEST_DATA = DEV_TEST_DATA + env = None - def setUp(self): - self.mpc_client = MPCClient(signer=LocalSigner(self.mpc_api_secret), - env=self.ENV, - debug=False) + @classmethod + def setUpClass(cls): + cls.mpc_client = MPCClient(signer=LocalSigner(cls.mpc_api_secret), + env=cls.env, + debug=False) def test_get_supported_chains(self): response = self.mpc_client.get_supported_chains() - print(response) + # print(response) self.assertTrue(response.success) def test_get_supported_coins(self): chain_code = "GETH" response = self.mpc_client.get_supported_coins(chain_code=chain_code) - print(response) + # print(response) self.assertTrue(response.success) def test_get_wallet_supported_coins(self): response = self.mpc_client.get_wallet_supported_coins() - print(response) + # print(response) self.assertTrue(response.success) def test_is_valid_address(self): coin = "GETH" address = "0x3ede1e59a3f3a66de4260df7ba3029b515337e5c" response = self.mpc_client.is_valid_address(coin=coin, address=address) - print(response) + # print(response) self.assertTrue(response.success) def test_generate_addresses(self): chain_code = "GETH" count = 2 response = self.mpc_client.generate_addresses(chain_code=chain_code, count=count) - print(response) + # print(response) self.assertTrue(response.success) def test_update_address_description(self): coin = "GETH" - address = "0x6da78aece2d350ca4078ee66a61c3f4cc49eb9bf" - description = "test1" + address = "0x6a060efe0ff887f4e24dc2d2098020abf28bcce4" + description = "test" response = self.mpc_client.update_address_description(coin=coin, address=address, description=description) - print(response) + # print(response) self.assertTrue(response.success) def test_list_addresses(self): chain_code = "GETH" response = self.mpc_client.list_addresses(chain_code=chain_code) - print(response) + # print(response) self.assertTrue(response.success) def test_get_balance(self): - address = "0x3ede1e59a3f3a66de4260df7ba3029b515337e5c" + address = "0x6a060efe0ff887f4e24dc2d2098020abf28bcce4" coin = "GETH" response = self.mpc_client.get_balance(address=address, coin=coin) - print(response) + # print(response) self.assertTrue(response.success) def test_list_balances(self): @@ -75,47 +70,50 @@ def test_list_balances(self): chain_code = "GETH" page_index = 0 page_length = 50 - response = self.mpc_client.list_balances(coin=coin, chain_code=chain_code, page_index=page_index, page_length=page_length) - print(response) + response = self.mpc_client.list_balances(coin=coin, chain_code=chain_code, page_index=page_index, + page_length=page_length) + # print(response) self.assertTrue(response.success) def test_create_transaction(self): coin = "GETH" request_id = str(int(time.time() * 1000)) - from_addr = "0x3ede1e59a3f3a66de4260df7ba3029b515337e5c" - to_addr = "0xEEACb7a5e53600c144C0b9839A834bb4b39E540c" + from_addr = "0x6a060efe0ff887f4e24dc2d2098020abf28bcce4" + to_addr = "0x6a060efe0ff887f4e24dc2d2098020abf28bcce4" amount = "10" response = self.mpc_client.create_transaction(coin=coin, request_id=request_id, from_addr=from_addr, to_addr=to_addr, amount=amount) - print(response) + # print(response) self.assertTrue(response.success) def test_transactions_by_request_ids(self): - request_ids = "1668678820274" + request_ids = "web_send_by_user_453_1702345165920" response = self.mpc_client.transactions_by_request_ids(request_ids=request_ids) - print(response) + # print(response) self.assertTrue(response.success) def test_transactions_by_tx_hash(self): - tx_hash = "0x1e14311142db1f5b02e587f0e00643f7fd460c81e73dffff65cf501123fb99dd" + tx_hash = "0xc8592ef7821b1d21d92309f902a583cf4f64bbcf500056e663c9e95900f31bbe" response = self.mpc_client.transactions_by_tx_hash(tx_hash=tx_hash) - print(response) + # print(response) self.assertTrue(response.success) def test_list_transactions(self): response = self.mpc_client.list_transactions() - print(response) + # print(response) self.assertTrue(response.success) + # 有bug，待解决；https://pha.1cobo.com/T58503 def test_list_tss_node_requests(self): + self.skipTest("有bug，待解决") response = self.mpc_client.list_tss_node_requests() print(response) self.assertTrue(response.success) def test_list_spendable(self): - coin = "BTC" + coin = "GETH" response = self.mpc_client.list_spendable(coin=coin) - print(response) + # print(response) self.assertTrue(response.success) def test_estimate_fee(self): @@ -123,44 +121,23 @@ def test_estimate_fee(self): amount = 10000 address = "0xEEACb7a5e53600c144C0b9839A834bb4b39E540c" response = self.mpc_client.estimate_fee(coin=coin, amount=amount, address=address) - print(response) + # print(response) self.assertTrue(response.success) def test_retry_double_check(self): - request_id = "123" + request_id = "web_send_by_user_453_1702345165920" response = self.mpc_client.retry_double_check(request_id=request_id) - print(response) + # print(response) self.assertFalse(response.success) def test_sign_messages_by_request_ids(self): - request_ids = "1690349242683,1690268795963,1690187858862" + request_ids = "web_send_by_user_453_1702345165920,1702350828108" response = self.mpc_client.sign_messages_by_request_ids(request_ids=request_ids) - print(response) + # print(response) + self.assertTrue(response.success) def test_sign_messages_by_cobo_ids(self): - cobo_ids = "20230726132723000341052000008222,20230725150636000308867000003494,20230725135301000361318000002480" + cobo_ids = "20231212111349000140669000006194,20231212094205000140669000003110" response = self.mpc_client.sign_messages_by_cobo_ids(cobo_ids=cobo_ids) - print(response) - - def test_get_max_send_amount(self): - response = self.mpc_client.get_max_send_amount(coin="GETH", fee_rate=0, to_address='0xEEACb7a5e53600c144C0b9839A834bb4b39E540c') - print(response) - - -if __name__ == '__main__': - if len(sys.argv) > 1: - parser = argparse.ArgumentParser() - parser.add_argument("--env", nargs='?', default="develop") - parser.add_argument("--mpcApiSecret", type=str, required=True) - args = parser.parse_args() - env = args.env if args.env else "develop" - mpc_api_secret = args.mpcApiSecret - - MPCClientTest.mpc_api_secret = mpc_api_secret - MPCClientTest.ENV = DEV_ENV - MPCClientTest.TEST_DATA = DEV_TEST_DATA - - # unittest.main() - runner = unittest.TextTestRunner() - suite = unittest.TestLoader().loadTestsFromTestCase(MPCClientTest) - runner.run(suite) + # print(response) + self.assertTrue(response.success) \ No newline at end of file diff --git a/test_web3.py b/tests/testcase/test_web3_client.py similarity index 79% rename from test_web3.py rename to tests/testcase/test_web3_client.py index 9d5a7ee..073e855 100644 --- a/test_web3.py +++ b/tests/testcase/test_web3_client.py @@ -1,25 +1,22 @@ import json +import unittest +import time from cobo_custody.client.web3_client import Web3Client -from cobo_custody.config import DEV_ENV -from cobo_custody.config import DEV_TEST_DATA - -import unittest from cobo_custody.signer.local_signer import LocalSigner -import sys -import argparse -import time class Web3ClientTest(unittest.TestCase): + web3_api_secret = "" - ENV = DEV_ENV - TEST_DATA = DEV_TEST_DATA + env = None + test_data = "" - def setUp(self): - self.web3_client = Web3Client(signer=LocalSigner(self.web3_api_secret), - env=self.ENV, - debug=False) + @classmethod + def setUpClass(cls): + cls.web3_client = Web3Client(signer=LocalSigner(cls.web3_api_secret), + env=cls.env, + debug=False) def test_get_web3_supported_chains(self): response = self.web3_client.get_web3_supported_chains() @@ -46,7 +43,8 @@ def test_get_web3_supported_contracts(self): def test_get_web3_supported_contract_methods(self): chain_code = "GETH" contract_address = "0x7851dcc90e79f3f2c59915e7f4d6fabd8d3d305b" - response = self.web3_client.get_web3_supported_contract_methods(chain_code=chain_code, contract_address=contract_address) + response = self.web3_client.get_web3_supported_contract_methods(chain_code=chain_code, + contract_address=contract_address) print(response) self.assertTrue(response.success) @@ -61,7 +59,8 @@ def test_get_web3_address_list(self): chain_code = "GETH" page_index = 0 page_length = 20 - response = self.web3_client.get_web3_address_list(chain_code=chain_code, page_index=page_index, page_length=page_length) + response = self.web3_client.get_web3_address_list(chain_code=chain_code, page_index=page_index, + page_length=page_length) print(response) self.assertTrue(response.success) @@ -113,7 +112,8 @@ def test_web3_contract(self): args = json.dumps(["0x040149e133077aebcfe4594e00638135eb4bc77f", 1]) amount = 0 response = self.web3_client.web3_contract(chain_code=chain_code, request_id=request_id, wallet_addr=wallet_addr, - contract_addr=contract_addr, method_id=method_id, method_name=method_name, + contract_addr=contract_addr, method_id=method_id, + method_name=method_name, args=args, amount=amount) print(response) self.assertTrue(response.success) @@ -130,22 +130,3 @@ def test_list_web3_wallet_transactions(self): response = self.web3_client.list_web3_wallet_transactions(address=address, chain_code=chain_code) print(response) self.assertTrue(response.success) - - -if __name__ == '__main__': - if len(sys.argv) > 1: - parser = argparse.ArgumentParser() - parser.add_argument("--env", nargs='?', default="develop") - parser.add_argument("--web3ApiSecret", type=str, required=True) - args = parser.parse_args() - env = args.env if args.env else "develop" - web3_api_secret = args.web3ApiSecret - - Web3ClientTest.web3_api_secret = web3_api_secret - Web3ClientTest.ENV = DEV_ENV - Web3ClientTest.TEST_DATA = DEV_TEST_DATA - - # unittest.main() - runner = unittest.TextTestRunner() - suite = unittest.TestLoader().loadTestsFromTestCase(Web3ClientTest) - runner.run(suite) \ No newline at end of file diff --git a/tox.ini b/tox.ini new file mode 100644 index 0000000..ababfea --- /dev/null +++ b/tox.ini @@ -0,0 +1,8 @@ +[tox] +envlist = py310 + +[testenv] +base_python = python3.10 +commands = + pip install -r tests/test_requirements.txt + python tests/run_test.py {posargs} From 7cd8e645fa40f5dd8a745f8339607e4820ad4da3 Mon Sep 17 00:00:00 2001 From: Cchangyixue <153605971+Cchangyixue@users.noreply.github.com> Date: Wed, 13 Dec 2023 09:57:21 +0800 Subject: [PATCH 2/5] Update run_test.py --- tests/run_test.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/run_test.py b/tests/run_test.py index 5707892..b992820 100644 --- a/tests/run_test.py +++ b/tests/run_test.py @@ -1,3 +1,5 @@ +import sys +sys.path.append("..") import unittest import argparse from cobo_custody.config import DEV_ENV From 0558224124c5e3b9d3202d54b7e04fa99f765484 Mon Sep 17 00:00:00 2001 From: Cchangyixue <153605971+Cchangyixue@users.noreply.github.com> Date: Wed, 13 Dec 2023 10:00:51 +0800 Subject: [PATCH 3/5] Update run_test.py --- tests/run_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/run_test.py b/tests/run_test.py index b992820..652d3b0 100644 --- a/tests/run_test.py +++ b/tests/run_test.py @@ -1,5 +1,5 @@ import sys -sys.path.append("..") +sys.path.insert(0, sys.path[0]+"/../") import unittest import argparse from cobo_custody.config import DEV_ENV From 294905ac79b4b0cc1dbfe232f13b19421f90ae0f Mon Sep 17 00:00:00 2001 From: Cchangyixue <153605971+Cchangyixue@users.noreply.github.com> Date: Wed, 13 Dec 2023 11:38:48 +0800 Subject: [PATCH 4/5] Update unittest.yml --- .github/workflows/unittest.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/unittest.yml b/.github/workflows/unittest.yml index 9eedc72..ed10704 100644 --- a/.github/workflows/unittest.yml +++ b/.github/workflows/unittest.yml @@ -16,15 +16,15 @@ on: push: branches: [ main ] pull_request: - branches: - - main - - 'main/**' + branches: [ main ] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 + with: + ref: ${{ github.event.pull_request.head.sha }} - name: Set up Python 3.10 uses: actions/setup-python@v3 with: From 8a37bceeb7274e7ce64c47c41ca24c820fce6147 Mon Sep 17 00:00:00 2001 From: Cchangyixue <153605971+Cchangyixue@users.noreply.github.com> Date: Wed, 13 Dec 2023 11:48:47 +0800 Subject: [PATCH 5/5] Update setup.py --- setup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/setup.py b/setup.py index 015279f..9bbdf6c 100644 --- a/setup.py +++ b/setup.py @@ -1,7 +1,7 @@ from setuptools import setup setup( name="cobo_custody", - version="0.32", + version="0.33", author="Cobo", author_email="support@cobo.com", description="Cobo Custody restful api",