2024-06-11 11:36:20 +00:00
|
|
|
import random
|
|
|
|
import pytest
|
|
|
|
import jsonschema
|
|
|
|
import json
|
|
|
|
from conftest import option, user_1, user_2
|
|
|
|
from test_cases import RpcTestCase, TransactionTestCase
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.wallet
|
|
|
|
@pytest.mark.tx
|
2024-08-13 12:46:48 +00:00
|
|
|
@pytest.mark.rpc
|
2024-06-11 11:36:20 +00:00
|
|
|
class TestTransactionRpc(TransactionTestCase):
|
|
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
|
|
"method, params",
|
|
|
|
[
|
|
|
|
(
|
2024-09-18 07:55:47 +00:00
|
|
|
"wallet_checkRecentHistoryForChainIDs",
|
|
|
|
[[31337], ["0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266"]],
|
2024-06-11 11:36:20 +00:00
|
|
|
),
|
|
|
|
(
|
2024-09-18 07:55:47 +00:00
|
|
|
"wallet_getPendingTransactionsForIdentities",
|
|
|
|
[[{"chainId": None, "hash": None}]],
|
2024-06-11 11:36:20 +00:00
|
|
|
),
|
|
|
|
],
|
|
|
|
)
|
|
|
|
def test_tx_(self, method, params):
|
|
|
|
_id = str(random.randint(1, 9999))
|
|
|
|
|
|
|
|
if method in ["wallet_getPendingTransactionsForIdentities"]:
|
|
|
|
params[0][0]["chainId"] = self.network_id
|
|
|
|
params[0][0]["hash"] = self.tx_hash
|
|
|
|
|
|
|
|
response = self.rpc_request(method, params, _id)
|
|
|
|
self.verify_is_valid_json_rpc_response(response)
|
|
|
|
with open(f"{option.base_dir}/schemas/{method}", "r") as schema:
|
|
|
|
jsonschema.validate(instance=response.json(), schema=json.load(schema))
|
|
|
|
|
|
|
|
def test_create_multi_transaction(self):
|
|
|
|
response = self.wallet_create_multi_transaction()
|
2024-09-18 16:54:17 +00:00
|
|
|
self.verify_is_valid_json_rpc_response(response)
|
|
|
|
|
2024-06-11 11:36:20 +00:00
|
|
|
# how to create schema:
|
|
|
|
# from schema_builder import CustomSchemaBuilder
|
2024-07-04 10:52:50 +00:00
|
|
|
# CustomSchemaBuilder(method).create_schema(response.json())
|
2024-09-18 16:54:17 +00:00
|
|
|
|
|
|
|
with open(f"{option.base_dir}/schemas/wallet_createMultiTransaction/transferTx_positive", "r") as schema:
|
2024-06-11 11:36:20 +00:00
|
|
|
jsonschema.validate(instance=response.json(), schema=json.load(schema))
|
|
|
|
|
2024-09-18 16:54:17 +00:00
|
|
|
@pytest.mark.parametrize(
|
|
|
|
"method, changed_values, expected_error_code, expected_error_text",
|
|
|
|
[
|
|
|
|
(
|
|
|
|
"transferTx_value_not_enough_balance",
|
|
|
|
{'value': '0x21e438ea8139cd35004'}, -32000, "Insufficient funds for gas",
|
|
|
|
),
|
|
|
|
(
|
|
|
|
"transferTx_from_from_invalid_string",
|
|
|
|
{'from': 'some_invalid_address'}, -32602, "cannot unmarshal hex string without 0x prefix",
|
|
|
|
),
|
|
|
|
],
|
|
|
|
)
|
|
|
|
def test_create_multi_transaction_validation(self, method,
|
|
|
|
changed_values,
|
|
|
|
expected_error_code, expected_error_text):
|
|
|
|
response = self.wallet_create_multi_transaction(**changed_values)
|
|
|
|
self.verify_is_json_rpc_error(response)
|
|
|
|
actual_error_code, actual_error_text = response.json()['error']['code'], response.json()['error']['message']
|
|
|
|
assert expected_error_code == actual_error_code, f"got code: {actual_error_code} instead of expected: {expected_error_code}"
|
|
|
|
assert expected_error_text in actual_error_text, f"got error: {actual_error_text} that does not include: {expected_error_text}"
|
|
|
|
|
|
|
|
with open(f"{option.base_dir}/schemas/wallet_createMultiTransaction/transferTx_error", "r") as schema:
|
|
|
|
jsonschema.validate(instance=response.json(), schema=json.load(schema))
|
|
|
|
|
2024-06-11 11:36:20 +00:00
|
|
|
|
|
|
|
@pytest.mark.wallet
|
2024-08-13 12:46:48 +00:00
|
|
|
@pytest.mark.rpc
|
2024-06-11 11:36:20 +00:00
|
|
|
class TestRpc(RpcTestCase):
|
|
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
|
|
"method, params",
|
2024-09-18 07:55:47 +00:00
|
|
|
[
|
2024-06-11 11:36:20 +00:00
|
|
|
("wallet_startWallet", []),
|
|
|
|
("wallet_getEthereumChains", []),
|
|
|
|
("wallet_getTokenList", []),
|
|
|
|
("wallet_getCryptoOnRamps", []),
|
2024-09-18 07:55:47 +00:00
|
|
|
("wallet_getCachedCurrencyFormats", []),
|
|
|
|
("wallet_fetchAllCurrencyFormats", [])
|
2024-06-11 11:36:20 +00:00
|
|
|
],
|
|
|
|
)
|
|
|
|
def test_(self, method, params):
|
|
|
|
_id = str(random.randint(1, 8888))
|
|
|
|
|
|
|
|
response = self.rpc_request(method, params, _id)
|
|
|
|
self.verify_is_valid_json_rpc_response(response)
|
|
|
|
with open(f"{option.base_dir}/schemas/{method}", "r") as schema:
|
|
|
|
jsonschema.validate(instance=response.json(), schema=json.load(schema))
|