Consolidate fixtures

Author: AlanCoding

tests/conftest.py

@@ -95,37 +95,47 @@ def pg_dispatcher() -> DispatcherMain:
 def test_settings():
     return DispatcherSettings(BASIC_CONFIG)
 
+
+@pytest.fixture
+def adispatcher_factory():
+    @contextlib.asynccontextmanager
+    async def _rf(config):
+        dispatcher = None
+        try:
+            settings = DispatcherSettings(config)
+            dispatcher = from_settings(settings=settings)
+
+            await dispatcher.connect_signals()
+            await dispatcher.start_working()
+            await dispatcher.wait_for_producers_ready()
+            await dispatcher.pool.events.workers_ready.wait()
+
+            assert dispatcher.pool.finished_count == 0  # sanity
+            assert dispatcher.control_count == 0
+
+            yield dispatcher
+        finally:
+            if dispatcher:
+                try:
+                    await dispatcher.shutdown()
+                    await dispatcher.cancel_tasks()
+                except Exception:
+                    logger.exception('shutdown had error')
+    return _rf
+
+
 @pytest_asyncio.fixture(
     loop_scope="function",
     scope="function",
     params=['ProcessManager', 'ForkServerManager'],
     ids=["fork", "forkserver"],
 )
-async def apg_dispatcher(request) -> AsyncIterator[DispatcherMain]:
-    dispatcher = None
-    try:
-        this_test_config = BASIC_CONFIG.copy()
-        this_test_config.setdefault('service', {})
-        this_test_config['service']['process_manager_cls'] = request.param
-        this_settings = DispatcherSettings(this_test_config)
-        dispatcher = from_settings(settings=this_settings)
-
-        await dispatcher.connect_signals()
-        await dispatcher.start_working()
-        await dispatcher.wait_for_producers_ready()
-        await dispatcher.pool.events.workers_ready.wait()
-
-        assert dispatcher.pool.finished_count == 0  # sanity
-        assert dispatcher.control_count == 0
-
+async def apg_dispatcher(request, adispatcher_factory) -> AsyncIterator[DispatcherMain]:
+    this_test_config = BASIC_CONFIG.copy()
+    this_test_config.setdefault('service', {})
+    this_test_config['service']['process_manager_cls'] = request.param
+    async with adispatcher_factory(this_test_config) as dispatcher:
         yield dispatcher
-    finally:
-        if dispatcher:
-            try:
-                await dispatcher.shutdown()
-                await dispatcher.cancel_tasks()
-            except Exception:
-                logger.exception('shutdown had error')
 
 
 @pytest_asyncio.fixture(loop_scope="function", scope="function")

tests/integration/test_socket_use.py

@@ -20,37 +20,19 @@ def sock_path(tmp_path_factory):
 
 
 @pytest.fixture(scope='session')
-def socket_settings(sock_path):
-    config = {"version": 2, "brokers": {"socket": {"socket_path": sock_path}}, "service": {"main_kwargs": {"node_id": "socket-test-server"}}}
-    return DispatcherSettings(config)
+def socket_config(sock_path):
+    return {"version": 2, "brokers": {"socket": {"socket_path": sock_path}}, "service": {"main_kwargs": {"node_id": "socket-test-server"}}}
 
 
-@pytest_asyncio.fixture
-async def asock_dispatcher(socket_settings) -> AsyncIterator[DispatcherMain]:
-    dispatcher = None
-    try:
-        dispatcher = from_settings(settings=socket_settings)
-        sock_pth = dispatcher.producers[0].broker.socket_path
-        assert not os.path.exists(sock_pth)
-
-        await dispatcher.connect_signals()
-        await dispatcher.start_working()
-        await dispatcher.wait_for_producers_ready()
-        await dispatcher.pool.events.workers_ready.wait()
-
-        assert os.path.exists(sock_pth)
+@pytest.fixture(scope='session')
+def socket_settings(socket_config):
+    return DispatcherSettings(socket_config)
 
-        assert dispatcher.pool.finished_count == 0  # sanity
-        assert dispatcher.control_count == 0
 
+@pytest_asyncio.fixture
+async def asock_dispatcher(socket_config, adispatcher_factory) -> AsyncIterator[DispatcherMain]:
+    async with adispatcher_factory(socket_config) as dispatcher:
         yield dispatcher
-    finally:
-        if dispatcher:
-            try:
-                await dispatcher.shutdown()
-                await dispatcher.cancel_tasks()
-            except Exception:
-                logger.exception('shutdown had error')
 
 
 @pytest_asyncio.fixture