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

import concurrent.futures
import json
import logging
import subprocess
from unittest.mock import MagicMock

import pytest
from deltachat_rpc_client import Chat, EventType, SpecialContactId, events
from deltachat_rpc_client.rpc import JsonRpcError


def test_system_info(rpc) -> None:
    system_info = rpc.get_system_info()
    assert "arch" in system_info
    assert "deltachat_core_version" in system_info


def test_sleep(rpc) -> None:
    """Test that long-running task does not block short-running task from completion."""
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        sleep_5_future = executor.submit(rpc.sleep, 5.0)
        sleep_3_future = executor.submit(rpc.sleep, 3.0)
        done, pending = concurrent.futures.wait(
            [sleep_5_future, sleep_3_future],
            return_when=concurrent.futures.FIRST_COMPLETED,
        )
        assert sleep_3_future in done
        assert sleep_5_future in pending


def test_email_address_validity(rpc) -> None:
    valid_addresses = [
        "email@example.com",
        "36aa165ae3406424e0c61af17700f397cad3fe8ab83d682d0bddf3338a5dd52e@yggmail@yggmail",
    ]
    invalid_addresses = ["email@", "example.com", "emai221"]

    for addr in valid_addresses:
        assert rpc.check_email_validity(addr)
    for addr in invalid_addresses:
        assert not rpc.check_email_validity(addr)


def test_acfactory(acfactory) -> None:
    account = acfactory.new_configured_account()
    while True:
        event = account.wait_for_event()
        if event.kind == EventType.CONFIGURE_PROGRESS:
            assert event.progress != 0  # Progress 0 indicates error.
            if event.progress == 1000:  # Success
                break
        else:
            print(event)
    print("Successful configuration")


def test_configure_starttls(acfactory) -> None:
    account = acfactory.new_preconfigured_account()

    # Use STARTTLS
    account.set_config("mail_security", "2")
    account.set_config("send_security", "2")
    account.configure()
    assert account.is_configured()


def test_account(acfactory) -> None:
    alice, bob = acfactory.get_online_accounts(2)

    bob_addr = bob.get_config("addr")
    alice_contact_bob = alice.create_contact(bob_addr, "Bob")
    alice_chat_bob = alice_contact_bob.create_chat()
    alice_chat_bob.send_text("Hello!")

    while True:
        event = bob.wait_for_event()
        if event.kind == EventType.INCOMING_MSG:
            chat_id = event.chat_id
            msg_id = event.msg_id
            break

    message = bob.get_message_by_id(msg_id)
    snapshot = message.get_snapshot()
    assert snapshot.chat_id == chat_id
    assert snapshot.text == "Hello!"
    bob.mark_seen_messages([message])

    assert alice != bob
    assert repr(alice)
    assert alice.get_info().level
    assert alice.get_size()
    assert alice.is_configured()
    assert not alice.get_avatar()
    assert alice.get_contact_by_addr(bob_addr) == alice_contact_bob
    assert alice.get_contacts()
    assert alice.get_contacts(snapshot=True)
    assert alice.self_contact
    assert alice.get_chatlist()
    assert alice.get_chatlist(snapshot=True)
    assert alice.get_qr_code()
    assert alice.get_fresh_messages()
    assert alice.get_next_messages()

    # Test sending empty message.
    assert len(bob.wait_next_messages()) == 0
    alice_chat_bob.send_text("")
    messages = bob.wait_next_messages()
    assert len(messages) == 1
    message = messages[0]
    snapshot = message.get_snapshot()
    assert snapshot.text == ""
    bob.mark_seen_messages([message])

    group = alice.create_group("test group")
    group.add_contact(alice_contact_bob)
    group_msg = group.send_message(text="hello")
    assert group_msg == alice.get_message_by_id(group_msg.id)
    assert group == alice.get_chat_by_id(group.id)
    alice.delete_messages([group_msg])

    alice.set_config("selfstatus", "test")
    assert alice.get_config("selfstatus") == "test"
    alice.update_config(selfstatus="test2")
    assert alice.get_config("selfstatus") == "test2"

    assert not alice.get_blocked_contacts()
    alice_contact_bob.block()
    blocked_contacts = alice.get_blocked_contacts()
    assert blocked_contacts
    assert blocked_contacts[0].contact == alice_contact_bob

    bob.remove()
    alice.stop_io()


def test_chat(acfactory) -> None:
    alice, bob = acfactory.get_online_accounts(2)

    bob_addr = bob.get_config("addr")
    alice_contact_bob = alice.create_contact(bob_addr, "Bob")
    alice_chat_bob = alice_contact_bob.create_chat()
    alice_chat_bob.send_text("Hello!")

    while True:
        event = bob.wait_for_event()
        if event.kind == EventType.INCOMING_MSG:
            chat_id = event.chat_id
            msg_id = event.msg_id
            break
    message = bob.get_message_by_id(msg_id)
    snapshot = message.get_snapshot()
    assert snapshot.chat_id == chat_id
    assert snapshot.text == "Hello!"
    bob_chat_alice = bob.get_chat_by_id(chat_id)

    assert alice_chat_bob != bob_chat_alice
    assert repr(alice_chat_bob)
    alice_chat_bob.delete()
    assert not bob_chat_alice.can_send()
    bob_chat_alice.accept()
    assert bob_chat_alice.can_send()
    bob_chat_alice.block()
    bob_chat_alice = snapshot.sender.create_chat()
    bob_chat_alice.mute()
    bob_chat_alice.unmute()
    bob_chat_alice.pin()
    bob_chat_alice.unpin()
    bob_chat_alice.archive()
    bob_chat_alice.unarchive()
    with pytest.raises(JsonRpcError):  # can't set name for 1:1 chats
        bob_chat_alice.set_name("test")
    bob_chat_alice.set_ephemeral_timer(300)
    bob_chat_alice.get_encryption_info()

    group = alice.create_group("test group")
    group.add_contact(alice_contact_bob)
    group.get_qr_code()

    snapshot = group.get_basic_snapshot()
    assert snapshot.name == "test group"
    group.set_name("new name")
    snapshot = group.get_full_snapshot()
    assert snapshot.name == "new name"

    msg = group.send_message(text="hi")
    assert (msg.get_snapshot()).text == "hi"
    group.forward_messages([msg])

    group.set_draft(text="test draft")
    draft = group.get_draft()
    assert draft.text == "test draft"
    group.remove_draft()
    assert not group.get_draft()

    assert group.get_messages()
    group.get_fresh_message_count()
    group.mark_noticed()
    assert group.get_contacts()
    group.remove_contact(alice_chat_bob)
    group.get_locations()


def test_contact(acfactory) -> None:
    alice, bob = acfactory.get_online_accounts(2)

    bob_addr = bob.get_config("addr")
    alice_contact_bob = alice.create_contact(bob_addr, "Bob")

    assert alice_contact_bob == alice.get_contact_by_id(alice_contact_bob.id)
    assert repr(alice_contact_bob)
    alice_contact_bob.block()
    alice_contact_bob.unblock()
    alice_contact_bob.set_name("new name")
    alice_contact_bob.get_encryption_info()
    snapshot = alice_contact_bob.get_snapshot()
    assert snapshot.address == bob_addr
    alice_contact_bob.create_chat()


def test_message(acfactory) -> None:
    alice, bob = acfactory.get_online_accounts(2)

    bob_addr = bob.get_config("addr")
    alice_contact_bob = alice.create_contact(bob_addr, "Bob")
    alice_chat_bob = alice_contact_bob.create_chat()
    alice_chat_bob.send_text("Hello!")

    while True:
        event = bob.wait_for_event()
        if event.kind == EventType.INCOMING_MSG:
            chat_id = event.chat_id
            msg_id = event.msg_id
            break

    message = bob.get_message_by_id(msg_id)
    snapshot = message.get_snapshot()
    assert snapshot.chat_id == chat_id
    assert snapshot.text == "Hello!"
    assert not snapshot.is_bot
    assert repr(message)

    with pytest.raises(JsonRpcError):  # chat is not accepted
        snapshot.chat.send_text("hi")
    snapshot.chat.accept()
    snapshot.chat.send_text("hi")

    message.mark_seen()
    message.send_reaction("😎")
    reactions = message.get_reactions()
    assert reactions
    snapshot = message.get_snapshot()
    assert reactions == snapshot.reactions


def test_is_bot(acfactory) -> None:
    """Test that we can recognize messages submitted by bots."""
    alice, bob = acfactory.get_online_accounts(2)

    bob_addr = bob.get_config("addr")
    alice_contact_bob = alice.create_contact(bob_addr, "Bob")
    alice_chat_bob = alice_contact_bob.create_chat()

    # Alice becomes a bot.
    alice.set_config("bot", "1")
    alice_chat_bob.send_text("Hello!")

    while True:
        event = bob.wait_for_event()
        if event.kind == EventType.INCOMING_MSG:
            msg_id = event.msg_id
            message = bob.get_message_by_id(msg_id)
            snapshot = message.get_snapshot()
            assert snapshot.chat_id == event.chat_id
            assert snapshot.text == "Hello!"
            assert snapshot.is_bot
            break


def test_bot(acfactory) -> None:
    mock = MagicMock()
    user = (acfactory.get_online_accounts(1))[0]
    bot = acfactory.new_configured_bot()
    bot2 = acfactory.new_configured_bot()

    assert bot.is_configured()
    assert bot.account.get_config("bot") == "1"

    hook = lambda e: mock.hook(e.msg_id) and None, events.RawEvent(EventType.INCOMING_MSG)
    bot.add_hook(*hook)
    event = acfactory.process_message(from_account=user, to_client=bot, text="Hello!")
    snapshot = bot.account.get_message_by_id(event.msg_id).get_snapshot()
    assert not snapshot.is_bot
    mock.hook.assert_called_once_with(event.msg_id)
    bot.remove_hook(*hook)

    def track(e):
        mock.hook(e.message_snapshot.id)

    mock.hook.reset_mock()
    hook = track, events.NewMessage(r"hello")
    bot.add_hook(*hook)
    bot.add_hook(track, events.NewMessage(command="/help"))
    event = acfactory.process_message(from_account=user, to_client=bot, text="hello")
    mock.hook.assert_called_with(event.msg_id)
    event = acfactory.process_message(from_account=user, to_client=bot, text="hello!")
    mock.hook.assert_called_with(event.msg_id)
    acfactory.process_message(from_account=bot2.account, to_client=bot, text="hello")
    assert len(mock.hook.mock_calls) == 2  # bot messages are ignored between bots
    acfactory.process_message(from_account=user, to_client=bot, text="hey!")
    assert len(mock.hook.mock_calls) == 2
    bot.remove_hook(*hook)

    mock.hook.reset_mock()
    acfactory.process_message(from_account=user, to_client=bot, text="hello")
    event = acfactory.process_message(from_account=user, to_client=bot, text="/help")
    mock.hook.assert_called_once_with(event.msg_id)


def test_wait_next_messages(acfactory) -> None:
    alice = acfactory.new_configured_account()

    # Create a bot account so it does not receive device messages in the beginning.
    bot = acfactory.new_preconfigured_account()
    bot.set_config("bot", "1")
    bot.configure()

    # There are no old messages and the call returns immediately.
    assert not bot.wait_next_messages()

    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        # Bot starts waiting for messages.
        next_messages_task = executor.submit(bot.wait_next_messages)

        bot_addr = bot.get_config("addr")
        alice_contact_bot = alice.create_contact(bot_addr, "Bob")
        alice_chat_bot = alice_contact_bot.create_chat()
        alice_chat_bot.send_text("Hello!")

        next_messages = next_messages_task.result()
        assert len(next_messages) == 1
        snapshot = next_messages[0].get_snapshot()
        assert snapshot.text == "Hello!"


def test_import_export(acfactory, tmp_path) -> None:
    alice = acfactory.new_configured_account()
    alice.export_backup(tmp_path)

    files = list(tmp_path.glob("*.tar"))
    alice2 = acfactory.get_unconfigured_account()
    alice2.import_backup(files[0])

    assert alice2.manager.get_system_info()


def test_openrpc_command_line() -> None:
    """Test that "deltachat-rpc-server --openrpc" command returns an OpenRPC specification."""
    out = subprocess.run(["deltachat-rpc-server", "--openrpc"], capture_output=True, check=True).stdout
    openrpc = json.loads(out)
    assert "openrpc" in openrpc
    assert "methods" in openrpc


def test_provider_info(rpc) -> None:
    account_id = rpc.add_account()

    provider_info = rpc.get_provider_info(account_id, "example.org")
    assert provider_info["id"] == "example.com"

    provider_info = rpc.get_provider_info(account_id, "uep7oiw4ahtaizuloith.org")
    assert provider_info is None

    # Test MX record resolution.
    provider_info = rpc.get_provider_info(account_id, "github.com")
    assert provider_info["id"] == "gmail"

    # Disable MX record resolution.
    rpc.set_config(account_id, "socks5_enabled", "1")
    provider_info = rpc.get_provider_info(account_id, "github.com")
    assert provider_info is None


def test_qr_setup_contact(acfactory) -> None:
    alice, bob = acfactory.get_online_accounts(2)

    qr_code, _svg = alice.get_qr_code()
    bob.secure_join(qr_code)

    while True:
        event = alice.wait_for_event()
        if event["kind"] == "SecurejoinInviterProgress" and event["progress"] == 1000:
            return


@pytest.mark.xfail()
def test_verified_group_recovery(acfactory, rpc) -> None:
    ac1, ac2, ac3 = acfactory.get_online_accounts(3)

    logging.info("ac1 creates verified group")
    chat = ac1.create_group("Verified group", protect=True)
    assert chat.get_basic_snapshot().is_protected

    logging.info("ac2 joins verified group")
    qr_code, _svg = chat.get_qr_code()
    ac2.secure_join(qr_code)
    while True:
        event = ac1.wait_for_event()
        if event.kind == "SecurejoinInviterProgress" and event["progress"] == 1000:
            break

    # ac1 has ac2 directly verified.
    ac1_contact_ac2 = ac1.get_contact_by_addr(ac2.get_config("addr"))
    assert ac1_contact_ac2.get_snapshot().verifier_id == SpecialContactId.SELF

    logging.info("ac3 joins verified group")
    ac3_chat = ac3.secure_join(qr_code)
    while True:
        event = ac1.wait_for_event()
        if event.kind == "SecurejoinInviterProgress" and event["progress"] == 1000:
            break

    logging.info("ac2 logs in on a new device")
    ac2 = acfactory.resetup_account(ac2)

    logging.info("ac2 reverifies with ac3")
    qr_code, _svg = ac3.get_qr_code()
    ac2.secure_join(qr_code)

    while True:
        event = ac3.wait_for_event()
        if event.kind == "SecurejoinInviterProgress" and event["progress"] == 1000:
            break

    logging.info("ac3 sends a message to the group")
    assert len(ac3_chat.get_contacts()) == 3
    ac3_chat.send_text("Hi!")

    msg_id = ac2.wait_for_incoming_msg_event().msg_id
    message = ac2.get_message_by_id(msg_id)
    snapshot = message.get_snapshot()
    logging.info("Received message %s", snapshot.text)
    assert snapshot.text == "Hi!"

    # ac1 contact cannot be verified by ac2 because ac3 did not gossip ac1 key in the "Hi!" message.
    ac1_contact = ac2.get_contact_by_addr(ac1.get_config("addr"))
    assert not ac1_contact.get_snapshot().is_verified

    ac3_contact_id_ac1 = rpc.lookup_contact_id_by_addr(ac3.id, ac1.get_config("addr"))
    ac3_chat.remove_contact(ac3_contact_id_ac1)
    ac3_chat.add_contact(ac3_contact_id_ac1)

    msg_id = ac2.wait_for_incoming_msg_event().msg_id
    message = ac2.get_message_by_id(msg_id)
    snapshot = message.get_snapshot()
    logging.info("ac2 got event message: %s", snapshot.text)
    assert "removed" in snapshot.text

    event = ac2.wait_for_incoming_msg_event()
    msg_id = event.msg_id
    chat_id = event.chat_id
    message = ac2.get_message_by_id(msg_id)
    snapshot = message.get_snapshot()
    logging.info("ac2 got event message: %s", snapshot.text)
    assert "added" in snapshot.text

    assert ac1_contact.get_snapshot().is_verified

    chat = Chat(ac2, chat_id)
    chat.send_text("Works again!")

    msg_id = ac3.wait_for_incoming_msg_event().msg_id
    message = ac3.get_message_by_id(msg_id)
    snapshot = message.get_snapshot()
    assert snapshot.text == "Works again!"

    ac1.wait_for_incoming_msg_event()  # Hi!
    ac1.wait_for_incoming_msg_event()  # Member removed
    ac1.wait_for_incoming_msg_event()  # Member added
    snapshot = ac1.get_message_by_id(ac1.wait_for_incoming_msg_event().msg_id).get_snapshot()
    assert snapshot.text == "Works again!"

    # ac2 is now verified by ac3 for ac1
    ac1_contact_ac3 = ac1.get_contact_by_addr(ac3.get_config("addr"))
    assert ac1_contact_ac2.get_snapshot().verifier_id == ac1_contact_ac3.id


def test_verified_group_member_added_recovery(acfactory) -> None:
    ac1, ac2, ac3 = acfactory.get_online_accounts(3)

    logging.info("ac1 creates verified group")
    chat = ac1.create_group("Verified group", protect=True)
    assert chat.get_basic_snapshot().is_protected

    logging.info("ac2 joins verified group")
    qr_code, _svg = chat.get_qr_code()
    ac2.secure_join(qr_code)
    while True:
        event = ac1.wait_for_event()
        if event.kind == "SecurejoinInviterProgress" and event["progress"] == 1000:
            break

    # ac1 has ac2 directly verified.
    ac1_contact_ac2 = ac1.get_contact_by_addr(ac2.get_config("addr"))
    assert ac1_contact_ac2.get_snapshot().verifier_id == SpecialContactId.SELF

    logging.info("ac3 joins verified group")
    ac3_chat = ac3.secure_join(qr_code)
    while True:
        event = ac1.wait_for_event()
        if event.kind == "SecurejoinInviterProgress" and event["progress"] == 1000:
            break

    logging.info("ac2 logs in on a new device")
    ac2 = acfactory.resetup_account(ac2)

    logging.info("ac2 reverifies with ac3")
    qr_code, _svg = ac3.get_qr_code()
    ac2.secure_join(qr_code)

    while True:
        event = ac3.wait_for_event()
        if event.kind == "SecurejoinInviterProgress" and event["progress"] == 1000:
            break

    logging.info("ac3 sends a message to the group")
    assert len(ac3_chat.get_contacts()) == 3
    ac3_chat.send_text("Hi!")

    msg_id = ac2.wait_for_incoming_msg_event().msg_id
    message = ac2.get_message_by_id(msg_id)
    snapshot = message.get_snapshot()
    logging.info("Received message %s", snapshot.text)
    assert snapshot.text == "Hi!"

    ac1.wait_for_incoming_msg_event()  # Hi!

    ac3_contact_ac2 = ac3.get_contact_by_addr(ac2.get_config("addr"))
    ac3_chat.remove_contact(ac3_contact_ac2)
    ac3_chat.add_contact(ac3_contact_ac2)

    msg_id = ac2.wait_for_incoming_msg_event().msg_id
    message = ac2.get_message_by_id(msg_id)
    snapshot = message.get_snapshot()
    assert "removed" in snapshot.text

    snapshot = ac1.get_message_by_id(ac1.wait_for_incoming_msg_event().msg_id).get_snapshot()
    assert "removed" in snapshot.text

    event = ac2.wait_for_incoming_msg_event()
    msg_id = event.msg_id
    chat_id = event.chat_id
    message = ac2.get_message_by_id(msg_id)
    snapshot = message.get_snapshot()
    logging.info("ac2 got event message: %s", snapshot.text)
    assert "added" in snapshot.text

    snapshot = ac1.get_message_by_id(ac1.wait_for_incoming_msg_event().msg_id).get_snapshot()
    assert "added" in snapshot.text

    logging.info("ac2 address is %s", ac2.get_config("addr"))
    ac1_contact_ac2 = ac1.get_contact_by_addr(ac2.get_config("addr"))
    ac1_contact_ac2_snapshot = ac1_contact_ac2.get_snapshot()
    # assert ac1_contact_ac2_snapshot.is_verified
    assert ac1_contact_ac2_snapshot.verifier_id == ac1.get_contact_by_addr(ac3.get_config("addr")).id

    chat = Chat(ac2, chat_id)
    chat.send_text("Works again!")

    msg_id = ac3.wait_for_incoming_msg_event().msg_id
    message = ac3.get_message_by_id(msg_id)
    snapshot = message.get_snapshot()
    assert snapshot.text == "Works again!"

    snapshot = ac1.get_message_by_id(ac1.wait_for_incoming_msg_event().msg_id).get_snapshot()
    assert snapshot.text == "Works again!"

    # ac2 is now verified by ac3 for ac1
    ac1_contact_ac3 = ac1.get_contact_by_addr(ac3.get_config("addr"))
    assert ac1_contact_ac2.get_snapshot().verifier_id == ac1_contact_ac3.id