Add a test case for listener with a process running with --trusted-peer

minode/tests/test_network.py

@@ -8,8 +8,11 @@ import time
 from contextlib import contextmanager
 
 from minode import connection, main, shared
+from minode.listener import Listener
 from minode.manager import Manager
 
+from .test_process import TestProcessProto
+
 
 logging.basicConfig(
     level=logging.INFO,
@@ -31,6 +34,27 @@ def time_offset(offset):
         time.time = time_call
 
 
+@contextmanager
+def run_listener(host='localhost', port=8444):
+    """
+    Run the Listener with zero connection limit and
+    reset variables in shared after its stop.
+    """
+    connection_limit = shared.connection_limit
+    shared.connection_limit = 0
+    try:
+        listener = Listener(host, port)
+        listener.start()
+        yield listener
+    except OSError:
+        yield
+    finally:
+        shared.connection_limit = connection_limit
+        shared.connections.clear()
+        shared.shutting_down = True
+        time.sleep(1)
+
+
 class TestNetwork(unittest.TestCase):
     """Test case starting connections"""
 
@@ -71,7 +95,7 @@ class TestNetwork(unittest.TestCase):
             c = connection.Connection(*node)
             c.start()
             connection_started = time.time()
-            while c.status not in ('disconnecting', 'disconnected', 'failed'):
+            while c.status not in ('disconnected', 'failed'):
                 # The addr of established connection is added to nodes pool
                 if unknown and c.status == 'fully_established':
                     unknown = False
@@ -116,3 +140,65 @@ class TestNetwork(unittest.TestCase):
 
         time_offset_connections(nodes, 4000)
         time_offset_connections(nodes, -4000)
+
+
+class TestListener(TestProcessProto):
+    """A separate test case for Listener with a process with --trusted-peer"""
+    _process_cmd = ['minode', '--trusted-peer', '127.0.0.1']
+
+    def setUp(self):
+        shared.shutting_down = False
+
+    @classmethod
+    def tearDownClass(cls):
+        super().tearDownClass()
+        shared.shutting_down = False
+
+    def test_listener(self):
+        """Start Listener and try to connect"""
+        with run_listener() as listener:
+            if not listener:
+                self.fail('Failed to start listener')
+
+            c = connection.Connection('127.0.0.1', 8444)
+            shared.connections.add(c)
+
+            for _ in range(30):
+                if len(shared.connections) > 1:
+                    self.fail('The listener ignored connection limit')
+                time.sleep(0.5)
+
+            shared.connection_limit = 2
+            c.start()
+            started = time.time()
+            while c.status not in ('disconnected', 'failed'):
+                if c.status == 'fully_established':
+                    self.fail('Connected to itself')
+                if time.time() - started > 90:
+                    c.status = 'disconnecting'
+                time.sleep(0.2)
+
+            server = None
+            started = time.time()
+            while not server:
+                time.sleep(0.2)
+                if time.time() - started > 90:
+                    self.fail('Failed to establish the connection')
+                for c in shared.connections:
+                    if c.status == 'fully_established':
+                        server = c
+            self.assertTrue(server.server)
+
+            while not self.process.connections():
+                time.sleep(0.2)
+                if time.time() - started > 90:
+                    self.fail('Failed to connect to listener')
+
+            client = self.process.connections()[0]
+            self.assertEqual(client.raddr[0], '127.0.0.1')
+            self.assertEqual(client.raddr[1], 8444)
+            self.assertEqual(server.host, client.laddr[0])
+            # self.assertEqual(server.port, client.laddr[1])
+            server.status = 'disconnecting'
+
+        self.assertFalse(listener.is_alive())