tests: robust picking of random port to allow for parallel test running

doc/DEVELOPMENT.org

@@ -80,6 +80,8 @@ You can also run them via tox:
 - it's useful to pass =--pdb= to pytest, so you don't have to restart webdriver all over
 - after initial run, it's best pass =--skip-pkg-install= to tox to speedup repetitive test runs
 - when you're in development/debugging process, perhaps best to run the firefox version of tests -- Chrome has a 5-ish seconds lag after installing the extension
+- since these tests can be quite slow, you can benefit from installing =pytest-xdist= plugin and using =-n= flag to parallelize
+- to debug flaky tests, it's useful to install =pytest-repeat= in addition, e.g. =--repeat=5 -n 5=
 
 * Releasing
 ** AMO (addons.mozilla.org)

src/promnesia/tests/common.py

@@ -1,11 +1,12 @@
-from contextlib import contextmanager
+from contextlib import closing, contextmanager
 import gc
 import inspect
 import os
 from pathlib import Path
+import socket
 import sys
 from textwrap import dedent
-from typing import NoReturn, TypeVar
+from typing import Iterator, NoReturn, TypeVar
 
 import pytest
 
@@ -107,3 +108,27 @@ def write_config(path: Path, gen, **kwargs) -> None:
         assert k in cfg_src, k
         cfg_src = cfg_src.replace(k, repr(str(v)))  # meh
     path.write_text(cfg_src)
+
+
+@contextmanager
+def free_port() -> Iterator[int]:
+    # this is a generator to make sure there are no race conditions between the time we call this and launch program
+    #
+    # also some relevant articles about this 'technique'
+    # - https://eklitzke.org/binding-on-port-zero
+    # - https://idea.popcount.org/2014-04-03-bind-before-connect
+    # - https://blog.cloudflare.com/the-quantum-state-of-a-tcp-port
+    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+        if sys.platform == 'linux':
+            # Ok, so from what I've been reading, SO_REUSEADDR should only be necessary in the program that reuses the port
+            # However, this answer (or man socket) claims we need it on both sites in Linux? see https://superuser.com/a/587955/300795
+            # On other platforms is seems to cause tests to hang (at least on CI?)
+            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        # also not sure where REUSEADDR is set in uvicorn (e.g. here reuse_address isn't passed?)
+        # https://github.com/encode/uvicorn/blob/6d666d99a285153bc4613e811543c39eca57054a/uvicorn/server.py#L162C37-L162C50
+        # but from strace looks like it is called somewhere :shrug:
+
+        # assign euphemeral port
+        s.bind(('', 0))
+
+        yield s.getsockname()[1]

src/promnesia/tests/server_helper.py

@@ -7,17 +7,7 @@
 import requests
 
 from ..common import PathIsh
-from .common import tmp_popen, promnesia_bin
-
-
-# TODO use proper random port
-TEST_PORT = 16556
-
-
-def next_port() -> int:
-    global TEST_PORT
-    TEST_PORT += 1
-    return TEST_PORT
+from .common import tmp_popen, promnesia_bin, free_port
 
 
 @dataclass
@@ -33,31 +23,37 @@ def post(self, path: str, *, json: Optional[Dict[str, Any]] = None):
 
 @contextmanager
 def run_server(db: Optional[PathIsh] = None, *, timezone: Optional[str] = None) -> Iterator[Helper]:
-    port = str(next_port())
-    cmd = [
-        'serve',
-        '--quiet',
-        '--port', port,
-        *([] if timezone is None else ['--timezone', timezone]),
-        *([] if db is None else ['--db'  , str(db)]),
-    ]
     # TODO not sure, perhaps best to use a thread or something?
     # but for some tests makes more sense to test in a separate process
-    with tmp_popen(promnesia_bin(*cmd)) as server_process:
-        server = Helper(port=port)
-
-        # wait till ready
-        for _ in range(50):
-            try:
-                server.get('/status').json()
-                break
-            except:
-                time.sleep(0.1)
-        else:
-            raise RuntimeError("Cooldn't connect to '{st}' after 50 attempts")
-        print("Started server up, db: {db}".format(db=db), file=sys.stderr)
-
-        yield server
-
-        # TODO use logger!
-        print("Done with the server", file=sys.stderr)
+    with free_port() as pp:
+        port = str(pp)
+        args = [
+            'serve',
+            '--quiet',
+            '--port', port,
+            *([] if timezone is None else ['--timezone', timezone]),
+            *([] if db is None else ['--db'  , str(db)]),
+        ]
+        # import shlex
+        # pcmd = promnesia_bin(*cmd)
+        # print("COMMAND")
+        # print(' '.join(map(shlex.quote, map(str, pcmd))))
+        # breakpoint()
+        with tmp_popen(promnesia_bin(*args)) as server_process:
+            server = Helper(port=port)
+
+            # wait till ready
+            for _ in range(50):
+                try:
+                    server.get('/status').json()
+                    break
+                except:
+                    time.sleep(0.1)
+            else:
+                raise RuntimeError("Cooldn't connect to '{st}' after 50 attempts")
+            print("Started server up, db: {db}".format(db=db), file=sys.stderr)
+
+            yield server
+
+            # TODO use logger!
+            print("Done with the server", file=sys.stderr)