mirror of
https://github.com/logos-messaging/logos-messaging-interop-tests.git
synced 2026-05-27 13:02:15 +00:00
Revert "Revert "S18/S25 added""
This reverts commit 27fd9df46a97f141d3a662b6b128aa0fa593ced7.
This commit is contained in:
parent
27fd9df46a
commit
4307db2cc6
@ -737,3 +737,275 @@ class TestS26LightpushPeerChurn(StepsCommon):
|
|||||||
assert error_event is None, f"Unexpected message_error event during peer churn: {error_event}"
|
assert error_event is None, f"Unexpected message_error event during peer churn: {error_event}"
|
||||||
|
|
||||||
assert_event_invariants(sender_collector, request_id)
|
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)
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user