Check port availability

kolibri/utils/sanity_checks.py

@@ -1,14 +1,18 @@
 import logging
 import os
-import socket
 import sys
 
+import portend
+
 from .conf import OPTIONS
 from .server import get_status
+from .server import LISTEN_ADDRESS
 from .server import NotRunning
 
 logger = logging.getLogger(__name__)
 
+PORT_AVAILABILITY_CHECK_TIMEOUT = 2
+
 
 def check_other_kolibri_running(port):
     try:
@@ -24,26 +28,19 @@ def check_other_kolibri_running(port):
     except NotRunning:
         # In case that something other than Kolibri occupies the port,
         # check the port's availability.
-        check_port_availability('127.0.0.1', port)
+        check_port_availability(LISTEN_ADDRESS, port)
 
 
 def check_port_availability(host, port):
-    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    # This is to prevent the previous execution has left the socket
-    # in a TIME_WAIT start, and can't be immediately reused.
-    # From the bottom of https://docs.python.org/2/library/socket.html#example
-    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     try:
-        s.bind((host, port))
-        s.close()
-    except socket.error:
+        portend.free(host, port, timeout=PORT_AVAILABILITY_CHECK_TIMEOUT)
+    except portend.Timeout:
         # Port is occupied
         logger.error(
             "Port {} is occupied.\n"
             "Please check that you do not have other processes "
             "running on this port and try again.\n".format(port)
         )
-        s.close()
         sys.exit(1)
 
 

kolibri/utils/tests/test_sanity_check.py

@@ -1,5 +1,4 @@
-import socket
-
+import portend
 from django.test import TestCase
 from mock import patch
 
@@ -16,11 +15,11 @@ def test_other_kolibri_running(self, status_mock, logging_mock):
             logging_mock.assert_called()
 
     @patch('kolibri.utils.sanity_checks.logging.error')
-    @patch('kolibri.utils.sanity_checks.socket.socket')
+    @patch('kolibri.utils.sanity_checks.portend.free')
     @patch('kolibri.utils.sanity_checks.get_status')
-    def test_port_occupied(self, status_mock, socket_mock, logging_mock):
+    def test_port_occupied(self, status_mock, portend_mock, logging_mock):
         status_mock.side_effect = NotRunning('Kolibri not running')
-        socket_mock.return_value.bind.side_effect = socket.error
+        portend_mock.side_effect = portend.Timeout
         with self.assertRaises(SystemExit):
             cli.main({'start': True})
             logging_mock.assert_called()