From 91884998b980303a014880d2841b8cff31451b85 Mon Sep 17 00:00:00 2001 From: iphydf Date: Sat, 24 Feb 2024 23:54:36 +0000 Subject: [PATCH 1/6] test: Add test for conference invites. --- pytox/toxcore/tox.pyi | 6 +- pytox/toxcore/tox.pyx | 17 +- .../auto_tests/self_connection_status_test.py | 113 +------ test/auto_tests/test.py | 310 ++++++++++++++++++ 4 files changed, 330 insertions(+), 116 deletions(-) create mode 100644 test/auto_tests/test.py diff --git a/pytox/toxcore/tox.pyi b/pytox/toxcore/tox.pyi index ec528ce..8e42df0 100644 --- a/pytox/toxcore/tox.pyi +++ b/pytox/toxcore/tox.pyi @@ -1,7 +1,6 @@ # pytox.toxcore.tox from types import TracebackType from typing import Any -from typing import List from typing import Optional from typing import TypeVar import enum @@ -112,8 +111,8 @@ class Tox_Ptr: address: Any def bootstrap(self, host: str, port: int, public_key: bytes) -> bool: ... def conference_by_id(self, id: bytes) -> Tox_Conference_Number: ... + conference_chatlist: Any def conference_delete(self, conference_number: Tox_Conference_Number) -> None: ... - def conference_get_chatlist(self) -> List[Tox_Conference_Number]: ... def conference_get_id(self, conference_number: Tox_Conference_Number) -> bytes: ... def conference_get_offline_peer_count(self, conference_number: Tox_Conference_Number) -> int: ... def conference_get_peer_count(self, conference_number: Tox_Conference_Number) -> int: ... @@ -121,7 +120,7 @@ class Tox_Ptr: def conference_get_title(self, conference_number: Tox_Conference_Number) -> bytes: ... def conference_get_type(self, conference_number: Tox_Conference_Number) -> Tox_Conference_Type: ... def conference_invite(self, friend_number: Tox_Friend_Number, conference_number: Tox_Conference_Number) -> None: ... - def conference_join(self, conference_number: Tox_Conference_Number, cookie: bytes) -> Tox_Conference_Number: ... + def conference_join(self, friend_number: Tox_Friend_Number, cookie: bytes) -> Tox_Conference_Number: ... def conference_new(self) -> Tox_Conference_Number: ... def conference_offline_peer_get_last_active(self, conference_number: Tox_Conference_Number, offline_peer_number: Tox_Conference_Offline_Peer_Number) -> int: ... def conference_offline_peer_get_name(self, conference_number: Tox_Conference_Number, offline_peer_number: Tox_Conference_Offline_Peer_Number) -> bytes: ... @@ -150,6 +149,7 @@ class Tox_Ptr: def friend_get_status(self, friend_number: Tox_Friend_Number) -> Tox_User_Status: ... def friend_get_status_message(self, friend_number: Tox_Friend_Number) -> bytes: ... def friend_get_typing(self, friend_number: Tox_Friend_Number) -> bool: ... + friend_list: Any def friend_send_lossless_packet(self, friend_number: Tox_Friend_Number, data: bytes) -> None: ... def friend_send_lossy_packet(self, friend_number: Tox_Friend_Number, data: bytes) -> None: ... def friend_send_message(self, friend_number: Tox_Friend_Number, type_: Tox_Message_Type, message: bytes) -> Tox_Friend_Message_Id: ... diff --git a/pytox/toxcore/tox.pyx b/pytox/toxcore/tox.pyx index b0628e5..d5ede6b 100644 --- a/pytox/toxcore/tox.pyx +++ b/pytox/toxcore/tox.pyx @@ -591,6 +591,16 @@ cdef class Tox_Ptr: def friend_exists(self, friend_number: Tox_Friend_Number) -> bool: return tox_friend_exists(self._get(), friend_number) + @property + def friend_list(self) -> list[Tox_Friend_Number]: + cdef size_t size = tox_self_get_friend_list_size(self._get()) + cdef Tox_Friend_Number *data = malloc(size * sizeof(Tox_Friend_Number)) + try: + tox_self_get_friend_list(self._get(), data) + return [data[i] for i in range(size)] + finally: + free(data) + def friend_get_public_key(self, friend_number: Tox_Friend_Number) -> bytes: cdef size_t size = tox_public_key_size() cdef uint8_t *data = malloc(size * sizeof(uint8_t)) @@ -741,9 +751,9 @@ cdef class Tox_Ptr: if err: raise ApiException(Tox_Err_Conference_Invite(err)) - def conference_join(self, conference_number: Tox_Conference_Number, cookie: bytes) -> Tox_Conference_Number: + def conference_join(self, friend_number: Tox_Friend_Number, cookie: bytes) -> Tox_Conference_Number: cdef Tox_Err_Conference_Join err = TOX_ERR_CONFERENCE_JOIN_OK - cdef Tox_Conference_Number res = tox_conference_join(self._get(), conference_number, cookie, len(cookie), &err) + cdef Tox_Conference_Number res = tox_conference_join(self._get(), friend_number, cookie, len(cookie), &err) if err: raise ApiException(Tox_Err_Conference_Join(err)) return res @@ -860,7 +870,8 @@ cdef class Tox_Ptr: raise ApiException(Tox_Err_Conference_Peer_Query(err)) return res - def conference_get_chatlist(self) -> List[Tox_Conference_Number]: + @property + def conference_chatlist(self) -> list[Tox_Conference_Number]: cdef size_t size = tox_conference_get_chatlist_size(self._get()) cdef Tox_Conference_Number *data = malloc(size * sizeof(Tox_Conference_Number)) try: diff --git a/test/auto_tests/self_connection_status_test.py b/test/auto_tests/self_connection_status_test.py index 28b3fa0..da41087 100644 --- a/test/auto_tests/self_connection_status_test.py +++ b/test/auto_tests/self_connection_status_test.py @@ -2,7 +2,6 @@ import time import unittest from dataclasses import dataclass -from dataclasses import field from typing import Callable import pytox.toxcore.tox as core @@ -16,14 +15,7 @@ class TestException(Exception): @dataclass class FriendInfo: connection_status: core.Tox_Connection = core.TOX_CONNECTION_NONE - status: core.Tox_User_Status = core.TOX_USER_STATUS_NONE - status_message: bytes = b"" - name: bytes = b"" request_message: bytes = b"" - messages: list[tuple[core.Tox_Message_Type, bytes]] = field( - default_factory=list) - lossy_packets: list[bytes] = field(default_factory=list) - lossless_packets: list[bytes] = field(default_factory=list) class TestTox(core.Tox_Ptr): @@ -42,10 +34,6 @@ def handle_self_connection_status(self, self.connection_status_from_cb = connection_status raise TestException(connection_status) - def handle_friend_request(self, public_key: bytes, message: bytes) -> None: - friend_number = self.friend_add_norequest(public_key) - self.friends[friend_number].request_message = message - def handle_friend_connection_status( self, friend_number: int, @@ -53,30 +41,9 @@ def handle_friend_connection_status( ) -> None: self.friends[friend_number].connection_status = connection_status - def handle_friend_message(self, friend_number: int, - type_: core.Tox_Message_Type, - message: bytes) -> None: - self.friends[friend_number].messages.append((type_, message)) - - def handle_friend_lossy_packet(self, friend_number: int, - message: bytes) -> None: - self.friends[friend_number].lossy_packets.append(message) - raise Exception("OMG") - - def handle_friend_lossless_packet(self, friend_number: int, - message: bytes) -> None: - self.friends[friend_number].lossless_packets.append(message) - - def handle_friend_name(self, friend_number: int, name: bytes) -> None: - self.friends[friend_number].name = name - - def handle_friend_status(self, friend_number: int, - status: core.Tox_User_Status) -> None: - self.friends[friend_number].status = status - - def handle_friend_status_message(self, friend_number: int, - status_message: bytes) -> None: - self.friends[friend_number].status_message = status_message + def handle_friend_request(self, public_key: bytes, message: bytes) -> None: + friend_number = self.friend_add_norequest(public_key) + self.friends[friend_number].request_message = message class AutoTest(unittest.TestCase): @@ -149,80 +116,6 @@ def test_friend_add(self) -> None: b"lala lala lala la I'm Mr. Happy Face", ) - def test_friend_by_public_key(self) -> None: - with self.assertRaises(core.ApiException) as ex: - # We're not our own friend. - self.tox1.friend_by_public_key(self.tox1.public_key) - self.assertEqual(ex.exception.error, - core.TOX_ERR_FRIEND_BY_PUBLIC_KEY_NOT_FOUND) - - def test_send_message(self) -> None: - self._wait_for_friend_online() - self.tox1.friend_send_message(0, core.TOX_MESSAGE_TYPE_NORMAL, - b"hello there!") - friend = self.tox2.friends[self.tox2.friend_by_public_key( - self.tox1.public_key)] - self._iterate(100, lambda: not friend.messages) - self.assertEqual(friend.messages[0], - (core.TOX_MESSAGE_TYPE_NORMAL, b"hello there!")) - - # TODO(iphydf): This one doesn't pass. Investigate why. - # def test_send_lossy_packet(self) -> None: - # self._wait_for_friend_online() - # self.tox1.friend_send_lossy_packet(0, b"\xc0hello there!") - # friend = self.tox2.friends[self.tox2.friend_by_public_key( - # self.tox1.public_key)] - # self._iterate(100, lambda: not friend.lossy_packets) - # self.assertEqual(friend.lossy_packets[0], b"\xc0hello there!") - - def test_send_lossless_packet(self) -> None: - self._wait_for_friend_online() - self.tox1.friend_send_lossless_packet(0, b"\xa0general kenobi.") - friend = self.tox2.friends[self.tox2.friend_by_public_key( - self.tox1.public_key)] - self._iterate(100, lambda: not friend.lossless_packets) - self.assertEqual(friend.lossless_packets[0], b"\xa0general kenobi.") - - def test_status(self) -> None: - self._wait_for_friend_online() - self.assertEqual(self.tox1.status, core.TOX_USER_STATUS_NONE) - self.tox1.status = core.TOX_USER_STATUS_AWAY - self.assertEqual(self.tox1.status, core.TOX_USER_STATUS_AWAY) - friend_number = self.tox2.friend_by_public_key(self.tox1.public_key) - friend = self.tox2.friends[friend_number] - self._iterate(100, lambda: friend.status == core.TOX_USER_STATUS_NONE) - self.assertEqual(friend.status, core.TOX_USER_STATUS_AWAY) - self.assertEqual(self.tox2.friend_get_status(friend_number), - core.TOX_USER_STATUS_AWAY) - - def test_name(self) -> None: - self._wait_for_friend_online() - self.assertEqual(self.tox1.name, b"") - self.tox1.name = b"Now that's a name I haven't heard in a long time" - self.assertEqual(self.tox1.name, - b"Now that's a name I haven't heard in a long time") - friend_number = self.tox2.friend_by_public_key(self.tox1.public_key) - friend = self.tox2.friends[friend_number] - self._iterate(100, lambda: friend.name == b"") - self.assertEqual(friend.name, - b"Now that's a name I haven't heard in a long time") - self.assertEqual( - self.tox2.friend_get_name(friend_number), - b"Now that's a name I haven't heard in a long time", - ) - - def test_status_message(self) -> None: - self._wait_for_friend_online() - self.assertEqual(self.tox1.status_message, b"") - self.tox1.status_message = b"Python rocks!" - self.assertEqual(self.tox1.status_message, b"Python rocks!") - friend_number = self.tox2.friend_by_public_key(self.tox1.public_key) - friend = self.tox2.friends[friend_number] - self._iterate(100, lambda: friend.status_message == b"") - self.assertEqual(friend.status_message, b"Python rocks!") - self.assertEqual(self.tox2.friend_get_status_message(friend_number), - b"Python rocks!") - if __name__ == "__main__": unittest.main() diff --git a/test/auto_tests/test.py b/test/auto_tests/test.py new file mode 100644 index 0000000..3eca3e5 --- /dev/null +++ b/test/auto_tests/test.py @@ -0,0 +1,310 @@ +import collections +import time +import unittest +from dataclasses import dataclass +from dataclasses import field +from typing import Callable + +import pytox.toxcore.tox as core + + +@dataclass +class FriendInfo: + connection_status: core.Tox_Connection = core.TOX_CONNECTION_NONE + status: core.Tox_User_Status = core.TOX_USER_STATUS_NONE + status_message: bytes = b"" + name: bytes = b"" + request_message: bytes = b"" + typing: bool = False + messages: list[tuple[core.Tox_Message_Type, bytes]] = field( + default_factory=list) + # message_id -> read receipt + messages_sent: dict[int, bool] = field(default_factory=lambda: collections.defaultdict(bool)) + lossy_packets: list[bytes] = field(default_factory=list) + lossless_packets: list[bytes] = field(default_factory=list) + + +@dataclass +class ConferencePeerInfo: + name: bytes = b"" + + +@dataclass +class ConferenceInfo: + title: tuple[int, bytes] = (0, b"") + inviter: int = -1 + connected: bool = False + peers: dict[int, ConferencePeerInfo] = field(default_factory=lambda: collections.defaultdict(ConferencePeerInfo)) + messages: list[tuple[int, core.Tox_Message_Type, bytes]] = field(default_factory=list) + + +class TestTox(core.Tox_Ptr): + index: int + connection_status = core.TOX_CONNECTION_NONE + friends: dict[int, FriendInfo] + conferences: dict[int, ConferenceInfo] + + def __init__(self, index: int) -> None: + super(TestTox, self).__init__() + self.index = index + self.friends = collections.defaultdict(FriendInfo) + self.conferences = collections.defaultdict(ConferenceInfo) + + def handle_self_connection_status(self, + connection_status: core.Tox_Connection + ) -> None: + self.connection_status = connection_status + + def handle_friend_name(self, friend_number: int, name: bytes) -> None: + self.friends[friend_number].name = name + + def handle_friend_status_message(self, friend_number: int, + status_message: bytes) -> None: + self.friends[friend_number].status_message = status_message + + def handle_friend_status(self, friend_number: int, + status: core.Tox_User_Status) -> None: + self.friends[friend_number].status = status + + def handle_friend_connection_status( + self, + friend_number: int, + connection_status: core.Tox_Connection, + ) -> None: + self.friends[friend_number].connection_status = connection_status + + def handle_friend_typing(self, friend_number: int, typing: bool) -> None: + self.friends[friend_number].typing = typing + + def handle_friend_read_receipt(self, friend_number: int, message_id: int) -> None: + assert self.friends[friend_number].messages_sent[message_id] is False + self.friends[friend_number].messages_sent[message_id] = True + + def handle_friend_request(self, public_key: bytes, message: bytes) -> None: + friend_number = self.friend_add_norequest(public_key) + self.friends[friend_number].request_message = message + + def handle_friend_message(self, friend_number: int, + type_: core.Tox_Message_Type, + message: bytes) -> None: + self.friends[friend_number].messages.append((type_, message)) + + def handle_friend_lossy_packet(self, friend_number: int, + message: bytes) -> None: + self.friends[friend_number].lossy_packets.append(message) + raise Exception("OMG") + + def handle_friend_lossless_packet(self, friend_number: int, + message: bytes) -> None: + self.friends[friend_number].lossless_packets.append(message) + + def handle_file_recv_control(self, friend_number: int, file_number: int, + control: core.Tox_File_Control) -> None: + pass + + def handle_file_chunk_request(self, friend_number: int, file_number: int, + position: int, length: int) -> None: + pass + + def handle_file_recv(self, friend_number: int, file_number: int, kind: int, + file_size: int, filename: bytes) -> None: + pass + + def handle_file_recv_chunk(self, friend_number: int, file_number: int, + position: int, data: bytes) -> None: + pass + + def handle_conference_peer_name(self, conference_number: int, peer_number: int, + name: bytes) -> None: + self.conferences[conference_number].peers[peer_number].name = name + + def handle_conference_peer_list_changed(self, conference_number: int) -> None: + pass + + def handle_conference_invite(self, friend_number: int, type_: core.Tox_Conference_Type, + cookie: bytes) -> None: + assert(type_ == core.TOX_CONFERENCE_TYPE_TEXT) + conference_number = self.conference_join(friend_number, cookie) + self.conferences[conference_number].inviter = friend_number + + def handle_conference_connected(self, conference_number: int) -> None: + self.conferences[conference_number].connected = True + + def handle_conference_message(self, conference_number: int, peer_number: int, + type_: core.Tox_Message_Type, message: bytes) -> None: + self.conferences[conference_number].messages.append((peer_number, type_, message)) + + def handle_conference_title(self, conference_number: int, peer_number: int, title: bytes) -> None: + self.conferences[conference_number].title = (peer_number, title) + +class AutoTest(unittest.TestCase): + + # TODO(iphydf): Use per-test toxes. Use Tox_Time to speed this up. + tox1 = TestTox(1) + tox2 = TestTox(2) + tox3 = TestTox(3) + + def _iterate(self, max_iterate: int, cond: Callable[[], bool]) -> None: + for _ in range(0, max_iterate): + if not cond(): + return + self.tox1.iterate() + self.tox2.iterate() + self.tox3.iterate() + time.sleep(self.tox1.iteration_interval / 1000) + self.fail(f"condition not met after {max_iterate} iterations") + + def _wait_for_self_online(self) -> None: + def is_online() -> bool: + return bool( + self.tox1.connection_status == core.TOX_CONNECTION_NONE + or self.tox2.connection_status == core.TOX_CONNECTION_NONE + or self.tox3.connection_status == core.TOX_CONNECTION_NONE) + + # At most 20 seconds. + self._iterate( + 1000, + is_online, + ) + + def _wait_for_friend_online(self) -> None: + self._wait_for_self_online() + if not self.tox1.friend_list: + self.tox1.friend_add(self.tox2.address, + b"are you gonna be my best friend?") + self.tox2.friend_add(self.tox3.address, + b"lala lala lala la I'm Mr. Happy Face") + + def is_online() -> bool: + return (self.tox1.friends[0].connection_status == + core.TOX_CONNECTION_NONE + or self.tox2.friends[0].connection_status == + core.TOX_CONNECTION_NONE + or self.tox2.friends[1].connection_status == + core.TOX_CONNECTION_NONE + or self.tox3.friends[0].connection_status == + core.TOX_CONNECTION_NONE) + + # At most 5 seconds. + self._iterate(250, is_online) + + def test_friend_add(self) -> None: + self._wait_for_friend_online() + self.assertEqual( + self.tox2.friends[self.tox2.friend_by_public_key( + self.tox1.public_key)].request_message, + b"are you gonna be my best friend?", + ) + self.assertEqual( + self.tox3.friends[self.tox3.friend_by_public_key( + self.tox2.public_key)].request_message, + b"lala lala lala la I'm Mr. Happy Face", + ) + + def test_friend_by_public_key(self) -> None: + with self.assertRaises(core.ApiException) as ex: + # We're not our own friend. + self.tox1.friend_by_public_key(self.tox1.public_key) + self.assertEqual(ex.exception.error, + core.TOX_ERR_FRIEND_BY_PUBLIC_KEY_NOT_FOUND) + + def test_send_message(self) -> None: + self._wait_for_friend_online() + self.tox1.friend_send_message(0, core.TOX_MESSAGE_TYPE_NORMAL, + b"hello there!") + friend = self.tox2.friends[self.tox2.friend_by_public_key( + self.tox1.public_key)] + self._iterate(100, lambda: not friend.messages) + self.assertEqual(friend.messages[0], + (core.TOX_MESSAGE_TYPE_NORMAL, b"hello there!")) + + # TODO(iphydf): This one doesn't pass. Investigate why. + # def test_send_lossy_packet(self) -> None: + # self._wait_for_friend_online() + # self.tox1.friend_send_lossy_packet(0, b"\xc0hello there!") + # friend = self.tox2.friends[self.tox2.friend_by_public_key( + # self.tox1.public_key)] + # self._iterate(100, lambda: not friend.lossy_packets) + # self.assertEqual(friend.lossy_packets[0], b"\xc0hello there!") + + def test_send_lossless_packet(self) -> None: + self._wait_for_friend_online() + self.tox1.friend_send_lossless_packet(0, b"\xa0general kenobi.") + friend = self.tox2.friends[self.tox2.friend_by_public_key( + self.tox1.public_key)] + self._iterate(100, lambda: not friend.lossless_packets) + self.assertEqual(friend.lossless_packets[0], b"\xa0general kenobi.") + + def test_status(self) -> None: + self._wait_for_friend_online() + friend_number = self.tox2.friend_by_public_key(self.tox1.public_key) + friend = self.tox2.friends[friend_number] + self.assertEqual(self.tox1.status, core.TOX_USER_STATUS_NONE) + self.tox1.status = core.TOX_USER_STATUS_AWAY + self.assertEqual(self.tox1.status, core.TOX_USER_STATUS_AWAY) + self._iterate(100, lambda: friend.status == core.TOX_USER_STATUS_NONE) + self.assertEqual(friend.status, core.TOX_USER_STATUS_AWAY) + self.assertEqual(self.tox2.friend_get_status(friend_number), + core.TOX_USER_STATUS_AWAY) + + def test_name(self) -> None: + self._wait_for_friend_online() + friend_number = self.tox2.friend_by_public_key(self.tox1.public_key) + friend = self.tox2.friends[friend_number] + self.assertEqual(self.tox1.name, b"") + self.tox1.name = b"Now that's a name I haven't heard in a long time" + self.assertEqual(self.tox1.name, + b"Now that's a name I haven't heard in a long time") + self._iterate(100, lambda: friend.name == b"") + self.assertEqual(friend.name, + b"Now that's a name I haven't heard in a long time") + self.assertEqual( + self.tox2.friend_get_name(friend_number), + b"Now that's a name I haven't heard in a long time", + ) + + def test_status_message(self) -> None: + self._wait_for_friend_online() + friend_number = self.tox2.friend_by_public_key(self.tox1.public_key) + friend = self.tox2.friends[friend_number] + self.assertEqual(self.tox1.status_message, b"") + self.tox1.status_message = b"Python rocks!" + self.assertEqual(self.tox1.status_message, b"Python rocks!") + self._iterate(100, lambda: friend.status_message == b"") + self.assertEqual(friend.status_message, b"Python rocks!") + self.assertEqual(self.tox2.friend_get_status_message(friend_number), + b"Python rocks!") + + def test_typing(self) -> None: + self._wait_for_friend_online() + friend_number = self.tox2.friend_by_public_key(self.tox1.public_key) + friend = self.tox2.friends[friend_number] + self.assertFalse(friend.typing) + self.assertFalse(self.tox2.friend_get_typing(friend_number)) + self.tox1.set_typing(0, True) + self._iterate(100, lambda: not friend.typing) + self.assertTrue(self.tox2.friend_get_typing(friend_number)) + + def test_read_receipt(self) -> None: + self._wait_for_friend_online() + friend_number = self.tox1.friend_by_public_key(self.tox2.public_key) + friend = self.tox1.friends[friend_number] + msg_id = self.tox1.friend_send_message( + friend_number, core.TOX_MESSAGE_TYPE_NORMAL, b"hello there!") + self.assertFalse(friend.messages_sent[msg_id]) + self._iterate(100, lambda: not friend.messages_sent[msg_id]) + + def test_conference_invite(self) -> None: + self._wait_for_friend_online() + # tox1 creates conference. + cnum = self.tox1.conference_new() + self.assertListEqual(self.tox1.conference_chatlist, [cnum]) + # tox1 invites tox2 to conference. + self.tox1.conference_invite(self.tox1.friend_by_public_key(self.tox2.public_key), cnum) + # wait for tox2 to join conference. + self._iterate(100, lambda: not self.tox2.conference_chatlist) + self.assertEqual(len(self.tox2.conference_chatlist), 1) + + +if __name__ == "__main__": + unittest.main() From 0dad81463caf36b0116ac648be12f90b194022f8 Mon Sep 17 00:00:00 2001 From: "Restyled.io" Date: Sat, 24 Feb 2024 23:55:20 +0000 Subject: [PATCH 2/6] Restyled by autopep8 --- test/auto_tests/test.py | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/test/auto_tests/test.py b/test/auto_tests/test.py index 3eca3e5..40e7da9 100644 --- a/test/auto_tests/test.py +++ b/test/auto_tests/test.py @@ -19,7 +19,8 @@ class FriendInfo: messages: list[tuple[core.Tox_Message_Type, bytes]] = field( default_factory=list) # message_id -> read receipt - messages_sent: dict[int, bool] = field(default_factory=lambda: collections.defaultdict(bool)) + messages_sent: dict[int, bool] = field( + default_factory=lambda: collections.defaultdict(bool)) lossy_packets: list[bytes] = field(default_factory=list) lossless_packets: list[bytes] = field(default_factory=list) @@ -34,8 +35,10 @@ class ConferenceInfo: title: tuple[int, bytes] = (0, b"") inviter: int = -1 connected: bool = False - peers: dict[int, ConferencePeerInfo] = field(default_factory=lambda: collections.defaultdict(ConferencePeerInfo)) - messages: list[tuple[int, core.Tox_Message_Type, bytes]] = field(default_factory=list) + peers: dict[int, ConferencePeerInfo] = field( + default_factory=lambda: collections.defaultdict(ConferencePeerInfo)) + messages: list[tuple[int, core.Tox_Message_Type, bytes] + ] = field(default_factory=list) class TestTox(core.Tox_Ptr): @@ -132,11 +135,13 @@ def handle_conference_connected(self, conference_number: int) -> None: def handle_conference_message(self, conference_number: int, peer_number: int, type_: core.Tox_Message_Type, message: bytes) -> None: - self.conferences[conference_number].messages.append((peer_number, type_, message)) + self.conferences[conference_number].messages.append( + (peer_number, type_, message)) def handle_conference_title(self, conference_number: int, peer_number: int, title: bytes) -> None: self.conferences[conference_number].title = (peer_number, title) + class AutoTest(unittest.TestCase): # TODO(iphydf): Use per-test toxes. Use Tox_Time to speed this up. @@ -171,9 +176,9 @@ def _wait_for_friend_online(self) -> None: self._wait_for_self_online() if not self.tox1.friend_list: self.tox1.friend_add(self.tox2.address, - b"are you gonna be my best friend?") + b"are you gonna be my best friend?") self.tox2.friend_add(self.tox3.address, - b"lala lala lala la I'm Mr. Happy Face") + b"lala lala lala la I'm Mr. Happy Face") def is_online() -> bool: return (self.tox1.friends[0].connection_status == @@ -300,7 +305,8 @@ def test_conference_invite(self) -> None: cnum = self.tox1.conference_new() self.assertListEqual(self.tox1.conference_chatlist, [cnum]) # tox1 invites tox2 to conference. - self.tox1.conference_invite(self.tox1.friend_by_public_key(self.tox2.public_key), cnum) + self.tox1.conference_invite( + self.tox1.friend_by_public_key(self.tox2.public_key), cnum) # wait for tox2 to join conference. self._iterate(100, lambda: not self.tox2.conference_chatlist) self.assertEqual(len(self.tox2.conference_chatlist), 1) From a13ef31803a0fd1147170621023c9bcc0fd959b0 Mon Sep 17 00:00:00 2001 From: "Restyled.io" Date: Sat, 24 Feb 2024 23:55:30 +0000 Subject: [PATCH 3/6] Restyled by black --- .../auto_tests/self_connection_status_test.py | 49 +++-- test/auto_tests/test.py | 182 ++++++++++-------- 2 files changed, 125 insertions(+), 106 deletions(-) diff --git a/test/auto_tests/self_connection_status_test.py b/test/auto_tests/self_connection_status_test.py index da41087..6023713 100644 --- a/test/auto_tests/self_connection_status_test.py +++ b/test/auto_tests/self_connection_status_test.py @@ -28,16 +28,14 @@ def __init__(self, index: int) -> None: self.index = index self.friends = collections.defaultdict(FriendInfo) - def handle_self_connection_status(self, - connection_status: core.Tox_Connection - ) -> None: + def handle_self_connection_status( + self, connection_status: core.Tox_Connection + ) -> None: self.connection_status_from_cb = connection_status raise TestException(connection_status) def handle_friend_connection_status( - self, - friend_number: int, - connection_status: core.Tox_Connection, + self, friend_number: int, connection_status: core.Tox_Connection, ) -> None: self.friends[friend_number].connection_status = connection_status @@ -68,24 +66,22 @@ def is_online() -> bool: return bool( self.tox1.connection_status == core.TOX_CONNECTION_NONE or self.tox2.connection_status == core.TOX_CONNECTION_NONE - or self.tox3.connection_status == core.TOX_CONNECTION_NONE) + or self.tox3.connection_status == core.TOX_CONNECTION_NONE + ) # At most 20 seconds. self._iterate( - 1000, - is_online, + 1000, is_online, ) def _wait_for_friend_online(self) -> None: def is_online() -> bool: - return (self.tox1.friends[0].connection_status == - core.TOX_CONNECTION_NONE - or self.tox2.friends[0].connection_status == - core.TOX_CONNECTION_NONE - or self.tox2.friends[1].connection_status == - core.TOX_CONNECTION_NONE - or self.tox3.friends[0].connection_status == - core.TOX_CONNECTION_NONE) + return ( + self.tox1.friends[0].connection_status == core.TOX_CONNECTION_NONE + or self.tox2.friends[0].connection_status == core.TOX_CONNECTION_NONE + or self.tox2.friends[1].connection_status == core.TOX_CONNECTION_NONE + or self.tox3.friends[0].connection_status == core.TOX_CONNECTION_NONE + ) # At most 5 seconds. self._iterate(250, is_online) @@ -95,24 +91,23 @@ def test_connection_status_cb(self) -> None: # Test that exceptions can pass through C code. with self.assertRaises(TestException) as ex: self._wait_for_self_online() - self.assertEqual(tox.connection_status, - tox.connection_status_from_cb) + self.assertEqual(tox.connection_status, tox.connection_status_from_cb) self.assertEqual(ex.exception.status, tox.connection_status) def test_friend_add(self) -> None: - self.tox1.friend_add(self.tox2.address, - b"are you gonna be my best friend?") - self.tox2.friend_add(self.tox3.address, - b"lala lala lala la I'm Mr. Happy Face") + self.tox1.friend_add(self.tox2.address, b"are you gonna be my best friend?") + self.tox2.friend_add(self.tox3.address, b"lala lala lala la I'm Mr. Happy Face") self._wait_for_friend_online() self.assertEqual( - self.tox2.friends[self.tox2.friend_by_public_key( - self.tox1.public_key)].request_message, + self.tox2.friends[ + self.tox2.friend_by_public_key(self.tox1.public_key) + ].request_message, b"are you gonna be my best friend?", ) self.assertEqual( - self.tox3.friends[self.tox3.friend_by_public_key( - self.tox2.public_key)].request_message, + self.tox3.friends[ + self.tox3.friend_by_public_key(self.tox2.public_key) + ].request_message, b"lala lala lala la I'm Mr. Happy Face", ) diff --git a/test/auto_tests/test.py b/test/auto_tests/test.py index 40e7da9..3ccc628 100644 --- a/test/auto_tests/test.py +++ b/test/auto_tests/test.py @@ -16,11 +16,11 @@ class FriendInfo: name: bytes = b"" request_message: bytes = b"" typing: bool = False - messages: list[tuple[core.Tox_Message_Type, bytes]] = field( - default_factory=list) + messages: list[tuple[core.Tox_Message_Type, bytes]] = field(default_factory=list) # message_id -> read receipt messages_sent: dict[int, bool] = field( - default_factory=lambda: collections.defaultdict(bool)) + default_factory=lambda: collections.defaultdict(bool) + ) lossy_packets: list[bytes] = field(default_factory=list) lossless_packets: list[bytes] = field(default_factory=list) @@ -36,9 +36,11 @@ class ConferenceInfo: inviter: int = -1 connected: bool = False peers: dict[int, ConferencePeerInfo] = field( - default_factory=lambda: collections.defaultdict(ConferencePeerInfo)) - messages: list[tuple[int, core.Tox_Message_Type, bytes] - ] = field(default_factory=list) + default_factory=lambda: collections.defaultdict(ConferencePeerInfo) + ) + messages: list[tuple[int, core.Tox_Message_Type, bytes]] = field( + default_factory=list + ) class TestTox(core.Tox_Ptr): @@ -53,26 +55,26 @@ def __init__(self, index: int) -> None: self.friends = collections.defaultdict(FriendInfo) self.conferences = collections.defaultdict(ConferenceInfo) - def handle_self_connection_status(self, - connection_status: core.Tox_Connection - ) -> None: + def handle_self_connection_status( + self, connection_status: core.Tox_Connection + ) -> None: self.connection_status = connection_status def handle_friend_name(self, friend_number: int, name: bytes) -> None: self.friends[friend_number].name = name - def handle_friend_status_message(self, friend_number: int, - status_message: bytes) -> None: + def handle_friend_status_message( + self, friend_number: int, status_message: bytes + ) -> None: self.friends[friend_number].status_message = status_message - def handle_friend_status(self, friend_number: int, - status: core.Tox_User_Status) -> None: + def handle_friend_status( + self, friend_number: int, status: core.Tox_User_Status + ) -> None: self.friends[friend_number].status = status def handle_friend_connection_status( - self, - friend_number: int, - connection_status: core.Tox_Connection, + self, friend_number: int, connection_status: core.Tox_Connection, ) -> None: self.friends[friend_number].connection_status = connection_status @@ -87,58 +89,75 @@ def handle_friend_request(self, public_key: bytes, message: bytes) -> None: friend_number = self.friend_add_norequest(public_key) self.friends[friend_number].request_message = message - def handle_friend_message(self, friend_number: int, - type_: core.Tox_Message_Type, - message: bytes) -> None: + def handle_friend_message( + self, friend_number: int, type_: core.Tox_Message_Type, message: bytes + ) -> None: self.friends[friend_number].messages.append((type_, message)) - def handle_friend_lossy_packet(self, friend_number: int, - message: bytes) -> None: + def handle_friend_lossy_packet(self, friend_number: int, message: bytes) -> None: self.friends[friend_number].lossy_packets.append(message) raise Exception("OMG") - def handle_friend_lossless_packet(self, friend_number: int, - message: bytes) -> None: + def handle_friend_lossless_packet(self, friend_number: int, message: bytes) -> None: self.friends[friend_number].lossless_packets.append(message) - def handle_file_recv_control(self, friend_number: int, file_number: int, - control: core.Tox_File_Control) -> None: + def handle_file_recv_control( + self, friend_number: int, file_number: int, control: core.Tox_File_Control + ) -> None: pass - def handle_file_chunk_request(self, friend_number: int, file_number: int, - position: int, length: int) -> None: + def handle_file_chunk_request( + self, friend_number: int, file_number: int, position: int, length: int + ) -> None: pass - def handle_file_recv(self, friend_number: int, file_number: int, kind: int, - file_size: int, filename: bytes) -> None: + def handle_file_recv( + self, + friend_number: int, + file_number: int, + kind: int, + file_size: int, + filename: bytes, + ) -> None: pass - def handle_file_recv_chunk(self, friend_number: int, file_number: int, - position: int, data: bytes) -> None: + def handle_file_recv_chunk( + self, friend_number: int, file_number: int, position: int, data: bytes + ) -> None: pass - def handle_conference_peer_name(self, conference_number: int, peer_number: int, - name: bytes) -> None: + def handle_conference_peer_name( + self, conference_number: int, peer_number: int, name: bytes + ) -> None: self.conferences[conference_number].peers[peer_number].name = name def handle_conference_peer_list_changed(self, conference_number: int) -> None: pass - def handle_conference_invite(self, friend_number: int, type_: core.Tox_Conference_Type, - cookie: bytes) -> None: - assert(type_ == core.TOX_CONFERENCE_TYPE_TEXT) + def handle_conference_invite( + self, friend_number: int, type_: core.Tox_Conference_Type, cookie: bytes + ) -> None: + assert type_ == core.TOX_CONFERENCE_TYPE_TEXT conference_number = self.conference_join(friend_number, cookie) self.conferences[conference_number].inviter = friend_number def handle_conference_connected(self, conference_number: int) -> None: self.conferences[conference_number].connected = True - def handle_conference_message(self, conference_number: int, peer_number: int, - type_: core.Tox_Message_Type, message: bytes) -> None: + def handle_conference_message( + self, + conference_number: int, + peer_number: int, + type_: core.Tox_Message_Type, + message: bytes, + ) -> None: self.conferences[conference_number].messages.append( - (peer_number, type_, message)) + (peer_number, type_, message) + ) - def handle_conference_title(self, conference_number: int, peer_number: int, title: bytes) -> None: + def handle_conference_title( + self, conference_number: int, peer_number: int, title: bytes + ) -> None: self.conferences[conference_number].title = (peer_number, title) @@ -164,31 +183,29 @@ def is_online() -> bool: return bool( self.tox1.connection_status == core.TOX_CONNECTION_NONE or self.tox2.connection_status == core.TOX_CONNECTION_NONE - or self.tox3.connection_status == core.TOX_CONNECTION_NONE) + or self.tox3.connection_status == core.TOX_CONNECTION_NONE + ) # At most 20 seconds. self._iterate( - 1000, - is_online, + 1000, is_online, ) def _wait_for_friend_online(self) -> None: self._wait_for_self_online() if not self.tox1.friend_list: - self.tox1.friend_add(self.tox2.address, - b"are you gonna be my best friend?") - self.tox2.friend_add(self.tox3.address, - b"lala lala lala la I'm Mr. Happy Face") + self.tox1.friend_add(self.tox2.address, b"are you gonna be my best friend?") + self.tox2.friend_add( + self.tox3.address, b"lala lala lala la I'm Mr. Happy Face" + ) def is_online() -> bool: - return (self.tox1.friends[0].connection_status == - core.TOX_CONNECTION_NONE - or self.tox2.friends[0].connection_status == - core.TOX_CONNECTION_NONE - or self.tox2.friends[1].connection_status == - core.TOX_CONNECTION_NONE - or self.tox3.friends[0].connection_status == - core.TOX_CONNECTION_NONE) + return ( + self.tox1.friends[0].connection_status == core.TOX_CONNECTION_NONE + or self.tox2.friends[0].connection_status == core.TOX_CONNECTION_NONE + or self.tox2.friends[1].connection_status == core.TOX_CONNECTION_NONE + or self.tox3.friends[0].connection_status == core.TOX_CONNECTION_NONE + ) # At most 5 seconds. self._iterate(250, is_online) @@ -196,13 +213,15 @@ def is_online() -> bool: def test_friend_add(self) -> None: self._wait_for_friend_online() self.assertEqual( - self.tox2.friends[self.tox2.friend_by_public_key( - self.tox1.public_key)].request_message, + self.tox2.friends[ + self.tox2.friend_by_public_key(self.tox1.public_key) + ].request_message, b"are you gonna be my best friend?", ) self.assertEqual( - self.tox3.friends[self.tox3.friend_by_public_key( - self.tox2.public_key)].request_message, + self.tox3.friends[ + self.tox3.friend_by_public_key(self.tox2.public_key) + ].request_message, b"lala lala lala la I'm Mr. Happy Face", ) @@ -210,18 +229,18 @@ def test_friend_by_public_key(self) -> None: with self.assertRaises(core.ApiException) as ex: # We're not our own friend. self.tox1.friend_by_public_key(self.tox1.public_key) - self.assertEqual(ex.exception.error, - core.TOX_ERR_FRIEND_BY_PUBLIC_KEY_NOT_FOUND) + self.assertEqual( + ex.exception.error, core.TOX_ERR_FRIEND_BY_PUBLIC_KEY_NOT_FOUND + ) def test_send_message(self) -> None: self._wait_for_friend_online() - self.tox1.friend_send_message(0, core.TOX_MESSAGE_TYPE_NORMAL, - b"hello there!") - friend = self.tox2.friends[self.tox2.friend_by_public_key( - self.tox1.public_key)] + self.tox1.friend_send_message(0, core.TOX_MESSAGE_TYPE_NORMAL, b"hello there!") + friend = self.tox2.friends[self.tox2.friend_by_public_key(self.tox1.public_key)] self._iterate(100, lambda: not friend.messages) - self.assertEqual(friend.messages[0], - (core.TOX_MESSAGE_TYPE_NORMAL, b"hello there!")) + self.assertEqual( + friend.messages[0], (core.TOX_MESSAGE_TYPE_NORMAL, b"hello there!") + ) # TODO(iphydf): This one doesn't pass. Investigate why. # def test_send_lossy_packet(self) -> None: @@ -235,8 +254,7 @@ def test_send_message(self) -> None: def test_send_lossless_packet(self) -> None: self._wait_for_friend_online() self.tox1.friend_send_lossless_packet(0, b"\xa0general kenobi.") - friend = self.tox2.friends[self.tox2.friend_by_public_key( - self.tox1.public_key)] + friend = self.tox2.friends[self.tox2.friend_by_public_key(self.tox1.public_key)] self._iterate(100, lambda: not friend.lossless_packets) self.assertEqual(friend.lossless_packets[0], b"\xa0general kenobi.") @@ -249,8 +267,9 @@ def test_status(self) -> None: self.assertEqual(self.tox1.status, core.TOX_USER_STATUS_AWAY) self._iterate(100, lambda: friend.status == core.TOX_USER_STATUS_NONE) self.assertEqual(friend.status, core.TOX_USER_STATUS_AWAY) - self.assertEqual(self.tox2.friend_get_status(friend_number), - core.TOX_USER_STATUS_AWAY) + self.assertEqual( + self.tox2.friend_get_status(friend_number), core.TOX_USER_STATUS_AWAY + ) def test_name(self) -> None: self._wait_for_friend_online() @@ -258,11 +277,13 @@ def test_name(self) -> None: friend = self.tox2.friends[friend_number] self.assertEqual(self.tox1.name, b"") self.tox1.name = b"Now that's a name I haven't heard in a long time" - self.assertEqual(self.tox1.name, - b"Now that's a name I haven't heard in a long time") + self.assertEqual( + self.tox1.name, b"Now that's a name I haven't heard in a long time" + ) self._iterate(100, lambda: friend.name == b"") - self.assertEqual(friend.name, - b"Now that's a name I haven't heard in a long time") + self.assertEqual( + friend.name, b"Now that's a name I haven't heard in a long time" + ) self.assertEqual( self.tox2.friend_get_name(friend_number), b"Now that's a name I haven't heard in a long time", @@ -277,8 +298,9 @@ def test_status_message(self) -> None: self.assertEqual(self.tox1.status_message, b"Python rocks!") self._iterate(100, lambda: friend.status_message == b"") self.assertEqual(friend.status_message, b"Python rocks!") - self.assertEqual(self.tox2.friend_get_status_message(friend_number), - b"Python rocks!") + self.assertEqual( + self.tox2.friend_get_status_message(friend_number), b"Python rocks!" + ) def test_typing(self) -> None: self._wait_for_friend_online() @@ -295,7 +317,8 @@ def test_read_receipt(self) -> None: friend_number = self.tox1.friend_by_public_key(self.tox2.public_key) friend = self.tox1.friends[friend_number] msg_id = self.tox1.friend_send_message( - friend_number, core.TOX_MESSAGE_TYPE_NORMAL, b"hello there!") + friend_number, core.TOX_MESSAGE_TYPE_NORMAL, b"hello there!" + ) self.assertFalse(friend.messages_sent[msg_id]) self._iterate(100, lambda: not friend.messages_sent[msg_id]) @@ -306,7 +329,8 @@ def test_conference_invite(self) -> None: self.assertListEqual(self.tox1.conference_chatlist, [cnum]) # tox1 invites tox2 to conference. self.tox1.conference_invite( - self.tox1.friend_by_public_key(self.tox2.public_key), cnum) + self.tox1.friend_by_public_key(self.tox2.public_key), cnum + ) # wait for tox2 to join conference. self._iterate(100, lambda: not self.tox2.conference_chatlist) self.assertEqual(len(self.tox2.conference_chatlist), 1) From a554731ff67fae41c5ddc9afdc981d9f8f0d4742 Mon Sep 17 00:00:00 2001 From: "Restyled.io" Date: Sat, 24 Feb 2024 23:55:40 +0000 Subject: [PATCH 4/6] Restyled by isort --- test/auto_tests/test.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/test/auto_tests/test.py b/test/auto_tests/test.py index 3ccc628..17e8f25 100644 --- a/test/auto_tests/test.py +++ b/test/auto_tests/test.py @@ -1,8 +1,7 @@ import collections import time import unittest -from dataclasses import dataclass -from dataclasses import field +from dataclasses import dataclass, field from typing import Callable import pytox.toxcore.tox as core From 42f5dcc801b6781e8bdb62039a74b20392af51be Mon Sep 17 00:00:00 2001 From: "Restyled.io" Date: Sat, 24 Feb 2024 23:55:42 +0000 Subject: [PATCH 5/6] Restyled by reorder-python-imports --- test/auto_tests/test.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/test/auto_tests/test.py b/test/auto_tests/test.py index 17e8f25..3ccc628 100644 --- a/test/auto_tests/test.py +++ b/test/auto_tests/test.py @@ -1,7 +1,8 @@ import collections import time import unittest -from dataclasses import dataclass, field +from dataclasses import dataclass +from dataclasses import field from typing import Callable import pytox.toxcore.tox as core From e9aa5809080fd25ac0c5605488d5a7ef04b705ee Mon Sep 17 00:00:00 2001 From: "Restyled.io" Date: Sat, 24 Feb 2024 23:55:52 +0000 Subject: [PATCH 6/6] Restyled by yapf --- .../auto_tests/self_connection_status_test.py | 49 +++-- test/auto_tests/test.py | 192 +++++++++--------- 2 files changed, 119 insertions(+), 122 deletions(-) diff --git a/test/auto_tests/self_connection_status_test.py b/test/auto_tests/self_connection_status_test.py index 6023713..da41087 100644 --- a/test/auto_tests/self_connection_status_test.py +++ b/test/auto_tests/self_connection_status_test.py @@ -28,14 +28,16 @@ def __init__(self, index: int) -> None: self.index = index self.friends = collections.defaultdict(FriendInfo) - def handle_self_connection_status( - self, connection_status: core.Tox_Connection - ) -> None: + def handle_self_connection_status(self, + connection_status: core.Tox_Connection + ) -> None: self.connection_status_from_cb = connection_status raise TestException(connection_status) def handle_friend_connection_status( - self, friend_number: int, connection_status: core.Tox_Connection, + self, + friend_number: int, + connection_status: core.Tox_Connection, ) -> None: self.friends[friend_number].connection_status = connection_status @@ -66,22 +68,24 @@ def is_online() -> bool: return bool( self.tox1.connection_status == core.TOX_CONNECTION_NONE or self.tox2.connection_status == core.TOX_CONNECTION_NONE - or self.tox3.connection_status == core.TOX_CONNECTION_NONE - ) + or self.tox3.connection_status == core.TOX_CONNECTION_NONE) # At most 20 seconds. self._iterate( - 1000, is_online, + 1000, + is_online, ) def _wait_for_friend_online(self) -> None: def is_online() -> bool: - return ( - self.tox1.friends[0].connection_status == core.TOX_CONNECTION_NONE - or self.tox2.friends[0].connection_status == core.TOX_CONNECTION_NONE - or self.tox2.friends[1].connection_status == core.TOX_CONNECTION_NONE - or self.tox3.friends[0].connection_status == core.TOX_CONNECTION_NONE - ) + return (self.tox1.friends[0].connection_status == + core.TOX_CONNECTION_NONE + or self.tox2.friends[0].connection_status == + core.TOX_CONNECTION_NONE + or self.tox2.friends[1].connection_status == + core.TOX_CONNECTION_NONE + or self.tox3.friends[0].connection_status == + core.TOX_CONNECTION_NONE) # At most 5 seconds. self._iterate(250, is_online) @@ -91,23 +95,24 @@ def test_connection_status_cb(self) -> None: # Test that exceptions can pass through C code. with self.assertRaises(TestException) as ex: self._wait_for_self_online() - self.assertEqual(tox.connection_status, tox.connection_status_from_cb) + self.assertEqual(tox.connection_status, + tox.connection_status_from_cb) self.assertEqual(ex.exception.status, tox.connection_status) def test_friend_add(self) -> None: - self.tox1.friend_add(self.tox2.address, b"are you gonna be my best friend?") - self.tox2.friend_add(self.tox3.address, b"lala lala lala la I'm Mr. Happy Face") + self.tox1.friend_add(self.tox2.address, + b"are you gonna be my best friend?") + self.tox2.friend_add(self.tox3.address, + b"lala lala lala la I'm Mr. Happy Face") self._wait_for_friend_online() self.assertEqual( - self.tox2.friends[ - self.tox2.friend_by_public_key(self.tox1.public_key) - ].request_message, + self.tox2.friends[self.tox2.friend_by_public_key( + self.tox1.public_key)].request_message, b"are you gonna be my best friend?", ) self.assertEqual( - self.tox3.friends[ - self.tox3.friend_by_public_key(self.tox2.public_key) - ].request_message, + self.tox3.friends[self.tox3.friend_by_public_key( + self.tox2.public_key)].request_message, b"lala lala lala la I'm Mr. Happy Face", ) diff --git a/test/auto_tests/test.py b/test/auto_tests/test.py index 3ccc628..1ca1233 100644 --- a/test/auto_tests/test.py +++ b/test/auto_tests/test.py @@ -16,11 +16,11 @@ class FriendInfo: name: bytes = b"" request_message: bytes = b"" typing: bool = False - messages: list[tuple[core.Tox_Message_Type, bytes]] = field(default_factory=list) + messages: list[tuple[core.Tox_Message_Type, bytes]] = field( + default_factory=list) # message_id -> read receipt messages_sent: dict[int, bool] = field( - default_factory=lambda: collections.defaultdict(bool) - ) + default_factory=lambda: collections.defaultdict(bool)) lossy_packets: list[bytes] = field(default_factory=list) lossless_packets: list[bytes] = field(default_factory=list) @@ -36,11 +36,9 @@ class ConferenceInfo: inviter: int = -1 connected: bool = False peers: dict[int, ConferencePeerInfo] = field( - default_factory=lambda: collections.defaultdict(ConferencePeerInfo) - ) + default_factory=lambda: collections.defaultdict(ConferencePeerInfo)) messages: list[tuple[int, core.Tox_Message_Type, bytes]] = field( - default_factory=list - ) + default_factory=list) class TestTox(core.Tox_Ptr): @@ -55,33 +53,34 @@ def __init__(self, index: int) -> None: self.friends = collections.defaultdict(FriendInfo) self.conferences = collections.defaultdict(ConferenceInfo) - def handle_self_connection_status( - self, connection_status: core.Tox_Connection - ) -> None: + def handle_self_connection_status(self, + connection_status: core.Tox_Connection + ) -> None: self.connection_status = connection_status def handle_friend_name(self, friend_number: int, name: bytes) -> None: self.friends[friend_number].name = name - def handle_friend_status_message( - self, friend_number: int, status_message: bytes - ) -> None: + def handle_friend_status_message(self, friend_number: int, + status_message: bytes) -> None: self.friends[friend_number].status_message = status_message - def handle_friend_status( - self, friend_number: int, status: core.Tox_User_Status - ) -> None: + def handle_friend_status(self, friend_number: int, + status: core.Tox_User_Status) -> None: self.friends[friend_number].status = status def handle_friend_connection_status( - self, friend_number: int, connection_status: core.Tox_Connection, + self, + friend_number: int, + connection_status: core.Tox_Connection, ) -> None: self.friends[friend_number].connection_status = connection_status def handle_friend_typing(self, friend_number: int, typing: bool) -> None: self.friends[friend_number].typing = typing - def handle_friend_read_receipt(self, friend_number: int, message_id: int) -> None: + def handle_friend_read_receipt(self, friend_number: int, + message_id: int) -> None: assert self.friends[friend_number].messages_sent[message_id] is False self.friends[friend_number].messages_sent[message_id] = True @@ -89,54 +88,53 @@ def handle_friend_request(self, public_key: bytes, message: bytes) -> None: friend_number = self.friend_add_norequest(public_key) self.friends[friend_number].request_message = message - def handle_friend_message( - self, friend_number: int, type_: core.Tox_Message_Type, message: bytes - ) -> None: + def handle_friend_message(self, friend_number: int, + type_: core.Tox_Message_Type, + message: bytes) -> None: self.friends[friend_number].messages.append((type_, message)) - def handle_friend_lossy_packet(self, friend_number: int, message: bytes) -> None: + def handle_friend_lossy_packet(self, friend_number: int, + message: bytes) -> None: self.friends[friend_number].lossy_packets.append(message) raise Exception("OMG") - def handle_friend_lossless_packet(self, friend_number: int, message: bytes) -> None: + def handle_friend_lossless_packet(self, friend_number: int, + message: bytes) -> None: self.friends[friend_number].lossless_packets.append(message) - def handle_file_recv_control( - self, friend_number: int, file_number: int, control: core.Tox_File_Control - ) -> None: + def handle_file_recv_control(self, friend_number: int, file_number: int, + control: core.Tox_File_Control) -> None: pass - def handle_file_chunk_request( - self, friend_number: int, file_number: int, position: int, length: int - ) -> None: + def handle_file_chunk_request(self, friend_number: int, file_number: int, + position: int, length: int) -> None: pass def handle_file_recv( - self, - friend_number: int, - file_number: int, - kind: int, - file_size: int, - filename: bytes, + self, + friend_number: int, + file_number: int, + kind: int, + file_size: int, + filename: bytes, ) -> None: pass - def handle_file_recv_chunk( - self, friend_number: int, file_number: int, position: int, data: bytes - ) -> None: + def handle_file_recv_chunk(self, friend_number: int, file_number: int, + position: int, data: bytes) -> None: pass - def handle_conference_peer_name( - self, conference_number: int, peer_number: int, name: bytes - ) -> None: + def handle_conference_peer_name(self, conference_number: int, + peer_number: int, name: bytes) -> None: self.conferences[conference_number].peers[peer_number].name = name - def handle_conference_peer_list_changed(self, conference_number: int) -> None: + def handle_conference_peer_list_changed(self, + conference_number: int) -> None: pass - def handle_conference_invite( - self, friend_number: int, type_: core.Tox_Conference_Type, cookie: bytes - ) -> None: + def handle_conference_invite(self, friend_number: int, + type_: core.Tox_Conference_Type, + cookie: bytes) -> None: assert type_ == core.TOX_CONFERENCE_TYPE_TEXT conference_number = self.conference_join(friend_number, cookie) self.conferences[conference_number].inviter = friend_number @@ -145,19 +143,17 @@ def handle_conference_connected(self, conference_number: int) -> None: self.conferences[conference_number].connected = True def handle_conference_message( - self, - conference_number: int, - peer_number: int, - type_: core.Tox_Message_Type, - message: bytes, + self, + conference_number: int, + peer_number: int, + type_: core.Tox_Message_Type, + message: bytes, ) -> None: self.conferences[conference_number].messages.append( - (peer_number, type_, message) - ) + (peer_number, type_, message)) - def handle_conference_title( - self, conference_number: int, peer_number: int, title: bytes - ) -> None: + def handle_conference_title(self, conference_number: int, peer_number: int, + title: bytes) -> None: self.conferences[conference_number].title = (peer_number, title) @@ -183,29 +179,31 @@ def is_online() -> bool: return bool( self.tox1.connection_status == core.TOX_CONNECTION_NONE or self.tox2.connection_status == core.TOX_CONNECTION_NONE - or self.tox3.connection_status == core.TOX_CONNECTION_NONE - ) + or self.tox3.connection_status == core.TOX_CONNECTION_NONE) # At most 20 seconds. self._iterate( - 1000, is_online, + 1000, + is_online, ) def _wait_for_friend_online(self) -> None: self._wait_for_self_online() if not self.tox1.friend_list: - self.tox1.friend_add(self.tox2.address, b"are you gonna be my best friend?") - self.tox2.friend_add( - self.tox3.address, b"lala lala lala la I'm Mr. Happy Face" - ) + self.tox1.friend_add(self.tox2.address, + b"are you gonna be my best friend?") + self.tox2.friend_add(self.tox3.address, + b"lala lala lala la I'm Mr. Happy Face") def is_online() -> bool: - return ( - self.tox1.friends[0].connection_status == core.TOX_CONNECTION_NONE - or self.tox2.friends[0].connection_status == core.TOX_CONNECTION_NONE - or self.tox2.friends[1].connection_status == core.TOX_CONNECTION_NONE - or self.tox3.friends[0].connection_status == core.TOX_CONNECTION_NONE - ) + return (self.tox1.friends[0].connection_status == + core.TOX_CONNECTION_NONE + or self.tox2.friends[0].connection_status == + core.TOX_CONNECTION_NONE + or self.tox2.friends[1].connection_status == + core.TOX_CONNECTION_NONE + or self.tox3.friends[0].connection_status == + core.TOX_CONNECTION_NONE) # At most 5 seconds. self._iterate(250, is_online) @@ -213,15 +211,13 @@ def is_online() -> bool: def test_friend_add(self) -> None: self._wait_for_friend_online() self.assertEqual( - self.tox2.friends[ - self.tox2.friend_by_public_key(self.tox1.public_key) - ].request_message, + self.tox2.friends[self.tox2.friend_by_public_key( + self.tox1.public_key)].request_message, b"are you gonna be my best friend?", ) self.assertEqual( - self.tox3.friends[ - self.tox3.friend_by_public_key(self.tox2.public_key) - ].request_message, + self.tox3.friends[self.tox3.friend_by_public_key( + self.tox2.public_key)].request_message, b"lala lala lala la I'm Mr. Happy Face", ) @@ -229,18 +225,18 @@ def test_friend_by_public_key(self) -> None: with self.assertRaises(core.ApiException) as ex: # We're not our own friend. self.tox1.friend_by_public_key(self.tox1.public_key) - self.assertEqual( - ex.exception.error, core.TOX_ERR_FRIEND_BY_PUBLIC_KEY_NOT_FOUND - ) + self.assertEqual(ex.exception.error, + core.TOX_ERR_FRIEND_BY_PUBLIC_KEY_NOT_FOUND) def test_send_message(self) -> None: self._wait_for_friend_online() - self.tox1.friend_send_message(0, core.TOX_MESSAGE_TYPE_NORMAL, b"hello there!") - friend = self.tox2.friends[self.tox2.friend_by_public_key(self.tox1.public_key)] + self.tox1.friend_send_message(0, core.TOX_MESSAGE_TYPE_NORMAL, + b"hello there!") + friend = self.tox2.friends[self.tox2.friend_by_public_key( + self.tox1.public_key)] self._iterate(100, lambda: not friend.messages) - self.assertEqual( - friend.messages[0], (core.TOX_MESSAGE_TYPE_NORMAL, b"hello there!") - ) + self.assertEqual(friend.messages[0], + (core.TOX_MESSAGE_TYPE_NORMAL, b"hello there!")) # TODO(iphydf): This one doesn't pass. Investigate why. # def test_send_lossy_packet(self) -> None: @@ -254,7 +250,8 @@ def test_send_message(self) -> None: def test_send_lossless_packet(self) -> None: self._wait_for_friend_online() self.tox1.friend_send_lossless_packet(0, b"\xa0general kenobi.") - friend = self.tox2.friends[self.tox2.friend_by_public_key(self.tox1.public_key)] + friend = self.tox2.friends[self.tox2.friend_by_public_key( + self.tox1.public_key)] self._iterate(100, lambda: not friend.lossless_packets) self.assertEqual(friend.lossless_packets[0], b"\xa0general kenobi.") @@ -267,9 +264,8 @@ def test_status(self) -> None: self.assertEqual(self.tox1.status, core.TOX_USER_STATUS_AWAY) self._iterate(100, lambda: friend.status == core.TOX_USER_STATUS_NONE) self.assertEqual(friend.status, core.TOX_USER_STATUS_AWAY) - self.assertEqual( - self.tox2.friend_get_status(friend_number), core.TOX_USER_STATUS_AWAY - ) + self.assertEqual(self.tox2.friend_get_status(friend_number), + core.TOX_USER_STATUS_AWAY) def test_name(self) -> None: self._wait_for_friend_online() @@ -277,13 +273,11 @@ def test_name(self) -> None: friend = self.tox2.friends[friend_number] self.assertEqual(self.tox1.name, b"") self.tox1.name = b"Now that's a name I haven't heard in a long time" - self.assertEqual( - self.tox1.name, b"Now that's a name I haven't heard in a long time" - ) + self.assertEqual(self.tox1.name, + b"Now that's a name I haven't heard in a long time") self._iterate(100, lambda: friend.name == b"") - self.assertEqual( - friend.name, b"Now that's a name I haven't heard in a long time" - ) + self.assertEqual(friend.name, + b"Now that's a name I haven't heard in a long time") self.assertEqual( self.tox2.friend_get_name(friend_number), b"Now that's a name I haven't heard in a long time", @@ -298,9 +292,8 @@ def test_status_message(self) -> None: self.assertEqual(self.tox1.status_message, b"Python rocks!") self._iterate(100, lambda: friend.status_message == b"") self.assertEqual(friend.status_message, b"Python rocks!") - self.assertEqual( - self.tox2.friend_get_status_message(friend_number), b"Python rocks!" - ) + self.assertEqual(self.tox2.friend_get_status_message(friend_number), + b"Python rocks!") def test_typing(self) -> None: self._wait_for_friend_online() @@ -316,9 +309,9 @@ def test_read_receipt(self) -> None: self._wait_for_friend_online() friend_number = self.tox1.friend_by_public_key(self.tox2.public_key) friend = self.tox1.friends[friend_number] - msg_id = self.tox1.friend_send_message( - friend_number, core.TOX_MESSAGE_TYPE_NORMAL, b"hello there!" - ) + msg_id = self.tox1.friend_send_message(friend_number, + core.TOX_MESSAGE_TYPE_NORMAL, + b"hello there!") self.assertFalse(friend.messages_sent[msg_id]) self._iterate(100, lambda: not friend.messages_sent[msg_id]) @@ -329,8 +322,7 @@ def test_conference_invite(self) -> None: self.assertListEqual(self.tox1.conference_chatlist, [cnum]) # tox1 invites tox2 to conference. self.tox1.conference_invite( - self.tox1.friend_by_public_key(self.tox2.public_key), cnum - ) + self.tox1.friend_by_public_key(self.tox2.public_key), cnum) # wait for tox2 to join conference. self._iterate(100, lambda: not self.tox2.conference_chatlist) self.assertEqual(len(self.tox2.conference_chatlist), 1)