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

fix: scoped assertion for store test in fleet mode

Browse Source
This commit is contained in:
Roman 2026-05-04 13:08:42 +08:00
parent 57133df197
commit da0eac56a2
No known key found for this signature in database
GPG Key ID: 583BDF43C238B83E
1 changed files with 11 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
tests/store/test_get_messages.py
View File

@ -23,12 +23,17 @@ class TestGetMessages(StepsStore):
logger.error(f'Payload {payload["description"]} failed: {str(e)}')
failed_payloads.append(payload["description"])
assert not failed_payloads, f"Payloads failed: {failed_payloads}"
# In fleet mode the archive may also store background fleet messages; count
# only the messages that belong to the test by matching the test content topic.
test_msgs = [m for m in self.store_response.messages if m.get("message", {}).get("contentTopic") == self.test_content_topic]
assert len(test_msgs) == len(SAMPLE_INPUTS), (
f"Expected {len(SAMPLE_INPUTS)} test messages but found {len(test_msgs)} " f"(total in store: {len(self.store_response.messages)})"
)
# Content-topic-scoped query to accommodate fleet tests
for node in self.store_nodes:
scoped = self.get_messages_from_store(
node,
page_size=len(SAMPLE_INPUTS) + 10,
content_topics=self.test_content_topic,
ascending="true",
)
assert len(scoped.messages) == len(SAMPLE_INPUTS), (
f"Expected {len(SAMPLE_INPUTS)} test messages on {node.image} " f"but found {len(scoped.messages)}"
)
@pytest.mark.waku_test_fleet
def test_get_store_messages_with_different_content_topics(self):