From 070541922b2df5c755d8a017a6dd26b48839a498 Mon Sep 17 00:00:00 2001
From: Dmitri Bogomolov <4glitch@gmail.com>
Date: Sun, 7 Mar 2021 23:39:37 +0200
Subject: [PATCH] Add minimal process tests based on pybitmessage and a test
 script

---
 minode/tests/__init__.py     |  0
 minode/tests/test_process.py | 93 ++++++++++++++++++++++++++++++++++++
 tests.py                     | 18 +++++++
 3 files changed, 111 insertions(+)
 create mode 100644 minode/tests/__init__.py
 create mode 100644 minode/tests/test_process.py
 create mode 100644 tests.py

diff --git a/minode/tests/__init__.py b/minode/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/minode/tests/test_process.py b/minode/tests/test_process.py
new file mode 100644
index 0000000..4abfad6
--- /dev/null
+++ b/minode/tests/test_process.py
@@ -0,0 +1,93 @@
+import unittest
+import signal
+import subprocess
+import tempfile
+import time
+
+import psutil
+
+
+class TestProcessProto(unittest.TestCase):
+    """Test process attributes, common flow"""
+    _process_cmd = ['minode']
+    _connection_limit = 16
+    _listen = None
+    _listening_port = None
+
+    home = None
+
+    @classmethod
+    def setUpClass(cls):
+        if not cls.home:
+            cls.home = tempfile.gettempdir()
+        cmd = cls._process_cmd + [
+            '--data-dir', cls.home,
+            '--connection-limit', str(cls._connection_limit)
+        ]
+        if cls._listen is True:
+            if cls._listening_port:
+                cmd += ['-p', cls._listening_port]
+        elif cls._listen is False:
+            cmd += ['--no-incoming']
+        cls.process = psutil.Popen(cmd, stderr=subprocess.STDOUT)  # nosec
+
+    @classmethod
+    def _stop_process(cls, timeout=5):
+        cls.process.send_signal(signal.SIGTERM)
+        try:
+            cls.process.wait(timeout)
+        except psutil.TimeoutExpired:
+            return False
+        return True
+
+    @classmethod
+    def tearDownClass(cls):
+        """Ensures that pybitmessage stopped and removes files"""
+        try:
+            if not cls._stop_process(10):
+                try:
+                    cls.process.kill()
+                except psutil.NoSuchProcess:
+                    pass
+        except psutil.NoSuchProcess:
+            pass
+
+
+class TestProcessShutdown(TestProcessProto):
+    """Separate test case for SIGTERM"""
+    def test_shutdown(self):
+        """Send to minode SIGTERM and ensure it stopped"""
+        # longer wait time because it's not a benchmark
+        self.assertTrue(
+            self._stop_process(20),
+            '%s has not stopped in 20 sec' % ' '.join(self._process_cmd))
+
+
+class TestProcess(TestProcessProto):
+    """The test case for minode process"""
+    def test_connections(self):
+        """Check minode process connections"""
+        _started = time.time()
+        connections = []
+        for t in range(40):
+            connections = self.process.connections()
+            if len(connections) > self._connection_limit / 2:
+                _time_to_connect = round(time.time() - _started)
+                break
+            time.sleep(0.5)
+        else:
+            self.fail(
+                'Failed establish at least %s connections in 20 sec'
+                % (self._connection_limit / 2))
+        for t in range(_time_to_connect * 2):
+            self.assertLessEqual(
+                len(connections), self._connection_limit + 1,  # one listening
+                'Opened more connections than required by --connection-limit')
+            time.sleep(0.5)
+        for c in connections:
+            if c.status == 'LISTEN':
+                if self._listen is False:
+                    return self.fail(
+                        'Listening while started with --no-incoming')
+                self.assertEqual(c.laddr[1], self._listening_port or 8444)
+                break
diff --git a/tests.py b/tests.py
new file mode 100644
index 0000000..ebc59b2
--- /dev/null
+++ b/tests.py
@@ -0,0 +1,18 @@
+#!/usr/bin/env python
+"""Custom tests runner script"""
+import random  # noseq
+import sys
+import unittest
+
+
+def unittest_discover():
+    """Explicit test suite creation"""
+    loader = unittest.defaultTestLoader
+    # randomize the order of tests in test cases
+    loader.sortTestMethodsUsing = lambda a, b: random.randint(-1, 1)
+    return loader.discover('minode.tests')
+
+
+if __name__ == "__main__":
+    result = unittest.TextTestRunner(verbosity=2).run(unittest_discover())
+    sys.exit(not result.wasSuccessful())