logos-messaging/logos-messaging-interop-tests
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-messaging-interop-tests.git synced 2026-01-02 14:03:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Selecting initial smoke tests set

Browse Source
This commit is contained in:
aya 2024-11-27 11:36:41 +02:00
parent 348a14cdea
commit 55fa433b72
5 changed files with 11 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
tests/discv5/test_discv5.py
View File

@ -1,3 +1,5 @@
import pytest
from src.env_vars import NODE_1, NODE_2
from src.libs.custom_logger import get_custom_logger
from src.node.waku_node import WakuNode
@ -11,6 +13,7 @@ logger = get_custom_logger(__name__)
class TestDiscv5(StepsRelay, StepsFilter, StepsStore, StepsLightPush):
@pytest.mark.key(name="smoke_tests")
def running_a_node(self, image, **kwargs):
node = WakuNode(image, f"node{len(self.main_nodes) + 1}_{self.test_id}")
node.start(**kwargs)
@ -25,6 +28,7 @@ class TestDiscv5(StepsRelay, StepsFilter, StepsStore, StepsLightPush):
def wait_for_lightpushed_message_to_be_stored(self):
self.check_light_pushed_message_reaches_receiving_peer(peer_list=[self.receiving_node1, self.receiving_node2])
@pytest.mark.key(name="smoke_tests")
def test_relay(self):
self.node1 = self.running_a_node(NODE_1, relay="true")
self.node2 = self.running_a_node(NODE_2, relay="true", discv5_bootstrap_node=self.node1.get_enr_uri())
@ -32,6 +36,7 @@ class TestDiscv5(StepsRelay, StepsFilter, StepsStore, StepsLightPush):
self.ensure_relay_subscriptions_on_nodes(self.main_nodes, [self.test_pubsub_topic])
self.wait_for_published_message_to_reach_relay_peer()
@pytest.mark.key(name="smoke_tests")
def test_filter(self):
self.node1 = self.running_a_node(NODE_1, relay="true", filter="true")
self.node2 = self.running_a_node(
@ -54,6 +59,7 @@ class TestDiscv5(StepsRelay, StepsFilter, StepsStore, StepsLightPush):
self.subscribe_to_pubsub_topics_via_relay(self.main_nodes)
self.wait_for_published_message_to_be_stored()
@pytest.mark.key(name="smoke_tests")
def test_lightpush(self):
self.receiving_node1 = self.running_a_node(NODE_1, lightpush="true", relay="true")
self.receiving_node2 = self.running_a_node(NODE_1, lightpush="false", relay="true", discv5_bootstrap_node=self.receiving_node1.get_enr_uri())

1
tests/e2e/test_e2e.py
View File

@ -304,6 +304,7 @@ class TestE2E(StepsFilter, StepsStore, StepsRelay, StepsLightPush):
assert len(response_list) == max_subscribed_nodes, "Received message count doesn't match sent "
@pytest.mark.skipif("go-waku" in NODE_2, reason="Test works only with nwaku")
@pytest.mark.key(name="smoke_tests")
def test_store_filter_interaction_with_six_nodes(self):
logger.debug("Create 6 nodes")
self.node4 = WakuNode(NODE_2, f"node4_{self.test_id}")

1
tests/filter/test_get_messages.py
View File

@ -11,6 +11,7 @@ logger = get_custom_logger(__name__)
# here we will also implicitly test filter push, see: https://rfc.vac.dev/spec/12/#messagepush
@pytest.mark.usefixtures("setup_main_relay_node", "setup_main_filter_node", "subscribe_main_nodes")
class TestFilterGetMessages(StepsFilter):
@pytest.mark.key(name="smoke_tests")
def test_filter_get_message_with_valid_payloads(self):
failed_payloads = []
for payload in SAMPLE_INPUTS:

1
tests/relay/test_multiple_nodes.py
View File

@ -4,6 +4,7 @@ from src.steps.relay import StepsRelay
@pytest.mark.usefixtures("setup_main_relay_nodes", "setup_optional_relay_nodes", "subscribe_main_relay_nodes")
class TestRelayMultipleNodes(StepsRelay):
@pytest.mark.key(name="smoke_tests")
def test_first_node_to_start_publishes(self, subscribe_optional_relay_nodes, relay_warm_up):
self.check_published_message_reaches_relay_peer()

2
tests/relay/test_subscribe.py
View File

@ -9,9 +9,11 @@ logger = get_custom_logger(__name__)
@pytest.mark.usefixtures("setup_main_relay_nodes")
class TestRelaySubscribe(StepsRelay):
@pytest.mark.key(name="smoke_tests")
def test_relay_no_subscription(self):
self.check_publish_without_relay_subscription(self.test_pubsub_topic)
@pytest.mark.key(name="smoke_tests")
def test_relay_subscribe_to_single_pubsub_topic(self):
self.ensure_relay_subscriptions_on_nodes(self.main_nodes, [self.test_pubsub_topic])
self.wait_for_published_message_to_reach_relay_peer()