import json import logging import time import random import threading import requests import docker import os from tenacity import retry, stop_after_delay, wait_fixed from clients.signals import SignalClient from clients.rpc import RpcClient from datetime import datetime from conftest import option from constants import user_1, DEFAULT_DISPLAY_NAME, USER_DIR class StatusBackend(RpcClient, SignalClient): def __init__(self, await_signals=[]): if option.status_backend_url: url = option.status_backend_url else: self.docker_client = docker.from_env() host_port = random.choice(option.status_backend_port_range) self.container = self._start_container(host_port) url = f"http://127.0.0.1:{host_port}" option.status_backend_port_range.remove(host_port) self.api_url = f"{url}/statusgo" self.ws_url = f"{url}".replace("http", "ws") self.rpc_url = f"{url}/statusgo/CallRPC" RpcClient.__init__(self, self.rpc_url) SignalClient.__init__(self, self.ws_url, await_signals) self._health_check() websocket_thread = threading.Thread(target=self._connect) websocket_thread.daemon = True websocket_thread.start() def _start_container(self, host_port): docker_project_name = option.docker_project_name timestamp = datetime.now().strftime("%Y%m%d%H%M%S") image_name = f"{docker_project_name}-status-backend:latest" container_name = f"{docker_project_name}-status-backend-{timestamp}" coverage_path = option.codecov_dir if option.codecov_dir else os.path.abspath("./coverage/binary") container_args = { "image": image_name, "detach": True, "name": container_name, "labels": {"com.docker.compose.project": docker_project_name}, "entrypoint": [ "status-backend", "--address", "0.0.0.0:3333", ], "ports": {"3333/tcp": host_port}, "environment": { "GOCOVERDIR": "/coverage/binary", }, "volumes": { coverage_path: { "bind": "/coverage/binary", "mode": "rw", } }, } if "FUNCTIONAL_TESTS_DOCKER_UID" in os.environ: container_args["user"] = os.environ["FUNCTIONAL_TESTS_DOCKER_UID"] container = self.docker_client.containers.run(**container_args) network = self.docker_client.networks.get( f"{docker_project_name}_default") network.connect(container) option.status_backend_containers.append(container.id) return container def _health_check(self): start_time = time.time() while True: try: self.api_valid_request(method="Fleets", data=[]) break except Exception as e: if time.time() - start_time > 20: raise Exception(e) time.sleep(1) def api_request(self, method, data, url=None): url = url if url else self.api_url url = f"{url}/{method}" logging.info( f"Sending POST request to url {url} with data: {json.dumps(data, sort_keys=True, indent=4)}") response = requests.post(url, json=data) logging.info(f"Got response: {response.content}") return response def verify_is_valid_api_response(self, response): assert response.status_code == 200, f"Got response {response.content}, status code {response.status_code}" assert response.content logging.info(f"Got response: {response.content}") try: error = response.json()["error"] assert not error, f"Error: {error}" except json.JSONDecodeError: raise AssertionError( f"Invalid JSON in response: {response.content}") except KeyError: pass def api_valid_request(self, method, data): response = self.api_request(method, data) self.verify_is_valid_api_response(response) return response def init_status_backend(self, data_dir=USER_DIR): method = "InitializeApplication" data = { "dataDir": data_dir, "logEnabled": True, "logLevel": "DEBUG", "apiLogging": True, } return self.api_valid_request(method, data) def create_account_and_login(self, data_dir=USER_DIR, display_name=DEFAULT_DISPLAY_NAME, password=user_1.password): method = "CreateAccountAndLogin" data = { "rootDataDir": data_dir, "kdfIterations": 256000, "displayName": display_name, "password": password, "customizationColor": "primary", "logEnabled": True, "logLevel": "DEBUG", } return self.api_valid_request(method, data) def restore_account_and_login(self, data_dir=USER_DIR, display_name=DEFAULT_DISPLAY_NAME, user=user_1, network_id=31337): method = "RestoreAccountAndLogin" data = { "rootDataDir": data_dir, "kdfIterations": 256000, "displayName": display_name, "password": user.password, "mnemonic": user.passphrase, "customizationColor": "blue", "logEnabled": True, "logLevel": "DEBUG", "testNetworksEnabled": False, "networkId": network_id, "networksOverride": [ { "ChainID": network_id, "ChainName": "Anvil", "DefaultRPCURL": "http://anvil:8545", "RPCURL": "http://anvil:8545", "ShortName": "eth", "NativeCurrencyName": "Ether", "NativeCurrencySymbol": "ETH", "NativeCurrencyDecimals": 18, "IsTest": False, "Layer": 1, "Enabled": True } ] } return self.api_valid_request(method, data) def login(self, keyUid, user=user_1): method = "LoginAccount" data = { "password": user.password, "keyUid": keyUid, "kdfIterations": 256000, } return self.api_valid_request(method, data) def logout(self, user=user_1): method = "Logout" return self.api_valid_request(method, {}) def restore_account_and_wait_for_rpc_client_to_start(self, timeout=60): self.restore_account_and_login() start_time = time.time() # ToDo: change this part for waiting for `node.login` signal when websockets are migrated to StatusBackend while time.time() - start_time <= timeout: try: self.rpc_valid_request(method='accounts_getKeypairs') return except AssertionError: time.sleep(3) raise TimeoutError( f"RPC client was not started after {timeout} seconds") @retry(stop=stop_after_delay(10), wait=wait_fixed(0.5), reraise=True) 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) def get_accounts(self, params=[]): method = "accounts_getAccounts" response = self.rpc_request(method, params) self.verify_is_valid_json_rpc_response(response) return response.json() def get_pubkey(self, display_name): response = self.get_accounts() accounts = response.get("result", []) for account in accounts: if account.get("name") == display_name: return account.get("public-key") raise ValueError( f"Public key not found for display name: {display_name}") def send_contact_request(self, params=[]): method = "wakuext_sendContactRequest" response = self.rpc_request(method, params) self.verify_is_valid_json_rpc_response(response) return response.json() def send_message(self, params=[]): method = "wakuext_sendOneToOneMessage" response = self.rpc_request(method, params) self.verify_is_valid_json_rpc_response(response) return response.json()