From abe86825e832c0a6bd0c5b9efea82677dc4331ad Mon Sep 17 00:00:00 2001 From: Aya Hassan Date: Tue, 19 May 2026 16:41:35 +0200 Subject: [PATCH] S18/S25 added --- .../test_send_lightpush_and_edge.py | 272 ++++++++++++++++++ 1 file changed, 272 insertions(+) diff --git a/tests/wrappers_tests/test_send_lightpush_and_edge.py b/tests/wrappers_tests/test_send_lightpush_and_edge.py index 529f5035a..d22568df5 100644 --- a/tests/wrappers_tests/test_send_lightpush_and_edge.py +++ b/tests/wrappers_tests/test_send_lightpush_and_edge.py @@ -737,3 +737,275 @@ class TestS26LightpushPeerChurn(StepsCommon): assert error_event is None, f"Unexpected message_error event during peer churn: {error_event}" assert_event_invariants(sender_collector, request_id) + + +class TestS18StagedTopologyReformation(StepsCommon): + """ + S18: relay absent at T0, lightpush absent at T0, both appear in stages. + + The sender has relay and lightpush enabled but starts fully isolated: + no relay peers and no lightpush peers. send() is called before any + peer exists. Peers then appear one stage at a time. The message must + succeed as soon as any valid path becomes available. + + Two orderings are covered, one per test method: + - lightpush appears first, then relay. + - relay appears first, then lightpush. + + Both prove the send service recovers from arbitrary staged topology + reformation. + + Common topology per stage: + [Sender] relay=True, lightpush=True, isolated at T0. + [LightpushPeer] relay=True, lightpush=True, staticnodes=[sender]. + [RelayPeer] relay=True, staticnodes=[sender]. + """ + + # Common config shared by the sender and both staged peers. + _COMMON = { + "relay": True, + "lightpush": True, + "store": False, + "filter": False, + "discv5Discovery": False, + "numShardsInNetwork": 1, + } + + def test_s18_lightpush_first_then_relay(self, node_config): + """Sender isolated at T0; lightpush peer appears, then relay peer.""" + sender_collector = EventCollector() + + node_config.update(self._COMMON) + + 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: + # send() before any peer exists must still return Ok(RequestId). + message = create_message_bindings( + payload=to_base64("S18 lightpush-first staged recovery"), + contentTopic="/test/1/s18-lightpush-first/proto", + ) + send_result = sender.send_message(message=message) + assert send_result.is_ok(), f"send() must return Ok(RequestId) while isolated, got: {send_result.err()}" + + request_id = send_result.ok_value + assert request_id, "send() returned an empty RequestId" + + # No peers yet: the message must not propagate. + delay(SERVICE_DOWN_SETTLE_S) + early_propagated = wait_for_propagated(sender_collector, request_id, timeout_s=0) + assert early_propagated is None, f"message_propagated arrived before any peer joined: {early_propagated}" + + # Stage 1: lightpush peer appears. + lightpush_config = { + **node_config, + "staticnodes": [get_node_multiaddr(sender)], + "portsShift": 1, + } + lightpush_result = WrapperManager.create_and_start(config=lightpush_config) + assert lightpush_result.is_ok(), f"Failed to start lightpush peer: {lightpush_result.err()}" + + with lightpush_result.ok_value: + # Stage 2: relay peer appears. + relay_config = { + **node_config, + "lightpush": False, + "staticnodes": [get_node_multiaddr(sender)], + "portsShift": 2, + } + 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: + # The message must succeed once any valid path is available. + propagated = wait_for_propagated( + collector=sender_collector, + request_id=request_id, + timeout_s=RECOVERY_TIMEOUT_S, + ) + assert propagated is not None, ( + f"No message_propagated within {RECOVERY_TIMEOUT_S}s " + f"after peers appeared in stages. " + f"Collected events: {sender_collector.events}" + ) + assert propagated["requestId"] == request_id + + error = wait_for_error(sender_collector, request_id, timeout_s=0) + assert error is None, f"Unexpected message_error after staged recovery: {error}" + + assert_event_invariants(sender_collector, request_id) + + def test_s18_relay_first_then_lightpush(self, node_config): + """Sender isolated at T0; relay peer appears, then lightpush peer.""" + sender_collector = EventCollector() + + node_config.update(self._COMMON) + + 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: + # send() before any peer exists must still return Ok(RequestId). + message = create_message_bindings( + payload=to_base64("S18 relay-first staged recovery"), + contentTopic="/test/1/s18-relay-first/proto", + ) + send_result = sender.send_message(message=message) + assert send_result.is_ok(), f"send() must return Ok(RequestId) while isolated, got: {send_result.err()}" + + request_id = send_result.ok_value + assert request_id, "send() returned an empty RequestId" + + # No peers yet: the message must not propagate. + delay(SERVICE_DOWN_SETTLE_S) + early_propagated = wait_for_propagated(sender_collector, request_id, timeout_s=0) + assert early_propagated is None, f"message_propagated arrived before any peer joined: {early_propagated}" + + # Stage 1: relay peer appears. + relay_config = { + **node_config, + "lightpush": False, + "staticnodes": [get_node_multiaddr(sender)], + "portsShift": 1, + } + 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: + # Stage 2: lightpush peer appears. + lightpush_config = { + **node_config, + "staticnodes": [get_node_multiaddr(sender)], + "portsShift": 2, + } + lightpush_result = WrapperManager.create_and_start(config=lightpush_config) + assert lightpush_result.is_ok(), f"Failed to start lightpush peer: {lightpush_result.err()}" + + with lightpush_result.ok_value: + # The message must succeed once any valid path is available. + propagated = wait_for_propagated( + collector=sender_collector, + request_id=request_id, + timeout_s=RECOVERY_TIMEOUT_S, + ) + assert propagated is not None, ( + f"No message_propagated within {RECOVERY_TIMEOUT_S}s " + f"after peers appeared in stages. " + f"Collected events: {sender_collector.events}" + ) + assert propagated["requestId"] == request_id + + error = wait_for_error(sender_collector, request_id, timeout_s=0) + assert error is None, f"Unexpected message_error after staged recovery: {error}" + + assert_event_invariants(sender_collector, request_id) + + +class TestS25EphemeralLightpushWithStore(StepsCommon): + """ + S25 — Ephemeral message over lightpush with a reachable store peer. + + The sender is an Edge node, so it has no local relay and publishes only + via lightpush. A docker store peer is reachable and joined to the + lightpush peer's relay mesh, so the message does reach a store. + + Because ephemeral messages are never store-validated, the expected result + is Propagated only — no Sent — even though the store peer is reachable. + This is the lightpush-transport counterpart of S24 and proves the + ephemeral rule is transport-independent. + + Topology mirrors S11: + [LightpushPeer] wrapper, relay=True, lightpush=True, store=False + [StorePeer] docker WakuNode, relay=true, store=true, joined to + the lightpush peer's shard-0 relay mesh + [Edge] wrapper, mode="Edge", reliabilityEnabled=True, + staticnodes=[lightpush_peer, store_peer] + """ + + def test_s25_ephemeral_lightpush_with_store(self): + sender_collector = EventCollector() + + common = { + "numShardsInNetwork": 1, + } + + lightpush_config = build_node_config( + relay=True, + lightpush=True, + **common, + ) + + lightpush_result = WrapperManager.create_and_start(config=lightpush_config) + assert lightpush_result.is_ok(), f"Failed to start lightpush peer: {lightpush_result.err()}" + + with lightpush_result.ok_value as lightpush_peer: + lightpush_multiaddr = get_node_multiaddr(lightpush_peer) + + # Docker store peer joined to the lightpush peer's shard-0 relay + # mesh, so messages propagated by the lightpush peer are archived. + store_peer = WakuNode(NODE_1, f"s25_store_{self.test_id}") + store_peer.start(relay="true", store="true") + self.add_node_peer(store_peer, [lightpush_multiaddr]) + store_peer.set_relay_subscriptions([STORE_PEER_PUBSUB_TOPIC]) + store_multiaddr = store_peer.get_multiaddr_with_id() + + edge_config = build_node_config( + mode="Edge", + staticnodes=[lightpush_multiaddr, store_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("S25 ephemeral lightpush + store payload"), + contentTopic="/test/1/s25-ephemeral-lightpush/proto", + ephemeral=True, + ) + + 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 + + # Ephemeral messages are never store-validated, so no Sent + # event must arrive even though the store peer is reachable. + sent = wait_for_sent( + collector=sender_collector, + request_id=request_id, + timeout_s=NO_SENT_OBSERVATION_S, + ) + assert sent is None, ( + f"Unexpected message_sent event for an ephemeral message. " + f"Ephemeral messages must never be store-validated.\n" + f"Sent event: {sent}\n" + f"Collected events: {sender_collector.events}" + ) + + 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)