test_: one to one messages (#6119)

* test_: one to one messages

* test_: use default display name

* test_: fix f-string format

* test_: fix signal log save

* test_: put signal saving under flag

* test_: addressed review comments

* test_: address review comment

.gitignore

@@ -107,7 +107,11 @@ __pycache__/
 report/results.xml
 tests-functional/coverage
 tests-functional/reports
+tests-functional/signals
 tests-functional/*.log
+pyrightconfig.json
+.venv
+
 
 # generated files
 mock

tests-functional/clients/rpc.py

@@ -2,7 +2,7 @@ import json
 import logging
 import jsonschema
 import requests
-
+from tenacity import retry, stop_after_delay, wait_fixed
 from conftest import option
 from json import JSONDecodeError
 
@@ -42,6 +42,7 @@ class RpcClient:
         assert response.content
         self._check_decode_and_key_errors_in_response(response, "error")
 
+    @retry(stop=stop_after_delay(10), wait=wait_fixed(0.5), reraise=True)
     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, "id": request_id}
@@ -50,7 +51,10 @@ class RpcClient:
         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)
         try:
-            logging.info(f"Got response: {json.dumps(response.json(), sort_keys=True, indent=4)}")
+            resp_json = response.json()
+            logging.info(f"Got response: {json.dumps(resp_json, sort_keys=True, indent=4)}")
+            if resp_json.get("error"):
+                assert "JSON-RPC client is unavailable" != resp_json["error"]
         except JSONDecodeError:
             logging.info(f"Got response: {response.content}")
         return response

tests-functional/clients/signals.py

@@ -3,10 +3,26 @@ import logging
 import time
 
 import websocket
+import os
+from pathlib import Path
+from constants import SIGNALS_DIR, LOG_SIGNALS_TO_FILE
+from datetime import datetime
+from enum import Enum
 
+class SignalType(Enum):
+    MESSAGES_NEW = "messages.new"
+    MESSAGE_DELIVERED = "message.delivered"
+    NODE_READY = "node.ready"
+    NODE_STARTED = "node.started"
+    NODE_LOGIN = "node.login"
+    MEDIASERVER_STARTED = "mediaserver.started"
+    WALLET_SUGGESTED_ROUTES = "wallet.suggested.routes"
+    WALLET_ROUTER_SIGN_TRANSACTIONS = "wallet.router.sign-transactions"
+    WALLET_ROUTER_SENDING_TRANSACTIONS_STARTED = "wallet.router.sending-transactions-started"
+    WALLET_TRANSACTION_STATUS_CHANGED = "wallet.transaction.status-changed"
+    WALLET_ROUTER_TRANSACTIONS_SENT = "wallet.router.transactions-sent"
 
 class SignalClient:
-
     def __init__(self, ws_url, await_signals):
         self.url = f"{ws_url}/signals"
 
@@ -24,14 +40,20 @@ class SignalClient:
                 "accept_fn": None
             } for signal in self.await_signals
         }
+        if LOG_SIGNALS_TO_FILE:
+            self.signal_file_path = os.path.join(SIGNALS_DIR, f"signal_{ws_url.split(':')[-1]}_{datetime.now().strftime('%H%M%S')}.log")
+            Path(SIGNALS_DIR).mkdir(parents=True, exist_ok=True)
 
     def on_message(self, ws, signal):
-        signal = json.loads(signal)
+        signal_data = json.loads(signal)
-        signal_type = signal.get("type")
+        if LOG_SIGNALS_TO_FILE:
+            self.write_signal_to_file(signal_data)
+
+        signal_type = signal_data.get("type")
         if signal_type in self.await_signals:
             accept_fn = self.received_signals[signal_type]["accept_fn"]
-            if not accept_fn or accept_fn(signal):
+            if not accept_fn or accept_fn(signal_data):
-                self.received_signals[signal_type]["received"].append(signal)
+                self.received_signals[signal_type]["received"].append(signal_data)
 
     def check_signal_type(self, signal_type):
         if signal_type not in self.await_signals:
@@ -65,6 +87,24 @@ class SignalClient:
             return self.received_signals[signal_type]["received"][-1]
         return self.received_signals[signal_type]["received"][-delta_count:]
 
+    def find_signal_containing_pattern(self, signal_type, event_pattern, timeout=20):
+        start_time = time.time()
+        while True:
+            if time.time() - start_time >= timeout:
+                raise TimeoutError(
+                    f"Signal {signal_type} containing {event_pattern} is not received in {timeout} seconds"
+                )
+            if not self.received_signals.get(signal_type):
+                time.sleep(0.2)
+                continue
+            for event in self.received_signals[signal_type]["received"]:
+                if event_pattern in str(event):
+                    logging.info(
+                        f"Signal {signal_type} containing {event_pattern} is received in {round(time.time() - start_time)} seconds"
+                    )
+                    return event
+            time.sleep(0.2)
+
     def _on_error(self, ws, error):
         logging.error(f"Error: {error}")
 
@@ -81,3 +121,8 @@ class SignalClient:
                                     on_close=self._on_close)
         ws.on_open = self._on_open
         ws.run_forever()
+
+    def write_signal_to_file(self, signal_data):
+        with open(self.signal_file_path, "a+") as file:
+            json.dump(signal_data, file)
+            file.write("\n")

tests-functional/clients/status_backend.py

@@ -4,12 +4,13 @@ import time
 import random
 import threading
 import requests
+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
+from constants import user_1, DEFAULT_DISPLAY_NAME
 
 
 
@@ -66,7 +67,7 @@ class StatusBackend(RpcClient, SignalClient):
         }
         return self.api_valid_request(method, data)
 
-    def create_account_and_login(self, display_name="Mr_Meeseeks", password=user_1.password):
+    def create_account_and_login(self, display_name=DEFAULT_DISPLAY_NAME, password=user_1.password):
         data_dir = f"dataDir_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
         method = "CreateAccountAndLogin"
         data = {
@@ -78,7 +79,7 @@ class StatusBackend(RpcClient, SignalClient):
         }
         return self.api_valid_request(method, data)
 
-    def restore_account_and_login(self, display_name="Mr_Meeseeks", user=user_1):
+    def restore_account_and_login(self, display_name=DEFAULT_DISPLAY_NAME, user=user_1):
         method = "RestoreAccountAndLogin"
         data = {
             "rootDataDir": "/",
@@ -118,6 +119,7 @@ class StatusBackend(RpcClient, SignalClient):
                 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)
@@ -139,3 +141,29 @@ class StatusBackend(RpcClient, SignalClient):
         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()

tests-functional/constants.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+import os
 
 
 @dataclass
@@ -21,3 +22,8 @@ user_2 = Account(
     password="Strong12345",
     passphrase="test test test test test test test test test test nest junk"
 )
+DEFAULT_DISPLAY_NAME = "Mr_Meeseeks"
+PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../"))
+TESTS_DIR = os.path.join(PROJECT_ROOT, "tests-functional")
+SIGNALS_DIR = os.path.join(TESTS_DIR, "signals")
+LOG_SIGNALS_TO_FILE = False # used for debugging purposes

tests-functional/requirements.txt

@@ -4,3 +4,5 @@ pytest==6.2.4
 requests==2.31.0
 genson~=1.2.2
 websocket-client~=1.4.2
+tenacity~=9.0.0
+pytest-dependency~=0.6.0

tests-functional/tests/test_cases.py

@@ -1,3 +1,4 @@
+from contextlib import contextmanager
 import json
 import logging
 import threading
@@ -6,10 +7,10 @@ from collections import namedtuple
 
 import pytest
 
-from clients.signals import SignalClient
+from clients.signals import SignalClient, SignalType
 from clients.status_backend import RpcClient, StatusBackend
 from conftest import option
-from constants import user_1, user_2
+from constants import user_1, user_2, DEFAULT_DISPLAY_NAME
 
 
 class StatusDTestCase:
@@ -24,7 +25,7 @@ class StatusDTestCase:
 class StatusBackendTestCase:
 
     await_signals = [
-        "node.ready"
+        SignalType.NODE_READY.value
     ]
 
     def setup_class(self):
@@ -32,7 +33,7 @@ class StatusBackendTestCase:
 
         self.rpc_client.init_status_backend()
         self.rpc_client.restore_account_and_login()
-        self.rpc_client.wait_for_signal("node.ready")
+        self.rpc_client.wait_for_signal(SignalType.NODE_READY.value)
 
         self.network_id = 31337
 
@@ -150,3 +151,63 @@ class SignalTestCase(StatusDTestCase):
         websocket_thread = threading.Thread(target=self.signal_client._connect)
         websocket_thread.daemon = True
         websocket_thread.start()
+
+
+class NetworkConditionTestCase:
+
+    @contextmanager
+    def add_latency(self):
+        pass
+        #TODO: To be implemented when we have docker exec capability
+
+    @contextmanager
+    def add_packet_loss(self):
+        pass
+        #TODO: To be implemented when we have docker exec capability
+
+    @contextmanager
+    def add_low_bandwith(self):
+        pass
+        #TODO: To be implemented when we have docker exec capability
+
+    @contextmanager
+    def node_pause(self, node):
+        pass
+        #TODO: To be implemented when we have docker exec capability
+
+class OneToOneMessageTestCase(NetworkConditionTestCase):
+
+    def initialize_backend(self, await_signals, display_name=DEFAULT_DISPLAY_NAME, url=None):
+        backend = StatusBackend(await_signals=await_signals, url=url)
+        backend.init_status_backend()
+        backend.create_account_and_login(display_name=display_name)
+        backend.start_messenger()
+        return backend
+
+
+    def validate_event_against_response(self, event, fields_to_validate, response):
+        messages_in_event = event["event"]["messages"]
+        assert len(messages_in_event) > 0, "No messages found in the event"
+        response_chat = response["result"]["chats"][0]
+
+        message_id = response_chat["lastMessage"]["id"]
+        message = next((message for message in messages_in_event if message["id"] == message_id), None)
+        assert message, f"Message with ID {message_id} not found in the event"
+
+        message_mismatch = []
+        for response_field, event_field in fields_to_validate.items():
+            response_value = response_chat["lastMessage"][response_field]
+            event_value = message[event_field]
+            if response_value != event_value:
+                message_mismatch.append(
+                    f"Field '{response_field}': Expected '{response_value}', Found '{event_value}'"
+                )
+
+        if not message_mismatch:
+            return
+
+        raise AssertionError(
+            "Some Sender RPC responses are not matching the signals received by the receiver.\n"
+            "Details of mismatches:\n" +
+            "\n".join(message_mismatch)
+        )

tests-functional/tests/test_init_status_app.py

@@ -1,8 +1,8 @@
 from test_cases import StatusBackend
 import pytest
+from clients.signals import SignalType
 import os
 
-
 @pytest.mark.create_account
 @pytest.mark.rpc
 class TestInitialiseApp:
@@ -12,10 +12,10 @@ class TestInitialiseApp:
 
         await_signals = [
 
-            "mediaserver.started",
+            SignalType.MEDIASERVER_STARTED.value,
-            "node.started",
+            SignalType.NODE_STARTED.value,
-            "node.ready",
+            SignalType.NODE_READY.value,
-            "node.login",
+            SignalType.NODE_LOGIN.value,
         ]
 
         backend_client = StatusBackend(await_signals)
@@ -24,13 +24,13 @@ class TestInitialiseApp:
 
         assert backend_client is not None
         backend_client.verify_json_schema(
-            backend_client.wait_for_signal("mediaserver.started"), "signal_mediaserver_started")
+            backend_client.wait_for_signal(SignalType.MEDIASERVER_STARTED.value), "signal_mediaserver_started")
         backend_client.verify_json_schema(
-            backend_client.wait_for_signal("node.started"), "signal_node_started")
+            backend_client.wait_for_signal(SignalType.NODE_STARTED.value), "signal_node_started")
         backend_client.verify_json_schema(
-            backend_client.wait_for_signal("node.ready"), "signal_node_ready")
+            backend_client.wait_for_signal(SignalType.NODE_READY.value), "signal_node_ready")
         backend_client.verify_json_schema(
-            backend_client.wait_for_signal("node.login"), "signal_node_login")
+            backend_client.wait_for_signal(SignalType.NODE_LOGIN.value), "signal_node_login")
 
 
 @pytest.mark.rpc

tests-functional/tests/test_one_to_one_messages.py

@@ -0,0 +1,75 @@
+from time import sleep
+from uuid import uuid4
+import pytest
+from test_cases import OneToOneMessageTestCase
+from constants import DEFAULT_DISPLAY_NAME
+from clients.signals import SignalType
+
+@pytest.mark.rpc
+class TestOneToOneMessages(OneToOneMessageTestCase):
+
+    @pytest.fixture(scope="class", autouse=True)
+    def setup_nodes(self, request):
+        await_signals = [
+            SignalType.MESSAGES_NEW.value,
+            SignalType.MESSAGE_DELIVERED.value,
+        ]
+        request.cls.sender = self.sender = self.initialize_backend(await_signals=await_signals)
+        request.cls.receiver = self.receiver = self.initialize_backend(await_signals=await_signals)
+
+    @pytest.mark.dependency(name="test_one_to_one_message_baseline")
+    def test_one_to_one_message_baseline(self, message_count=1):
+        pk_receiver = self.receiver.get_pubkey(DEFAULT_DISPLAY_NAME)
+
+        self.sender.send_contact_request([{"id": pk_receiver, "message": "contact_request"}])
+
+        sent_messages = []
+        for i in range(message_count):
+            message_text = f"test_message_{i+1}_{uuid4()}"
+            response = self.sender.send_message([{"id": pk_receiver, "message": message_text}])
+            sent_messages.append((message_text, response))
+            sleep(0.01)
+
+        for i, (message_text, response) in enumerate(sent_messages):
+            messages_new_event = self.receiver.find_signal_containing_pattern(SignalType.MESSAGES_NEW.value, event_pattern=message_text, timeout=60)
+            self.validate_event_against_response(
+                messages_new_event,
+                fields_to_validate={"text": "text"},
+                response=response,
+            )
+
+    @pytest.mark.dependency(depends=["test_one_to_one_message_baseline"])
+    def test_multiple_one_to_one_messages(self):
+        self.test_one_to_one_message_baseline(message_count=50)
+
+    @pytest.mark.dependency(depends=["test_one_to_one_message_baseline"])
+    @pytest.mark.skip(reason="Skipping until add_latency is implemented")
+    def test_one_to_one_message_with_latency(self):
+        with self.add_latency():
+            self.test_one_to_one_message_baseline()
+
+    @pytest.mark.dependency(depends=["test_one_to_one_message_baseline"])
+    @pytest.mark.skip(reason="Skipping until add_packet_loss is implemented")
+    def test_one_to_one_message_with_packet_loss(self):
+        with self.add_packet_loss():
+            self.test_one_to_one_message_baseline()
+
+    @pytest.mark.dependency(depends=["test_one_to_one_message_baseline"])
+    @pytest.mark.skip(reason="Skipping until add_low_bandwith is implemented")
+    def test_one_to_one_message_with_low_bandwidth(self):
+        with self.add_low_bandwith():
+            self.test_one_to_one_message_baseline()
+
+    @pytest.mark.dependency(depends=["test_one_to_one_message_baseline"])
+    @pytest.mark.skip(reason="Skipping until node_pause is implemented")
+    def test_one_to_one_message_with_node_pause_30_seconds(self):
+        pk_receiver = self.receiver.get_pubkey("Receiver")
+        self.sender.send_contact_request([{"id": pk_receiver, "message": "contact_request"}])
+        with self.node_pause(self.receiver):
+            message_text = f"test_message_{uuid4()}"
+            self.sender.send_message([{"id": pk_receiver, "message": message_text}])
+            sleep(30)
+        self.receiver.find_signal_containing_pattern(SignalType.MESSAGES_NEW.value, event_pattern=message_text)
+        self.sender.wait_for_signal("messages.delivered")
+
+    

tests-functional/tests/test_router.py

@@ -5,18 +5,18 @@ import pytest
 from conftest import option
 from constants import user_1, user_2
 from test_cases import SignalTestCase
-
+from clients.signals import SignalType
 
 @pytest.mark.rpc
 @pytest.mark.transaction
 @pytest.mark.wallet
 class TestTransactionFromRoute(SignalTestCase):
     await_signals = [
-        "wallet.suggested.routes",
+        SignalType.WALLET_SUGGESTED_ROUTES.value,
-        "wallet.router.sign-transactions",
+        SignalType.WALLET_ROUTER_SIGN_TRANSACTIONS.value,
-        "wallet.router.sending-transactions-started",
+        SignalType.WALLET_ROUTER_SENDING_TRANSACTIONS_STARTED.value,
-        "wallet.transaction.status-changed",
+        SignalType.WALLET_TRANSACTION_STATUS_CHANGED.value,
-        "wallet.router.transactions-sent"
+        SignalType.WALLET_ROUTER_TRANSACTIONS_SENT.value,
     ]
 
     def test_tx_from_route(self):
@@ -44,7 +44,7 @@ class TestTransactionFromRoute(SignalTestCase):
         ]
         response = self.rpc_client.rpc_valid_request(method, params)
 
-        routes = self.signal_client.wait_for_signal("wallet.suggested.routes")
+        routes = self.signal_client.wait_for_signal(SignalType.WALLET_SUGGESTED_ROUTES.value)
         assert routes['event']['Uuid'] == _uuid
 
         method = "wallet_buildTransactionsFromRoute"
@@ -57,7 +57,7 @@ class TestTransactionFromRoute(SignalTestCase):
         response = self.rpc_client.rpc_valid_request(method, params)
 
         wallet_router_sign_transactions = self.signal_client.wait_for_signal(
-            "wallet.router.sign-transactions")
+            SignalType.WALLET_ROUTER_SIGN_TRANSACTIONS.value)
 
         assert wallet_router_sign_transactions['event']['signingDetails']['signOnKeycard'] == False
         transaction_hashes = wallet_router_sign_transactions['event']['signingDetails']['hashes']
@@ -98,7 +98,7 @@ class TestTransactionFromRoute(SignalTestCase):
         response = self.rpc_client.rpc_valid_request(method, params)
 
         tx_status = self.signal_client.wait_for_signal(
-            "wallet.transaction.status-changed")
+            SignalType.WALLET_TRANSACTION_STATUS_CHANGED.value)
 
         assert tx_status["event"]["chainId"] == 31337
         assert tx_status["event"]["status"] == "Success"