From fa6a85ba726585426bb3a0852bd8c9aa40136930 Mon Sep 17 00:00:00 2001 From: zebra-lucky Date: Sat, 2 Nov 2024 23:23:13 +0200 Subject: [PATCH] test_onionmc.py: add client factory/custom message tests --- .../joinmarket/tests/jmdaemon/test_onionmc.py | 32 ++++++++++++++++++- 1 file changed, 31 insertions(+), 1 deletion(-) diff --git a/electrum/plugins/joinmarket/tests/jmdaemon/test_onionmc.py b/electrum/plugins/joinmarket/tests/jmdaemon/test_onionmc.py index 70ace4839..f2c0f285c 100644 --- a/electrum/plugins/joinmarket/tests/jmdaemon/test_onionmc.py +++ b/electrum/plugins/joinmarket/tests/jmdaemon/test_onionmc.py @@ -9,7 +9,7 @@ from electrum.plugins.joinmarket.jmdaemon import ( from electrum.plugins.joinmarket.jmdaemon.onionmc import ( PEER_STATUS_UNCONNECTED, PEER_STATUS_CONNECTED, PEER_STATUS_HANDSHAKED, PEER_STATUS_DISCONNECTED, NOT_SERVING_ONION_HOSTNAME, OnionPeerError, - OnionCustomMessage, JM_MESSAGE_TYPES) + OnionCustomMessageDecodingError, OnionCustomMessage, JM_MESSAGE_TYPES) from electrum.plugins.joinmarket.jmdaemon.onionmc_support import ( TorClientService) @@ -154,6 +154,36 @@ class OnionBaseTestCase(JMTestCase): await cb(self.mc) +class OnionCustomMessageTestCase(OnionBaseTestCase): + + async def test_from_string_decode(self): + with self.assertRaises(OnionCustomMessageDecodingError): + OnionCustomMessage.from_string_decode(b'') + msg = OnionCustomMessage.from_string_decode( + b'{"line": "dummymsg", "type": 1}') + assert msg.text == 'dummymsg' + assert msg.msgtype == 1 + + +class OnionClientFactoryTestCase(OnionBaseTestCase): + + async def test_register_disconnection(self): + peer = self.peer + factory = peer.factory + factory.register_disconnection(peer.reconnecting_service.protocol) + + async def test_send(self): + factory = self.peer.factory + msg = OnionCustomMessage('dummymsg', JM_MESSAGE_TYPES["pubmsg"]) + assert factory.send(msg) + + async def test_receive_message(self): + peer = self.peer + factory = peer.factory + msg = OnionCustomMessage('dummymsg', JM_MESSAGE_TYPES["pubmsg"]) + factory.receive_message(msg, peer.reconnecting_service.protocol) + + class OnionPeerTestCase(OnionBaseTestCase): async def test_update_status(self):