test_: Test private chat messages (#6259)

* test_: private chat messages

* test_: fix

* test_: apply network conditions

* test_: fix low bandwidth command

* test_: unskip remaining tests

* test_: fix test name

* test_: run tests in parallel

* test_: fix deps

* test_: remove dependencies

* test_: unique container name

* test_: increase port range

* test_: fix container cleanup

* test_: prepare for code review

* test_: debug rpc tests

* test_: debug rpc tests

* test_: revert port change to test failures

* test_: try larger port

* test_: address code review

* test_: port checks

* test_: add missing newline

* test_: cleanup all containers

* test_: fix pylint

_assets/ci/Jenkinsfile.tests-rpc

@@ -79,7 +79,7 @@ pipeline {
       cleanup {
         script {
           sh '''
-            docker ps -a --filter "name=status-go-func-tests-${BUILD_ID}" -q | xargs -r docker rm
+            docker ps -a --filter "name=status-go-func-tests-${BUILD_ID}" -q | xargs -r docker rm -f
             make git-clean
           '''
         }

_assets/scripts/run_functional_tests.sh

@@ -30,7 +30,7 @@ project_name="status-go-func-tests-${identifier}"
 export STATUS_BACKEND_URLS=$(eval echo http://${project_name}-status-backend-{1..${STATUS_BACKEND_COUNT}}:3333 | tr ' ' ,)
 
 # Remove orphans
-docker ps -a --filter "status-go-func-tests-${identifier}" --filter "status=exited" -q | xargs -r docker rm
+docker ps -a --filter "status-go-func-tests-${identifier}" --filter "status=exited" -q | xargs -r docker rm -f
 
 # Run docker
 echo -e "${GRN}Running tests${RST}, HEAD: $(git rev-parse HEAD)"

tests-functional/clients/services/wakuext.py

@@ -38,3 +38,8 @@ class WakuextService(Service):
         params = [None, group_chat_name, pubkey_list]
         response = self.rpc_request("createGroupChatWithMembers", params)
         return response.json()
+
+    def send_group_chat_message(self, group_id: str, message: str):
+        params = [{"id": group_id, "message": message}]
+        response = self.rpc_request("sendGroupChatMessage", params)
+        return response.json()

tests-functional/clients/status_backend.py

@@ -11,6 +11,8 @@ import requests
 import docker
 import docker.errors
 import os
+
+from tenacity import retry, stop_after_delay, wait_fixed
 from clients.services.wallet import WalletService
 from clients.services.wakuext import WakuextService
 from clients.services.accounts import AccountService
@@ -29,16 +31,24 @@ class StatusBackend(RpcClient, SignalClient):
     container = None
 
     def __init__(self, await_signals=[], privileged=False):
-
         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)
+            retries = 5
-
+            ports_tried = []
-            self.container = self._start_container(host_port, privileged)
+            for _ in range(retries):
-            url = f"http://127.0.0.1:{host_port}"
+                try:
-            option.status_backend_port_range.remove(host_port)
+                    host_port = random.choice(option.status_backend_port_range)
+                    ports_tried.append(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)
+                    break
+                except Exception:
+                    continue
+            else:
+                raise RuntimeError(f"Failed to start container on ports: {ports_tried}")
 
         self.base_url = url
         self.api_url = f"{url}/statusgo"
@@ -99,7 +109,7 @@ class StatusBackend(RpcClient, SignalClient):
         network = self.docker_client.networks.get(f"{docker_project_name}_default")
         network.connect(container)
 
-        option.status_backend_containers.append(container.id)
+        option.status_backend_containers.append(self)
         return container
 
     def wait_for_healthy(self, timeout=10):
@@ -312,3 +322,17 @@ class StatusBackend(RpcClient, SignalClient):
 
     def find_public_key(self):
         self.public_key = self.node_login_event.get("event", {}).get("settings", {}).get("public-key")
+
+    @retry(stop=stop_after_delay(5), wait=wait_fixed(0.1), reraise=True)
+    def kill(self):
+        if not self.container:
+            return
+        logging.info(f"Killing container with id {self.container.short_id}")
+        self.container.kill()
+        try:
+            self.container.remove()
+        except Exception as e:
+            logging.warning(f"Failed to remove container {self.container.short_id}: {e}")
+        finally:
+            self.container = None
+            logging.info("Container stopped.")

tests-functional/conftest.py

@@ -1,8 +1,7 @@
 import os
-import docker
-
 from dataclasses import dataclass, field
 from typing import List
+import pytest
 
 
 def pytest_addoption(parser):
@@ -61,21 +60,26 @@ def pytest_configure(config):
     executor_number = int(os.getenv("EXECUTOR_NUMBER", 5))
     base_port = 7000
     range_size = 100
+    max_port = 65535
+    min_port = 1024
 
     start_port = base_port + (executor_number * range_size)
+    end_port = start_port + 20000
 
-    option.status_backend_port_range = list(range(start_port, start_port + range_size - 1))
+    # Ensure generated ports are within the valid range
+    if start_port < min_port or end_port > max_port:
+        raise ValueError(f"Generated port range ({start_port}-{end_port}) is outside the allowed range ({min_port}-{max_port}).")
+
+    option.status_backend_port_range = list(range(start_port, end_port))
     option.status_backend_containers = []
 
     option.base_dir = os.path.dirname(os.path.abspath(__file__))
 
 
-def pytest_unconfigure():
+@pytest.fixture(scope="function", autouse=True)
-    docker_client = docker.from_env()
+def close_status_backend_containers(request):
-    for container_id in option.status_backend_containers:
+    yield
-        try:
+    if hasattr(request.node.instance, "reuse_container"):
-            container = docker_client.containers.get(container_id)
+        return
-            container.stop(timeout=30)
+    for container in option.status_backend_containers:
-            container.remove()
+        container.kill()  # type: ignore
-        except Exception as e:
-            print(e)

tests-functional/tests/reliability/test_contact_request.py

@@ -1,7 +1,7 @@
 from time import sleep
 from uuid import uuid4
 import pytest
-from test_cases import MessengerTestCase
+from tests.test_cases import MessengerTestCase
 from clients.signals import SignalType
 from resources.enums import MessageContentType
 
@@ -9,9 +9,7 @@ from resources.enums import MessageContentType
 @pytest.mark.reliability
 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
+    def test_contact_request_baseline(self, execution_number=1, network_condition=None):
-    @pytest.mark.dependency(name="test_contact_request_baseline")
-    def test_contact_request_baseline(self, execution_number=1):
         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)
@@ -21,6 +19,9 @@ class TestContactRequests(MessengerTestCase):
         if sender.public_key in str(existing_contacts):
             pytest.skip("Contact request was already sent for this sender<->receiver. Skipping test!!")
 
+        if network_condition:
+            network_condition(receiver)
+
         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]
 
@@ -44,43 +45,20 @@ class TestContactRequests(MessengerTestCase):
             expected_message=expected_message,
         )
 
-    @pytest.mark.skip(
-        reason=(
-            "Skipping because of error 'Not enough status-backend containers, "
-            "please add more'. Unkipping when we merge "
-            "https://github.com/status-im/status-go/pull/6159"
-        )
-    )
     @pytest.mark.parametrize("execution_number", range(10))
-    @pytest.mark.dependency(depends=["test_contact_request_baseline"])
     def test_multiple_contact_requests(self, execution_number):
         self.test_contact_request_baseline(execution_number=execution_number)
 
-    @pytest.mark.dependency(depends=["test_contact_request_baseline"])
+    @pytest.mark.parametrize("execution_number", range(10))
-    @pytest.mark.skip(reason="Skipping until add_latency is implemented")
+    def test_contact_request_with_latency(self, execution_number):
-    def test_contact_request_with_latency(self):
+        self.test_contact_request_baseline(execution_number=execution_number, network_condition=self.add_latency)
-        # with self.add_latency():
-        #     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():
+        self.test_contact_request_baseline(execution_number=10, network_condition=self.add_packet_loss)
-        #     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():
+        self.test_contact_request_baseline(execution_number=10, network_condition=self.add_low_bandwith)
-        #     self.test_contact_request_baseline()
-        # to be done in the next PR
-        pass
 
-    @pytest.mark.dependency(depends=["test_contact_request_baseline"])
     def test_contact_request_with_node_pause_30_seconds(self):
         sender = self.initialize_backend(await_signals=self.await_signals)
         receiver = self.initialize_backend(await_signals=self.await_signals)

tests-functional/tests/reliability/test_create_private_groups.py

@@ -1,7 +1,7 @@
 from time import sleep
 from uuid import uuid4
 import pytest
-from test_cases import MessengerTestCase
+from tests.test_cases import MessengerTestCase
 from clients.signals import SignalType
 from resources.enums import MessageContentType
 
@@ -10,8 +10,6 @@ from resources.enums import MessageContentType
 @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()
 
@@ -38,35 +36,9 @@ class TestCreatePrivateGroups(MessengerTestCase):
                 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()
 

tests-functional/tests/reliability/test_one_to_one_messages.py

@@ -1,7 +1,7 @@
 from time import sleep
 from uuid import uuid4
 import pytest
-from test_cases import MessengerTestCase
+from tests.test_cases import MessengerTestCase
 from clients.signals import SignalType
 from resources.enums import MessageContentType
 
@@ -10,8 +10,6 @@ from resources.enums import MessageContentType
 @pytest.mark.reliability
 class TestOneToOneMessages(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_one_to_one_message_baseline")
     def test_one_to_one_message_baseline(self, message_count=1):
         sent_messages = []
         for i in range(message_count):
@@ -33,35 +31,21 @@ class TestOneToOneMessages(MessengerTestCase):
                 expected_message=expected_message,
             )
 
-    @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():
+        with self.add_latency(self.receiver):
-        #     self.test_one_to_one_message_baseline()
+            self.test_one_to_one_message_baseline(message_count=50)
-        # 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.receiver):
-        #     self.test_one_to_one_message_baseline()
+            self.test_one_to_one_message_baseline(message_count=50)
-        # 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.receiver):
-        #     self.test_one_to_one_message_baseline()
+            self.test_one_to_one_message_baseline(message_count=50)
-        # to be done in the next PR
-        pass
 
-    @pytest.mark.dependency(depends=["test_one_to_one_message_baseline"])
     def test_one_to_one_message_with_node_pause_30_seconds(self):
         with self.node_pause(self.receiver):
             message_text = f"test_message_{uuid4()}"

tests-functional/tests/reliability/test_private_groups_messages.py

@@ -0,0 +1,61 @@
+from time import sleep
+from uuid import uuid4
+import pytest
+from tests.test_cases import MessengerTestCase
+from clients.signals import SignalType
+from resources.enums import MessageContentType
+
+
+@pytest.mark.usefixtures("setup_two_nodes")
+@pytest.mark.reliability
+class TestPrivateGroupMessages(MessengerTestCase):
+
+    def test_private_group_messages_baseline(self, message_count=1):
+        self.make_contacts()
+        self.private_group_id = self.join_private_group()
+
+        sent_messages = []
+        for i in range(message_count):
+            message_text = f"test_message_{i+1}_{uuid4()}"
+            response = self.sender.wakuext_service.send_group_chat_message(self.private_group_id, message_text)
+            expected_message = self.get_message_by_content_type(response, content_type=MessageContentType.TEXT_PLAIN.value)[0]
+            sent_messages.append(expected_message)
+            sleep(0.01)
+
+        for _, expected_message in enumerate(sent_messages):
+            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,
+                fields_to_validate={"text": "text"},
+                expected_message=expected_message,
+            )
+
+    def test_multiple_group_chat_messages(self):
+        self.test_private_group_messages_baseline(message_count=50)
+
+    def test_multiple_group_chat_messages_with_latency(self):
+        with self.add_latency(self.receiver):
+            self.test_private_group_messages_baseline(message_count=50)
+
+    def test_multiple_group_chat_messages_with_packet_loss(self):
+        with self.add_packet_loss(self.receiver):
+            self.test_private_group_messages_baseline(message_count=50)
+
+    def test_multiple_group_chat_messages_with_low_bandwidth(self):
+        with self.add_low_bandwith(self.receiver):
+            self.test_private_group_messages_baseline(message_count=50)
+
+    def test_private_group_messages_with_node_pause_30_seconds(self):
+        self.make_contacts()
+        self.private_group_id = self.join_private_group()
+
+        with self.node_pause(self.receiver):
+            message_text = f"test_message_{uuid4()}"
+            self.sender.wakuext_service.send_group_chat_message(self.private_group_id, message_text)
+            sleep(30)
+        self.receiver.find_signal_containing_pattern(SignalType.MESSAGES_NEW.value, event_pattern=message_text)
+        self.sender.wait_for_signal(SignalType.MESSAGE_DELIVERED.value)

tests-functional/tests/test_cases.py

@@ -4,6 +4,7 @@ import logging
 import threading
 import time
 from collections import namedtuple
+from uuid import uuid4
 
 import pytest
 
@@ -24,6 +25,7 @@ class StatusDTestCase:
 
 class StatusBackendTestCase:
 
+    reuse_container = True  # Skip close_status_backend_containers cleanup
     await_signals = [SignalType.NODE_LOGIN.value]
 
     network_id = 31337
@@ -36,6 +38,10 @@ class StatusBackendTestCase:
         self.rpc_client.restore_account_and_login()
         self.rpc_client.wait_for_login()
 
+    def teardown_class(self):
+        for container in option.status_backend_containers:
+            container.kill()
+
 
 class WalletTestCase(StatusBackendTestCase):
 
@@ -172,9 +178,9 @@ class NetworkConditionTestCase:
             node.container_exec("tc qdisc del dev eth0 root netem")
 
     @contextmanager
-    def add_low_bandwith(self, node, rate="1mbit", burst="32kbit"):
+    def add_low_bandwith(self, node, rate="1mbit", burst="32kbit", limit="12500"):
         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}")
+        node.container_exec(f"apk add iproute2 && tc qdisc add dev eth0 root tbf rate {rate} burst {burst} limit {limit}")
         try:
             yield
         finally:
@@ -200,13 +206,13 @@ class MessengerTestCase(NetworkConditionTestCase):
         SignalType.NODE_LOGIN.value,
     ]
 
-    @pytest.fixture(scope="class", autouse=False)
+    @pytest.fixture(scope="function", 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 = StatusBackend(await_signals=await_signals, privileged=True)
         backend.init_status_backend()
         backend.create_account_and_login()
         backend.find_public_key()
@@ -264,3 +270,13 @@ class MessengerTestCase(NetworkConditionTestCase):
             return matched_messages
         else:
             raise ValueError(f"Failed to find a message with contentType '{content_type}' in response")
+
+    def join_private_group(self):
+        private_group_name = f"private_group_{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]
+        self.receiver.find_signal_containing_pattern(SignalType.MESSAGES_NEW.value, event_pattern=expected_message.get("id"), timeout=60)
+        return response.get("result", {}).get("chats", [])[0].get("id")