Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-05-02 12:56:30 +03:00
Code Issues Packages Projects Releases Wiki Activity

test: fix test_securejoin_after_contact_resetup flakiness

Browse Source
This commit is contained in:
link2xt
2024-10-18 12:10:45 +00:00
parent 5e13b4c736
commit 65b970a191
1 changed files with 23 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
deltachat-rpc-client/tests/test_securejoin.py
View File

@@ -1,4 +1,5 @@
import logging import logging
import time
import pytest import pytest
@@ -569,6 +570,7 @@ def test_securejoin_after_contact_resetup(acfactory) -> None:
# ac1 waits for member added message and creates a QR code. # ac1 waits for member added message and creates a QR code.
snapshot = ac1.get_message_by_id(ac1.wait_for_incoming_msg_event().msg_id).get_snapshot() snapshot = ac1.get_message_by_id(ac1.wait_for_incoming_msg_event().msg_id).get_snapshot()
assert snapshot.text == "Member Me ({}) added by {}.".format(ac1.get_config("addr"), ac3.get_config("addr"))
ac1_qr_code = snapshot.chat.get_qr_code() ac1_qr_code = snapshot.chat.get_qr_code()
# ac2 verifies ac1 # ac2 verifies ac1
@@ -583,17 +585,29 @@ def test_securejoin_after_contact_resetup(acfactory) -> None:
# ac1 resetups the account. # ac1 resetups the account.
ac1 = acfactory.resetup_account(ac1) ac1 = acfactory.resetup_account(ac1)
# ac1 sends a message to ac2. # Loop sending message from ac1 to ac2
ac1_contact_ac2 = ac1.create_contact(ac2.get_config("addr"), "") # until ac2 accepts new ac1 key.
ac1_chat_ac2 = ac1_contact_ac2.create_chat() #
ac1_chat_ac2.send_text("Hello!") # This may not happen immediately because resetup of ac1
# rewinds "smeared timestamp" so Date: header for messages
# sent by new ac1 are in the past compared to the last Date:
# header sent by old ac1.
while True:
# ac1 sends a message to ac2.
ac1_contact_ac2 = ac1.create_contact(ac2.get_config("addr"), "")
ac1_chat_ac2 = ac1_contact_ac2.create_chat()
ac1_chat_ac2.send_text("Hello!")
# ac2 receives a message. # ac2 receives a message.
snapshot = ac2.get_message_by_id(ac2.wait_for_incoming_msg_event().msg_id).get_snapshot() snapshot = ac2.get_message_by_id(ac2.wait_for_incoming_msg_event().msg_id).get_snapshot()
assert snapshot.text == "Hello!" assert snapshot.text == "Hello!"
logging.info("ac2 received Hello!")
# ac1 is no longer verified for ac2 as new Autocrypt key is not the same as old verified key. # ac1 is no longer verified for ac2 as new Autocrypt key is not the same as old verified key.
assert not ac2_contact_ac1.get_snapshot().is_verified logging.info("ac2 addr={}, ac1 addr={}".format(ac2.get_config("addr"), ac1.get_config("addr")))
if not ac2_contact_ac1.get_snapshot().is_verified:
break
time.sleep(1)
# ac1 goes offline. # ac1 goes offline.
ac1.remove() ac1.remove()