Get multi-server working

Author: AlanCoding

test_app/tests/task/conftest.py

@@ -2,6 +2,7 @@
 import multiprocessing
 import contextlib
 import time
+from uuid import uuid4
 
 import pytest
 
@@ -10,7 +11,7 @@
 
 
 
-async def asyncio_target(queue_in, queue_out, config):
+async def asyncio_target(queue_in, queue_out, config, this_uuid):
     try:
         dispatcher = DispatcherMain(config)
 
@@ -20,31 +21,31 @@ async def asyncio_target(queue_in, queue_out, config):
         queue_out.put('ready')
 
 
-        print('dispatcher server listening on queue_in')
+        print(f'{this_uuid} dispatcher server listening on queue_in')
         loop = asyncio.get_event_loop()
         message = await loop.run_in_executor(None, queue_in.get)
 
-        print(f'got message, will shut down: {message}')
+        print(f'{this_uuid} got message, will shut down: {message}')
     finally:
         await dispatcher.shutdown()
         await dispatcher.cancel_tasks()
 
 
-def subprocess_target(queue_in, queue_out, config):
+def subprocess_target(queue_in, queue_out, config, this_uuid):
     loop = asyncio.get_event_loop()
     try:
-        loop.run_until_complete(asyncio_target(queue_in, queue_out, config))
+        loop.run_until_complete(asyncio_target(queue_in, queue_out, config, this_uuid))
     except Exception:
         import traceback
 
         traceback.print_exc()
         # We are in a subprocess here, so even if we handle the exception
         # the main process will not know and still wait forever
         # so give them a kick on our way out
-        print('sending error message after error')
+        print(f'{this_uuid} sending error message after error')
         queue_out.put('error')
     finally:
-        print('closing asyncio loop')
+        print(f'{this_uuid} closing asyncio loop')
         loop.close()
 
 
@@ -54,21 +55,28 @@ def __init__(self):
         self.queue_in = multiprocessing.Queue()
         self.queue_out = multiprocessing.Queue()
 
-    def start_in_subprocess(self, config):
-        process = multiprocessing.Process(target=subprocess_target, args=(self.queue_in, self.queue_out, config))
+    def start_in_subprocess(self, config, server_uuid):
+        process = multiprocessing.Process(target=subprocess_target, args=(self.queue_in, self.queue_out, config, server_uuid))
         process.start()
         return process
 
     @contextlib.contextmanager
-    def with_server(self, config):
-        process = self.start_in_subprocess(config)
+    def with_server(self, config, server_uuid=None):
+        if server_uuid is None:
+            server_uuid = str(uuid4())
+        process = self.start_in_subprocess(config, server_uuid)
         msg = self.queue_out.get()
         if msg != 'ready':
             raise RuntimeError('never got ready message from subprocess')
         try:
             yield self
         finally:
             self.queue_in.put('stop')
+            process.join(timeout=2)
+            if not process.is_alive():
+                return  # exited because we told it to
+
+            # Subprocess did not behave
             process.terminate()  # SIGTERM
             # Poll to close process resources, due to race condition where it is not still running
             for i in range(3):

test_app/tests/task/test_multi_server.py

@@ -0,0 +1,24 @@
+from uuid import uuid4
+import time
+
+import pytest
+
+from ansible_base.task import get_config
+
+from test_app.tasks import create_uuid_entry
+from test_app.models import UUIDModel
+
+
+@pytest.mark.django_db
+def test_run_task_two_servers(dispatcher_subprocess):
+    with dispatcher_subprocess(get_config(), server_uuid=1) as server1:
+        with dispatcher_subprocess(get_config(), server_uuid=2) as server2:
+            my_uuid = str(uuid4())
+            create_uuid_entry.delay(uuid=my_uuid)
+
+            for i in range(20):
+                if UUIDModel.objects.filter(id=my_uuid).exists():
+                    break
+                time.sleep(0.05)
+            else:
+                assert f'Task never inserted UUID entry with {my_uuid}'