Add a test for PoW interrruption, make a separate test case for PoW

minode/tests/test_proofofwork.py

@@ -0,0 +1,76 @@
+"""Special tests for PoW"""
+import base64
+import logging
+import multiprocessing
+import os
+import queue
+import signal
+import threading
+import time
+import unittest
+
+from minode import proofofwork, shared, structure
+
+
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='[%(asctime)s] [%(levelname)s] %(message)s')
+
+multiprocessing.set_start_method('spawn')
+
+
+class TestProofofwork(unittest.TestCase):
+    """Test components of proof of work"""
+
+    def setUp(self):
+        shared.objects = {}
+
+    def test_proofofwork(self):
+        """Check the main proofofwork call and worker"""
+        shared.vector_advertise_queue = queue.Queue()
+        obj = structure.Object(
+            b'\x00' * 8, int(time.time() + 300), 42, 1,
+            shared.stream, b'HELLO')
+        start_time = time.time()
+        proofofwork.do_pow_and_publish(obj)
+        try:
+            vector = shared.vector_advertise_queue.get(timeout=300)
+        except queue.Empty:
+            self.fail("Couldn't make work in 300 sec")
+        else:
+            time.sleep(1)
+            try:
+                result = shared.objects[vector]
+            except KeyError:
+                self.fail(
+                    "Couldn't found object with vector %s"
+                    " %s sec after pow start" % (
+                        base64.b16encode(vector), time.time() - start_time))
+            self.assertTrue(result.is_valid())
+            self.assertEqual(result.object_type, 42)
+            self.assertEqual(result.object_payload, b'HELLO')
+
+        q = queue.Queue()
+        # pylint: disable=protected-access
+        proofofwork._pow_worker(obj.pow_target(), obj.pow_initial_hash(), q)
+        try:
+            nonce = q.get(timeout=5)
+        except queue.Empty:
+            self.fail("No nonce found in the queue")
+
+        obj = structure.Object(
+            nonce, obj.expires_time, obj.object_type, obj.version,
+            obj.stream_number, obj.object_payload)
+        self.assertTrue(obj.is_valid())
+
+    def test_interrupt(self):
+        """Send signals to _pow_worker process"""
+        obj = structure.Object(
+            b'\x00' * 8, int(time.time() + 300), 42, 1, 2, os.urandom(4096))
+        threading.Thread(target=proofofwork._worker, args=(obj, )).start()
+        time.sleep(1)
+        worker = multiprocessing.active_children()[0]
+        # worker.terminate()
+        os.kill(worker.pid, signal.SIGINT)
+        time.sleep(1)
+        self.assertFalse(worker.is_alive())

minode/tests/test_structure.py

@@ -1,7 +1,5 @@
 """Tests for structures"""
-import base64
 import logging
-import queue
 import struct
 import time
 import unittest
@@ -154,41 +152,3 @@ class TestStructure(unittest.TestCase):
         obj.object_payload = \
             b'TIGER, tiger, burning bright. In the forests of the night'
         self.assertFalse(obj.is_valid())
-
-    def test_proofofwork(self):
-        """Check the main proofofwork call and worker"""
-        shared.vector_advertise_queue = queue.Queue()
-        obj = structure.Object(
-            b'\x00' * 8, int(time.time() + 300), 42, 1,
-            shared.stream, b'HELLO')
-        start_time = time.time()
-        proofofwork.do_pow_and_publish(obj)
-        try:
-            vector = shared.vector_advertise_queue.get(timeout=300)
-        except queue.Empty:
-            self.fail("Couldn't make work in 300 sec")
-        else:
-            time.sleep(1)
-            try:
-                result = shared.objects[vector]
-            except KeyError:
-                self.fail(
-                    "Couldn't found object with vector %s"
-                    " %s sec after pow start" % (
-                        base64.b16encode(vector), time.time() - start_time))
-            self.assertTrue(result.is_valid())
-            self.assertEqual(result.object_type, 42)
-            self.assertEqual(result.object_payload, b'HELLO')
-
-        q = queue.Queue()
-        # pylint: disable=protected-access
-        proofofwork._pow_worker(obj.pow_target(), obj.pow_initial_hash(), q)
-        try:
-            nonce = q.get(timeout=5)
-        except queue.Empty:
-            self.fail("No nonce found in the queue")
-
-        obj = structure.Object(
-            nonce, obj.expires_time, obj.object_type, obj.version,
-            obj.stream_number, obj.object_payload)
-        self.assertTrue(obj.is_valid())