diff --git a/tests/wrappers_tests/test_send_e2e.py b/tests/wrappers_tests/test_send_e2e.py index c65c77c20..9bd3062d0 100644 --- a/tests/wrappers_tests/test_send_e2e.py +++ b/tests/wrappers_tests/test_send_e2e.py @@ -1,90 +1,154 @@ +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_error, + wait_for_sent, ) -CONTENT_TOPIC = "/test/1/s17-relay-late-join/proto" + PROPAGATED_TIMEOUT_S = 30.0 +SENT_TIMEOUT_S = 10.0 + +DEFAULT_CONTENT_TOPIC = "/test/1/default/proto" +DEFAULT_PAYLOAD = "Default Payload" -class TestS17RelayPeersAppearLater(StepsCommon): - """ - S17 – No delivery peers at T0, relay peers appear later. - """ +@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_send_before_relay_peers_exist_then_peer_joins(self, node_config): + def test_s17_send_before_relay_peers_joins(self, node_config): """ - S17: send() is called while the sender has no peers. - A relay peer is then brought online with the sender's address as a - static peer, causing the sender's retry loop to deliver the message. + 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}) + 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) + 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()}" - sender_node = sender_result.ok_value - # try: - # Step 1: send while isolated — must return Ok(RequestId) immediately - send_result = sender_node.send_message(message=self.create_message(payload=to_base64(f"Message"))) - assert send_result.is_ok(), f"send() must return Ok(RequestId) even with no peers, got: {send_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: get sender's multiaddr so the relay peer can dial back to it - sender_multiaddr = get_node_multiaddr(sender_node) + request_id = send_result.ok_value + assert request_id, "send() returned an empty RequestId" - # Step 3: start the relay peer with the sender listed as a static peer - relay_config = node_config.copy() - relay_config["staticPeers"] = [sender_multiaddr] + # 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()}" - relay_peer = relay_result.ok_value - try: - # Step 4: wait for a Propagated event — the sender's retry loop should - # deliver the message now that a relay peer is reachable + with relay_result.ok_value: propagated_event = wait_for_propagated( collector=sender_collector, request_id=request_id, timeout_s=PROPAGATED_TIMEOUT_S, ) - - # Step 5: check no Error event arrived - error_event = wait_for_error( + 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=0, + 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}" ) - for event in sender_collector.get_events_for_request(request_id): - assert event.get("requestId") == request_id, f"Event carries wrong requestId: {event}" + 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}" ) - assert error_event is None, ( - f"MessageErrorEvent arrived before Propagated — violates S17 expectations.\n" - f"Error : {error_event}\n" - f"Propagated: {propagated_event}" + 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}" ) - - finally: - relay_peer.stop_and_destroy() - - finally: - sender_node.stop_and_destroy() - - """