diff --git a/tests/wrappers_tests/test_send_e2e.py b/tests/wrappers_tests/test_send_e2e.py index 9bd3062d0..fb794ab6e 100644 --- a/tests/wrappers_tests/test_send_e2e.py +++ b/tests/wrappers_tests/test_send_e2e.py @@ -1,9 +1,11 @@ from time import time_ns import pytest +from src.env_vars import NODE_2 from src.steps.common import StepsCommon - -from src.libs.common import to_base64 +from src.libs.common import delay, to_base64 +from src.libs.custom_logger import get_custom_logger +from src.node.waku_node import WakuNode from src.node.wrappers_manager import WrapperManager from src.node.wrapper_helpers import ( EventCollector, @@ -11,26 +13,21 @@ from src.node.wrapper_helpers import ( wait_for_propagated, wait_for_sent, ) +from src.steps.store import StepsStore + +logger = get_custom_logger(__name__) PROPAGATED_TIMEOUT_S = 30.0 SENT_TIMEOUT_S = 10.0 - +NO_SENT_OBSERVATION_S = 5.0 DEFAULT_CONTENT_TOPIC = "/test/1/default/proto" DEFAULT_PAYLOAD = "Default Payload" +SENT_AFTER_STORE_TIMEOUT_S = 60.0 @pytest.mark.smoke -class TestSendBeforeRelay(StepsCommon): - def create_message(self, payload=None, content_topic=DEFAULT_CONTENT_TOPIC): - if payload is None: - payload = to_base64(DEFAULT_PAYLOAD) - return { - "payload": payload, - "contentTopic": content_topic, - "timestamp": time_ns(), - } - +class TestSendBeforeRelay(StepsStore): def test_s17_send_before_relay_peers_joins(self, node_config): """ S17: sender starts isolated, calls send() @@ -55,7 +52,7 @@ class TestSendBeforeRelay(StepsCommon): assert sender_result.is_ok(), f"Failed to start sender: {sender_result.err()}" with sender_result.ok_value as sender_node: - message = self.create_message(payload=to_base64("Message")) + message = self.create_message() send_result = sender_node.send_message(message=message) assert send_result.is_ok(), f"send() must return Ok(RequestId) even with no peers, got: {send_result.err()}" @@ -115,7 +112,7 @@ class TestSendBeforeRelay(StepsCommon): assert sender_result.is_ok(), f"Failed to start sender: {sender_result.err()}" with sender_result.ok_value as sender_node: - send_result = sender_node.send_message(message=self.create_message(payload=to_base64("Message"))) + send_result = sender_node.send_message(message=self.create_message()) assert send_result.is_ok(), f"send() must return Ok(RequestId) even with no peers, got: {send_result.err()}" request_id = send_result.ok_value @@ -152,3 +149,102 @@ class TestSendBeforeRelay(StepsCommon): f"Sent event: {sent_event}\n" f"Collected events: {sender_collector.events}" ) + + def test_s19_store_peer_appears_after_propagation(self, node_config): + """ + S19: a store peer comes online later. + question for Zoltan , is reliability = true mandatory for the store peer ? + what is the effect of the reliability here ? + - send() returns Ok(RequestId) immediately + - Propagated --- relay peer + - Sent when store peer is reachable + """ + sender_collector = EventCollector() + + sender_collector = EventCollector() + + node_config.update( + { + "relay": True, + "store": False, + "discv5Discovery": False, + "numShardsInNetwork": 1, + "reliability": True, + } + ) + + sender_result = WrapperManager.create_and_start( + config=node_config, + event_cb=sender_collector.event_callback, + ) + assert sender_result.is_ok(), f"Failed to start sender: {sender_result.err()}" + + with sender_result.ok_value as sender_node: + # Relay-only wrapper peer so propagation can complete without any + # store peer being reachable yet. + relay_config = { + **node_config, + "staticnodes": [get_node_multiaddr(sender_node)], + "portsshift": 1, + "store": False, + "reliability": False, + } + + relay_result = WrapperManager.create_and_start(config=relay_config) + assert relay_result.is_ok(), f"Failed to start relay peer: {relay_result.err()}" + + with relay_result.ok_value as relay_peer: + # Step 1: send(). Must return Ok(RequestId) immediately. + message = self.create_message() + send_result = sender_node.send_message(message=message) + assert send_result.is_ok(), f"send() must return Ok(RequestId), got: {send_result.err()}" + + request_id = send_result.ok_value + assert request_id, "send() returned an empty RequestId" + + # Step 2: Propagated should arrive via the relay peer. + propagated_event = wait_for_propagated( + collector=sender_collector, + request_id=request_id, + timeout_s=PROPAGATED_TIMEOUT_S, + ) + assert propagated_event is not None, ( + f"No MessagePropagatedEvent received within {PROPAGATED_TIMEOUT_S}s. " f"Collected events: {sender_collector.events}" + ) + + # Step 3: with no store peer reachable, Sent must not arrive yet. + early_sent_event = wait_for_sent( + collector=sender_collector, + request_id=request_id, + timeout_s=NO_SENT_OBSERVATION_S, + ) + assert early_sent_event is None, f"MessageSentEvent arrived before any store peer was reachable. " f"Event: {early_sent_event}" + + # Step 4: bring a docker store node online, wired into the + # existing mesh via REST. Teardown is handled by the autouse + # close_open_nodes fixture in tests/conftest.py. + store_node = WakuNode(NODE_2, f"store_node") + store_node.start(relay="true", store="true", discv5_discovery="false") + store_node.set_relay_subscriptions([self.test_pubsub_topic]) + relay_multiaddr = get_node_multiaddr(relay_peer) + sender_multiaddr = get_node_multiaddr(sender_node) + store_node.add_peers([relay_multiaddr, sender_multiaddr]) + delay(3) + + sent_event = wait_for_sent( + collector=sender_collector, + request_id=request_id, + timeout_s=SENT_AFTER_STORE_TIMEOUT_S, + ) + assert sent_event is not None, ( + f"No MessageSentEvent received within {SENT_AFTER_STORE_TIMEOUT_S}s " + f"after store peer joined. Collected events: {sender_collector.events}" + ) + + self.check_published_message_is_stored( + store_node=store_node, + pubsub_topic=self.test_pubsub_topic, + messages_to_check=[message], + page_size=5, + ascending="true", + )