Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-04-05 23:22:11 +03:00
Code Issues Packages Projects Releases Wiki Activity

feat: key-contacts

Browse Source 
This change introduces a new type of contacts
identified by their public key fingerprint
rather than an e-mail address.

Encrypted chats now stay encrypted
and unencrypted chats stay unencrypted.
For example, 1:1 chats with key-contacts
are encrypted and 1:1 chats with address-contacts
are unencrypted.
Groups that have a group ID are encrypted
and can only contain key-contacts
while groups that don't have a group ID ("adhoc groups")
are unencrypted and can only contain address-contacts.

JSON-RPC API `reset_contact_encryption` is removed.
Python API `Contact.reset_encryption` is removed.
"Group tracking plugin" in legacy Python API was removed because it
relied on parsing email addresses from system messages with regexps.

Co-authored-by: Hocuri <hocuri@gmx.de>
Co-authored-by: iequidoo <dgreshilov@gmail.com>
Co-authored-by: B. Petersen <r10s@b44t.com>
This commit is contained in:
link2xt
2025-06-26 14:07:39 +00:00
parent 7ac04d0204
commit 416131b4a2
84 changed files with 4735 additions and 6338 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
python/examples/group_tracking.py
View File

@@ -1,49 +0,0 @@
# content of group_tracking.py
from deltachat import account_hookimpl, run_cmdline
class GroupTrackingPlugin:
@account_hookimpl
def ac_incoming_message(self, message):
print("process_incoming message", message)
if message.text.strip() == "/quit":
message.account.shutdown()
else:
# unconditionally accept the chat
message.create_chat()
addr = message.get_sender_contact().addr
text = message.text
message.chat.send_text(f"echoing from {addr}:\n{text}")
@account_hookimpl
def ac_outgoing_message(self, message):
print("ac_outgoing_message:", message)
@account_hookimpl
def ac_configure_completed(self, success):
print("ac_configure_completed:", success)
@account_hookimpl
def ac_chat_modified(self, chat):
print("ac_chat_modified:", chat.id, chat.get_name())
for member in chat.get_contacts():
print(f"chat member: {member.addr}")
@account_hookimpl
def ac_member_added(self, chat, contact, actor, message):
print(f"ac_member_added {contact.addr} to chat {chat.id} from {actor or message.get_sender_contact().addr}")
for member in chat.get_contacts():
print(f"chat member: {member.addr}")
@account_hookimpl
def ac_member_removed(self, chat, contact, actor, message):
print(f"ac_member_removed {contact.addr} from chat {chat.id} by {actor or message.get_sender_contact().addr}")
def main(argv=None):
run_cmdline(argv=argv, account_plugins=[GroupTrackingPlugin()])
if __name__ == "__main__":
main()

55
python/examples/test_examples.py
View File

@@ -1,10 +1,7 @@
import echo_and_quit
import group_tracking
import py
import pytest
from deltachat.events import FFIEventLogger
@pytest.fixture(scope="session")
def datadir():
@@ -36,55 +33,3 @@ def test_echo_quit_plugin(acfactory, lp):
lp.sec("send quit sequence")
bot_chat.send_text("/quit")
botproc.wait()
def test_group_tracking_plugin(acfactory, lp):
lp.sec("creating one group-tracking bot and two temp accounts")
botproc = acfactory.run_bot_process(group_tracking)
ac1, ac2 = acfactory.get_online_accounts(2)
ac1.add_account_plugin(FFIEventLogger(ac1))
ac2.add_account_plugin(FFIEventLogger(ac2))
lp.sec("creating bot test group with bot")
bot_chat = ac1.qr_setup_contact(botproc.qr)
ac1._evtracker.wait_securejoin_joiner_progress(1000)
bot_contact = bot_chat.get_contacts()[0]
ch = ac1.create_group_chat("bot test group")
ch.add_contact(bot_contact)
ch.send_text("hello")
botproc.fnmatch_lines(
"""
*ac_chat_modified*bot test group*
""",
)
lp.sec("adding third member {}".format(ac2.get_config("addr")))
contact3 = ac1.create_contact(ac2)
ch.add_contact(contact3)
reply = ac1._evtracker.wait_next_incoming_message()
assert "hello" in reply.text
lp.sec("now looking at what the bot received")
botproc.fnmatch_lines(
"""
*ac_member_added {}*from*{}*
""".format(
contact3.addr,
ac1.get_config("addr"),
),
)
lp.sec("contact successfully added, now removing")
ch.remove_contact(contact3)
botproc.fnmatch_lines(
"""
*ac_member_removed {}*from*{}*
""".format(
contact3.addr,
ac1.get_config("addr"),
),
)

2
python/src/deltachat/account.py
View File

@@ -293,6 +293,8 @@ class Account:
return Contact(self, contact_id)
def get_contact(self, obj) -> Optional[Contact]:
if isinstance(obj, Account):
return self.create_contact(obj)
if isinstance(obj, Contact):
return obj
(_, addr) = self.get_contact_addr_and_name(obj)

8
python/src/deltachat/chat.py
View File

@@ -417,7 +417,13 @@ class Chat:
:raises ValueError: if contact could not be added
:returns: None
"""
contact = self.account.create_contact(obj)
from .contact import Contact
if isinstance(obj, Contact):
contact = obj
else:
contact = self.account.create_contact(obj)
ret = lib.dc_add_contact_to_chat(self.account._dc_context, self.id, contact.id)
if ret != 1:
raise ValueError(f"could not add contact {contact!r} to chat")

11
python/src/deltachat/events.py
View File

@@ -13,7 +13,6 @@ from .account import Account
from .capi import ffi, lib
from .cutil import from_optional_dc_charpointer
from .hookspec import account_hookimpl
from .message import map_system_message
def get_dc_event_name(integer, _DC_EVENTNAME_MAP={}):
@@ -304,21 +303,15 @@ class EventThread(threading.Thread):
elif name == "DC_EVENT_INCOMING_MSG":
msg = account.get_message_by_id(ffi_event.data2)
if msg is not None:
yield map_system_message(msg) or ("ac_incoming_message", {"message": msg})
yield ("ac_incoming_message", {"message": msg})
elif name == "DC_EVENT_MSGS_CHANGED":
if ffi_event.data2 != 0:
msg = account.get_message_by_id(ffi_event.data2)
if msg is not None:
if msg.is_outgoing():
res = map_system_message(msg)
if res and res[0].startswith("ac_member"):
yield res
yield "ac_outgoing_message", {"message": msg}
elif msg.is_in_fresh():
yield map_system_message(msg) or (
"ac_incoming_message",
{"message": msg},
)
yield "ac_incoming_message", {"message": msg}
elif name == "DC_EVENT_REACTIONS_CHANGED":
assert ffi_event.data1 > 0
msg = account.get_message_by_id(ffi_event.data2)

54
python/src/deltachat/message.py
View File

@@ -2,7 +2,6 @@
import json
import os
import re
from datetime import datetime, timezone
from typing import Optional, Union
@@ -504,56 +503,3 @@ def get_viewtype_code_from_name(view_type_name):
raise ValueError(
f"message typecode not found for {view_type_name!r}, available {list(_view_type_mapping.keys())!r}",
)
#
# some helper code for turning system messages into hook events
#
def map_system_message(msg):
if msg.is_system_message():
res = parse_system_add_remove(msg.text)
if not res:
return None
action, affected, actor = res
affected = msg.account.get_contact_by_addr(affected)
actor = None if actor == "me" else msg.account.get_contact_by_addr(actor)
d = {"chat": msg.chat, "contact": affected, "actor": actor, "message": msg}
return "ac_member_" + res[0], d
def extract_addr(text):
m = re.match(r".*\((.+@.+)\)", text)
if m:
text = m.group(1)
text = text.rstrip(".")
return text.strip()
def parse_system_add_remove(text):
"""return add/remove info from parsing the given system message text.
returns a (action, affected, actor) triple
"""
# You removed member a@b.
# You added member a@b.
# Member Me (x@y) removed by a@b.
# Member x@y added by a@b
# Member With space (tmp1@x.org) removed by tmp2@x.org.
# Member With space (tmp1@x.org) removed by Another member (tmp2@x.org).",
# Group left by some one (tmp1@x.org).
# Group left by tmp1@x.org.
text = text.lower()
m = re.match(r"member (.+) (removed|added) by (.+)", text)
if m:
affected, action, actor = m.groups()
return action, extract_addr(affected), extract_addr(actor)
m = re.match(r"you (removed|added) member (.+)", text)
if m:
action, affected = m.groups()
return action, extract_addr(affected), "me"
if text.startswith("group left by "):
addr = extract_addr(text[13:])
if addr:
return "removed", addr, addr

77
python/tests/test_0_complex_or_slow.py
View File

@@ -187,83 +187,6 @@ def test_qr_verified_group_and_chatting(acfactory, lp):
assert msg.is_encrypted()
def test_undecipherable_group(acfactory, lp):
"""Test how group messages that cannot be decrypted are
handled.
Group name is encrypted and plaintext subject is set to "..." in
this case, so we should assign the messages to existing chat
instead of creating a new one. Since there is no existing group
chat, the messages should be assigned to 1-1 chat with the sender
of the message.
"""
lp.sec("creating and configuring three accounts")
ac1, ac2, ac3 = acfactory.get_online_accounts(3)
acfactory.introduce_each_other([ac1, ac2, ac3])
lp.sec("ac3 reinstalls DC and generates a new key")
ac3.stop_io()
acfactory.remove_preconfigured_keys()
ac4 = acfactory.new_online_configuring_account(cloned_from=ac3)
acfactory.wait_configured(ac4)
# Create contacts to make sure incoming messages are not treated as contact requests
chat41 = ac4.create_chat(ac1)
chat42 = ac4.create_chat(ac2)
ac4.start_io()
ac4._evtracker.wait_idle_inbox_ready()
lp.sec("ac1: creating group chat with 2 other members")
chat = ac1.create_group_chat("title", contacts=[ac2, ac3])
lp.sec("ac1: send message to new group chat")
msg = chat.send_text("hello")
lp.sec("ac2: checking that the chat arrived correctly")
msg = ac2._evtracker.wait_next_incoming_message()
assert msg.text == "hello"
assert msg.is_encrypted(), "Message is not encrypted"
# ac4 cannot decrypt the message.
# Error message should be assigned to the chat with ac1.
lp.sec("ac4: checking that message is assigned to the sender chat")
error_msg = ac4._evtracker.wait_next_incoming_message()
assert error_msg.error # There is an error decrypting the message
assert error_msg.chat == chat41
lp.sec("ac2: sending a reply to the chat")
msg.chat.send_text("reply")
reply = ac1._evtracker.wait_next_incoming_message()
assert reply.text == "reply"
assert reply.is_encrypted(), "Reply is not encrypted"
lp.sec("ac4: checking that reply is assigned to ac2 chat")
error_reply = ac4._evtracker.wait_next_incoming_message()
assert error_reply.error # There is an error decrypting the message
assert error_reply.chat == chat42
# Test that ac4 replies to error messages don't appear in the
# group chat on ac1 and ac2.
lp.sec("ac4: replying to ac1 and ac2")
# Otherwise reply becomes a contact request.
chat41.send_text("I can't decrypt your message, ac1!")
chat42.send_text("I can't decrypt your message, ac2!")
msg = ac1._evtracker.wait_next_incoming_message()
assert msg.error is None
assert msg.text == "I can't decrypt your message, ac1!"
assert msg.is_encrypted(), "Message is not encrypted"
assert msg.chat == ac1.create_chat(ac3)
msg = ac2._evtracker.wait_next_incoming_message()
assert msg.error is None
assert msg.text == "I can't decrypt your message, ac2!"
assert msg.is_encrypted(), "Message is not encrypted"
assert msg.chat == ac2.create_chat(ac4)
def test_ephemeral_timer(acfactory, lp):
ac1, ac2 = acfactory.get_online_accounts(2)

124
python/tests/test_1_online.py
View File

@@ -767,7 +767,7 @@ def test_mdn_asymmetric(acfactory, lp):
assert len(list(ac1.direct_imap.conn.fetch(AND(seen=True)))) == 1
def test_send_and_receive_will_encrypt_decrypt(acfactory, lp):
def test_send_receive_encrypt(acfactory, lp):
ac1, ac2 = acfactory.get_online_accounts(2)
ac1.get_device_chat().mark_noticed()
@@ -798,12 +798,11 @@ def test_send_and_receive_will_encrypt_decrypt(acfactory, lp):
msg3.mark_seen()
assert not list(ac1.get_fresh_messages())
lp.sec("create group chat with two members, one of which has no encrypt state")
lp.sec("create group chat with two members")
chat = ac1.create_group_chat("encryption test")
chat.add_contact(ac2)
chat.add_contact(ac1.create_contact("notexisting@testrun.org"))
msg = chat.send_text("test not encrypt")
assert not msg.is_encrypted()
assert msg.is_encrypted()
ac1._evtracker.get_matching("DC_EVENT_SMTP_MESSAGE_SENT")
@@ -1139,9 +1138,9 @@ def test_import_export_online_all(acfactory, tmp_path, data, lp):
lp.sec("create some chat content")
some1_addr = some1.get_config("addr")
chat1 = ac1.create_contact(some1_addr, name="some1").create_chat()
chat1 = ac1.create_contact(some1).create_chat()
chat1.send_text("msg1")
assert len(ac1.get_contacts(query="some1")) == 1
assert len(ac1.get_contacts()) == 1
original_image_path = data.get_path("d.png")
chat1.send_image(original_image_path)
@@ -1153,7 +1152,7 @@ def test_import_export_online_all(acfactory, tmp_path, data, lp):
chat1.send_file(str(path))
def assert_account_is_proper(ac):
contacts = ac.get_contacts(query="some1")
contacts = ac.get_contacts()
assert len(contacts) == 1
contact2 = contacts[0]
assert contact2.addr == some1_addr
@@ -1288,79 +1287,6 @@ def test_set_get_contact_avatar(acfactory, data, lp):
assert msg6.get_sender_contact().get_profile_image() is None
def test_add_remove_member_remote_events(acfactory, lp):
ac1, ac2, ac3 = acfactory.get_online_accounts(3)
ac1_addr = ac1.get_config("addr")
ac3_addr = ac3.get_config("addr")
# activate local plugin for ac2
in_list = queue.Queue()
class EventHolder:
def __init__(self, **kwargs) -> None:
self.__dict__.update(kwargs)
class InPlugin:
@account_hookimpl
def ac_incoming_message(self, message):
# we immediately accept the sender because
# otherwise we won't see member_added contacts
message.create_chat()
@account_hookimpl
def ac_chat_modified(self, chat):
in_list.put(EventHolder(action="chat-modified", chat=chat))
@account_hookimpl
def ac_member_added(self, chat, contact, message):
in_list.put(EventHolder(action="added", chat=chat, contact=contact, message=message))
@account_hookimpl
def ac_member_removed(self, chat, contact, message):
in_list.put(EventHolder(action="removed", chat=chat, contact=contact, message=message))
ac2.add_account_plugin(InPlugin())
lp.sec("ac1: create group chat with ac2")
chat = ac1.create_group_chat("hello", contacts=[ac2])
lp.sec("ac1: send a message to group chat to promote the group")
chat.send_text("afterwards promoted")
ev = in_list.get()
assert ev.action == "chat-modified"
assert chat.is_promoted()
assert sorted(x.addr for x in chat.get_contacts()) == sorted(x.addr for x in ev.chat.get_contacts())
lp.sec("ac1: add address2")
# note that if the above create_chat() would not
# happen we would not receive a proper member_added event
contact2 = chat.add_contact(ac3)
ev = in_list.get()
assert ev.action == "chat-modified"
ev = in_list.get()
assert ev.action == "chat-modified"
ev = in_list.get()
assert ev.action == "added"
assert ev.message.get_sender_contact().addr == ac1_addr
assert ev.contact.addr == ac3_addr
lp.sec("ac1: remove address2")
chat.remove_contact(contact2)
ev = in_list.get()
assert ev.action == "chat-modified"
ev = in_list.get()
assert ev.action == "removed"
assert ev.contact.addr == contact2.addr
assert ev.message.get_sender_contact().addr == ac1_addr
lp.sec("ac1: remove ac2 contact from chat")
chat.remove_contact(ac2)
ev = in_list.get()
assert ev.action == "chat-modified"
ev = in_list.get()
assert ev.action == "removed"
assert ev.message.get_sender_contact().addr == ac1_addr
def test_system_group_msg_from_blocked_user(acfactory, lp):
"""
Tests that a blocked user removes you from a group.
@@ -1760,44 +1686,6 @@ def test_configure_error_msgs_invalid_server(acfactory):
assert "configuration" not in ev.data2.lower()
def test_name_changes(acfactory):
ac1, ac2 = acfactory.get_online_accounts(2)
ac1.set_config("displayname", "Account 1")
chat12 = acfactory.get_accepted_chat(ac1, ac2)
contact = None
def update_name():
"""Send a message from ac1 to ac2 to update the name"""
nonlocal contact
chat12.send_text("Hello")
msg = ac2._evtracker.wait_next_incoming_message()
contact = msg.get_sender_contact()
return contact.name
assert update_name() == "Account 1"
ac1.set_config("displayname", "Account 1 revision 2")
assert update_name() == "Account 1 revision 2"
# Explicitly rename contact on ac2 to "Renamed"
ac2.create_contact(contact, name="Renamed")
assert contact.name == "Renamed"
ev = ac2._evtracker.get_matching("DC_EVENT_CONTACTS_CHANGED")
assert ev.data1 == contact.id
# ac1 also renames itself into "Renamed"
assert update_name() == "Renamed"
ac1.set_config("displayname", "Renamed")
assert update_name() == "Renamed"
# Contact name was set to "Renamed" explicitly before,
# so it should not be changed.
ac1.set_config("displayname", "Renamed again")
updated_name = update_name()
assert updated_name == "Renamed"
def test_status(acfactory):
"""Test that status is transferred over the network."""
ac1, ac2 = acfactory.get_online_accounts(2)

187
python/tests/test_3_offline.py
View File

@@ -1,51 +1,11 @@
import os
import time
from datetime import datetime, timedelta, timezone
import pytest
import deltachat as dc
from deltachat.tracker import ImexFailed
from deltachat import Account, account_hookimpl, Message
@pytest.mark.parametrize(
("msgtext", "res"),
[
(
"Member Me (tmp1@x.org) removed by tmp2@x.org.",
("removed", "tmp1@x.org", "tmp2@x.org"),
),
(
"Member With space (tmp1@x.org) removed by tmp2@x.org.",
("removed", "tmp1@x.org", "tmp2@x.org"),
),
(
"Member With space (tmp1@x.org) removed by Another member (tmp2@x.org).",
("removed", "tmp1@x.org", "tmp2@x.org"),
),
(
"Member With space (tmp1@x.org) removed by me",
("removed", "tmp1@x.org", "me"),
),
(
"Group left by some one (tmp1@x.org).",
("removed", "tmp1@x.org", "tmp1@x.org"),
),
("Group left by tmp1@x.org.", ("removed", "tmp1@x.org", "tmp1@x.org")),
(
"Member tmp1@x.org added by tmp2@x.org.",
("added", "tmp1@x.org", "tmp2@x.org"),
),
("Member nothing bla bla", None),
("Another unknown system message", None),
],
)
def test_parse_system_add_remove(msgtext, res):
from deltachat.message import parse_system_add_remove
out = parse_system_add_remove(msgtext)
assert out == res
from deltachat import Account, Message
class TestOfflineAccountBasic:
@@ -177,14 +137,15 @@ class TestOfflineContact:
def test_get_contacts_and_delete(self, acfactory):
ac1 = acfactory.get_pseudo_configured_account()
contact1 = ac1.create_contact("some1@example.org", name="some1")
ac2 = acfactory.get_pseudo_configured_account()
contact1 = ac1.create_contact(ac2)
contacts = ac1.get_contacts()
assert len(contacts) == 1
assert contact1 in contacts
assert not ac1.get_contacts(query="some2")
assert ac1.get_contacts(query="some1")
assert not ac1.get_contacts(query="some1")
assert len(ac1.get_contacts(with_self=True)) == 2
assert contact1 in ac1.get_contacts()
assert ac1.delete_contact(contact1)
assert contact1 not in ac1.get_contacts()
@@ -199,9 +160,9 @@ class TestOfflineContact:
def test_create_chat_flexibility(self, acfactory):
ac1 = acfactory.get_pseudo_configured_account()
ac2 = acfactory.get_pseudo_configured_account()
chat1 = ac1.create_chat(ac2)
chat2 = ac1.create_chat(ac2.get_self_contact().addr)
assert chat1 == chat2
chat1 = ac1.create_chat(ac2) # This creates a key-contact chat
chat2 = ac1.create_chat(ac2.get_self_contact().addr) # This creates address-contact chat
assert chat1 != chat2
ac3 = acfactory.get_unconfigured_account()
with pytest.raises(ValueError):
ac1.create_chat(ac3)
@@ -259,17 +220,18 @@ class TestOfflineChat:
ac1 = acfactory.get_pseudo_configured_account()
ac2 = acfactory.get_pseudo_configured_account()
chat = ac1.create_group_chat(name="title1")
with pytest.raises(ValueError):
chat.add_contact(ac2.get_self_contact())
contact = chat.add_contact(ac2)
assert contact.addr == ac2.get_config("addr")
assert contact.name == ac2.get_config("displayname")
assert contact.account == ac1
chat.remove_contact(ac2)
def test_group_chat_creation(self, ac1):
contact1 = ac1.create_contact("some1@example.org", name="some1")
contact2 = ac1.create_contact("some2@example.org", name="some2")
def test_group_chat_creation(self, acfactory):
ac1 = acfactory.get_pseudo_configured_account()
ac2 = acfactory.get_pseudo_configured_account()
ac3 = acfactory.get_pseudo_configured_account()
contact1 = ac1.create_contact(ac2)
contact2 = ac1.create_contact(ac3)
chat = ac1.create_group_chat(name="title1", contacts=[contact1, contact2])
assert chat.get_name() == "title1"
assert contact1 in chat.get_contacts()
@@ -316,13 +278,14 @@ class TestOfflineChat:
qr = chat.get_join_qr()
assert ac2.check_qr(qr).is_ask_verifygroup
def test_removing_blocked_user_from_group(self, ac1, lp):
def test_removing_blocked_user_from_group(self, ac1, acfactory, lp):
"""
Test that blocked contact is not unblocked when removed from a group.
See https://github.com/deltachat/deltachat-core-rust/issues/2030
"""
lp.sec("Create a group chat with a contact")
contact = ac1.create_contact("some1@example.org")
ac2 = acfactory.get_pseudo_configured_account()
contact = ac1.create_contact(ac2)
group = ac1.create_group_chat("title", contacts=[contact])
group.send_text("First group message")
@@ -334,10 +297,6 @@ class TestOfflineChat:
group.remove_contact(contact)
assert contact.is_blocked()
lp.sec("ac1 adding blocked contact unblocks it")
group.add_contact(contact)
assert not contact.is_blocked()
def test_get_set_profile_image_simple(self, ac1, data):
chat = ac1.create_group_chat(name="title1")
p = data.get_path("d.png")
@@ -480,7 +439,8 @@ class TestOfflineChat:
backupdir = tmp_path / "backup"
backupdir.mkdir()
ac1 = acfactory.get_pseudo_configured_account()
chat = ac1.create_contact("some1 <some1@example.org>").create_chat()
ac_contact = acfactory.get_pseudo_configured_account()
chat = ac1.create_contact(ac_contact).create_chat()
# send a text message
msg = chat.send_text("msg1")
# send a binary file
@@ -495,10 +455,10 @@ class TestOfflineChat:
assert os.path.exists(path)
ac2 = acfactory.get_unconfigured_account()
ac2.import_all(path)
contacts = ac2.get_contacts(query="some1")
contacts = ac2.get_contacts()
assert len(contacts) == 1
contact2 = contacts[0]
assert contact2.addr == "some1@example.org"
assert contact2.addr == ac_contact.get_config("addr")
chat2 = contact2.create_chat()
messages = chat2.get_messages()
assert len(messages) == 2
@@ -511,8 +471,9 @@ class TestOfflineChat:
backupdir = tmp_path / "backup"
backupdir.mkdir()
ac1 = acfactory.get_pseudo_configured_account(passphrase=passphrase1)
ac2 = acfactory.get_pseudo_configured_account()
chat = ac1.create_contact("some1 <some1@example.org>").create_chat()
chat = ac1.create_contact(ac2).create_chat()
# send a text message
msg = chat.send_text("msg1")
# send a binary file
@@ -533,10 +494,10 @@ class TestOfflineChat:
ac2.import_all(path)
# check data integrity
contacts = ac2.get_contacts(query="some1")
contacts = ac2.get_contacts()
assert len(contacts) == 1
contact2 = contacts[0]
assert contact2.addr == "some1@example.org"
contact2_addr = contact2.addr
chat2 = contact2.create_chat()
messages = chat2.get_messages()
assert len(messages) == 2
@@ -550,10 +511,10 @@ class TestOfflineChat:
ac2.open(passphrase2)
# check data integrity
contacts = ac2.get_contacts(query="some1")
contacts = ac2.get_contacts()
assert len(contacts) == 1
contact2 = contacts[0]
assert contact2.addr == "some1@example.org"
assert contact2.addr == contact2_addr
chat2 = contact2.create_chat()
messages = chat2.get_messages()
assert len(messages) == 2
@@ -566,8 +527,9 @@ class TestOfflineChat:
backupdir = tmp_path / "backup"
backupdir.mkdir()
ac1 = acfactory.get_pseudo_configured_account()
ac_contact = acfactory.get_pseudo_configured_account()
chat = ac1.create_contact("some1 <some1@example.org>").create_chat()
chat = ac1.create_contact(ac_contact).create_chat()
# send a text message
msg = chat.send_text("msg1")
# send a binary file
@@ -589,10 +551,10 @@ class TestOfflineChat:
ac2.import_all(path, passphrase)
# check data integrity
contacts = ac2.get_contacts(query="some1")
contacts = ac2.get_contacts()
assert len(contacts) == 1
contact2 = contacts[0]
assert contact2.addr == "some1@example.org"
assert contact2.addr == ac_contact.get_config("addr")
chat2 = contact2.create_chat()
messages = chat2.get_messages()
assert len(messages) == 2
@@ -611,7 +573,8 @@ class TestOfflineChat:
backupdir.mkdir()
ac1 = acfactory.get_pseudo_configured_account()
chat = ac1.create_contact("some1 <some1@example.org>").create_chat()
ac_contact = acfactory.get_pseudo_configured_account()
chat = ac1.create_contact(ac_contact).create_chat()
# send a text message
msg = chat.send_text("msg1")
# send a binary file
@@ -634,10 +597,10 @@ class TestOfflineChat:
ac2.import_all(path, bak_passphrase)
# check data integrity
contacts = ac2.get_contacts(query="some1")
contacts = ac2.get_contacts()
assert len(contacts) == 1
contact2 = contacts[0]
assert contact2.addr == "some1@example.org"
assert contact2.addr == ac_contact.get_config("addr")
chat2 = contact2.create_chat()
messages = chat2.get_messages()
assert len(messages) == 2
@@ -651,10 +614,10 @@ class TestOfflineChat:
ac2.open(acct_passphrase)
# check data integrity
contacts = ac2.get_contacts(query="some1")
contacts = ac2.get_contacts()
assert len(contacts) == 1
contact2 = contacts[0]
assert contact2.addr == "some1@example.org"
assert contact2.addr == ac_contact.get_config("addr")
chat2 = contact2.create_chat()
messages = chat2.get_messages()
assert len(messages) == 2
@@ -681,78 +644,10 @@ class TestOfflineChat:
assert not res.is_ask_verifygroup()
assert res.contact_id == 10
def test_group_chat_many_members_add_remove(self, ac1, lp):
lp.sec("ac1: creating group chat with 10 other members")
chat = ac1.create_group_chat(name="title1")
# promote chat
chat.send_text("hello")
assert chat.is_promoted()
def test_audit_log_view_without_daymarker(self, acfactory, lp):
ac1 = acfactory.get_pseudo_configured_account()
ac2 = acfactory.get_pseudo_configured_account()
# activate local plugin
in_list = []
class InPlugin:
@account_hookimpl
def ac_member_added(self, chat, contact, actor):
in_list.append(("added", chat, contact, actor))
@account_hookimpl
def ac_member_removed(self, chat, contact, actor):
in_list.append(("removed", chat, contact, actor))
ac1.add_account_plugin(InPlugin())
# perform add contact many times
contacts = []
for i in range(10):
lp.sec("create contact")
contact = ac1.create_contact(f"some{i}@example.org")
contacts.append(contact)
lp.sec("add contact")
chat.add_contact(contact)
assert chat.num_contacts() == 11
# let's make sure the events perform plugin hooks
def wait_events(cond):
now = time.time()
while time.time() < now + 5:
if cond():
break
time.sleep(0.1)
else:
pytest.fail("failed to get events")
wait_events(lambda: len(in_list) == 10)
assert len(in_list) == 10
chat_contacts = chat.get_contacts()
for in_cmd, in_chat, in_contact, in_actor in in_list:
assert in_cmd == "added"
assert in_chat == chat
assert in_contact in chat_contacts
assert in_actor is None
chat_contacts.remove(in_contact)
assert chat_contacts[0].id == 1 # self contact
in_list[:] = []
lp.sec("ac1: removing two contacts and checking things are right")
chat.remove_contact(contacts[9])
chat.remove_contact(contacts[3])
assert chat.num_contacts() == 9
wait_events(lambda: len(in_list) == 2)
assert len(in_list) == 2
assert in_list[0][0] == "removed"
assert in_list[0][1] == chat
assert in_list[0][2] == contacts[9]
assert in_list[1][0] == "removed"
assert in_list[1][1] == chat
assert in_list[1][2] == contacts[3]
def test_audit_log_view_without_daymarker(self, ac1, lp):
lp.sec("ac1: test audit log (show only system messages)")
chat = ac1.create_group_chat(name="audit log sample data")
@@ -761,7 +656,7 @@ class TestOfflineChat:
assert chat.is_promoted()
lp.sec("create test data")
chat.add_contact(ac1.create_contact("some-1@example.org"))
chat.add_contact(ac2)
chat.set_name("audit log test group")
chat.send_text("a message in between")