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

fixes

Browse Source
This commit is contained in:
Florin Barbu 2024-05-21 14:07:33 +03:00
parent 6398249ad2
commit 68cd209e1b
No known key found for this signature in database
GPG Key ID: 593D6DBC6D9E5095
2 changed files with 14 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/test_data.py
View File

@ -149,19 +149,4 @@ SAMPLE_TIMESTAMPS = [
{"description": "Missing", "value": None, "valid_for": ["gowaku"]}, {"description": "Missing", "value": None, "valid_for": ["gowaku"]},
] ]
STORE_TIMESTAMPS_PASS = [
{"description": "3 sec Past", "value": int((datetime.now() - timedelta(seconds=3)).timestamp() * 1e9)},
{"description": "1 sec Past", "value": int((datetime.now() - timedelta(seconds=1)).timestamp() * 1e9)},
{"description": "0.1 sec Past", "value": int((datetime.now() - timedelta(seconds=0.1)).timestamp() * 1e9)},
{"description": "0.1 sec Future", "value": int((datetime.now() + timedelta(seconds=0.1)).timestamp() * 1e9)},
{"description": "2 sec Future", "value": int((datetime.now() + timedelta(seconds=2)).timestamp() * 1e9)},
{"description": "10 sec Future", "value": int((datetime.now() + timedelta(seconds=10)).timestamp() * 1e9)},
]
STORE_TIMESTAMPS_FAIL = [
{"description": "20 sec Past", "value": int((datetime.now() - timedelta(seconds=20)).timestamp() * 1e9)},
{"description": "40 sec Future", "value": int((datetime.now() + timedelta(seconds=40)).timestamp() * 1e9)},
]
PUBSUB_TOPICS_RLN = ["/waku/2/rs/1/0"] PUBSUB_TOPICS_RLN = ["/waku/2/rs/1/0"]

17
tests/store/test_time_filter.py
View File

@ -2,7 +2,6 @@ import pytest
from datetime import timedelta, datetime from datetime import timedelta, datetime
from src.libs.custom_logger import get_custom_logger from src.libs.custom_logger import get_custom_logger
from src.steps.store import StepsStore from src.steps.store import StepsStore
from src.test_data import STORE_TIMESTAMPS_PASS, STORE_TIMESTAMPS_FAIL
logger = get_custom_logger(__name__) logger = get_custom_logger(__name__)
@ -13,8 +12,16 @@ logger = get_custom_logger(__name__)
@pytest.mark.usefixtures("node_setup") @pytest.mark.usefixtures("node_setup")
class TestTimeFilter(StepsStore): class TestTimeFilter(StepsStore):
def test_messages_with_timestamps_close_to_now(self): def test_messages_with_timestamps_close_to_now(self):
ts = [
{"description": "3 sec Past", "value": int((datetime.now() - timedelta(seconds=3)).timestamp() * 1e9)},
{"description": "1 sec Past", "value": int((datetime.now() - timedelta(seconds=1)).timestamp() * 1e9)},
{"description": "0.1 sec Past", "value": int((datetime.now() - timedelta(seconds=0.1)).timestamp() * 1e9)},
{"description": "0.1 sec Future", "value": int((datetime.now() + timedelta(seconds=0.1)).timestamp() * 1e9)},
{"description": "2 sec Future", "value": int((datetime.now() + timedelta(seconds=2)).timestamp() * 1e9)},
{"description": "10 sec Future", "value": int((datetime.now() + timedelta(seconds=10)).timestamp() * 1e9)},
]
failed_timestamps = [] failed_timestamps = []
for timestamp in STORE_TIMESTAMPS_PASS: for timestamp in ts:
logger.debug(f'Running test with payload {timestamp["description"]}') logger.debug(f'Running test with payload {timestamp["description"]}')
message = self.create_message(timestamp=timestamp["value"]) message = self.create_message(timestamp=timestamp["value"])
try: try:
@ -27,7 +34,11 @@ class TestTimeFilter(StepsStore):
def test_messages_with_timestamps_far_from_now(self): def test_messages_with_timestamps_far_from_now(self):
success_timestamps = [] success_timestamps = []
for timestamp in STORE_TIMESTAMPS_FAIL: ts = [
{"description": "20 sec Past", "value": int((datetime.now() - timedelta(seconds=20)).timestamp() * 1e9)},
{"description": "40 sec Future", "value": int((datetime.now() + timedelta(seconds=40)).timestamp() * 1e9)},
]
for timestamp in ts:
logger.debug(f'Running test with payload {timestamp["description"]}') logger.debug(f'Running test with payload {timestamp["description"]}')
message = self.create_message(timestamp=timestamp["value"]) message = self.create_message(timestamp=timestamp["value"])
try: try: