status-im
/
status-go
mirror of https://github.com/status-im/status-go.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

test_: added status-backend support, fixed signals

This commit is contained in:
Anton 2024-10-07 17:40:18 +02:00 committed by Anton Danchenko
parent 451b22af55
commit 27d0b0f0c3
13 changed files with 355 additions and 195 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
tests-functional/conftest.py
View File

@ -4,23 +4,29 @@ from dataclasses import dataclass
def pytest_addoption(parser):
parser.addoption(
"--rpc_url",
"--rpc_url_statusd",
action="store",
help="",
default="http://0.0.0.0:3333",
)
parser.addoption(
"--rpc_url_2",
"--rpc_url_status_backend",
action="store",
help="",
default="http://0.0.0.0:3334",
)
parser.addoption(
"--ws_url",
"--ws_url_statusd",
action="store",
help="",
default="ws://0.0.0.0:8354",
)
parser.addoption(
"--ws_url_status_backend",
action="store",
help="",
default="ws://0.0.0.0:3334",
)
parser.addoption(
"--anvil_url",
action="store",
@ -39,14 +45,17 @@ class Account():
address: str
private_key: str
password: str
user_1 = Account(
address="0xf39fd6e51aad88f6f4ce6ab8827279cfffb92266",
private_key="0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80",
password="Strong12345"
)
user_2 = Account(
address="0x70997970c51812dc3a010c7d01b50e0d17dc79c8",
private_key="0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d",
password="Strong12345"
)
@dataclass

2
tests-functional/docker-compose.status-go.local.yml
View File

@ -6,6 +6,6 @@ services:
ports:
- 3333:3333
- 8354:8354
status-go-no-funds:
status-backend:
ports:
- 3334:3333

27
tests-functional/docker-compose.test.status-go.yml
View File

@ -30,22 +30,18 @@ services:
- ./coverage/binary:/coverage/binary
stop_signal: SIGINT
# TODO: Remove this duplication when implemented: https://github.com/status-im/status-go/issues/5803
status-go-no-funds:
user: ${FUNCTIONAL_TESTS_DOCKER_UID}
status-backend:
user: ${INTEGRATION_TESTS_DOCKER_UID}
build:
context: ../
dockerfile: _assets/build/Dockerfile
args:
build_tags: gowaku_no_rln,enable_private_api
build_target: statusd
build_target: status-backend
build_flags: -cover
entrypoint: [
"statusd",
"-c", "/static/configs/config.json",
"--seed-phrase", "test test test test test test test test test test test takoe",
"--password", "Strong12345",
"--dir", "/tmp/status-go-data", # Keep in sync with `config.json/DataDir` value. Later this arg will not be needed.
"status-backend",
"--address", "0.0.0.0:3333",
]
healthcheck:
test: ["CMD-SHELL", "curl -X POST --data '{\"jsonrpc\":\"2.0\",\"method\":\"net_version\",\"params\":[],\"id\":1}' -H 'Content-Type: application/json' http://0.0.0.0:3333 || exit 1"]
@ -63,7 +59,7 @@ services:
depends_on:
status-go:
condition: service_healthy
status-go-no-funds:
status-backend:
condition: service_healthy
deploy-communities-contracts:
condition: service_completed_successfully
@ -72,12 +68,13 @@ services:
dockerfile: Dockerfile.tests-rpc
entrypoint: [
"pytest",
"-m", "wallet or ethclient",
"--rpc_url=http://status-go:3333",
"--rpc_url_2=http://status-go-no-funds:3333",
"-m", "rpc",
"--anvil_url=http://anvil:8545",
"--ws_url=ws://status-go:8354",
"--junitxml=/tests-rpc/reports/report.xml",
"--rpc_url_statusd=http://status-go:3333",
"--rpc_url_status_backend=http://status-backend:3333",
"--ws_url_statusd=ws://status-go:8354",
"--ws_url_status_backend=ws://status-backend:3333",
"--junitxml=/tests-rpc/reports/report.xml"
]
volumes:
- .:/tests-rpc

1
tests-functional/pytest.ini
View File

@ -11,3 +11,4 @@ markers =
wakuext
accounts
ethclient
init

49
tests-functional/tests/clients/signals.py Normal file
View File

@ -0,0 +1,49 @@
import websocket
import time
import json
import logging
class SignalClient:
def __init__(self, ws_url, await_signals):
self.ws_url = ws_url
self.await_signals = await_signals
self.received_signals = {
signal: [] for signal in self.await_signals
}
def on_message(self, ws, signal):
signal = json.loads(signal)
if signal.get("type") in self.await_signals:
self.received_signals[signal["type"]].append(signal)
def wait_for_signal(self, signal_type, timeout=20):
start_time = time.time()
while not self.received_signals.get(signal_type):
time_passed = time.time() - start_time
if time_passed >= timeout:
raise TimeoutError(
f"Signal {signal_type} is not received in {timeout} seconds")
time.sleep(0.2)
return self.received_signals[signal_type][0]
def _on_error(self, ws, error):
logging.info(f"Error: {error}")
def _on_close(self, ws, close_status_code, close_msg):
logging.info(f"Connection closed: {close_status_code}, {close_msg}")
def _on_open(self, ws):
logging.info("Connection opened")
def _connect(self):
self.url = f"{self.ws_url}/signals"
ws = websocket.WebSocketApp(self.url,
on_message=self.on_message,
on_error=self._on_error,
on_close=self._on_close)
ws.on_open = self._on_open
ws.run_forever()

141
tests-functional/tests/clients/status_backend.py Normal file
View File

@ -0,0 +1,141 @@
import jsonschema
import json
import requests
from datetime import datetime
from conftest import option, user_1
from clients.signals import SignalClient
class RpcClient:
def __init__(
self, rpc_url, client=requests.Session()
):
self.client = client
self.rpc_url = rpc_url
def _try_except_JSONDecodeError_KeyError(self, response, key: str):
try:
return response.json()[key]
except json.JSONDecodeError:
raise AssertionError(
f"Invalid JSON in response: {response.content}")
except KeyError:
raise AssertionError(
f"Key '{key}' not found in the JSON response: {response.content}")
def verify_is_valid_json_rpc_response(self, response, _id=None):
assert response.status_code == 200
assert response.content
self._try_except_JSONDecodeError_KeyError(response, "result")
if _id:
try:
if _id != response.json()["id"]:
raise AssertionError(
f"got id: {response.json()['id']} instead of expected id: {_id}"
)
except KeyError:
raise AssertionError(f"no id in response {response.json()}")
return response
def verify_is_json_rpc_error(self, response):
assert response.status_code == 200
assert response.content
self._try_except_JSONDecodeError_KeyError(response, "error")
def rpc_request(self, method, params=[], _id=None, url=None):
url = url if url else self.rpc_url
data = {"jsonrpc": "2.0", "method": method}
if params:
data["params"] = params
data["id"] = _id if _id else 13
response = self.client.post(url, json=data)
return response
def rpc_valid_request(self, method, params=[], _id=None, url=None):
response = self.rpc_request(method, params, _id, url)
self.verify_is_valid_json_rpc_response(response, _id)
return response
def verify_json_schema(self, response, method):
with open(f"{option.base_dir}/schemas/{method}", "r") as schema:
jsonschema.validate(instance=response.json(),
schema=json.load(schema))
class StatusBackend(RpcClient, SignalClient):
def __init__(self, await_signals):
self.api_url = f"{option.rpc_url_status_backend}/statusgo"
self.ws_url = f"{option.ws_url_status_backend}"
self.rpc_url = f"{option.rpc_url_status_backend}/statusgo/CallRPC"
RpcClient.__init__(self, self.rpc_url)
SignalClient.__init__(self, self.ws_url, await_signals)
def api_request(self, method, data, url=None):
url = url if url else self.api_url
url = f"{url}/{method}"
response = requests.post(url, json=data)
return response
def verify_is_valid_api_response(self, response):
assert response.status_code == 200
assert response.content
try:
assert not response.json()["error"]
except json.JSONDecodeError:
raise AssertionError(
f"Invalid JSON in response: {response.content}")
except KeyError:
pass
def api_valid_request(self, method, data, url=None):
response = self.api_request(method, data)
self.verify_is_valid_api_response(response)
return response
def init_status_backend(self, data_dir="/"):
method = "InitializeApplication"
data = {
"dataDir": data_dir
}
return self.api_valid_request(method, data)
def create_account_and_login(self, display_name="Mr_Meeseeks", password=user_1.password):
data_dir = f"dataDir_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
method = "CreateAccountAndLogin"
data = {
"rootDataDir": data_dir,
"kdfIterations": 256000,
"displayName": display_name,
"password": password,
"customizationColor": "primary"
}
return self.api_valid_request(method, data)
def start_messenger(self, params=[]):
method = "wakuext_startMessenger"
response = self.rpc_request(method, params)
json_response = response.json()
if 'error' in json_response:
assert json_response['error']['code'] == -32000
assert json_response['error']['message'] == "messenger already started"
return
self.verify_is_valid_json_rpc_response(response)
def start_wallet(self, params=[]):
method = "wallet_startWallet"
response = self.rpc_request(method, params)
self.verify_is_valid_json_rpc_response(response)
def get_settings(self, params=[]):
method = "settings_getSettings"
response = self.rpc_request(method, params)
self.verify_is_valid_json_rpc_response(response)

10
tests-functional/tests/test_accounts.py
View File

@ -3,12 +3,12 @@ import pytest
import jsonschema
import json
from conftest import option
from test_cases import RpcTestCase
from test_cases import StatusDTestCase
@pytest.mark.accounts
@pytest.mark.rpc
class TestAccounts(RpcTestCase):
class TestAccounts(StatusDTestCase):
@pytest.mark.parametrize(
"method, params",
@ -22,7 +22,5 @@ class TestAccounts(RpcTestCase):
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))
response = self.rpc_client.rpc_valid_request(method, params, _id)
self.rpc_client.verify_json_schema(response, method)

141
tests-functional/tests/test_cases.py
View File

@ -1,74 +1,25 @@
import json
import websocket
import threading
import logging
import jsonschema
import time
import requests
from conftest import option, user_1, user_2
import pytest
import logging
from conftest import option, user_1, user_2
from clients.signals import SignalClient
from clients.status_backend import RpcClient, StatusBackend
from collections import namedtuple
class RpcTestCase:
class StatusDTestCase:
network_id = 31337
def setup_method(self):
pass
self.rpc_client = RpcClient(
option.rpc_url_statusd
)
def _try_except_JSONDecodeError_KeyError(self, response, key: str):
try:
return response.json()[key]
except json.JSONDecodeError:
raise AssertionError(
f"Invalid JSON in response: {response.content}")
except KeyError:
raise AssertionError(
f"Key '{key}' not found in the JSON response: {response.content}")
class WalletTestCase(StatusDTestCase):
def verify_is_valid_json_rpc_response(self, response, _id=None):
assert response.status_code == 200
assert response.content
self._try_except_JSONDecodeError_KeyError(response, "result")
if _id:
try:
if _id != response.json()["id"]:
raise AssertionError(
f"got id: {response.json()['id']} instead of expected id: {_id}"
)
except KeyError:
raise AssertionError(f"no id in response {response.json()}")
return response
def verify_is_json_rpc_error(self, response):
assert response.status_code == 200
assert response.content
self._try_except_JSONDecodeError_KeyError(response, "error")
def rpc_request(self, method, params=[], _id=None, client=None, url=None):
client = client if client else requests.Session()
url = url if url else option.rpc_url
data = {"jsonrpc": "2.0", "method": method}
if params:
data["params"] = params
data["id"] = _id if _id else 13
response = client.post(url, json=data)
return response
def rpc_valid_request(self, method, params=[], _id=None, client=None, url=None):
response = self.rpc_request(method, params, _id, client, url)
self.verify_is_valid_json_rpc_response(response, _id)
return response
def verify_json_schema(self, response, method):
with open(f"{option.base_dir}/schemas/{method}", "r") as schema:
jsonschema.validate(instance=response.json(),
schema=json.load(schema))
class WalletTestCase(RpcTestCase):
def setup_method(self):
super().setup_method()
@ -89,7 +40,7 @@ class WalletTestCase(RpcTestCase):
if key in transferTx_data:
transferTx_data[key] = new_value
else:
print(
logging.info(
f"Warning: The key '{key}' does not exist in the transferTx parameters and will be ignored.")
params = [
{
@ -109,13 +60,13 @@ class WalletTestCase(RpcTestCase):
],
f"{option.password}",
]
return self.rpc_request(method, params)
return self.rpc_client.rpc_request(method, params)
def send_valid_multi_transaction(self, **kwargs):
response = self.wallet_create_multi_transaction(**kwargs)
tx_hash = None
self.verify_is_valid_json_rpc_response(response)
self.rpc_client.verify_is_valid_json_rpc_response(response)
try:
tx_hash = response.json(
)["result"]["hashes"][str(self.network_id)][0]
@ -124,14 +75,19 @@ class WalletTestCase(RpcTestCase):
return tx_hash
class TransactionTestCase(WalletTestCase):
def setup_method(self):
super().setup_method()
self.tx_hash = self.send_valid_multi_transaction()
class EthApiTestCase(WalletTestCase):
class EthRpcTestCase(WalletTestCase):
@pytest.fixture(autouse=True, scope='class')
def tx_data(self):
self.rpc_client = RpcClient(
option.rpc_url_statusd
)
tx_hash = self.send_valid_multi_transaction()
self.wait_until_tx_not_pending(tx_hash)
@ -148,17 +104,17 @@ class EthApiTestCase(WalletTestCase):
def get_block_header(self, block_number):
method = "ethclient_headerByNumber"
params = [self.network_id, block_number]
return self.rpc_valid_request(method, params)
return self.rpc_client.rpc_valid_request(method, params)
def get_transaction_receipt(self, tx_hash):
method = "ethclient_transactionReceipt"
params = [self.network_id, tx_hash]
return self.rpc_valid_request(method, params)
return self.rpc_client.rpc_valid_request(method, params)
def wait_until_tx_not_pending(self, tx_hash, timeout=10):
method = "ethclient_transactionByHash"
params = [self.network_id, tx_hash]
response = self.rpc_valid_request(method, params)
response = self.rpc_client.rpc_valid_request(method, params)
start_time = time.time()
while response.json()["result"]["isPending"] == True:
@ -167,53 +123,22 @@ class EthApiTestCase(WalletTestCase):
raise TimeoutError(
f"Tx {tx_hash} is still pending after {timeout} seconds")
time.sleep(0.5)
response = self.rpc_valid_request(method, params)
response = self.rpc_client.rpc_valid_request(method, params)
return response.json()["result"]["tx"]
class SignalTestCase(RpcTestCase):
class SignalTestCase(StatusDTestCase):
await_signals = []
received_signals = {}
def on_message(self, ws, signal):
signal = json.loads(signal)
if signal.get("type") in self.await_signals:
self.received_signals[signal["type"]] = signal
def wait_for_signal(self, signal_type, timeout=10):
start_time = time.time()
while signal_type not in self.received_signals:
time_passed = time.time() - start_time
if time_passed >= timeout:
raise TimeoutError(
f"Signal {signal_type} is not received in {timeout} seconds")
time.sleep(0.5)
return self.received_signals[signal_type]
def _on_error(self, ws, error):
logging.info(f"Error: {error}")
def _on_close(self, ws, close_status_code, close_msg):
logging.info(f"Connection closed: {close_status_code}, {close_msg}")
def _on_open(self, ws):
logging.info("Connection opened")
def _connect(self):
self.url = f"{option.ws_url}/signals"
ws = websocket.WebSocketApp(self.url,
on_message=self.on_message,
on_error=self._on_error,
on_close=self._on_close)
ws.on_open = self._on_open
ws.run_forever()
def setup_method(self):
super().setup_method()
self.signal_client = SignalClient(option.ws_url_statusd, self.await_signals)
websocket_thread = threading.Thread(target=self._connect)
websocket_thread = threading.Thread(target=self.signal_client._connect)
websocket_thread.daemon = True
websocket_thread.start()
class StatusBackendTestCase:
def setup_method(self):
pass

21
tests-functional/tests/test_eth_api.py
View File

@ -1,6 +1,6 @@
import pytest
from conftest import option
from test_cases import EthApiTestCase
from test_cases import EthRpcTestCase
def validateHeader(header, block_number, block_hash):
assert header["number"] == block_number
@ -21,33 +21,34 @@ def validateReceipt(receipt, tx_hash, block_number, block_hash):
@pytest.mark.rpc
@pytest.mark.ethclient
class TestRpc(EthApiTestCase):
class TestEth(EthRpcTestCase):
def test_block_number(self):
self.rpc_valid_request("ethclient_blockNumber", [self.network_id])
self.rpc_client.rpc_valid_request("ethclient_blockNumber", [self.network_id])
def test_suggest_gas_price(self):
self.rpc_valid_request("ethclient_suggestGasPrice", [self.network_id])
self.rpc_client.rpc_valid_request("ethclient_suggestGasPrice", [self.network_id])
def test_header_by_number(self, tx_data):
response = self.rpc_valid_request("ethclient_headerByNumber", [self.network_id, tx_data.block_number])
response = self.rpc_client.rpc_valid_request("ethclient_headerByNumber", [self.network_id, tx_data.block_number])
validateHeader(response.json()["result"], tx_data.block_number, tx_data.block_hash)
def test_block_by_number(self, tx_data):
response = self.rpc_valid_request("ethclient_blockByNumber", [self.network_id, tx_data.block_number])
response = self.rpc_client.rpc_valid_request("ethclient_blockByNumber", [self.network_id, tx_data.block_number])
validateBlock(response.json()["result"], tx_data.block_number, tx_data.block_hash, tx_data.tx_hash)
def test_header_by_hash(self, tx_data):
response = self.rpc_valid_request("ethclient_headerByHash", [self.network_id, tx_data.block_hash])
response = self.rpc_client.rpc_valid_request("ethclient_headerByHash", [self.network_id, tx_data.block_hash])
validateHeader(response.json()["result"], tx_data.block_number, tx_data.block_hash)
def test_block_by_hash(self, tx_data):
response = self.rpc_valid_request("ethclient_blockByHash", [self.network_id, tx_data.block_hash])
response = self.rpc_client.rpc_valid_request("ethclient_blockByHash", [self.network_id, tx_data.block_hash])
validateBlock(response.json()["result"], tx_data.block_number, tx_data.block_hash, tx_data.tx_hash)
def test_transaction_by_hash(self, tx_data):
response = self.rpc_valid_request("ethclient_transactionByHash", [self.network_id, tx_data.tx_hash])
response = self.rpc_client.rpc_valid_request("ethclient_transactionByHash", [self.network_id, tx_data.tx_hash])
validateTransaction(response.json()["result"], tx_data.tx_hash)
def test_transaction_receipt(self, tx_data):
response = self.rpc_valid_request("ethclient_transactionReceipt", [self.network_id, tx_data.tx_hash])
response = self.rpc_client.rpc_valid_request("ethclient_transactionReceipt", [self.network_id, tx_data.tx_hash])
validateReceipt(response.json()["result"], tx_data.tx_hash, tx_data.block_number, tx_data.block_hash)

53
tests-functional/tests/test_init_status_app.py Normal file
View File

@ -0,0 +1,53 @@
import pytest
import threading
from conftest import option
from clients.status_backend import StatusBackend
from test_cases import StatusBackendTestCase
@pytest.mark.create_account
@pytest.mark.rpc
class TestInitialiseApp(StatusBackendTestCase):
@pytest.fixture(scope="session", autouse=True)
def init_status_backend(self):
await_signals = [
"mediaserver.started",
"node.started",
"node.ready",
"node.login",
"wallet", # TODO: a test per event of a different type
]
self.backend_client = StatusBackend(
await_signals
)
websocket_thread = threading.Thread(
target=self.backend_client._connect)
websocket_thread.daemon = True
websocket_thread.start()
self.backend_client.init_status_backend()
self.backend_client.create_account_and_login()
yield self.backend_client
@pytest.mark.init
def test_init_app(self, init_status_backend):
# this test is going to fail on every call except first since status-backend will be already initialized
backend_client = init_status_backend
assert backend_client is not None
mediaserver_started = backend_client.wait_for_signal(
"mediaserver.started")
port = mediaserver_started['event']['port']
assert type(port) is int, f"Port is not an integer, found {type(port)}"
backend_client.wait_for_signal("node.started")
backend_client.wait_for_signal("node.ready")
backend_client.wait_for_signal("node.login")

30
tests-functional/tests/test_router.py
View File

@ -41,10 +41,9 @@ class TestTransactionFromRoute(SignalTestCase):
"fromLockedAmount": {}
}
]
response = self.rpc_request(method, params)
self.verify_is_valid_json_rpc_response(response)
response = self.rpc_client.rpc_valid_request(method, params)
routes = self.wait_for_signal("wallet.suggested.routes")
routes = self.signal_client.wait_for_signal("wallet.suggested.routes")
assert routes['event']['Uuid'] == _uuid
method = "wallet_buildTransactionsFromRoute"
@ -54,15 +53,13 @@ class TestTransactionFromRoute(SignalTestCase):
"slippagePercentage": 0
}
]
response = self.rpc_request(method, params)
self.verify_is_valid_json_rpc_response(response)
response = self.rpc_client.rpc_valid_request(method, params)
self.wait_for_signal("wallet.router.sign-transactions")
wallet_router_sign_transactions = self.signal_client.wait_for_signal(
"wallet.router.sign-transactions")
assert self.received_signals[
'wallet.router.sign-transactions']['event']['signingDetails']['signOnKeycard'] == False
transaction_hashes = self.received_signals[
'wallet.router.sign-transactions']['event']['signingDetails']['hashes']
assert wallet_router_sign_transactions['event']['signingDetails']['signOnKeycard'] == False
transaction_hashes = wallet_router_sign_transactions['event']['signingDetails']['hashes']
assert transaction_hashes, "Transaction hashes are empty!"
@ -77,8 +74,7 @@ class TestTransactionFromRoute(SignalTestCase):
option.password
]
response = self.rpc_request(method, params)
self.verify_is_valid_json_rpc_response(response)
response = self.rpc_client.rpc_valid_request(method, params)
if response.json()["result"].startswith("0x"):
tx_signature = response.json()["result"][2:]
@ -98,11 +94,10 @@ class TestTransactionFromRoute(SignalTestCase):
"Signatures": tx_signatures
}
]
response = self.rpc_request(method, params)
self.verify_is_valid_json_rpc_response(response)
tx_status = self.wait_for_signal("wallet.transaction.status-changed")
response = self.rpc_client.rpc_valid_request(method, params)
tx_status = self.signal_client.wait_for_signal(
"wallet.transaction.status-changed")
assert tx_status["event"]["chainId"] == 31337
assert tx_status["event"]["status"] == "Success"
@ -111,8 +106,7 @@ class TestTransactionFromRoute(SignalTestCase):
method = "eth_getTransactionByHash"
params = [tx_hash]
response = self.rpc_request(method, params, url=option.anvil_url)
self.verify_is_valid_json_rpc_response(response)
response = self.rpc_client.rpc_valid_request(method, params, url=option.anvil_url)
tx_details = response.json()["result"]
assert tx_details["value"] == amount_in

35
tests-functional/tests/test_waku_rpc.py
View File

@ -4,14 +4,14 @@ import jsonschema
import json
import time
from conftest import option, user_1, user_2
from test_cases import RpcTestCase, TransactionTestCase
from test_cases import StatusDTestCase, TransactionTestCase
from dataclasses import dataclass, field
from typing import List, Optional
@pytest.mark.rpc
@pytest.mark.wakuext
class TestRpc(RpcTestCase):
@pytest.mark.skip("to be reworked via status-backend")
class TestRpc(StatusDTestCase):
@pytest.mark.parametrize(
"method, params",
@ -22,15 +22,12 @@ class TestRpc(RpcTestCase):
def test_(self, method, params):
_id = str(random.randint(1, 8888))
response = self.rpc_request(method, params, _id, url=option.rpc_url_2)
self.verify_is_valid_json_rpc_response(response)
self.verify_json_schema(response, method)
response = self.rpc_client.rpc_valid_request(method, params, _id, url=option.rpc_url_2)
self.rpc_client.verify_json_schema(response, method)
@pytest.mark.rpc
@pytest.mark.accounts
@pytest.mark.wakuext
class TestRpcMessaging(RpcTestCase):
@pytest.mark.skip("to be reworked via status-backend")
class TestRpcMessaging(StatusDTestCase):
@dataclass
class User:
@ -46,10 +43,10 @@ class TestRpcMessaging(RpcTestCase):
# get chat public key
for user in self.user_1, self.user_2:
response = self.rpc_request(
response = self.rpc_client.rpc_request(
"accounts_getAccounts", [], _id, url=user.rpc_url
)
self.verify_is_valid_json_rpc_response(response)
self.rpc_client.verify_is_valid_json_rpc_response(response)
user.chat_public_key = next(
(
@ -64,7 +61,7 @@ class TestRpcMessaging(RpcTestCase):
for sender in self.user_1, self.user_2:
for receiver in self.user_1, self.user_2:
if sender != receiver:
response = self.rpc_request(
response = self.rpc_client.rpc_request(
method="wakuext_sendContactRequest",
params=[
{
@ -76,12 +73,12 @@ class TestRpcMessaging(RpcTestCase):
url=sender.rpc_url,
)
self.verify_is_valid_json_rpc_response(response)
self.rpc_client.verify_is_valid_json_rpc_response(response)
sender.chat_id = response.json()["result"]["chats"][0]["lastMessage"]["id"]
# accept contact requests
for user in self.user_1, self.user_2:
response = self.rpc_request(
response = self.rpc_client.rpc_request(
method="wakuext_acceptContactRequest",
params=[
{
@ -91,17 +88,17 @@ class TestRpcMessaging(RpcTestCase):
_id=99,
url=user.rpc_url,
)
self.verify_is_valid_json_rpc_response(response)
self.rpc_client.verify_is_valid_json_rpc_response(response)
# verify contacts
for user in (self.user_1, self.user_2), (self.user_2, self.user_1):
response = self.rpc_request(
response = self.rpc_client.rpc_request(
method="wakuext_contacts",
params=[],
_id=99,
url=user[0].rpc_url,
)
self.verify_is_valid_json_rpc_response(response)
self.rpc_client.verify_is_valid_json_rpc_response(response)
response = response.json()
assert response["result"][0]["added"] == True

23
tests-functional/tests/test_wallet_rpc.py
View File

@ -3,7 +3,7 @@ import pytest
import jsonschema
import json
from conftest import option, user_1, user_2
from test_cases import RpcTestCase, TransactionTestCase
from test_cases import StatusDTestCase, TransactionTestCase
@pytest.mark.wallet
@ -31,14 +31,12 @@ class TestTransactionRpc(TransactionTestCase):
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))
response = self.rpc_client.rpc_valid_request(method, params, _id)
self.rpc_client.verify_json_schema(response, method)
def test_create_multi_transaction(self):
response = self.wallet_create_multi_transaction()
self.verify_is_valid_json_rpc_response(response)
self.rpc_client.verify_is_valid_json_rpc_response(response)
# how to create schema:
# from schema_builder import CustomSchemaBuilder
@ -64,18 +62,17 @@ class TestTransactionRpc(TransactionTestCase):
changed_values,
expected_error_code, expected_error_text):
response = self.wallet_create_multi_transaction(**changed_values)
self.verify_is_json_rpc_error(response)
self.rpc_client.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))
self.rpc_client.verify_json_schema(response, "wallet_createMultiTransaction/transferTx_error")
@pytest.mark.wallet
@pytest.mark.rpc
class TestRpc(RpcTestCase):
class TestRpc(StatusDTestCase):
@pytest.mark.parametrize(
"method, params",
@ -90,7 +87,5 @@ class TestRpc(RpcTestCase):
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))
response = self.rpc_client.rpc_valid_request(method, params, _id)
self.rpc_client.verify_json_schema(response, method)