chatmail-core/deltachat-rpc-client/tests/test_multitransport.py

import time
import urllib.parse

import pytest

from deltachat_rpc_client import EventType
from deltachat_rpc_client.const import ChatType, DownloadState
from deltachat_rpc_client.rpc import JsonRpcError


def alice_with_two_transports_and_bob(acf):
    alice, bob = acf.get_online_accounts(2)
    alice.add_transport_from_qr(acf.get_account_qr())
    alice.bring_online()
    return alice, alice.create_chat(bob), bob.create_chat(alice)


def messages_with_text(chat, text):
    return [msg for msg in chat.get_messages() if msg.get_snapshot().text == text]


def wait_for_imap_message(imap):
    while not imap.get_all_messages():
        time.sleep(1)


def test_add_second_address(acf) -> None:
    account = acf.new_configured_account()
    assert len(account.list_transports()) == 1

    qr = acf.get_account_qr()
    account.add_transport_from_qr(qr)
    assert len(account.list_transports()) == 2

    account.add_transport_from_qr(qr)
    assert len(account.list_transports()) == 3

    first_addr = account.list_transports()[0]["addr"]
    second_addr = account.list_transports()[1]["addr"]
    third_addr = account.list_transports()[2]["addr"]

    assert account.get_config("configured_addr") == first_addr
    account.delete_transport(first_addr)
    assert len(account.list_transports()) == 2
    assert account.get_config("configured_addr") != first_addr

    account.delete_transport(second_addr)
    assert len(account.list_transports()) == 1

    with pytest.raises(JsonRpcError):
        account.delete_transport(third_addr)


def test_change_address(acf) -> None:
    """Test Alice configuring a second transport and removing the first one."""
    alice, bob = acf.get_online_accounts(2)

    bob.create_chat(alice)

    alice_chat_bob = alice.create_chat(bob)
    alice_chat_bob.send_text("Hello!")

    msg1 = bob.wait_for_incoming_msg().get_snapshot()
    sender_addr1 = msg1.sender.get_snapshot().address

    alice.stop_io()
    old_alice_addr = alice.list_transports()[0]["addr"]
    alice_vcard = alice.self_contact.make_vcard()
    assert old_alice_addr in alice_vcard
    qr = acf.get_account_qr()
    alice.add_transport_from_qr(qr)
    new_alice_addr = alice.list_transports()[1]["addr"]

    alice.delete_transport(old_alice_addr)

    alice_vcard = alice.self_contact.make_vcard()
    assert old_alice_addr not in alice_vcard
    assert new_alice_addr in alice_vcard
    alice.start_io()

    alice_chat_bob.send_text("Hello again!")

    msg2 = bob.wait_for_incoming_msg().get_snapshot()
    sender_addr2 = msg2.sender.get_snapshot().address

    assert msg1.sender == msg2.sender
    assert sender_addr1 != sender_addr2
    assert sender_addr1 == old_alice_addr
    assert sender_addr2 == new_alice_addr


def test_remove_transport_keep_messages(acf) -> None:
    """Test that deleting current sending transport keeps queued messages."""
    alice, bob = acf.get_online_accounts(2)

    qr = acf.get_account_qr()
    alice.add_transport_from_qr(qr)

    alice.stop_io()
    alice_chat_bob = alice.create_chat(bob)
    alice_chat_bob.send_text("Hello!")

    new_alice_addr = alice.list_transports()[1]["addr"]
    alice.delete_transport(alice.list_transports()[0]["addr"])
    alice.start_io()

    bob_msg = bob.wait_for_incoming_msg().get_snapshot()
    assert bob_msg.sender.get_snapshot().address == new_alice_addr


def test_download_on_demand(acf, rpcdata) -> None:
    alice, bob = acf.get_online_accounts(2)
    alice.set_config("download_limit", "1")

    alice.stop_io()
    qr = acf.get_account_qr()
    alice.add_transport_from_qr(qr)
    alice.start_io()

    alice.create_chat(bob)
    chat_bob_alice = bob.create_chat(alice)
    chat_bob_alice.send_message(file=rpcdata.get_path("image/screenshot.jpg"))
    msg = alice.wait_for_incoming_msg()
    snapshot = msg.get_snapshot()
    assert snapshot.download_state == DownloadState.AVAILABLE
    chat_id = snapshot.chat_id
    # Actually the message isn't available yet. Wait somehow for the post-message to arrive.
    chat_bob_alice.send_message("Now you can download my previous message")
    alice.wait_for_incoming_msg()
    alice._rpc.download_full_message(alice.id, msg.id)
    for dstate in [DownloadState.IN_PROGRESS, DownloadState.DONE]:
        event = alice.wait_for_event(EventType.MSGS_CHANGED)
        assert event.chat_id == chat_id
        assert event.msg_id == msg.id
        assert msg.get_snapshot().download_state == dstate


def test_reconfigure_transport(acf) -> None:
    """Test that reconfiguring the transport works."""
    account = acf.get_online_account()

    [transport] = account.list_transports()
    account.add_or_update_transport(transport)


def test_transport_synchronization(acf, log) -> None:
    """Test synchronization of transports between devices."""

    def wait_for_io_started(ac):
        while True:
            ev = ac.wait_for_event(EventType.INFO)
            if "scheduler is running" in ev.msg:
                return

    def wait_transports(ac, n):
        while len(ac.list_transports()) != n:
            ac.wait_for_event(EventType.TRANSPORTS_MODIFIED)

    ac1, ac2 = acf.get_online_accounts(2)
    ac1_clone = ac1.clone()
    ac1_clone.bring_online()

    qr = acf.get_account_qr()

    ac1.add_transport_from_qr(qr)
    wait_transports(ac1_clone, 2)
    wait_for_io_started(ac1_clone)
    assert len(ac1.list_transports()) == 2

    ac1_clone.add_transport_from_qr(qr)
    wait_transports(ac1, 3)
    wait_for_io_started(ac1)
    assert len(ac1_clone.list_transports()) == 3

    log.section("ac1 clone removes second transport")
    [transport1, transport2, transport3] = ac1_clone.list_transports()
    addr3 = transport3["addr"]
    ac1_clone.delete_transport(transport2["addr"])

    wait_transports(ac1, 2)
    wait_for_io_started(ac1)
    [transport1, transport3] = ac1.list_transports()

    log.section("ac1 changes the sending transport")
    ac1.set_config("configured_addr", transport3["addr"])

    log.section("ac1 removes the first transport")
    ac1.delete_transport(transport1["addr"])

    wait_transports(ac1_clone, 1)
    wait_for_io_started(ac1_clone)
    [transport3] = ac1_clone.list_transports()
    assert transport3["addr"] == addr3
    assert ac1_clone.get_config("configured_addr") == addr3

    ac2_chat = ac2.create_chat(ac1)
    ac2_chat.send_text("Hello!")

    assert ac1.wait_for_incoming_msg().get_snapshot().text == "Hello!"
    assert ac1_clone.wait_for_incoming_msg().get_snapshot().text == "Hello!"


def test_transport_sync_new_as_primary(acf, log) -> None:
    """Test that a transport promoted on one device is usable on other devices."""
    ac1, bob = acf.get_online_accounts(2)
    ac1_clone = ac1.clone()
    ac1_clone.bring_online()

    qr = acf.get_account_qr()

    ac1.add_transport_from_qr(qr)
    ac1.wait_for_event(EventType.TRANSPORTS_MODIFIED)
    ac1_transports = ac1.list_transports()
    assert len(ac1_transports) == 2
    [transport1, transport2] = ac1_transports
    ac1_clone.wait_for_event(EventType.TRANSPORTS_MODIFIED)
    assert len(ac1_clone.list_transports()) == 2
    assert ac1_clone.get_config("configured_addr") == transport1["addr"]

    log.section("ac1 changes the primary transport")
    ac1.set_config("configured_addr", transport2["addr"])

    log.section("ac1_clone receives a message via the new transport")
    ac1_chat = ac1.create_chat(bob)
    ac1_chat.send_text("Hello!")
    bob_chat_id = bob.wait_for_incoming_msg_event().chat_id
    bob_chat = bob.get_chat_by_id(bob_chat_id)
    bob_chat.accept()
    bob_chat.send_text("hello back")
    assert ac1_clone.wait_for_incoming_msg().get_snapshot().text == "hello back"


def test_recognize_self_address(acf) -> None:
    alice, bob = acf.get_online_accounts(2)

    bob_chat = bob.create_chat(alice)

    qr = acf.get_account_qr()
    alice.add_transport_from_qr(qr)

    new_alice_addr = alice.list_transports()[1]["addr"]
    alice.set_config("configured_addr", new_alice_addr)

    bob_chat.send_text("Hello!")
    msg = alice.wait_for_incoming_msg().get_snapshot()
    assert msg.chat == alice.create_chat(bob)


def test_transport_limit(acf) -> None:
    """Test transports limit."""
    account = acf.get_online_account()
    qr = acf.get_account_qr()

    limit = 5

    for _ in range(1, limit):
        account.add_transport_from_qr(qr)

    assert len(account.list_transports()) == limit

    with pytest.raises(JsonRpcError):
        account.add_transport_from_qr(qr)

    second_addr = account.list_transports()[1]["addr"]

    account.delete_transport(second_addr)
    account.add_transport_from_qr(qr)
    with pytest.raises(JsonRpcError):
        account.add_transport_from_qr(qr)


def test_message_info_imap_urls(acf) -> None:
    """Test that message info contains IMAP URLs of where the message was received."""
    alice, bob = acf.get_online_accounts(2)

    qr = acf.get_account_qr()
    for _ in range(3):
        alice.add_transport_from_qr(qr)
        # Wait for all transports to go IDLE after adding each one.
        alice.bring_online()

    # Enable multi-device mode so messages are not deleted immediately.
    alice.set_config("bcc_self", "1")

    # Bob creates chat, learning about Alice's currently selected transport.
    # This is where he will send the message.
    bob_chat = bob.create_chat(alice)

    # Alice switches to another transport and removes the rest of the transports.
    new_alice_addr = alice.list_transports()[1]["addr"]
    alice.set_config("configured_addr", new_alice_addr)
    removed_addrs = []
    for transport in alice.list_transports():
        if transport["addr"] != new_alice_addr:
            alice.delete_transport(transport["addr"])
            removed_addrs.append(transport["addr"])
    alice.stop_io()
    alice.start_io()

    bob_chat.send_text("Hello!")

    msg = alice.wait_for_incoming_msg()
    msg_info = msg.get_info()
    assert new_alice_addr in msg_info
    for removed_addr in removed_addrs:
        assert removed_addr not in msg_info
    assert f"{new_alice_addr}/INBOX" in msg_info


def test_remove_primary_transport(acf, log) -> None:
    """Test that after removing the primary relay, Alice can still receive messages."""
    alice, alice_chat, bob_chat = alice_with_two_transports_and_bob(acf)

    log.section("Alice sets up second transport")
    [transport1, transport2] = alice.list_transports()

    bob_chat.send_text("Hello!")
    msg1 = alice.wait_for_incoming_msg().get_snapshot()
    assert msg1.text == "Hello!"

    log.section("Alice removes the primary relay")
    alice.delete_transport(transport1["addr"])
    assert alice.get_config("configured_addr") == transport2["addr"]
    alice.stop_io()
    alice.start_io()

    bob_chat.send_text("Hello again!")
    msg2 = alice.wait_for_incoming_msg().get_snapshot()
    assert msg2.text == "Hello again!"
    assert msg2.chat.get_basic_snapshot().chat_type == ChatType.SINGLE
    assert msg2.chat == alice_chat


def test_qr_works_after_removing_primary_transport(acf, log) -> None:
    log.section("Alice setups an account and adds two additional relays")
    alice = acf.new_configured_account()
    relay_qr = acf.get_account_qr()
    alice.add_transport_from_qr(relay_qr)
    alice.add_transport_from_qr(relay_qr)

    first_addr = alice.list_transports()[0]["addr"]
    second_addr = alice.list_transports()[1]["addr"]
    third_addr = alice.list_transports()[2]["addr"]

    log.section("Alice creates a QR code")
    chat_qr = alice.get_qr_code()
    chat_qr_unquoted = urllib.parse.unquote(chat_qr)
    assert f"&a={first_addr}" in chat_qr_unquoted
    assert f"&r={third_addr},{second_addr}" in chat_qr_unquoted

    log.section("Alice removes first and second transport")
    alice.set_config("configured_addr", third_addr)
    alice.delete_transport(first_addr)
    alice.delete_transport(second_addr)

    log.section("Bob scans the QR code, which still works")
    alice.bring_online()
    bob = acf.get_online_account()
    bob.secure_join(chat_qr)
    alice.wait_for_securejoin_inviter_success()
    bob.wait_for_securejoin_joiner_success()


def test_background_fetch_from_second_transport(acf, direct_imap, dc):
    alice, alice_chat, bob_chat = alice_with_two_transports_and_bob(acf)
    [transport1, transport2] = alice.list_transports()
    assert alice.get_config("configured_addr") == transport1["addr"]

    alice.stop_io()
    bob_chat.send_text("hello")
    imap1 = direct_imap(alice, transport1["addr"], transport1["password"])
    wait_for_imap_message(direct_imap(alice, transport2["addr"], transport2["password"]))
    wait_for_imap_message(imap1)

    # Leave the message on the second transport only.
    imap1.delete("1:*")

    dc.background_fetch(300)
    assert len(messages_with_text(alice_chat, "hello")) == 1


def test_background_fetch_no_duplicates(acf, direct_imap, dc):
    alice, alice_chat, bob_chat = alice_with_two_transports_and_bob(acf)

    alice.stop_io()
    bob_chat.send_text("hello")
    for transport in alice.list_transports():
        wait_for_imap_message(direct_imap(alice, transport["addr"], transport["password"]))

    dc.background_fetch(300)
    assert len(messages_with_text(alice_chat, "hello")) == 1


def test_multitransport_mdn(acf):
    """Test sending an MDN right after configuring two transports."""
    alice, bob = acf.get_online_accounts(2)
    alice.add_transport_from_qr(acf.get_account_qr())
    alice.bring_online()
    alice.create_chat(bob)
    bob_msg = bob.create_chat(alice).send_text("Hello!")
    alice.wait_for_incoming_msg().mark_seen()
    assert bob.wait_for_event(EventType.MSG_READ).msg_id == bob_msg.id