Method run_briefly

tests/test_dirty_worker.py:639–655

()

            # Start the server in background
            async def run_briefly():
                # Remove existing socket
                if os.path.exists(socket_path):
                    os.unlink(socket_path)

                worker._server = await asyncio.start_unix_server(
                    worker.handle_connection,
                    path=socket_path
                )
                os.chmod(socket_path, 0o600)

                # Verify socket exists
                assert os.path.exists(socket_path)

                # Close immediately
                worker._server.close()
                await worker._server.wait_closed()

            await run_briefly()