Refactored launch_server into a utility

example/private/launch_server.py

@@ -0,0 +1,57 @@
+#!/usr/bin/python3
+#
+# Copyright (c) 2019-2024 Ruben Perez Hidalgo (rubenperez038 at gmail dot com)
+#
+# Distributed under the Boost Software License, Version 1.0. (See accompanying
+# file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+#
+
+# Utility to run the example TCP and HTTP servers
+
+import os
+import re
+from subprocess import Popen, PIPE, STDOUT
+from contextlib import contextmanager
+
+
+_is_win = os.name == 'nt'
+
+
+# Returns the port the server is listening at
+def _parse_server_start_line(line: str) -> int:
+    m = re.match(r'Server listening at 0\.0\.0\.0:([0-9]+)', line)
+    if m is None:
+        raise RuntimeError('Unexpected server start line')
+    return int(m.group(1))
+
+
+@contextmanager
+def launch_server(exe: str, host: str):
+    # Launch server and let it choose a free port for us.
+    # This prevents port clashes during b2 parallel test runs
+    server = Popen([exe, 'example_user', 'example_password', host, '0'], stdout=PIPE, stderr=STDOUT)
+    assert server.stdout is not None
+    with server:
+        try:
+            # Wait until the server is ready
+            ready_line = server.stdout.readline().decode()
+            print(ready_line, end='', flush=True)
+            if ready_line.startswith('Sorry'): # C++ standard unsupported, skip the test
+                exit(0)
+            yield _parse_server_start_line(ready_line)
+        finally:
+            print('Terminating server...', flush=True)
+
+            # In Windows, there is no sane way to cleanly terminate the process.
+            # Sending a Ctrl-C terminates all process attached to the console (including ourselves
+            # and any parent test runner). Running the process in a separate terminal doesn't allow
+            # access to stdout, which is problematic, too.
+            # terminate() sends SIGTERM in Unix, and uses TerminateProcess in Windows
+            server.terminate()
+
+            # Print any output the process generated
+            print('Server stdout: \n', server.stdout.read().decode(), flush=True)
+
+    # The return code is only relevant in Unix, as in Windows we used TerminateProcess
+    if not _is_win and server.returncode != 0:
+        raise RuntimeError('Server did not exit cleanly. retcode={}'.format(server.returncode))

example/private/run_batch_inserts.py

@@ -8,7 +8,6 @@
 
 from subprocess import run
 import argparse
-from typing import List
 from os import path
 
 # Helper to run the batch inserts example

example/private/run_connection_pool.py

@@ -9,12 +9,11 @@
 import requests
 import random
 import argparse
-from subprocess import PIPE, STDOUT, Popen
-from contextlib import contextmanager
-import re
-import os
+import sys
+from os import path
 
-_is_win = os.name == 'nt'
+sys.path.append(path.abspath(path.dirname(path.realpath(__file__))))
+from launch_server import launch_server
 
 
 def _check_response(res: requests.Response):
@@ -27,50 +26,6 @@ def _random_string() -> str:
     return bytes(random.getrandbits(8) for _ in range(8)).hex()
 
 
-# Returns the port the server is listening at
-def _parse_server_start_line(line: str) -> int:
-    m = re.match(r'Server listening at 0\.0\.0\.0:([0-9]+)', line)
-    if m is None:
-        raise RuntimeError('Unexpected server start line')
-    return int(m.group(1))
-
-
-@contextmanager
-def _launch_server(exe: str, host: str):
-    # Launch server and let it choose a free port for us.
-    # This prevents port clashes during b2 parallel test runs
-    server = Popen([exe, 'example_user', 'example_password', host, '0'], stdout=PIPE, stderr=STDOUT)
-    assert server.stdout is not None
-    with server:
-        try:
-            # Wait until the server is ready
-            ready_line = server.stdout.readline().decode()
-            print(ready_line, end='', flush=True)
-            if ready_line.startswith('Sorry'): # C++14 unsupported, skip the test
-                exit(0)
-            yield _parse_server_start_line(ready_line)
-        finally:
-            print('Terminating server...', flush=True)
-
-            # In Windows, there is no sane way to cleanly terminate the process.
-            # Sending a Ctrl-C terminates all process attached to the console (including ourselves
-            # and any parent test runner). Running the process in a separate terminal doesn't allow
-            # access to stdout, which is problematic, too.
-            if _is_win:
-                # kill is an alias for TerminateProcess with the given exit code
-                os.kill(server.pid, 9999)
-            else:
-                # Send SIGTERM
-                server.terminate()
-
-            # Print any output the process generated
-            print('Server stdout: \n', server.stdout.read().decode(), flush=True)
-
-    # Verify that it exited gracefully
-    if (_is_win and server.returncode != 9999) or (not _is_win and server.returncode):
-        raise RuntimeError('Server did not exit cleanly. retcode={}'.format(server.returncode))
-
-
 def _call_endpoints(port: int):
     base_url = 'http://127.0.0.1:{}'.format(port)
 
@@ -134,7 +89,7 @@ def main():
     args = parser.parse_args()
 
     # Launch the server
-    with _launch_server(args.executable, args.host) as listening_port:
+    with launch_server(args.executable, args.host) as listening_port:
     # Run the tests
         _call_endpoints(listening_port)
 

example/private/run_tutorial_connection_pool.py

@@ -7,54 +7,13 @@
 #
 
 import argparse
-from subprocess import PIPE, STDOUT, Popen
-from contextlib import contextmanager
-import re
-import os
+import sys
+from os import path
 import socket
 import struct
 
-_is_win = os.name == 'nt'
-
-
-# Returns the port the server is listening at
-def _parse_server_start_line(line: str) -> int:
-    m = re.match(r'Server listening at 0\.0\.0\.0:([0-9]+)', line)
-    if m is None:
-        raise RuntimeError('Unexpected server start line')
-    return int(m.group(1))
-
-
-@contextmanager
-def _launch_server(exe: str, host: str):
-    # Launch server and let it choose a free port for us.
-    # This prevents port clashes during b2 parallel test runs
-    server = Popen([exe, 'example_user', 'example_password', host, '0'], stdout=PIPE, stderr=STDOUT)
-    assert server.stdout is not None
-    with server:
-        try:
-            # Wait until the server is ready
-            ready_line = server.stdout.readline().decode()
-            print(ready_line, end='', flush=True)
-            if ready_line.startswith('Sorry'): # C++ standard unsupported, skip the test
-                exit(0)
-            yield _parse_server_start_line(ready_line)
-        finally:
-            print('Terminating server...', flush=True)
-
-            # In Windows, there is no sane way to cleanly terminate the process.
-            # Sending a Ctrl-C terminates all process attached to the console (including ourselves
-            # and any parent test runner). Running the process in a separate terminal doesn't allow
-            # access to stdout, which is problematic, too.
-            # terminate() sends SIGTERM in Unix, and uses TerminateProcess in Windows
-            server.terminate()
-
-            # Print any output the process generated
-            print('Server stdout: \n', server.stdout.read().decode(), flush=True)
-
-    # The return code is only relevant in Unix, as in Windows we used TerminateProcess
-    if not _is_win and server.returncode != 0:
-        raise RuntimeError('Server did not exit cleanly. retcode={}'.format(server.returncode))
+sys.path.append(path.abspath(path.dirname(path.realpath(__file__))))
+from launch_server import launch_server
 
 
 class _Runner:
@@ -108,7 +67,7 @@ def main():
     args = parser.parse_args()
 
     # Launch the server
-    with _launch_server(args.executable, args.host) as listening_port:
+    with launch_server(args.executable, args.host) as listening_port:
     # Run the tests
         _Runner(listening_port).run(args.test_errors)