A test case for the network start checks that:

- all the threads are started,
  - it opens connections and updates stats.

A base class for partial run essentially mimics bitmessagemain.

src/tests/partial.py

@@ -0,0 +1,35 @@
+"""A test case for partial run class definition"""
+
+import os
+import sys
+import unittest
+
+from pybitmessage import pathmagic
+
+
+class TestPartialRun(unittest.TestCase):
+    """
+    A base class for test cases running some parts of the app,
+    e.g. separate threads or packages.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        cls.dirs = (os.path.abspath(os.curdir), pathmagic.setup())
+
+        import bmconfigparser
+        import state
+
+        from debug import logger  # noqa:F401 pylint: disable=unused-variable
+
+        state.shutdown = 0
+        cls.state = state
+        bmconfigparser.config = cls.config = bmconfigparser.BMConfigParser()
+        cls.config.read()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.state.shutdown = 1
+        # deactivate pathmagic
+        os.chdir(cls.dirs[0])
+        sys.path.remove(cls.dirs[1])

src/tests/test_network.py

@@ -0,0 +1,74 @@
+"""Test network module"""
+
+import threading
+import time
+
+from .common import skip_python3
+from .partial import TestPartialRun
+
+skip_python3()
+
+
+class TestNetwork(TestPartialRun):
+    """A test case for running the network subsystem"""
+
+    @classmethod
+    def setUpClass(cls):
+        super(TestNetwork, cls).setUpClass()
+
+        cls.state.maximumNumberOfHalfOpenConnections = 4
+
+        cls.config.set('bitmessagesettings', 'sendoutgoingconnections', 'True')
+
+        # config variable is still used inside of the network ):
+        import network
+        from network import connectionpool, stats
+
+        # beware of singleton
+        connectionpool.config = cls.config
+        cls.pool = network.BMConnectionPool()
+        cls.stats = stats
+
+        network.start(cls.config, cls.state)
+
+    def test_threads(self):
+        """Ensure all the network threads started"""
+        threads = {
+            "AddrBroadcaster", "Asyncore", "Downloader", "InvBroadcaster",
+            "Uploader"}
+        extra = (
+            self.config.getint('threads', 'receive')
+            + self.config.safeGetBoolean('bitmessagesettings', 'udp'))
+        for thread in threading.enumerate():
+            try:
+                threads.remove(thread.name)
+            except KeyError:
+                extra -= (
+                    thread.name == "Announcer"
+                    or thread.name.startswith("ReceiveQueue_"))
+
+        self.assertEqual(len(threads), 0)
+        self.assertEqual(extra, 0)
+
+    def test_stats(self):
+        """Check that network starts connections and updates stats"""
+        pl = 0
+        for _ in range(30):
+            if pl == 0:
+                pl = len(self.pool)
+            if (
+                self.stats.receivedBytes() > 0 and self.stats.sentBytes() > 0
+                and pl > 0
+                # and len(self.stats.connectedHostsList()) > 0
+            ):
+                break
+            time.sleep(1)
+        else:
+            self.fail('Have not started any connection in 30 sec')
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestNetwork, cls).tearDownClass()
+        for thread in threading.enumerate():
+            if thread.name == "Asyncore":
+                thread.stopThread()