Yet another strategy to get a free port from the server

Author: elpaso

tests/src/python/qgis_wrapped_server.py

@@ -100,9 +100,9 @@ def do_POST(self):
 
 
 if __name__ == '__main__':
-    print('Starting server on %s:%s, use <Ctrl-C> to stop' %
-          (QGIS_SERVER_HOST, QGIS_SERVER_PORT))
     server = HTTPServer((QGIS_SERVER_HOST, QGIS_SERVER_PORT), Handler)
+    print('Starting server on %s:%s, use <Ctrl-C> to stop' %
+          (QGIS_SERVER_HOST, server.server_port), flush=True)
 
     def signal_handler(signal, frame):
         global qgs_app

tests/src/python/test_authmanager_endpoint.py

@@ -17,6 +17,7 @@
 """
 import os
 import sys
+import re
 import subprocess
 import tempfile
 import random
@@ -29,7 +30,6 @@
 # This will get replaced with a git SHA1 when you do a git archive
 __revision__ = '$Format:%H$'
 
-from time import sleep
 from urllib.parse import quote
 from shutil import rmtree
 
@@ -96,13 +96,18 @@ def setUpClass(cls):
         server_path = os.path.dirname(os.path.realpath(__file__)) + \
             '/qgis_wrapped_server.py'
         cls.server = subprocess.Popen([sys.executable, server_path],
-                                      env=os.environ)
+                                      env=os.environ, stdout=subprocess.PIPE)
+        line = cls.server.stdout.readline()
+        cls.port = int(re.findall(b':(\d+)', line)[0])
+        assert cls.port != 0
+        # Wait for the server process to start
         assert waitServer('http://127.0.0.1:%s' % cls.port), "Server is not responding!"
 
     @classmethod
     def tearDownClass(cls):
         """Run after all tests"""
         cls.server.terminate()
+        cls.server.wait()
         rmtree(QGIS_AUTH_DB_DIR_PATH)
         del cls.server
 

tests/src/python/test_offline_editing_wfs.py

@@ -31,6 +31,7 @@
 
 import os
 import sys
+import re
 import subprocess
 from shutil import copytree, rmtree
 import tempfile
@@ -49,11 +50,7 @@
 try:
     QGIS_SERVER_WFST_DEFAULT_PORT = os.environ['QGIS_SERVER_WFST_DEFAULT_PORT']
 except:
-    import socket
-    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    s.bind(("", 0))
-    QGIS_SERVER_WFST_DEFAULT_PORT = s.getsockname()[1]
-    s.close()
+    QGIS_SERVER_WFST_DEFAULT_PORT = '0' # Auto
 
 qgis_app = start_app()
 
@@ -97,7 +94,10 @@ def tearDownClass(cls):
     def setUp(self):
         """Run before each test."""
         self.server = subprocess.Popen([sys.executable, self.server_path],
-                                       env=os.environ)
+                                       env=os.environ, stdout=subprocess.PIPE)
+        line = self.server.stdout.readline()
+        self.port = int(re.findall(b':(\d+)', line)[0])
+        assert self.port != 0
         # Wait for the server process to start
         assert waitServer('http://127.0.0.1:%s' % self.port), "Server is not responding!"
         self._setUp()
@@ -108,15 +108,13 @@ def tearDown(self):
         self._clearLayer(self._getOnlineLayer('test_point'))
         # Kill the server
         self.server.terminate()
+        self.server.wait()
         del self.server
-        # Wait for the server process to stop
-        sleep(2)
         # Delete the sqlite db
         os.unlink(os.path.join(self.temp_path, 'offlineDbFile.sqlite'))
         self._tearDown()
 
-    @classmethod
-    def _getOnlineLayer(cls, type_name, layer_name=None):
+    def _getOnlineLayer(self, type_name, layer_name=None):
         """
         Return a new WFS layer, overriding the WFS cache
         """
@@ -125,14 +123,14 @@ def _getOnlineLayer(cls, type_name, layer_name=None):
         parms = {
             'srsname': 'EPSG:4326',
             'typename': type_name,
-            'url': 'http://127.0.0.1:%s/%s/?map=%s' % (cls.port,
-                                                       cls.counter,
-                                                       cls.project_path),
+            'url': 'http://127.0.0.1:%s/%s/?map=%s' % (self.port,
+                                                       self.counter,
+                                                       self.project_path),
             'version': 'auto',
             'table': '',
             #'sql': '',
         }
-        cls.counter += 1
+        self.counter += 1
         uri = ' '.join([("%s='%s'" % (k, v)) for k, v in list(parms.items())])
         wfs_layer = QgsVectorLayer(uri, layer_name, 'WFS')
         assert wfs_layer.isValid()

tests/src/python/test_qgsserver_wfst.py

@@ -37,6 +37,7 @@
 
 import os
 import sys
+import re
 import subprocess
 from shutil import copytree, rmtree
 import tempfile
@@ -98,13 +99,18 @@ def setUpClass(cls):
         server_path = os.path.dirname(os.path.realpath(__file__)) + \
             '/qgis_wrapped_server.py'
         cls.server = subprocess.Popen([sys.executable, server_path],
-                                      env=os.environ)
+                                      env=os.environ, stdout=subprocess.PIPE)
+        line = cls.server.stdout.readline()
+        cls.port = int(re.findall(b':(\d+)', line)[0])
+        assert cls.port != 0
+        # Wait for the server process to start
         assert waitServer('http://127.0.0.1:%s' % cls.port), "Server is not responding!"
 
     @classmethod
     def tearDownClass(cls):
         """Run after all tests"""
         cls.server.terminate()
+        cls.server.wait()
         del cls.server
         # Clear all test layers
         for ln in ['test_point', 'test_polygon', 'test_linestring']: