fix: can't pickle mp.Queue -> mp.Manager().Queue()

Author: nkemnitz

taskqueue/taskqueue.py

@@ -513,8 +513,6 @@ def soloprocess_upload(QueueClass, queue_name, tasks):
   tq = QueueClass(queue_name, progress=False)
   return tq.insert(tasks, skip_insert_counter=True)
 
-error_queue = mp.Queue()
-
 def multiprocess_upload(QueueClass, queue_name, tasks, parallel=True, total=None):
   if parallel is True:
     parallel = mp.cpu_count()
@@ -570,6 +568,7 @@ def capturing_soloprocess_upload(*args, **kwargs):
   # Don't fork, spawn entirely new processes. This
   # avoids accidental deadlocks.
   mp.set_start_method("spawn", force=True)
+  error_queue = mp.Manager().Queue()
 
   ct = 0
   with tqdm(desc="Upload", total=total) as pbar:

test/test_taskqueue.py

@@ -4,7 +4,6 @@
 import time
 
 from moto import mock_aws
-
 from six.moves import range
 import pytest
 
@@ -18,7 +17,7 @@
 from taskqueue.queueables import totask
 from taskqueue.queueablefns import tofunc, UnregisteredFunctionError, func2task
 
-@pytest.fixture(scope='function')
+@pytest.fixture(scope='session')
 def aws_credentials():
   """Mocked AWS Credentials for moto."""
   os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
@@ -35,6 +34,21 @@ def sqs(aws_credentials):
     client.create_queue(QueueName='test-pull-queue')
     yield client
 
+@pytest.fixture(scope='session')
+def sqs_server(aws_credentials):
+  from moto.server import ThreadedMotoServer
+  import boto3
+  server = ThreadedMotoServer(port=0)
+  server.start()
+  host, port = server.get_host_and_port()
+  endpoint = f"http://{host}:{port}"
+  os.environ['AWS_ENDPOINT_URL'] = endpoint
+  client = boto3.client('sqs', endpoint_url=endpoint)
+  client.create_queue(QueueName='test-pull-queue')
+  yield client
+  del os.environ['AWS_ENDPOINT_URL']
+  server.stop()
+
 QURLS = {
   'sqs': 'test-pull-queue',
   'fq': '/tmp/removeme/taskqueue/fq',
@@ -269,7 +283,7 @@ def test_local_taskqueue():
   assert tq.insert(epts) == 200
 
 @pytest.mark.parametrize('protocol', PROTOCOL)
-def test_parallel_insert_all(sqs, protocol):
+def test_parallel_insert_all(sqs_server, protocol):
   import pathos_issue
 
   path = getpath(protocol)