tests/artifactservice.py: run in separate processraoul/1023-grpc-forking

Forking after creating a grpc channel can cause issues so these tests
are changed to avoid this.

May fix #1023

1 files changed, 98 insertions, 49 deletions

diff --git a/tests/artifactcache/artifactservice.py b/tests/artifactcache/artifactservice.py
index 5a7a3cdd5..791083b17 100644
--- a/tests/artifactcache/artifactservice.py
+++ b/tests/artifactcache/artifactservice.py
@@ -18,7 +18,9 @@
 #
 import os
 import pytest
+import signal
 from urllib.parse import urlparse
+from multiprocessing import Process, Queue
 
 import grpc
 
@@ -30,46 +32,63 @@ from buildstream._protos.build.bazel.remote.execution.v2 \
 from buildstream._protos.build.bazel.remote.execution.v2 \
     import remote_execution_pb2_grpc as re_pb2_grpc
 from buildstream import utils
+from buildstream import _signals
 
 from tests.testutils.artifactshare import create_artifact_share
 
 
+# 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 test_artifact_get_not_found(tmpdir):
     sharedir = os.path.join(str(tmpdir), "share")
     with create_artifact_share(sharedir) as share:
         # set up artifact service stub
         url = urlparse(share.repo)
-        channel = grpc.insecure_channel("{}:{}".format(url.hostname, url.port))
-        artifact_stub = ArtifactServiceStub(channel)
-
-        # Run GetArtifact and check it throws a not found error
-        request = GetArtifactRequest()
-        request.cache_key = "@artifact/something/not_there"
+        queue = Queue()
+        process = Process(target=_queue_wrapper, args=(_artifact_request, queue, url))
         try:
-            artifact_stub.GetArtifact(request)
-        except grpc.RpcError as e:
-            assert e.code() == grpc.StatusCode.NOT_FOUND
-            assert e.details() == "Artifact proto not found"
-        else:
-            assert False
+            with _signals.blocked([signal.SIGINT], ignore=False):
+                process.start()
+            error = queue.get()
+            process.join()
+        except KeyboardInterrupt:
+            utils._kill_process_tree(process.pid)
+            raise
+
+        assert not error
+
+
+def _artifact_request(url, queue):
+    channel = grpc.insecure_channel("{}:{}".format(url.hostname, url.port))
+    artifact_stub = ArtifactServiceStub(channel)
+
+    # Run GetArtifact and check it throws a not found error
+    request = GetArtifactRequest()
+    request.cache_key = "@artifact/something/not_there"
+    try:
+        artifact_stub.GetArtifact(request)
+    except grpc.RpcError as e:
+        assert e.code() == grpc.StatusCode.NOT_FOUND
+        assert e.details() == "Artifact proto not found"
+        queue.put(None)
+    else:
+        assert False
+
 
 # Successfully getting the artifact
 @pytest.mark.parametrize("files", ["present", "absent", "invalid"])
 def test_update_artifact(tmpdir, files):
     sharedir = os.path.join(str(tmpdir), "share")
     with create_artifact_share(sharedir) as share:
-        url = urlparse(share.repo)
-        channel = grpc.insecure_channel("{}:{}".format(url.hostname, url.port))
-        artifact_stub = ArtifactServiceStub(channel)
-
-        # initialise an artifact
-        artifact = Artifact()
-        artifact.version = 0
-        artifact.build_success = True
-        artifact.strong_key = "abcdefghijklmnop"
-        artifact.files.hash = "hashymchashash"
-        artifact.files.size_bytes = 10
-
         # put files object
         if files == "present":
             directory = re_pb2.Directory()
@@ -79,31 +98,61 @@ def test_update_artifact(tmpdir, files):
         elif files == "absent":
             digest = utils._message_digest("abcdefghijklmnop".encode("utf-8"))
 
-        artifact.files.CopyFrom(digest)
+        url = urlparse(share.repo)
+        queue = Queue()
+        process = Process(target=_queue_wrapper, args=(_get_artifact, queue, url, files, digest))
 
-        # Put it in the artifact share with an UpdateArtifactRequest
-        request = UpdateArtifactRequest()
-        request.artifact.CopyFrom(artifact)
-        request.cache_key = "a-cache-key"
+        try:
+            with _signals.blocked([signal.SIGINT], ignore=False):
+                process.start()
+            error = queue.get()
+            process.join()
+        except KeyboardInterrupt:
+            utils._kill_process_tree(process.pid)
+            raise
 
-        # should return the same artifact back
-        if files == "present":
-            response = artifact_stub.UpdateArtifact(request)
-            assert response == artifact
-        else:
-            try:
-                artifact_stub.UpdateArtifact(request)
-            except grpc.RpcError as e:
-                assert e.code() == grpc.StatusCode.FAILED_PRECONDITION
-                if files == "absent":
-                    assert e.details() == "Artifact files specified but no files found"
-                elif files == "invalid":
-                    assert e.details() == "Artifact files specified but directory not found"
-                return
-
-        # If we uploaded the artifact check GetArtifact
-        request = GetArtifactRequest()
-        request.cache_key = "a-cache-key"
-
-        response = artifact_stub.GetArtifact(request)
+        assert not error
+
+
+def _get_artifact(url, files, digest, queue):
+    channel = grpc.insecure_channel("{}:{}".format(url.hostname, url.port))
+    artifact_stub = ArtifactServiceStub(channel)
+
+    # initialise an artifact
+    artifact = Artifact()
+    artifact.version = 0
+    artifact.build_success = True
+    artifact.strong_key = "abcdefghijklmnop"
+    artifact.files.hash = "hashymchashash"
+    artifact.files.size_bytes = 10
+
+    artifact.files.CopyFrom(digest)
+
+    # Put it in the artifact share with an UpdateArtifactRequest
+    request = UpdateArtifactRequest()
+    request.artifact.CopyFrom(artifact)
+    request.cache_key = "a-cache-key"
+
+    # should return the same artifact back
+    if files == "present":
+        response = artifact_stub.UpdateArtifact(request)
         assert response == artifact
+    else:
+        try:
+            artifact_stub.UpdateArtifact(request)
+        except grpc.RpcError as e:
+            assert e.code() == grpc.StatusCode.FAILED_PRECONDITION
+            if files == "absent":
+                assert e.details() == "Artifact files specified but no files found"
+            elif files == "invalid":
+                assert e.details() == "Artifact files specified but directory not found"
+            queue.put(None)
+            return
+
+    # If we uploaded the artifact check GetArtifact
+    request = GetArtifactRequest()
+    request.cache_key = "a-cache-key"
+
+    response = artifact_stub.GetArtifact(request)
+    assert response == artifact
+    queue.put(None)