from time import time_ns import pytest from src.steps.common import StepsCommon from src.libs.common import to_base64 from src.node.wrappers_manager import WrapperManager from src.node.wrapper_helpers import ( EventCollector, get_node_multiaddr, wait_for_propagated, wait_for_sent, ) PROPAGATED_TIMEOUT_S = 30.0 SENT_TIMEOUT_S = 10.0 DEFAULT_CONTENT_TOPIC = "/test/1/default/proto" DEFAULT_PAYLOAD = "Default Payload" @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(), } def test_s17_send_before_relay_peers_joins(self, node_config): """ S17: sender starts isolated, calls send() - send() returns Ok(RequestId) immediately - Propagated event eventually arrives """ sender_collector = EventCollector() node_config.update( { "relay": True, "store": False, "discv5Discovery": False, "numShardsInNetwork": 1, } ) 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: message = self.create_message(payload=to_base64("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()}" request_id = send_result.ok_value assert request_id, "send() returned an empty RequestId" # Step 2: start a relay peer with store enabled. relay_config = { **node_config, "staticnodes": [get_node_multiaddr(sender_node)], "portsshift": 1, "store": True, } 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: 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"after relay peer joined. Collected events: {sender_collector.events}" ) sent_event = wait_for_sent( collector=sender_collector, request_id=request_id, timeout_s=SENT_TIMEOUT_S, ) assert sent_event is not None, ( f"No MessageSentEvent received within {SENT_TIMEOUT_S}s " f"from a store-enabled relay peer. Collected events: {sender_collector.events}" ) def test_s17_no_sent_event_when_relay_has_no_store(self, node_config): """ S17 negative: relay peerstore=false, there shouldn't be a Sent event,. """ sender_collector = EventCollector() node_config.update( { "relay": True, "store": False, "discv5Discovery": False, "numShardsInNetwork": 1, } ) 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: send_result = sender_node.send_message(message=self.create_message(payload=to_base64("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 assert request_id, "send() returned an empty RequestId" relay_config = { **node_config, "staticnodes": [get_node_multiaddr(sender_node)], "portsshift": 1, "store": 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: 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"after relay peer joined. Collected events: {sender_collector.events}" ) sent_event = wait_for_sent( collector=sender_collector, request_id=request_id, timeout_s=SENT_TIMEOUT_S, ) assert sent_event is None, ( f"Unexpected MessageSentEvent received when relay peer has store=false.\n" f"Sent event: {sent_event}\n" f"Collected events: {sender_collector.events}" )