mirror of
https://github.com/logos-messaging/logos-messaging-interop-tests.git
synced 2026-08-25 07:31:17 +00:00
remove some RC04 variants /add RC05
This commit is contained in:
@@ -0,0 +1,91 @@
|
||||
import time
|
||||
|
||||
from src.libs.common import delay, to_base64
|
||||
from src.node.wrappers_manager import WrapperManager
|
||||
from src.node.wrapper_helpers import (
|
||||
EventCollector,
|
||||
create_message_bindings,
|
||||
get_node_multiaddr,
|
||||
wait_for_connected,
|
||||
)
|
||||
|
||||
CHANNEL_ID = "rc05-channel"
|
||||
CONTENT_TOPIC = "/test/1/rc05-channel/proto"
|
||||
SENDER_A = "rc05-sender-a"
|
||||
SENDER_B = "rc05-sender-b"
|
||||
|
||||
CHANNEL_RECEIVED_EVENT = "channel_message_received"
|
||||
|
||||
# Let the gossipsub mesh + content-topic subscription settle before A sends.
|
||||
MESH_SETTLE_S = 10
|
||||
DELIVERY_TIMEOUT_S = 30.0
|
||||
|
||||
|
||||
def wait_for_channel_received(collector, channel_id, timeout_s, poll_interval_s=0.5):
|
||||
"""Poll until a channel_message_received event for `channel_id` arrives, else None."""
|
||||
deadline = time.monotonic() + timeout_s
|
||||
while True:
|
||||
for event in collector.snapshot():
|
||||
if event.get("eventType") == CHANNEL_RECEIVED_EVENT and event.get("channelId") == channel_id:
|
||||
return event
|
||||
if time.monotonic() >= deadline:
|
||||
return None
|
||||
time.sleep(poll_interval_s)
|
||||
|
||||
|
||||
class TestChannelDelivery:
|
||||
def test_rc05_basic_a_to_b_delivery(self, node_config):
|
||||
"""RC05: a message sent by A on a channel is delivered to B on the matching channel + topic.
|
||||
|
||||
Two relay-connected nodes both create the same (channelId, contentTopic);
|
||||
B subscribes to the topic. A channel_send()s a payload; B must fire a
|
||||
channel_message_received event carrying that payload intact. Baseline
|
||||
two-party delivery — the positive control for RC06/RC07 ingress drops.
|
||||
"""
|
||||
payload_b64 = to_base64("rc05 hello from A")
|
||||
|
||||
node_config.update(
|
||||
{
|
||||
"relay": True,
|
||||
"numShardsInNetwork": 1,
|
||||
}
|
||||
)
|
||||
|
||||
receiver_collector = EventCollector()
|
||||
receiver_result = WrapperManager.create_and_start(config=node_config, event_cb=receiver_collector.event_callback)
|
||||
assert receiver_result.is_ok(), f"Failed to start receiver: {receiver_result.err()}"
|
||||
|
||||
with receiver_result.ok_value as receiver:
|
||||
sender_config = {
|
||||
**node_config,
|
||||
"staticnodes": [get_node_multiaddr(receiver)],
|
||||
"portsShift": 1,
|
||||
}
|
||||
sender_collector = EventCollector()
|
||||
sender_result = WrapperManager.create_and_start(config=sender_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:
|
||||
assert wait_for_connected(sender_collector) is not None, "Sender did not reach Connected/PartiallyConnected state"
|
||||
|
||||
subscribe_result = receiver.subscribe_content_topic(CONTENT_TOPIC)
|
||||
assert subscribe_result.is_ok(), f"receiver subscribe_content_topic failed: {subscribe_result.err()}"
|
||||
|
||||
receiver_create = receiver.channel_create(CHANNEL_ID, CONTENT_TOPIC, SENDER_B)
|
||||
assert receiver_create.is_ok(), f"receiver channel_create failed: {receiver_create.err()}"
|
||||
|
||||
sender_create = sender.channel_create(CHANNEL_ID, CONTENT_TOPIC, SENDER_A)
|
||||
assert sender_create.is_ok(), f"sender channel_create failed: {sender_create.err()}"
|
||||
|
||||
delay(MESH_SETTLE_S)
|
||||
|
||||
send_result = sender.channel_send(CHANNEL_ID, create_message_bindings(payload=payload_b64))
|
||||
assert send_result.is_ok(), f"channel_send failed: {send_result.err()}"
|
||||
assert send_result.ok_value, f"channel_send returned an empty handle: {send_result.ok_value!r}"
|
||||
|
||||
received = wait_for_channel_received(receiver_collector, CHANNEL_ID, DELIVERY_TIMEOUT_S)
|
||||
assert received is not None, (
|
||||
f"No {CHANNEL_RECEIVED_EVENT} on B for {CHANNEL_ID} within {DELIVERY_TIMEOUT_S}s. "
|
||||
f"Collected events: {receiver_collector.snapshot()}"
|
||||
)
|
||||
assert received["payload"] == payload_b64, f"delivered payload mismatch: got {received['payload']!r}, sent {payload_b64!r}"
|
||||
@@ -1,5 +1,3 @@
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from src.node.wrappers_manager import WrapperManager
|
||||
from src.node.wrapper_helpers import EventCollector, create_message_bindings
|
||||
@@ -14,8 +12,7 @@ UNKNOWN_CHANNEL_ID = "rc02-unknown-channel"
|
||||
RC03_CHANNEL_ID = "rc03-channel"
|
||||
|
||||
RC04_CHANNEL_ID = "rc04-channel"
|
||||
RC04_EPHEMERAL_CHANNEL_ID = "rc04-ephemeral-channel"
|
||||
RC04_DISTINCT_CHANNEL_ID = "rc04-distinct-channel"
|
||||
RC04_CLOSED_CHANNEL_ID = "rc04-closed-channel"
|
||||
|
||||
# Give the reliable channel manager a moment to settle a freshly-created
|
||||
# channel before we act on it.
|
||||
@@ -106,41 +103,6 @@ class TestChannelLifecycle:
|
||||
finally:
|
||||
node.stop_and_destroy()
|
||||
|
||||
def test_rc03_close_random_id_leaves_open_channel_intact(self, node_config):
|
||||
"""RC03 variant: closing a random unknown id while a real channel is open.
|
||||
|
||||
A genuine channel is created and left open; closing a random id Errs with
|
||||
"unknown channel: <id>" and must not disturb the open channel, which then
|
||||
still closes cleanly. No events are expected.
|
||||
"""
|
||||
node_config.update({"mode": "Core"})
|
||||
|
||||
channel_id = f"rc03-open-{uuid.uuid4()}"
|
||||
random_id = f"rc03-random-{uuid.uuid4()}"
|
||||
|
||||
collector = EventCollector()
|
||||
create_node_result = WrapperManager.create_and_start(config=node_config, event_cb=collector.event_callback)
|
||||
assert create_node_result.is_ok(), f"Failed to create and start node: {create_node_result.err()}"
|
||||
node = create_node_result.ok_value
|
||||
|
||||
try:
|
||||
create_result = node.channel_create(channel_id, CONTENT_TOPIC, SENDER_ID)
|
||||
assert create_result.is_ok(), f"channel_create failed: {create_result.err()}"
|
||||
|
||||
delay(CHANNEL_SETTLE_S)
|
||||
|
||||
random_close_result = node.channel_close(random_id)
|
||||
assert random_close_result.is_err(), f"channel_close on a random id must fail, got Ok({random_close_result.ok_value!r})"
|
||||
assert f"unknown channel: {random_id}" in random_close_result.err(), f"unexpected error message: {random_close_result.err()!r}"
|
||||
|
||||
# The real channel was left untouched and still closes cleanly.
|
||||
close_result = node.channel_close(channel_id)
|
||||
assert close_result.is_ok(), f"open channel_close failed after a bad-id close: {close_result.err()}"
|
||||
|
||||
assert collector.events == [], f"expected no events, got: {collector.events}"
|
||||
finally:
|
||||
node.stop_and_destroy()
|
||||
|
||||
def test_rc04_empty_payload_rejected_and_send_returns_handle(self, node_config):
|
||||
"""RC04: empty payload is rejected; a non-empty send returns a handle synchronously.
|
||||
|
||||
@@ -170,58 +132,33 @@ class TestChannelLifecycle:
|
||||
finally:
|
||||
node.stop_and_destroy()
|
||||
|
||||
def test_rc04_ephemeral_send_returns_handle(self, node_config):
|
||||
"""RC04 variant: an ephemeral channel_send still returns a handle synchronously.
|
||||
def test_rc04_send_after_close_rejected(self, node_config):
|
||||
"""RC04 variant: sending after the channel is closed is rejected.
|
||||
|
||||
The ephemeral flag only affects downstream persistence / terminal events,
|
||||
not the send contract, so channel_send returns Ok(channelReqId) immediately.
|
||||
Exercises the teardown -> send ordering (distinct from RC02's
|
||||
never-created id): once channel_close removes the id, channel_send Errs
|
||||
with "unknown channel: <id>". No events are expected.
|
||||
"""
|
||||
node_config.update({"mode": "Core"})
|
||||
|
||||
create_node_result = WrapperManager.create_and_start(config=node_config)
|
||||
collector = EventCollector()
|
||||
create_node_result = WrapperManager.create_and_start(config=node_config, event_cb=collector.event_callback)
|
||||
assert create_node_result.is_ok(), f"Failed to create and start node: {create_node_result.err()}"
|
||||
node = create_node_result.ok_value
|
||||
|
||||
try:
|
||||
create_result = node.channel_create(RC04_EPHEMERAL_CHANNEL_ID, CONTENT_TOPIC, SENDER_ID)
|
||||
create_result = node.channel_create(RC04_CLOSED_CHANNEL_ID, CONTENT_TOPIC, SENDER_ID)
|
||||
assert create_result.is_ok(), f"channel_create failed: {create_result.err()}"
|
||||
|
||||
delay(CHANNEL_SETTLE_S)
|
||||
|
||||
send_result = node.channel_send(RC04_EPHEMERAL_CHANNEL_ID, create_message_bindings(ephemeral=True))
|
||||
assert send_result.is_ok(), f"ephemeral channel_send failed: {send_result.err()}"
|
||||
assert send_result.ok_value, f"channel_send must return a non-empty channelReqId handle, got: {send_result.ok_value!r}"
|
||||
finally:
|
||||
node.stop_and_destroy()
|
||||
close_result = node.channel_close(RC04_CLOSED_CHANNEL_ID)
|
||||
assert close_result.is_ok(), f"channel_close failed: {close_result.err()}"
|
||||
|
||||
def test_rc04_sequential_sends_return_distinct_handles(self, node_config):
|
||||
"""RC04 variant: back-to-back sends each return a distinct handle.
|
||||
send_result = node.channel_send(RC04_CLOSED_CHANNEL_ID, create_message_bindings())
|
||||
assert send_result.is_err(), f"channel_send after close must fail, got Ok({send_result.ok_value!r})"
|
||||
assert f"unknown channel: {RC04_CLOSED_CHANNEL_ID}" in send_result.err(), f"unexpected error message: {send_result.err()!r}"
|
||||
|
||||
Every channel_send is its own request, so two sends on the same channel
|
||||
return two different (non-empty) channelReqId handles.
|
||||
"""
|
||||
node_config.update({"mode": "Core"})
|
||||
|
||||
create_node_result = WrapperManager.create_and_start(config=node_config)
|
||||
assert create_node_result.is_ok(), f"Failed to create and start node: {create_node_result.err()}"
|
||||
node = create_node_result.ok_value
|
||||
|
||||
try:
|
||||
create_result = node.channel_create(RC04_DISTINCT_CHANNEL_ID, CONTENT_TOPIC, SENDER_ID)
|
||||
assert create_result.is_ok(), f"channel_create failed: {create_result.err()}"
|
||||
|
||||
delay(CHANNEL_SETTLE_S)
|
||||
|
||||
first_result = node.channel_send(RC04_DISTINCT_CHANNEL_ID, create_message_bindings())
|
||||
assert first_result.is_ok(), f"first channel_send failed: {first_result.err()}"
|
||||
assert first_result.ok_value, f"first channel_send returned an empty handle: {first_result.ok_value!r}"
|
||||
|
||||
second_result = node.channel_send(RC04_DISTINCT_CHANNEL_ID, create_message_bindings())
|
||||
assert second_result.is_ok(), f"second channel_send failed: {second_result.err()}"
|
||||
assert second_result.ok_value, f"second channel_send returned an empty handle: {second_result.ok_value!r}"
|
||||
|
||||
assert (
|
||||
first_result.ok_value != second_result.ok_value
|
||||
), f"channel_send handles must be distinct, got the same id twice: {first_result.ok_value!r}"
|
||||
assert collector.events == [], f"expected no events, got: {collector.events}"
|
||||
finally:
|
||||
node.stop_and_destroy()
|
||||
|
||||
Reference in New Issue
Block a user