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-09 09:23:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: switch to use cluster ID 1 for RLN

Browse Source
This commit is contained in:
Roman 2024-04-15 15:32:48 +08:00
parent e6d56d1a4f
commit e3981ffd9d
No known key found for this signature in database
GPG Key ID: B8FE070B54E11B75
2 changed files with 18 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/test_data.py
View File

@ -148,3 +148,5 @@ SAMPLE_TIMESTAMPS = [
{"description": "ISO 8601 timestamp", "value": "2023-12-26T10:58:51", "valid_for": []},
{"description": "Missing", "value": None, "valid_for": ["gowaku"]},
]
PUBSUB_TOPICS_RLN = ["/waku/2/rs/1/0"]

28
tests/relay/test_rln.py
View File

@ -2,23 +2,27 @@ import os
import pytest
from src.env_vars import RLN_CREDENTIALS
from src.libs.common import delay, to_base64
from src.libs.custom_logger import get_custom_logger
from src.steps.relay import StepsRelay
from src.test_data import SAMPLE_INPUTS, PUBSUB_TOPICS_RLN
logger = get_custom_logger(__name__)
@pytest.mark.usefixtures()
@pytest.mark.usefixtures("register_main_rln_relay_nodes", "setup_main_rln_relay_nodes", "subscribe_main_relay_nodes")
class TestRelayRLN(StepsRelay):
def test_register_rln(self):
logger.debug("Running register RLN test for main relay nodes")
key_stores_found = 0
test_pubsub_topic = PUBSUB_TOPICS_RLN[0]
if RLN_CREDENTIALS is None:
pytest.skip("RLN_CREDENTIALS not set, skipping test")
for k in range(1, 6):
self.register_rln_single_node(rln_creds_source=RLN_CREDENTIALS, rln_creds_id=f"{k}")
self.check_rln_registration(k)
key_stores_found += 1
assert key_stores_found == 5, f"Invalid number of RLN keystores found, expected 5 found {key_stores_found}"
def test_publish_valid_payloads_at_slow_pace(self):
failed_payloads = []
for payload in SAMPLE_INPUTS:
logger.debug(f'Running test with payload {payload["description"]}')
message = self.create_message(payload=to_base64(payload["value"]))
try:
self.check_published_message_reaches_relay_peer(message)
except Exception as e:
logger.error(f'Payload {payload["description"]} failed: {str(e)}')
failed_payloads.append(payload["description"])
delay(1)
assert not failed_payloads, f"Payloads failed: {failed_payloads}"