diff --git a/tests/wrappers_tests/test_send_e2e.py b/tests/wrappers_tests/test_send_e2e.py index 18a8da5aa..d250c333b 100644 --- a/tests/wrappers_tests/test_send_e2e.py +++ b/tests/wrappers_tests/test_send_e2e.py @@ -9,6 +9,7 @@ from src.node.waku_node import WakuNode from src.node.wrappers_manager import WrapperManager from src.node.wrapper_helpers import ( EventCollector, + assert_event_invariants, create_message_bindings, get_node_multiaddr, wait_for_propagated, @@ -16,6 +17,7 @@ from src.node.wrapper_helpers import ( wait_for_error, ) from src.steps.store import StepsStore +from tests.wrappers_tests.conftest import build_node_config logger = get_custom_logger(__name__) @@ -97,6 +99,8 @@ class TestSendBeforeRelay(StepsStore): f"from a store-enabled relay peer. Collected events: {sender_collector.events}" ) + assert_event_invariants(sender_collector, request_id) + 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,. @@ -158,6 +162,8 @@ class TestSendBeforeRelay(StepsStore): f"Collected events: {sender_collector.events}" ) + assert_event_invariants(sender_collector, request_id) + def test_s19_store_peer_appears_after_propagation(self, node_config): """ S19: a store peer comes online later. @@ -169,8 +175,6 @@ class TestSendBeforeRelay(StepsStore): """ sender_collector = EventCollector() - sender_collector = EventCollector() - node_config.update( { "relay": True, @@ -254,6 +258,8 @@ class TestSendBeforeRelay(StepsStore): ascending="true", ) + assert_event_invariants(sender_collector, request_id) + def test_s21_error_when_retry_window_expires(self, node_config): """ S21: delivery retry window expires before any valid path recovers. @@ -305,6 +311,170 @@ class TestSendBeforeRelay(StepsStore): f"Full event: {error_event}" ) + assert_event_invariants(sender_collector, request_id) + + +class TestS07CoreSenderRelayAndStore(StepsCommon): + """ + S07 — Core sender with relay peers and store peer, reliability enabled. + Sender relays message to a store-capable peer; delivery service validates + the message reached the store via p2p reliability check. + Expected: Propagated, then Sent. + """ + + def test_s07_relay_propagation_with_store_validation(self, node_config): + sender_collector = EventCollector() + + node_config.update( + { + "relay": True, + "store": False, + "lightpush": False, + "filter": False, + "discv5Discovery": False, + "numShardsInNetwork": 1, + "reliabilityEnabled": 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: + peer_config = { + **node_config, + "staticnodes": [get_node_multiaddr(sender)], + "portsshift": 1, + "store": True, + } + + peer_result = WrapperManager.create_and_start(config=peer_config) + assert peer_result.is_ok(), f"Failed to start store peer: {peer_result.err()}" + + with peer_result.ok_value: + message = create_message_bindings( + payload=to_base64("S07 relay+store test payload"), + contentTopic="/test/1/s07-relay-store/proto", + ) + + send_result = sender.send_message(message=message) + assert send_result.is_ok(), f"send() failed: {send_result.err()}" + + request_id = send_result.ok_value + assert request_id, "send() returned an empty RequestId" + + propagated = wait_for_propagated( + collector=sender_collector, + request_id=request_id, + timeout_s=PROPAGATED_TIMEOUT_S, + ) + assert propagated is not None, ( + f"No message_propagated event within {PROPAGATED_TIMEOUT_S}s. " f"Collected events: {sender_collector.events}" + ) + assert propagated["requestId"] == request_id + + sent = wait_for_sent( + collector=sender_collector, + request_id=request_id, + timeout_s=SENT_TIMEOUT_S, + ) + assert sent is not None, ( + f"No message_sent event within {SENT_TIMEOUT_S}s after propagation. " f"Collected events: {sender_collector.events}" + ) + assert sent["requestId"] == request_id + + error = wait_for_error(sender_collector, request_id, timeout_s=0) + assert error is None, f"Unexpected message_error event: {error}" + + assert_event_invariants(sender_collector, request_id) + + +class TestS10EdgeSenderLightpushOnly(StepsCommon): + """ + S10 — Edge sender with lightpush path only, no store peer. + Edge sender has no local relay; it publishes via a lightpush service node. + Expected: Propagated only (no Sent, no Error). + """ + + def test_s10_edge_lightpush_propagation(self, node_config): + sender_collector = EventCollector() + + common = { + "store": False, + "filter": False, + "discv5Discovery": False, + "numShardsInNetwork": 1, + } + + # 1. Service node: relay + lightpush server (gateway for the edge sender). + service_config = build_node_config(relay=True, lightpush=True, **common) + + service_result = WrapperManager.create_and_start(config=service_config) + assert service_result.is_ok(), f"Failed to start service node: {service_result.err()}" + + with service_result.ok_value as service_node: + service_multiaddr = get_node_multiaddr(service_node) + + # 2. Relay peer: forms gossipsub mesh with the service node so it + # has someone to relay-publish to after receiving lightpush. + relay_config = build_node_config( + relay=True, + staticnodes=[service_multiaddr], + **common, + ) + + 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: + # 3. Edge sender: lightpush client only, no relay. + edge_config = build_node_config( + mode="Edge", + relay=False, + lightpushnode=service_multiaddr, + staticnodes=[service_multiaddr], + **common, + ) + + edge_result = WrapperManager.create_and_start( + config=edge_config, + event_cb=sender_collector.event_callback, + ) + assert edge_result.is_ok(), f"Failed to start edge sender: {edge_result.err()}" + + with edge_result.ok_value as edge_sender: + message = create_message_bindings( + payload=to_base64("S10 edge lightpush test payload"), + contentTopic="/test/1/s10-edge-lightpush/proto", + ) + + send_result = edge_sender.send_message(message=message) + assert send_result.is_ok(), f"send() failed: {send_result.err()}" + + request_id = send_result.ok_value + assert request_id, "send() returned an empty RequestId" + + propagated = wait_for_propagated( + collector=sender_collector, + request_id=request_id, + timeout_s=PROPAGATED_TIMEOUT_S, + ) + assert propagated is not None, ( + f"No message_propagated event within {PROPAGATED_TIMEOUT_S}s. " f"Collected events: {sender_collector.events}" + ) + assert propagated["requestId"] == request_id + + sent = wait_for_sent(sender_collector, request_id, timeout_s=NO_SENT_OBSERVATION_S) + assert sent is None, f"Unexpected message_sent event (no store peer): {sent}" + + error = wait_for_error(sender_collector, request_id, timeout_s=0) + assert error is None, f"Unexpected message_error event: {error}" + + assert_event_invariants(sender_collector, request_id) + class TestS06CoreSenderRelayOnly(StepsCommon): """ @@ -371,3 +541,5 @@ class TestS06CoreSenderRelayOnly(StepsCommon): sent = wait_for_sent(sender_collector, request_id, timeout_s=0) assert sent is None, f"Unexpected message_sent event (store is disabled): {sent}" + + assert_event_invariants(sender_collector, request_id)