PyBitmessage-2025-02-24/src/tests/core.py

"""
Tests for core and those that do not work outside
(because of import error for example)
"""

import os
import pickle  # nosec
import Queue
import random  # nosec
import string
import time
import unittest

import knownnodes
import state
from bmconfigparser import BMConfigParser
from helper_msgcoding import MsgEncode, MsgDecode
from network import asyncore_pollchoose as asyncore
from network.connectionpool import BMConnectionPool
from network.tcp import Socks4aBMConnection, Socks5BMConnection, TCPConnection
from queues import excQueue

knownnodes_file = os.path.join(state.appdata, 'knownnodes.dat')


def pickle_knownnodes():
    now = time.time()
    with open(knownnodes_file, 'wb') as dst:
        pickle.dump({
            stream: {
                state.Peer(
                    '%i.%i.%i.%i' % tuple([
                        random.randint(1, 255) for i in range(4)]),
                    8444): {'lastseen': now, 'rating': 0.1}
                for i in range(1, 4)  # 3 test nodes
            }
            for stream in range(1, 4)  # 3 test streams
        }, dst)


def cleanup():
    os.remove(knownnodes_file)


class TestCore(unittest.TestCase):
    """Test case, which runs in main pybitmessage thread"""

    def test_msgcoding(self):
        """test encoding and decoding (originally from helper_msgcoding)"""
        msg_data = {
            'subject': ''.join(
                random.choice(string.ascii_lowercase + string.digits)  # nosec
                for _ in range(40)),
            'body': ''.join(
                random.choice(string.ascii_lowercase + string.digits)  # nosec
                for _ in range(10000))
        }

        obj1 = MsgEncode(msg_data, 1)
        obj2 = MsgEncode(msg_data, 2)
        obj3 = MsgEncode(msg_data, 3)
        # print "1: %i 2: %i 3: %i" % (
        # len(obj1.data), len(obj2.data), len(obj3.data))

        obj1e = MsgDecode(1, obj1.data)
        # no subject in trivial encoding
        self.assertEqual(msg_data['body'], obj1e.body)

        obj2e = MsgDecode(2, obj2.data)
        self.assertEqual(msg_data['subject'], obj2e.subject)
        self.assertEqual(msg_data['body'], obj2e.body)

        obj3e = MsgDecode(3, obj3.data)
        self.assertEqual(msg_data['subject'], obj3e.subject)
        self.assertEqual(msg_data['body'], obj3e.body)

        try:
            MsgEncode({'body': 'A msg with no subject'}, 3)
        except Exception as e:
            self.fail(
                'Exception %s while trying to encode message'
                ' with no subject!' % e
            )

    @unittest.skip('Bad environment for asyncore.loop')
    def test_tcpconnection(self):
        """initial fill script from network.tcp"""
        BMConfigParser().set('bitmessagesettings', 'dontconnect', 'true')
        try:
            for peer in (state.Peer("127.0.0.1", 8448),):
                direct = TCPConnection(peer)
                while asyncore.socket_map:
                    print("loop, state = %s" % direct.state)
                    asyncore.loop(timeout=10, count=1)
        except:
            self.fail('Exception in test loop')

    @staticmethod
    def _wipe_knownnodes():
        with knownnodes.knownNodesLock:
            knownnodes.knownNodes = {stream: {} for stream in range(1, 4)}

    @staticmethod
    def _outdate_knownnodes():
        with knownnodes.knownNodesLock:
            for nodes in knownnodes.knownNodes.itervalues():
                for node in nodes.itervalues():
                    node['lastseen'] -= 2419205  # older than 28 days

    def test_knownnodes_pickle(self):
        """ensure that 3 nodes was imported for each stream"""
        pickle_knownnodes()
        self._wipe_knownnodes()
        knownnodes.readKnownNodes()
        for nodes in knownnodes.knownNodes.itervalues():
            self_count = n = 0
            for n, node in enumerate(nodes.itervalues()):
                if node.get('self'):
                    self_count += 1
            self.assertEqual(n - self_count, 2)

    def test_knownnodes_default(self):
        """test adding default knownnodes if nothing loaded"""
        cleanup()
        self._wipe_knownnodes()
        knownnodes.readKnownNodes()
        self.assertGreaterEqual(
            len(knownnodes.knownNodes[1]), len(knownnodes.DEFAULT_NODES))

    def test_0_cleaner(self):
        """test knownnodes starvation leading to IndexError in Asyncore"""
        self._outdate_knownnodes()
        # time.sleep(303)  # singleCleaner wakes up every 5 min
        knownnodes.cleanupKnownNodes()
        while True:
            try:
                thread, exc = excQueue.get(block=False)
            except Queue.Empty:
                return
            if thread == 'Asyncore' and isinstance(exc, IndexError):
                self.fail("IndexError because of empty knownNodes!")

    def test_bootstrap(self):
        """test bootstrapping"""
        BMConfigParser().set('bitmessagesettings', 'dontconnect', 'true')
        self._outdate_knownnodes()
        knownnodes.cleanupKnownNodes()
        # it's weird, knownnodes appear empty
        knownnodes.addKnownNode(1, state.Peer('127.0.0.1', 8444), is_self=True)
        time.sleep(0.25)
        BMConfigParser().remove_option('bitmessagesettings', 'dontconnect')
        proxy_type = BMConfigParser().safeGet(
            'bitmessagesettings', 'socksproxytype')
        if proxy_type == 'SOCKS5':
            connection_base = Socks5BMConnection
        elif proxy_type == 'SOCKS4a':
            connection_base = Socks4aBMConnection
        else:
            connection_base = TCPConnection
        _started = time.time()
        for _ in range(180):
            time.sleep(1)
            for peer, con in BMConnectionPool().outboundConnections.iteritems():
                if not peer.host.startswith('bootstrap'):
                    self.assertIsInstance(con, connection_base)
                    return
        else:  # pylint: disable=useless-else-on-loop
            self.fail(
                'Failed to connect during %s sec' % (time.time() - _started))


def run():
    """Starts all tests defined in this module"""
    loader = unittest.TestLoader()
    loader.sortTestMethodsUsing = None
    suite = loader.loadTestsFromTestCase(TestCore)
    return unittest.TextTestRunner(verbosity=2).run(suite)