test: add unit tests

Tests written using unittest [1] and hypothesis [2] modules.

How-to run:

$ make test_unittest

1. https://docs.python.org/3/library/unittest.html
2. https://hypothesis.readthedocs.io/en/latest/

Author: ligurio

Makefile

@@ -18,4 +18,9 @@ build_unit_tests:
 test_integration: build_unit_tests
 	test/test-run.py --builddir=build/ --var=build/test/var --force $(TEST_RUN_EXTRA_PARAMS)
 
-.PHONY: lint flake8 luacheck test
+test_unittest:
+	PYTHONPATH=. python test/unittest/*.py
+
+test: test_unittest test_integration 
+
+.PHONY: lint flake8 luacheck test test_integration test_unittest

requirements-test.txt

@@ -1 +1,2 @@
 flake8==3.7.9
+hypothesis==5.41.4

test/unittest/00000000000000000003.snap

@@ -0,0 +1,64 @@
+import unittest
+
+from hypothesis import assume, given, settings
+from hypothesis.strategies import integers
+import os
+import socket
+import random
+
+import lib.utils as utils
+
+
+class TestUtils(unittest.TestCase):
+    def test_extract_schema_from_snapshot(self):
+        snapshot_path = '00000000000000000003.snap'
+        v = utils.extract_schema_from_snapshot(snapshot_path)
+        self.assertEqual(v, [u'version', 2, 3, 1])
+
+    @settings(max_examples=5)
+    @given(port=integers(65000, 65535))
+    def test_check_port(self, port):
+        def open_socket(p):
+            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            s.bind(('localhost', p))
+            s.listen()
+            return s
+        status = utils.check_port(port, False, ipv4=True, ipv6=False)
+        self.assertEqual(status, True)
+        s = open_socket(port)
+        status = utils.check_port(port, False, ipv4=True, ipv6=False)
+        s.close()
+        self.assertEqual(status, False)
+
+    @settings(max_examples=2)
+    @given(start_port=integers(65000, 65535), end_port=integers(65000, 65535))
+    def test_find_port(self, start_port, end_port):
+        assume(start_port < end_port)
+        is_open_port_found = False
+        open_port = start_port
+        for p in range(start_port, end_port):
+            open_port = random.randint(start_port, end_port)
+            if utils.check_port(open_port, False, ipv4=True, ipv6=False):
+                is_open_port_found = True
+                break
+        if not is_open_port_found:
+            self.fail("Failed to found open port in range {}-{}".format(start_port, end_port))
+
+        os.environ['TEST_RUN_TCP_PORT_START'] = str(start_port)
+        os.environ['TEST_RUN_TCP_PORT_END'] = str(end_port)
+        open_sockets = []
+        for p in range(start_port, end_port):
+            if p != open_port:
+                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+                s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                s.bind(('localhost', p))
+                s.listen()
+                open_sockets.append(s)
+        found_open_port = utils.find_port()
+        for s in open_sockets:
+            s.close()
+        self.assertEqual(found_open_port, open_port)
+
+
+if __name__ == "__main__":
+    unittest.main()