test(functional): add #7132 community control node transfer test

tests-functional/clients/services/wallet.py

@@ -18,6 +18,9 @@ def get_balances_at_by_chain(self, addresses: list, tokens: list):
     def start_wallet(self):
         return self.rpc_request("startWallet")
 
+    def add_ethereum_chain(self, network: dict):
+        return self.rpc_request("addEthereumChain", [network])
+
     def get_derived_addresses_for_mnemonic(self, mnemonic: str, paths: list):
         params = [mnemonic, paths]
         return self.rpc_request("getDerivedAddressesForMnemonic", params)

tests-functional/clients/status_backend.py

@@ -42,6 +42,7 @@
 
 
 class StatusBackend(RpcClient, SignalClient, ApiClient):
+    name: str = ""
     container: StatusBackendContainer | None = None
     _media_server_port_gen = itertools.count(constants.STATUS_MEDIA_SERVER_PORT, 1)
     _connector_ws_port_gen = itertools.count(constants.STATUS_CONNECTOR_WS_PORT, 1)
@@ -202,25 +203,18 @@ def init_status_backend(self):
 
         return self.api_request_json(method, data)
 
-    def _set_networks(self, data, **kwargs):
-        self.network_id = kwargs.get("network_id", ANVIL_NETWORK_ID)
-
-        # Allow callers (fixtures/tests) to add additional networks on top of the default Anvil network.
-        # - networks_override: full replacement for networksOverride (list[dict])
-        # - extra_networks_override: appended to the default Anvil network (list[dict])
-        networks_override = kwargs.get("networks_override", None)
-        extra_networks_override = kwargs.get("extra_networks_override", []) or []
-
+    def _build_anvil_network(self, provider_type="embedded-direct", **kwargs):
+        network_id = kwargs.get("network_id", ANVIL_NETWORK_ID)
         anvil_network = {
-            "chainID": self.network_id,
+            "chainID": network_id,
             "chainName": "Anvil",
             "rpcProviders": [
                 {
-                    "chainId": self.network_id,
+                    "chainId": network_id,
                     "name": "Anvil Direct",
                     "url": Config.anvil_url,
                     "enableRpsLimiter": False,
-                    "type": "embedded-direct",
+                    "type": provider_type,
                     "enabled": True,
                     "authType": "no-auth",
                 }
@@ -235,7 +229,18 @@ def _set_networks(self, data, **kwargs):
             "isActive": True,
             "isDeactivatable": False,
         }
-        anvil_network = self._set_token_overrides(anvil_network, kwargs.get("token_overrides", []))
+        return self._set_token_overrides(anvil_network, kwargs.get("token_overrides", []))
+
+    def _set_networks(self, data, **kwargs):
+        self.network_id = kwargs.get("network_id", ANVIL_NETWORK_ID)
+
+        # Allow callers (fixtures/tests) to add additional networks on top of the default Anvil network.
+        # - networks_override: full replacement for networksOverride (list[dict])
+        # - extra_networks_override: appended to the default Anvil network (list[dict])
+        networks_override = kwargs.get("networks_override", None)
+        extra_networks_override = kwargs.get("extra_networks_override", []) or []
+
+        anvil_network = self._build_anvil_network(**kwargs)
 
         data["testNetworksEnabled"] = False
         data["networkId"] = self.network_id
@@ -244,6 +249,15 @@ def _set_networks(self, data, **kwargs):
         else:
             data["networksOverride"] = [anvil_network, *extra_networks_override]
 
+    def add_anvil_network(self, **kwargs):
+        # LoginAccount rebuilds the network list from defaults (status-im/status-go#6010, #5597), so a
+        # paired device that signs in via login() never gets the Anvil chain and drops token-gated
+        # community messages. wallet_addEthereumChain (Upsert) keeps only USER providers, so add the
+        # chain with a user provider — an embedded one is stripped, leaving the chain with no usable
+        # provider, which fails as "could not find any enabled RPC providers for chain: 31337".
+        network = self._build_anvil_network(provider_type="user", **kwargs)
+        return self.wallet_service.add_ethereum_chain(network)
+
     def _set_proxy_credentials(self, data):
         if "STATUS_BUILD_PROXY_USER" not in os.environ:
             return data

tests-functional/steps/community_control_node.py

@@ -0,0 +1,124 @@
+"""Control-node transfer steps (issue 7132): publish the owner token and promote a paired device to control node."""
+
+import logging
+import time
+from typing import Optional
+
+from clients.api import ApiResponseError
+from clients.services.wakuext import CommunityTokenPermissionType
+from clients.status_backend import StatusBackend
+from steps import community_tokens, messenger
+
+logger = logging.getLogger(__name__)
+
+# Go iota enums (protocol/communities/token).
+TOKEN_OWNER_PRIVILEGES_LEVEL = 0
+TOKEN_DEPLOY_STATE_DEPLOYED = 2
+PROTOBUF_COMMUNITY_TOKEN_ERC721 = 2
+
+# ErrOwnerTokenNeeded: promote is retryable only while the owner-token permission is still syncing.
+_PROMOTE_RETRYABLE_ERROR = "owner token is needed"
+
+
+def register_owner_token_on_backend(
+    backend: StatusBackend,
+    community_id: str,
+    owner_token_address: str,
+    *,
+    name: str = "TestOwnerToken",
+    symbol: str = "TOT",
+):
+    # May already exist → ignore the UNIQUE constraint.
+    try:
+        return backend.wakuext_service.save_community_token(
+            {
+                "tokenType": PROTOBUF_COMMUNITY_TOKEN_ERC721,
+                "communityId": community_id,
+                "address": owner_token_address,
+                "name": name,
+                "symbol": symbol,
+                "supply": "1",
+                "infiniteSupply": False,
+                "transferable": True,
+                "remoteSelfDestruct": False,
+                "chainId": backend.network_id,
+                "deployState": TOKEN_DEPLOY_STATE_DEPLOYED,
+                "image": "",
+                "decimals": 0,
+                "privilegesLevel": TOKEN_OWNER_PRIVILEGES_LEVEL,
+            }
+        )
+    except ApiResponseError as exc:
+        if "UNIQUE constraint failed" not in str(exc):
+            raise
+        return None
+
+
+def publish_owner_token_to_community(owner_backend: StatusBackend, community_id: str, owner_token_address: str):
+    # Adds the BECOME_TOKEN_OWNER permission that control-node transfer requires.
+    register_owner_token_on_backend(owner_backend, community_id, owner_token_address)
+
+    owner_backend.wakuext_service.add_community_token(community_id, owner_backend.network_id, owner_token_address)
+
+    owner_community = next(
+        (c for c in messenger.communities_list(owner_backend.wakuext_service.communities()) if c.get("id") == community_id),
+        None,
+    )
+    permission_types = community_tokens.community_permission_types(owner_community or {})
+    assert (
+        CommunityTokenPermissionType.BECOME_TOKEN_OWNER.value in permission_types
+    ), f"Owner token did not create a BECOME_TOKEN_OWNER permission; observed={permission_types}"
+
+
+def promote_to_control_node(
+    backend: StatusBackend,
+    community_id: str,
+    attempts: int = 30,
+    delay: int = 2,
+) -> dict:
+    """Promote *backend* to control node, polling while the owner-token permission is still syncing.
+
+    Retries only ErrOwnerTokenNeeded; any other error is a real bug and re-raises immediately."""
+    last_exc: Optional[ApiResponseError] = None
+    for attempt in range(attempts):
+        try:
+            response = backend.wakuext_service.promote_self_to_control_node(community_id)
+        except ApiResponseError as exc:
+            if _PROMOTE_RETRYABLE_ERROR not in str(exc):
+                raise
+            last_exc = exc
+            logger.debug(f"promote_self_to_control_node waiting for owner-token sync (attempt {attempt + 1}/{attempts}): {exc}")
+            time.sleep(delay)
+            continue
+        assert response is not None, "promote_self_to_control_node returned no response"
+        backend.wakuext_service.register_received_ownership_notification(community_id)
+        return response
+
+    raise AssertionError(f"promote_self_to_control_node never succeeded after {attempts} attempts; last error={last_exc}")
+
+
+def wait_until_local_control_node_state(
+    backend: StatusBackend,
+    community_id: str,
+    expected: bool = True,
+    attempts: int = 30,
+    delay: int = 2,
+) -> dict:
+    """Poll the backend's local control-node state until isControlNode equals *expected*.
+
+    isControlNode is a local per-device flag, so this reads from the local database only."""
+    community: Optional[dict] = None
+    observed: Optional[bool] = None
+    for attempt in range(attempts):
+        try:
+            community = messenger.fetch_community(backend, community_id, wait_for_response=True, try_database=True)
+        except ApiResponseError as exc:
+            logger.debug(f"fetch_community failed (attempt {attempt + 1}): {exc}")
+            community = None
+        observed = community.get("isControlNode") if isinstance(community, dict) else None
+        logger.info(f"control-node scan {attempt + 1}/{attempts}: isControlNode={observed} (want {expected})")
+        if isinstance(community, dict) and observed == expected:
+            return community
+        time.sleep(delay)
+
+    raise AssertionError(f"Backend never reached isControlNode={expected} after {attempts} attempts; last={observed}")

tests-functional/steps/community_tokens.py

@@ -33,17 +33,24 @@ def check_member_community_updated(
     expected_name: str,
     expected_description: str,
     community: Optional[dict] = None,
+    include_live_fetch: bool = True,
 ) -> bool:
-    """Return True when any fetched or cached community view matches the expected details."""
+    """Return True when any fetched or cached community view matches the expected details.
+
+    With include_live_fetch=False only the local DB is read (non-blocking). Use it where a live fetch
+    (wait_for_response=True) can block indefinitely under store contention.
+    """
     sources: List[Optional[dict]] = [community]
 
-    for try_database in (False, True):
+    # (wait_for_response, try_database) per fetch: live + db-with-fallback normally, else a non-blocking db read.
+    fetch_variants = [(True, False), (True, True)] if include_live_fetch else [(False, True)]
+    for wait_for_response, try_database in fetch_variants:
         try:
             sources.append(
                 messenger.fetch_community(
                     backend,
                     community_id,
-                    wait_for_response=True,
+                    wait_for_response=wait_for_response,
                     try_database=try_database,
                 )
             )
@@ -89,17 +96,23 @@ def wait_until_member_sees_community_update(
     attempts: int = 30,
     delay: int = 2,
     spectate: bool = False,
+    fetch_live: bool = True,
 ):
-    """Poll until backend sees the expected community name and description."""
-    if spectate:
-        try:
-            backend.wakuext_service.spectate_community(community_id)
-        except Exception as exc:
-            logger.debug(f"spectate_community failed for {community_id}: {exc}")
+    """Poll until backend sees the expected community name and description.
 
+    fetch_live=False reads only the local DB (non-blocking) — the update lands there once the backend
+    processes the message, so it avoids a live fetch that can hang under store contention.
+    """
     last_community = None
     for _ in range(attempts):
-        if check_member_community_updated(backend, community_id, expected_name, expected_description):
+        # Re-spectate each iteration to survive a subscription that raced a device reconnect.
+        if spectate:
+            try:
+                backend.wakuext_service.spectate_community(community_id)
+            except Exception as exc:
+                logger.debug(f"spectate_community failed for {community_id}: {exc}")
+
+        if check_member_community_updated(backend, community_id, expected_name, expected_description, include_live_fetch=fetch_live):
             return
         member_communities = backend.wakuext_service.communities()
         last_community = next(
@@ -108,7 +121,7 @@ def wait_until_member_sees_community_update(
         )
         time.sleep(delay)
 
-    assert check_member_community_updated(backend, community_id, expected_name, expected_description), (
+    assert check_member_community_updated(backend, community_id, expected_name, expected_description, include_live_fetch=fetch_live), (
         f"{backend} did not see the updated community name/description; "
         f"expected name={expected_name!r} description={expected_description!r}, "
         f"last observed name={(last_community or {}).get('name')!r} "

tests-functional/tests/conftest.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import re
 from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from uuid import uuid4
@@ -115,6 +116,7 @@ def factory(name="", **kwargs) -> StatusBackend:
 
         # Create backend
         backend = StatusBackend(privileged=privileged, ipv6=ipv6, **kwargs)
+        backend.name = name
         created_backends.append(backend)
         logging.debug(f"✅ [SETUP] {name.capitalize()} backend created")
 
@@ -125,8 +127,12 @@ def factory(name="", **kwargs) -> StatusBackend:
     # Cleanup all created backends concurrently
     logging.debug(f"🧹 [TEARDOWN] Cleaning up {len(created_backends)} backends for {cls_name or 'test'}")
 
+    def _log_sufix(b):
+        backend_name = re.sub(r"[^A-Za-z0-9_.-]", "-", getattr(b, "name", "") or "backend")
+        return f"{test_name}_{backend_name}"
+
     tasks = [
-        (f"backend-{len(created_backends) - i}", lambda b=backend: b.shutdown(log_sufix=test_name))
+        (f"{getattr(backend, 'name', '') or 'backend'}-{len(created_backends) - i}", lambda b=backend: b.shutdown(log_sufix=_log_sufix(b)))
         for i, backend in enumerate(reversed(created_backends))
     ]
     _parallel_teardown(tasks)