tests: Avoid hangs due to exceptions in subprocessesjmac/fix-test-hangs

2 files changed, 35 insertions, 21 deletions

diff --git a/tests/artifactcache/pull.py b/tests/artifactcache/pull.py
index dde451a8c..e9fa734ee 100644
--- a/tests/artifactcache/pull.py
+++ b/tests/artifactcache/pull.py
@@ -25,6 +25,16 @@ def message_handler(message, context):
     pass
 
 
+# Since parent processes wait for queue events, we need
+# to put something on it if the called process raises an
+# exception.
+def _queue_wrapper(target, queue, *args):
+    try:
+        target(*args, queue=queue)
+    except Exception as e:
+        queue.put(str(e))
+        raise
+
 def tree_maker(cas, tree, directory):
     if tree.root.ByteSize() == 0:
         tree.root.CopyFrom(directory)
@@ -97,9 +107,9 @@ def test_pull(cli, tmpdir, datafiles):
         queue = multiprocessing.Queue()
         # Use subprocess to avoid creation of gRPC threads in main BuildStream process
         # See https://github.com/grpc/grpc/blob/master/doc/fork_support.md for details
-        process = multiprocessing.Process(target=_test_pull,
-                                          args=(user_config_file, project_dir, artifact_dir,
-                                                'target.bst', element_key, queue))
+        process = multiprocessing.Process(target=_queue_wrapper,
+                                          args=(_test_pull, queue, user_config_file, project_dir,
+                                                artifact_dir, 'target.bst', element_key))
 
         try:
             # Keep SIGINT blocked in the child process
@@ -205,9 +215,9 @@ def test_pull_tree(cli, tmpdir, datafiles):
         queue = multiprocessing.Queue()
         # Use subprocess to avoid creation of gRPC threads in main BuildStream process
         # See https://github.com/grpc/grpc/blob/master/doc/fork_support.md for details
-        process = multiprocessing.Process(target=_test_push_tree,
-                                          args=(user_config_file, project_dir, artifact_dir,
-                                                artifact_digest, queue))
+        process = multiprocessing.Process(target=_queue_wrapper,
+                                          args=(_test_pull_tree, queue, user_config_file, project_dir,
+                                                artifact_dir, artifact_digest))
 
         try:
             # Keep SIGINT blocked in the child process
@@ -233,9 +243,9 @@ def test_pull_tree(cli, tmpdir, datafiles):
 
         queue = multiprocessing.Queue()
         # Use subprocess to avoid creation of gRPC threads in main BuildStream process
-        process = multiprocessing.Process(target=_test_pull_tree,
-                                          args=(user_config_file, project_dir, artifact_dir,
-                                                tree_digest, queue))
+        process = multiprocessing.Process(target=_queue_wrapper,
+                                          args=(_test_pull_tree, queue, user_config_file, project_dir,
+                                                artifact_dir, tree_digest))
 
         try:
             # Keep SIGINT blocked in the child process
diff --git a/tests/testutils/artifactshare.py b/tests/testutils/artifactshare.py
index 5194e83cf..85c51de85 100644
--- a/tests/testutils/artifactshare.py
+++ b/tests/testutils/artifactshare.py
@@ -67,22 +67,26 @@ class ArtifactShare():
     def run(self, q):
         pytest_cov.embed.cleanup_on_sigterm()
 
-        # Optionally mock statvfs
-        if self.total_space:
-            if self.free_space is None:
-                self.free_space = self.total_space
-            os.statvfs = self._mock_statvfs
+        try:
+            # Optionally mock statvfs
+            if self.total_space:
+                if self.free_space is None:
+                    self.free_space = self.total_space
+                os.statvfs = self._mock_statvfs
 
-        server = create_server(self.repodir, enable_push=True)
-        port = server.add_insecure_port('localhost:0')
+            server = create_server(self.repodir, enable_push=True)
+            port = server.add_insecure_port('localhost:0')
 
-        server.start()
+            server.start()
 
-        # Send port to parent
-        q.put(port)
+            # Send port to parent
+            q.put(port)
 
-        # Sleep until termination by signal
-        signal.pause()
+            # Sleep until termination by signal
+            signal.pause()
+        except Exception as e:
+            q.put(None)
+            raise
 
     # has_object():
     #