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_: a little bit of refactoring

This commit is contained in:
Yevheniia Berdnyk 2024-10-11 11:08:39 +03:00 committed by Anton Danchenko
parent 86cd41d04e
commit 4dbad723bf
12 changed files with 147 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
tests-functional/tests/clients/signals.py → tests-functional/clients/signals.py
View File

@ -1,14 +1,14 @@
import websocket
import time
import json
import logging
import time
import websocket
class SignalClient:
def __init__(self, ws_url, await_signals):
self.ws_url = ws_url
self.url = f"{ws_url}/signals"
self.await_signals = await_signals
self.received_signals = {
@ -23,15 +23,15 @@ class SignalClient:
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:
if time.time() - start_time >= timeout:
raise TimeoutError(
f"Signal {signal_type} is not received in {timeout} seconds")
f"Signal {signal_type} is not received in {timeout} seconds")
time.sleep(0.2)
logging.debug(f"Signal {signal_type} is received in {round(time.time() - start_time)} seconds")
return self.received_signals[signal_type][0]
def _on_error(self, ws, error):
logging.info(f"Error: {error}")
logging.error(f"Error: {error}")
def _on_close(self, ws, close_status_code, close_msg):
logging.info(f"Connection closed: {close_status_code}, {close_msg}")
@ -40,7 +40,6 @@ class SignalClient:
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,

29
tests-functional/tests/clients/status_backend.py → tests-functional/clients/status_backend.py
View File

@ -1,21 +1,22 @@
import jsonschema
import json
import requests
import logging
from datetime import datetime
from conftest import option, user_1
import jsonschema
import requests
from clients.signals import SignalClient
from conftest import option
from constants import user_1
class RpcClient:
def __init__(
self, rpc_url, client=requests.Session()
):
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):
def _check_decode_and_key_errors_in_response(self, response, key):
try:
return response.json()[key]
except json.JSONDecodeError:
@ -28,7 +29,7 @@ class RpcClient:
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")
self._check_decode_and_key_errors_in_response(response, "result")
if _id:
try:
@ -43,14 +44,14 @@ class RpcClient:
def verify_is_json_rpc_error(self, response):
assert response.status_code == 200
assert response.content
self._try_except_JSONDecodeError_KeyError(response, "error")
self._check_decode_and_key_errors_in_response(response, "error")
def rpc_request(self, method, params=[], _id=None, url=None):
def rpc_request(self, method, params=[], request_id=13, url=None):
url = url if url else self.rpc_url
data = {"jsonrpc": "2.0", "method": method}
data = {"jsonrpc": "2.0", "method": method, "id": request_id}
if params:
data["params"] = params
data["id"] = _id if _id else 13
logging.info(f"Sending POST request to url {url} with data: {json.dumps(data, sort_keys=True, indent=4)}")
response = self.client.post(url, json=data)
return response
@ -94,7 +95,7 @@ class StatusBackend(RpcClient, SignalClient):
except KeyError:
pass
def api_valid_request(self, method, data, url=None):
def api_valid_request(self, method, data):
response = self.api_request(method, data)
self.verify_is_valid_api_response(response)
return response

51
tests-functional/conftest.py
View File

@ -1,6 +1,9 @@
import os
import threading
from dataclasses import dataclass
import pytest as pytest
def pytest_addoption(parser):
parser.addoption(
@ -40,31 +43,45 @@ def pytest_addoption(parser):
default="Strong12345",
)
@dataclass
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
class Option:
pass
option = Option()
def pytest_configure(config):
global option
option = config.option
option.base_dir = os.path.dirname(os.path.abspath(__file__))
@pytest.fixture(scope="session", autouse=True)
def init_status_backend():
await_signals = [
"mediaserver.started",
"node.started",
"node.ready",
"node.login",
"wallet", # TODO: a test per event of a different type
]
from clients.status_backend import StatusBackend
backend_client = StatusBackend(
await_signals=await_signals
)
websocket_thread = threading.Thread(
target=backend_client._connect
)
websocket_thread.daemon = True
websocket_thread.start()
backend_client.init_status_backend()
backend_client.create_account_and_login()
yield backend_client

20
tests-functional/constants.py Normal file
View File

@ -0,0 +1,20 @@
from dataclasses import dataclass
@dataclass
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"
)

6
tests-functional/schema_builder.py
View File

@ -1,11 +1,13 @@
import json
import os
from conftest import option
from genson import SchemaBuilder
from conftest import option
class CustomSchemaBuilder(SchemaBuilder):
def __init__(self, schema_name):
super().__init__()
self.path = f"{option.base_dir}/schemas/{schema_name}"

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

@ -1,8 +1,7 @@
import random
import pytest
import jsonschema
import json
from conftest import option
from test_cases import StatusDTestCase

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

@ -1,16 +1,18 @@
import json
import logging
import threading
import time
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 StatusDTestCase:
import pytest
from clients.signals import SignalClient
from clients.status_backend import RpcClient
from conftest import option
from constants import user_1, user_2
class StatusDTestCase:
network_id = 31337
def setup_method(self):
@ -18,8 +20,9 @@ class StatusDTestCase:
option.rpc_url_statusd
)
class WalletTestCase(StatusDTestCase):
def setup_method(self):
super().setup_method()
@ -47,7 +50,7 @@ class WalletTestCase(StatusDTestCase):
"fromAddress": user_1.address,
"fromAmount": "0x5af3107a4000",
"fromAsset": "ETH",
"type": 0, # MultiTransactionSend
"type": 0, # MultiTransactionSend
"toAddress": user_2.address,
"toAsset": "ETH",
},
@ -69,17 +72,19 @@ class WalletTestCase(StatusDTestCase):
self.rpc_client.verify_is_valid_json_rpc_response(response)
try:
tx_hash = response.json(
)["result"]["hashes"][str(self.network_id)][0]
)["result"]["hashes"][str(self.network_id)][0]
except (KeyError, json.JSONDecodeError):
raise Exception(response.content)
return tx_hash
class TransactionTestCase(WalletTestCase):
def setup_method(self):
super().setup_method()
self.tx_hash = self.send_valid_multi_transaction()
class EthRpcTestCase(WalletTestCase):
@pytest.fixture(autouse=True, scope='class')
@ -97,7 +102,7 @@ class EthRpcTestCase(WalletTestCase):
block_hash = receipt.json()["result"]["blockHash"]
except (KeyError, json.JSONDecodeError):
raise Exception(receipt.content)
TxData = namedtuple("TxData", ["tx_hash", "block_number", "block_hash"])
return TxData(tx_hash, block_number, block_hash)
@ -126,8 +131,8 @@ class EthRpcTestCase(WalletTestCase):
response = self.rpc_client.rpc_valid_request(method, params)
return response.json()["result"]["tx"]
class SignalTestCase(StatusDTestCase):
class SignalTestCase(StatusDTestCase):
await_signals = []
def setup_method(self):
@ -138,6 +143,7 @@ class SignalTestCase(StatusDTestCase):
websocket_thread.daemon = True
websocket_thread.start()
class StatusBackendTestCase:
def setup_method(self):

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

@ -1,24 +1,29 @@
import pytest
from conftest import option
from test_cases import EthRpcTestCase
def validateHeader(header, block_number, block_hash):
def validate_header(header, block_number, block_hash):
assert header["number"] == block_number
assert header["hash"] == block_hash
def validateBlock(block, block_number, block_hash, expected_tx_hash):
validateHeader(block["header"], block_number, block_hash)
def validate_block(block, block_number, block_hash, expected_tx_hash):
validate_header(block["header"], block_number, block_hash)
tx_hashes = [tx["hash"] for tx in block["transactions"]]
assert expected_tx_hash in tx_hashes
def validateTransaction(tx, tx_hash):
def validate_transaction(tx, tx_hash):
assert tx["tx"]["hash"] == tx_hash
def validateReceipt(receipt, tx_hash, block_number, block_hash):
def validate_receipt(receipt, tx_hash, block_number, block_hash):
assert receipt["transactionHash"] == tx_hash
assert receipt["blockNumber"] == block_number
assert receipt["blockHash"] == block_hash
@pytest.mark.rpc
@pytest.mark.ethclient
class TestEth(EthRpcTestCase):
@ -30,25 +35,26 @@ class TestEth(EthRpcTestCase):
self.rpc_client.rpc_valid_request("ethclient_suggestGasPrice", [self.network_id])
def test_header_by_number(self, tx_data):
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)
response = self.rpc_client.rpc_valid_request("ethclient_headerByNumber",
[self.network_id, tx_data.block_number])
validate_header(response.json()["result"], tx_data.block_number, tx_data.block_hash)
def test_block_by_number(self, tx_data):
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)
validate_block(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_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)
validate_header(response.json()["result"], tx_data.block_number, tx_data.block_hash)
def test_block_by_hash(self, tx_data):
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)
validate_block(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_client.rpc_valid_request("ethclient_transactionByHash", [self.network_id, tx_data.tx_hash])
validateTransaction(response.json()["result"], tx_data.tx_hash)
validate_transaction(response.json()["result"], tx_data.tx_hash)
def test_transaction_receipt(self, tx_data):
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)
validate_receipt(response.json()["result"], tx_data.tx_hash, tx_data.block_number, tx_data.block_hash)

31
tests-functional/tests/test_init_status_app.py
View File

@ -1,7 +1,5 @@
import pytest
import threading
from conftest import option
from clients.status_backend import StatusBackend
from test_cases import StatusBackendTestCase
@ -9,33 +7,6 @@ from test_cases import StatusBackendTestCase
@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

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

@ -1,7 +1,9 @@
import pytest
import time
import uuid
from conftest import user_1, user_2, option
import pytest
from conftest import option
from constants import user_1, user_2
from test_cases import SignalTestCase
@ -9,7 +11,6 @@ from test_cases import SignalTestCase
@pytest.mark.transaction
@pytest.mark.wallet
class TestTransactionFromRoute(SignalTestCase):
await_signals = [
"wallet.suggested.routes",
"wallet.router.sign-transactions",

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

@ -1,13 +1,12 @@
import random
import pytest
import jsonschema
import json
import time
from conftest import option, user_1, user_2
from test_cases import StatusDTestCase, TransactionTestCase
from dataclasses import dataclass, field
from typing import List, Optional
from dataclasses import dataclass
from typing import Optional
import pytest
from conftest import option
from test_cases import StatusDTestCase
@pytest.mark.skip("to be reworked via status-backend")
@ -28,7 +27,6 @@ class TestRpc(StatusDTestCase):
@pytest.mark.skip("to be reworked via status-backend")
class TestRpcMessaging(StatusDTestCase):
@dataclass
class User:
rpc_url: str
@ -69,7 +67,7 @@ class TestRpcMessaging(StatusDTestCase):
"message": f"contact request from {sender.chat_public_key}: sent at {time.time()}",
}
],
_id=99,
request_id=99,
url=sender.rpc_url,
)
@ -85,7 +83,7 @@ class TestRpcMessaging(StatusDTestCase):
"id": user.chat_id,
}
],
_id=99,
request_id=99,
url=user.rpc_url,
)
self.rpc_client.verify_is_valid_json_rpc_response(response)
@ -95,7 +93,7 @@ class TestRpcMessaging(StatusDTestCase):
response = self.rpc_client.rpc_request(
method="wakuext_contacts",
params=[],
_id=99,
request_id=99,
url=user[0].rpc_url,
)
self.rpc_client.verify_is_valid_json_rpc_response(response)

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

@ -1,8 +1,10 @@
import random
import pytest
import jsonschema
import json
from conftest import option, user_1, user_2
import random
import jsonschema
import pytest
from conftest import option
from test_cases import StatusDTestCase, TransactionTestCase
@ -37,11 +39,11 @@ class TestTransactionRpc(TransactionTestCase):
def test_create_multi_transaction(self):
response = self.wallet_create_multi_transaction()
self.rpc_client.verify_is_valid_json_rpc_response(response)
# how to create schema:
# from schema_builder import CustomSchemaBuilder
# CustomSchemaBuilder(method).create_schema(response.json())
with open(f"{option.base_dir}/schemas/wallet_createMultiTransaction/transferTx_positive", "r") as schema:
jsonschema.validate(instance=response.json(), schema=json.load(schema))
@ -64,8 +66,10 @@ class TestTransactionRpc(TransactionTestCase):
response = self.wallet_create_multi_transaction(**changed_values)
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}"
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}"
self.rpc_client.verify_json_schema(response, "wallet_createMultiTransaction/transferTx_error")