test_: create private group tests (#6225)

* test_: create private group tests

* test_: set privileged False for jenkins

* test_: run baseline tests in rpc suite

* test_: address review comments

* test_: address review comments

tests-functional/clients/services/accounts.py

@@ -0,0 +1,15 @@
+from clients.rpc import RpcClient
+from clients.services.service import Service
+
+
+class AccountService(Service):
+    def __init__(self, client: RpcClient):
+        super().__init__(client, "accounts")
+
+    def get_accounts(self):
+        response = self.rpc_request("getAccounts")
+        return response.json()
+
+    def get_account_keypairs(self):
+        response = self.rpc_request("getKeypairs")
+        return response.json()

tests-functional/clients/services/service.py

@@ -9,4 +9,4 @@ class Service:
 
     def rpc_request(self, method: str, params=None):
         full_method_name = f"{self.name}_{method}"
-        return self.rpc_client.rpc_request(full_method_name, params)
+        return self.rpc_client.rpc_valid_request(full_method_name, params)

tests-functional/clients/services/settings.py

@@ -0,0 +1,11 @@
+from clients.rpc import RpcClient
+from clients.services.service import Service
+
+
+class SettingsService(Service):
+    def __init__(self, client: RpcClient):
+        super().__init__(client, "settings")
+
+    def get_settings(self):
+        response = self.rpc_request("getSettings")
+        return response.json()

tests-functional/clients/services/wakuext.py

@@ -0,0 +1,40 @@
+from clients.rpc import RpcClient
+from clients.services.service import Service
+
+
+class WakuextService(Service):
+    def __init__(self, client: RpcClient):
+        super().__init__(client, "wakuext")
+
+    def send_contact_request(self, contact_id: str, message: str):
+        params = [{"id": contact_id, "message": message}]
+        response = self.rpc_request("sendContactRequest", params)
+        return response.json()
+
+    def accept_contact_request(self, request_id: str):
+        params = [{"id": request_id}]
+        response = self.rpc_request("acceptContactRequest", params)
+        return response.json()
+
+    def get_contacts(self):
+        response = self.rpc_request("contacts")
+        return response.json()
+
+    def send_message(self, contact_id: str, message: str):
+        params = [{"id": contact_id, "message": message}]
+        response = self.rpc_request("sendOneToOneMessage", params)
+        return response.json()
+
+    def start_messenger(self):
+        response = self.rpc_request("startMessenger")
+        json_response = response.json()
+
+        if "error" in json_response:
+            assert json_response["error"]["code"] == -32000
+            assert json_response["error"]["message"] == "messenger already started"
+            return
+
+    def create_group_chat_with_members(self, pubkey_list: list, group_chat_name: str):
+        params = [None, group_chat_name, pubkey_list]
+        response = self.rpc_request("createGroupChatWithMembers", params)
+        return response.json()

tests-functional/clients/services/wallet.py

@@ -9,3 +9,6 @@ class WalletService(Service):
     def get_balances_at_by_chain(self, chains: list, addresses: list, tokens: list):
         params = [chains, addresses, tokens]
         return self.rpc_request("getBalancesByChain", params)
+
+    def start_wallet(self):
+        return self.rpc_request("startWallet")

tests-functional/clients/status_backend.py

@@ -1,6 +1,7 @@
 import io
 import json
 import logging
+import string
 import tarfile
 import tempfile
 import time
@@ -10,12 +11,15 @@ import requests
 import docker
 import docker.errors
 import os
-
+from clients.services.wallet import WalletService
-from tenacity import retry, stop_after_delay, wait_fixed
+from clients.services.wakuext import WakuextService
-from clients.signals import SignalClient
+from clients.services.accounts import AccountService
+from clients.services.settings import SettingsService
+from clients.signals import SignalClient, SignalType
 from clients.rpc import RpcClient
 from conftest import option
 from resources.constants import user_1, DEFAULT_DISPLAY_NAME, USER_DIR
+from docker.errors import APIError
 
 NANOSECONDS_PER_SECOND = 1_000_000_000
 
@@ -24,7 +28,7 @@ class StatusBackend(RpcClient, SignalClient):
 
     container = None
 
-    def __init__(self, await_signals=[]):
+    def __init__(self, await_signals=[], privileged=False):
 
         if option.status_backend_url:
             url = option.status_backend_url
@@ -32,7 +36,7 @@ class StatusBackend(RpcClient, SignalClient):
             self.docker_client = docker.from_env()
             host_port = random.choice(option.status_backend_port_range)
 
-            self.container = self._start_container(host_port)
+            self.container = self._start_container(host_port, privileged)
             url = f"http://127.0.0.1:{host_port}"
             option.status_backend_port_range.remove(host_port)
 
@@ -40,6 +44,7 @@ class StatusBackend(RpcClient, SignalClient):
         self.api_url = f"{url}/statusgo"
         self.ws_url = f"{url}".replace("http", "ws")
         self.rpc_url = f"{url}/statusgo/CallRPC"
+        self.public_key = ""
 
         RpcClient.__init__(self, self.rpc_url)
         SignalClient.__init__(self, self.ws_url, await_signals)
@@ -50,7 +55,12 @@ class StatusBackend(RpcClient, SignalClient):
         websocket_thread.daemon = True
         websocket_thread.start()
 
-    def _start_container(self, host_port):
+        self.wallet_service = WalletService(self)
+        self.wakuext_service = WakuextService(self)
+        self.accounts_service = AccountService(self)
+        self.settings_service = SettingsService(self)
+
+    def _start_container(self, host_port, privileged):
         docker_project_name = option.docker_project_name
 
         timestamp = int(time.time() * 1000)  # Keep in sync with run_functional_tests.sh
@@ -62,6 +72,7 @@ class StatusBackend(RpcClient, SignalClient):
         container_args = {
             "image": image_name,
             "detach": True,
+            "privileged": privileged,
             "name": container_name,
             "labels": {"com.docker.compose.project": docker_project_name},
             "entrypoint": [
@@ -182,21 +193,26 @@ class StatusBackend(RpcClient, SignalClient):
     def create_account_and_login(
         self,
         data_dir=USER_DIR,
-        display_name=DEFAULT_DISPLAY_NAME,
+        display_name=None,
         password=user_1.password,
     ):
+        self.display_name = (
+            display_name if display_name else f"DISP_NAME_{''.join(random.choices(string.ascii_letters + string.digits + '_-', k=10))}"
+        )
         method = "CreateAccountAndLogin"
         data = {
             "rootDataDir": data_dir,
             "kdfIterations": 256000,
-            "displayName": display_name,
+            "displayName": self.display_name,
             "password": password,
             "customizationColor": "primary",
             "logEnabled": True,
             "logLevel": "DEBUG",
         }
         data = self._set_proxy_credentials(data)
-        return self.api_valid_request(method, data)
+        resp = self.api_valid_request(method, data)
+        self.node_login_event = self.find_signal_containing_pattern(SignalType.NODE_LOGIN.value, event_pattern=self.display_name)
+        return resp
 
     def restore_account_and_login(
         self,
@@ -256,72 +272,34 @@ class StatusBackend(RpcClient, SignalClient):
         # 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")
+                self.accounts_service.get_account_keypairs()
                 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 container_pause(self):
-    def start_messenger(self, params=[]):
+        if not self.container:
-        method = "wakuext_startMessenger"
+            raise RuntimeError("Container is not initialized.")
-        response = self.rpc_request(method, params)
+        self.container.pause()
-        json_response = response.json()
+        logging.info(f"Container {self.container.name} paused.")
 
-        if "error" in json_response:
+    def container_unpause(self):
-            assert json_response["error"]["code"] == -32000
+        if not self.container:
-            assert json_response["error"]["message"] == "messenger already started"
+            raise RuntimeError("Container is not initialized.")
-            return
+        self.container.unpause()
+        logging.info(f"Container {self.container.name} unpaused.")
 
-        self.verify_is_valid_json_rpc_response(response)
+    def container_exec(self, command):
+        if not self.container:
+            raise RuntimeError("Container is not initialized.")
+        try:
+            exec_result = self.container.exec_run(cmd=["sh", "-c", command], stdout=True, stderr=True, tty=False)
+            if exec_result.exit_code != 0:
+                raise RuntimeError(f"Failed to execute command in container {self.container.id}:\n" f"OUTPUT: {exec_result.output.decode().strip()}")
+            return exec_result.output.decode().strip()
+        except APIError as e:
+            raise RuntimeError(f"API error during container execution: {str(e)}") from e
 
-    def start_wallet(self, params=[]):
+    def find_public_key(self):
-        method = "wallet_startWallet"
+        self.public_key = self.node_login_event.get("event", {}).get("settings", {}).get("public-key")
-        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, contact_id: str, message: str):
-        method = "wakuext_sendContactRequest"
-        params = [{"id": contact_id, "message": message}]
-        response = self.rpc_request(method, params)
-        self.verify_is_valid_json_rpc_response(response)
-        return response.json()
-
-    def accept_contact_request(self, chat_id: str):
-        method = "wakuext_acceptContactRequest"
-        params = [{"id": chat_id}]
-        response = self.rpc_request(method, params)
-        self.verify_is_valid_json_rpc_response(response)
-        return response.json()
-
-    def get_contacts(self):
-        method = "wakuext_contacts"
-        response = self.rpc_request(method)
-        self.verify_is_valid_json_rpc_response(response)
-        return response.json()
-
-    def send_message(self, contact_id: str, message: str):
-        method = "wakuext_sendOneToOneMessage"
-        params = [{"id": contact_id, "message": message}]
-        response = self.rpc_request(method, params)
-        self.verify_is_valid_json_rpc_response(response)
-        return response.json()

tests-functional/tests/reliability/test_contact_request.py

@@ -1,38 +1,28 @@
 from time import sleep
 from uuid import uuid4
 import pytest
-from test_cases import OneToOneMessageTestCase
+from test_cases import MessengerTestCase
-from resources.constants import DEFAULT_DISPLAY_NAME
 from clients.signals import SignalType
 from resources.enums import MessageContentType
 
 
-@pytest.mark.rpc
+@pytest.mark.reliability
-class TestContactRequests(OneToOneMessageTestCase):
+class TestContactRequests(MessengerTestCase):
 
+    @pytest.mark.rpc  # until we have dedicated functional tests for this we can still run this test as part of the functional tests suite
     @pytest.mark.dependency(name="test_contact_request_baseline")
     def test_contact_request_baseline(self, execution_number=1):
-
-        await_signals = [
-            SignalType.MESSAGES_NEW.value,
-            SignalType.MESSAGE_DELIVERED.value,
-        ]
-
         message_text = f"test_contact_request_{execution_number}_{uuid4()}"
+        sender = self.initialize_backend(await_signals=self.await_signals)
+        receiver = self.initialize_backend(await_signals=self.await_signals)
 
-        sender = self.initialize_backend(await_signals=await_signals)
+        existing_contacts = receiver.wakuext_service.get_contacts()
-        receiver = self.initialize_backend(await_signals=await_signals)
 
-        pk_sender = sender.get_pubkey(DEFAULT_DISPLAY_NAME)
+        if sender.public_key in str(existing_contacts):
-        pk_receiver = receiver.get_pubkey(DEFAULT_DISPLAY_NAME)
-
-        existing_contacts = receiver.get_contacts()
-
-        if pk_sender in str(existing_contacts):
             pytest.skip("Contact request was already sent for this sender<->receiver. Skipping test!!")
 
-        response = sender.send_contact_request(pk_receiver, message_text)
+        response = sender.wakuext_service.send_contact_request(receiver.public_key, message_text)
-        expected_message = self.get_message_by_content_type(response, content_type=MessageContentType.CONTACT_REQUEST.value)
+        expected_message = self.get_message_by_content_type(response, content_type=MessageContentType.CONTACT_REQUEST.value)[0]
 
         messages_new_event = receiver.find_signal_containing_pattern(
             SignalType.MESSAGES_NEW.value,
@@ -44,7 +34,9 @@ class TestContactRequests(OneToOneMessageTestCase):
         if "messages" in messages_new_event.get("event", {}):
             signal_messages_texts.extend(message["text"] for message in messages_new_event["event"]["messages"] if "text" in message)
 
-        assert f"@{pk_sender} sent you a contact request" in signal_messages_texts, "Couldn't find the signal corresponding to the contact request"
+        assert (
+            f"@{sender.public_key} sent you a contact request" in signal_messages_texts
+        ), "Couldn't find the signal corresponding to the contact request"
 
         self.validate_signal_event_against_response(
             signal_event=messages_new_event,
@@ -67,23 +59,28 @@ class TestContactRequests(OneToOneMessageTestCase):
     @pytest.mark.dependency(depends=["test_contact_request_baseline"])
     @pytest.mark.skip(reason="Skipping until add_latency is implemented")
     def test_contact_request_with_latency(self):
-        with self.add_latency():
+        # with self.add_latency():
-            self.test_contact_request_baseline()
+        #     self.test_contact_request_baseline()
+        # to be done in the next PR
+        pass
 
     @pytest.mark.dependency(depends=["test_contact_request_baseline"])
     @pytest.mark.skip(reason="Skipping until add_packet_loss is implemented")
     def test_contact_request_with_packet_loss(self):
-        with self.add_packet_loss():
+        # with self.add_packet_loss():
-            self.test_contact_request_baseline()
+        #     self.test_contact_request_baseline()
+        # to be done in the next PR
+        pass
 
     @pytest.mark.dependency(depends=["test_contact_request_baseline"])
     @pytest.mark.skip(reason="Skipping until add_low_bandwith is implemented")
     def test_contact_request_with_low_bandwidth(self):
-        with self.add_low_bandwith():
+        # with self.add_low_bandwith():
-            self.test_contact_request_baseline()
+        #     self.test_contact_request_baseline()
+        # to be done in the next PR
+        pass
 
     @pytest.mark.dependency(depends=["test_contact_request_baseline"])
-    @pytest.mark.skip(reason="Skipping until node_pause is implemented")
     def test_contact_request_with_node_pause_30_seconds(self):
         await_signals = [
             SignalType.MESSAGES_NEW.value,
@@ -91,11 +88,11 @@ class TestContactRequests(OneToOneMessageTestCase):
         ]
         sender = self.initialize_backend(await_signals=await_signals)
         receiver = self.initialize_backend(await_signals=await_signals)
-        pk_receiver = receiver.get_pubkey(DEFAULT_DISPLAY_NAME)
 
         with self.node_pause(receiver):
             message_text = f"test_contact_request_{uuid4()}"
-            sender.send_contact_request(pk_receiver, message_text)
+            response = sender.wakuext_service.send_contact_request(receiver.public_key, message_text)
+            expected_message = self.get_message_by_content_type(response, content_type=MessageContentType.CONTACT_REQUEST.value)[0]
             sleep(30)
-        receiver.find_signal_containing_pattern(SignalType.MESSAGES_NEW.value, event_pattern=message_text)
+        receiver.find_signal_containing_pattern(SignalType.MESSAGES_NEW.value, event_pattern=expected_message.get("id"))
-        sender.wait_for_signal("messages.delivered")
+        sender.wait_for_signal(SignalType.MESSAGE_DELIVERED.value)

tests-functional/tests/reliability/test_create_private_groups.py

@@ -0,0 +1,77 @@
+from time import sleep
+from uuid import uuid4
+import pytest
+from test_cases import MessengerTestCase
+from clients.signals import SignalType
+from resources.enums import MessageContentType
+
+
+@pytest.mark.usefixtures("setup_two_nodes")
+@pytest.mark.reliability
+class TestCreatePrivateGroups(MessengerTestCase):
+
+    @pytest.mark.rpc  # until we have dedicated functional tests for this we can still run this test as part of the functional tests suite
+    @pytest.mark.dependency(name="test_create_private_group_baseline")
+    def test_create_private_group_baseline(self, private_groups_count=1):
+        self.make_contacts()
+
+        private_groups = []
+        for i in range(private_groups_count):
+            private_group_name = f"private_group_{i+1}_{uuid4()}"
+            response = self.sender.wakuext_service.create_group_chat_with_members([self.receiver.public_key], private_group_name)
+
+            expected_group_creation_msg = f"@{self.sender.public_key} created the group {private_group_name}"
+            expected_message = self.get_message_by_content_type(
+                response, content_type=MessageContentType.SYSTEM_MESSAGE_CONTENT_PRIVATE_GROUP.value, message_pattern=expected_group_creation_msg
+            )[0]
+
+            private_groups.append(expected_message)
+            sleep(0.01)
+
+        for i, expected_message in enumerate(private_groups):
+            messages_new_event = self.receiver.find_signal_containing_pattern(
+                SignalType.MESSAGES_NEW.value, event_pattern=expected_message.get("id"), timeout=60
+            )
+            self.validate_signal_event_against_response(
+                signal_event=messages_new_event,
+                expected_message=expected_message,
+                fields_to_validate={"text": "text"},
+            )
+
+    @pytest.mark.dependency(depends=["test_create_private_group_baseline"])
+    def test_multiple_one_create_private_groups(self):
+        self.test_create_private_group_baseline(private_groups_count=50)
+
+    @pytest.mark.dependency(depends=["test_create_private_group_baseline"])
+    @pytest.mark.skip(reason="Skipping until add_latency is implemented")
+    def test_create_private_groups_with_latency(self):
+        # with self.add_latency():
+        #     self.test_create_private_group_baseline()
+        # to be done in the next PR
+        pass
+
+    @pytest.mark.dependency(depends=["test_create_private_group_baseline"])
+    @pytest.mark.skip(reason="Skipping until add_packet_loss is implemented")
+    def test_create_private_groups_with_packet_loss(self):
+        # with self.add_packet_loss():
+        #     self.test_create_private_group_baseline()
+        # to be done in the next PR
+        pass
+
+    @pytest.mark.dependency(depends=["test_create_private_group_baseline"])
+    @pytest.mark.skip(reason="Skipping until add_low_bandwith is implemented")
+    def test_create_private_groups_with_low_bandwidth(self):
+        # with self.add_low_bandwith():
+        #     self.test_create_private_group_baseline()
+        # to be done in the next PR
+        pass
+
+    @pytest.mark.dependency(depends=["test_create_private_group_baseline"])
+    def test_create_private_groups_with_node_pause_30_seconds(self):
+        self.make_contacts()
+
+        with self.node_pause(self.receiver):
+            private_group_name = f"private_group_{uuid4()}"
+            self.sender.wakuext_service.create_group_chat_with_members([self.receiver.public_key], private_group_name)
+            sleep(30)
+        self.receiver.find_signal_containing_pattern(SignalType.MESSAGES_NEW.value, event_pattern=private_group_name)

tests-functional/tests/reliability/test_one_to_one_messages.py

@@ -1,35 +1,23 @@
 from time import sleep
 from uuid import uuid4
 import pytest
-from test_cases import OneToOneMessageTestCase
+from test_cases import MessengerTestCase
-from resources.constants import DEFAULT_DISPLAY_NAME
 from clients.signals import SignalType
 from resources.enums import MessageContentType
 
 
-@pytest.mark.rpc
+@pytest.mark.usefixtures("setup_two_nodes")
-class TestOneToOneMessages(OneToOneMessageTestCase):
+@pytest.mark.reliability
-
+class TestOneToOneMessages(MessengerTestCase):
-    @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.rpc  # until we have dedicated functional tests for this we can still run this test as part of the functional tests suite
     @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(pk_receiver, "contact_request")
-
         sent_messages = []
         for i in range(message_count):
             message_text = f"test_message_{i+1}_{uuid4()}"
-            response = self.sender.send_message(pk_receiver, message_text)
+            response = self.sender.wakuext_service.send_message(self.receiver.public_key, message_text)
-            expected_message = self.get_message_by_content_type(response, content_type=MessageContentType.TEXT_PLAIN.value)
+            expected_message = self.get_message_by_content_type(response, content_type=MessageContentType.TEXT_PLAIN.value)[0]
             sent_messages.append(expected_message)
             sleep(0.01)
 
@@ -52,29 +40,32 @@ class TestOneToOneMessages(OneToOneMessageTestCase):
     @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():
+        # with self.add_latency():
-            self.test_one_to_one_message_baseline()
+        #     self.test_one_to_one_message_baseline()
+        # to be done in the next PR
+        pass
 
     @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():
+        # with self.add_packet_loss():
-            self.test_one_to_one_message_baseline()
+        #     self.test_one_to_one_message_baseline()
+        # to be done in the next PR
+        pass
 
     @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():
+        # with self.add_low_bandwith():
-            self.test_one_to_one_message_baseline()
+        #     self.test_one_to_one_message_baseline()
+        # to be done in the next PR
+        pass
 
     @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(pk_receiver, "contact_request")
         with self.node_pause(self.receiver):
             message_text = f"test_message_{uuid4()}"
-            self.sender.send_message(pk_receiver, message_text)
+            self.sender.wakuext_service.send_message(self.receiver.public_key, 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")
+        self.sender.wait_for_signal(SignalType.MESSAGE_DELIVERED.value)

tests-functional/tests/test_cases.py

@@ -11,7 +11,8 @@ from clients.services.wallet import WalletService
 from clients.signals import SignalClient, SignalType
 from clients.status_backend import RpcClient, StatusBackend
 from conftest import option
-from resources.constants import user_1, user_2, DEFAULT_DISPLAY_NAME
+from resources.constants import user_1, user_2
+from resources.enums import MessageContentType
 
 
 class StatusDTestCase:
@@ -151,47 +152,80 @@ class SignalTestCase(StatusDTestCase):
 class NetworkConditionTestCase:
 
     @contextmanager
-    def add_latency(self):
+    def add_latency(self, node, latency=300, jitter=50):
+        logging.info("Entering context manager: add_latency")
+        node.container_exec(f"apk add iproute2 && tc qdisc add dev eth0 root netem delay {latency}ms {jitter}ms distribution normal")
         try:
-            # TODO: To be implemented when we have docker exec capability
             yield
         finally:
-            pass
+            logging.info("Exiting context manager: add_latency")
+            node.container_exec("tc qdisc del dev eth0 root")
 
     @contextmanager
-    def add_packet_loss(self):
+    def add_packet_loss(self, node, packet_loss=2):
+        logging.info("Entering context manager: add_packet_loss")
+        node.container_exec(f"apk add iproute2 && tc qdisc add dev eth0 root netem loss {packet_loss}%")
         try:
-            # TODO: To be implemented when we have docker exec capability
             yield
         finally:
-            pass
+            logging.info("Exiting context manager: add_packet_loss")
+            node.container_exec("tc qdisc del dev eth0 root netem")
 
     @contextmanager
-    def add_low_bandwith(self):
+    def add_low_bandwith(self, node, rate="1mbit", burst="32kbit"):
+        logging.info("Entering context manager: add_low_bandwith")
+        node.container_exec(f"apk add iproute2 && tc qdisc add dev eth0 root tbf rate {rate} burst {burst}")
         try:
-            # TODO: To be implemented when we have docker exec capability
             yield
         finally:
-            pass
+            logging.info("Exiting context manager: add_low_bandwith")
+            node.container_exec("tc qdisc del dev eth0 root")
 
     @contextmanager
     def node_pause(self, node):
+        logging.info("Entering context manager: node_pause")
+        node.container_pause()
         try:
-            # TODO: To be implemented when we have docker exec capability
             yield
         finally:
-            pass
+            logging.info("Exiting context manager: node_pause")
+            node.container_unpause()
 
 
-class OneToOneMessageTestCase(NetworkConditionTestCase):
+class MessengerTestCase(NetworkConditionTestCase):
 
-    def initialize_backend(self, await_signals, display_name=DEFAULT_DISPLAY_NAME):
+    await_signals = [
+        SignalType.MESSAGES_NEW.value,
+        SignalType.MESSAGE_DELIVERED.value,
+        SignalType.NODE_LOGIN.value,
+    ]
+
+    @pytest.fixture(scope="class", autouse=False)
+    def setup_two_nodes(self, request):
+        request.cls.sender = self.sender = self.initialize_backend(await_signals=self.await_signals)
+        request.cls.receiver = self.receiver = self.initialize_backend(await_signals=self.await_signals)
+
+    def initialize_backend(self, await_signals):
         backend = StatusBackend(await_signals=await_signals)
         backend.init_status_backend()
-        backend.create_account_and_login(display_name=display_name)
+        backend.create_account_and_login()
-        backend.start_messenger()
+        backend.find_public_key()
+        backend.wakuext_service.start_messenger()
         return backend
 
+    def make_contacts(self):
+        existing_contacts = self.receiver.wakuext_service.get_contacts()
+
+        if self.sender.public_key in str(existing_contacts):
+            return
+
+        response = self.sender.wakuext_service.send_contact_request(self.receiver.public_key, "contact_request")
+        expected_message = self.get_message_by_content_type(response, content_type=MessageContentType.CONTACT_REQUEST.value)[0]
+        self.receiver.find_signal_containing_pattern(SignalType.MESSAGES_NEW.value, event_pattern=expected_message.get("id"))
+        self.receiver.wakuext_service.accept_contact_request(expected_message.get("id"))
+        accepted_signal = f"@{self.receiver.public_key} accepted your contact request"
+        self.sender.find_signal_containing_pattern(SignalType.MESSAGES_NEW.value, event_pattern=accepted_signal)
+
     def validate_signal_event_against_response(self, signal_event, fields_to_validate, expected_message):
         expected_message_id = expected_message.get("id")
         signal_event_messages = signal_event.get("event", {}).get("messages")
@@ -218,10 +252,15 @@ class OneToOneMessageTestCase(NetworkConditionTestCase):
             "Details of mismatches:\n" + "\n".join(message_mismatch)
         )
 
-    def get_message_by_content_type(self, response, content_type):
+    def get_message_by_content_type(self, response, content_type, message_pattern=""):
+        matched_messages = []
         messages = response.get("result", {}).get("messages", [])
         for message in messages:
-            if message.get("contentType") == content_type:
+            if message.get("contentType") != content_type:
-                return message
+                continue
-
+            if not message_pattern or message_pattern in str(message):
-        raise ValueError(f"Failed to find a message with contentType '{content_type}' in response")
+                matched_messages.append(message)
+        if matched_messages:
+            return matched_messages
+        else:
+            raise ValueError(f"Failed to find a message with contentType '{content_type}' in response")