test: fix Python freezes

I would get random freezes when running tests; I completely forgot that
logging & multiprocessing don't play well together.

2 files changed, 64 insertions, 11 deletions

diff --git a/test/py/lib/logging.py b/test/py/lib/logging.py
new file mode 100644
index 0000000..554ce1f
--- /dev/null
+++ b/test/py/lib/logging.py
@@ -0,0 +1,45 @@
+# Copyright (c) 2023 Egor Tensin <Egor.Tensin@gmail.com>
+# This file is part of the "cimple" project.
+# For details, see https://github.com/egor-tensin/cimple.
+# Distributed under the MIT License.
+
+from contextlib import contextmanager
+import logging
+import logging.config
+import logging.handlers
+import multiprocessing as mp
+
+
+@contextmanager
+def child_logging_thread():
+    # Delegating logging to the parent logger.
+    ctx = mp.get_context('spawn')
+    queue = ctx.Queue()
+    listener = logging.handlers.QueueListener(queue, logging.getLogger())
+    listener.start()
+    try:
+        yield queue
+    finally:
+        listener.stop()
+
+
+@contextmanager
+def configure_logging_in_child(queue):
+    config = {
+        'version': 1,
+        'handlers': {
+            'sink': {
+                'class': 'logging.handlers.QueueHandler',
+                'queue': queue,
+            },
+        },
+        'root': {
+            'handlers': ['sink'],
+            'level': 'DEBUG',
+        },
+    }
+    logging.config.dictConfig(config)
+    try:
+        yield
+    except Exception as e:
+        logging.exception(e)
diff --git a/test/py/test_repo.py b/test/py/test_repo.py
index f451e8f..e3a498d 100644
--- a/test/py/test_repo.py
+++ b/test/py/test_repo.py
@@ -3,11 +3,13 @@
 # For details, see https://github.com/egor-tensin/cimple.
 # Distributed under the MIT License.
 
-from multiprocessing import Process
+import logging
+import multiprocessing as mp
 import re
 
 import pytest
 
+from lib.logging import child_logging_thread, configure_logging_in_child
 from lib.process import LoggingEvent
 
 
@@ -27,6 +29,13 @@ class LoggingEventRunComplete(LoggingEvent):
             super().set()
 
 
+def client_runner(log_queue, client, runs_per_process, repo):
+    with configure_logging_in_child(log_queue):
+        logging.info('Executing %s clients', runs_per_process)
+        for i in range(runs_per_process):
+            client.run('run', repo.path, 'HEAD')
+
+
 def _test_repo_internal(env, repo, numof_processes, runs_per_process):
     numof_runs = numof_processes * runs_per_process
 
@@ -34,18 +43,17 @@ def _test_repo_internal(env, repo, numof_processes, runs_per_process):
     # Count the number of times the server receives the "run complete" message.
     env.server.logger.add_event(event)
 
-    def client_runner():
-        for i in range(runs_per_process):
-            env.client.run('run', repo.path, 'HEAD')
-
-    processes = [Process(target=client_runner) for i in range(numof_processes)]
-    for proc in processes:
-        proc.start()
+    with child_logging_thread() as log_queue:
+        ctx = mp.get_context('spawn')
+        args = (log_queue, env.client, runs_per_process, repo)
+        processes = [ctx.Process(target=client_runner, args=args) for i in range(numof_processes)]
+        for proc in processes:
+            proc.start()
 
-    event.wait()
+        event.wait()
 
-    for proc in processes:
-        proc.join()
+        for proc in processes:
+            proc.join()
 
     assert numof_runs == repo.count_run_files()