From b8b7e5454c3633c3fdfd526306b4a3f97c2cb4b3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=88=98=E5=AE=9D?= <po.lb@antfin.com>
Date: Wed, 11 May 2022 22:20:21 +0800
Subject: [PATCH] Use @pytest_asyncio.fixture instead of @pytest.fixture

---
 mars/conftest.py                                           | 3 ++-
 mars/dataframe/contrib/raydataset/tests/test_raydataset.py | 3 ++-
 mars/deploy/oscar/tests/test_fault_injection.py            | 3 ++-
 mars/deploy/oscar/tests/test_local.py                      | 5 +++--
 mars/deploy/oscar/tests/test_ray.py                        | 3 ++-
 mars/deploy/oscar/tests/test_ray_fault_injection.py        | 3 ++-
 mars/deploy/oscar/tests/test_ray_scheduling.py             | 3 ++-
 mars/oscar/backends/mars/tests/test_debug.py               | 3 ++-
 mars/oscar/backends/mars/tests/test_mars_actor_context.py  | 3 ++-
 mars/oscar/backends/ray/tests/test_ray_actor_context.py    | 3 ++-
 mars/oscar/backends/ray/tests/test_ray_actor_driver.py     | 3 ++-
 mars/services/cluster/tests/test_api.py                    | 3 ++-
 mars/services/cluster/tests/test_locator.py                | 3 ++-
 mars/services/cluster/tests/test_procinfo.py               | 3 ++-
 mars/services/cluster/tests/test_service.py                | 3 ++-
 mars/services/cluster/tests/test_uploader.py               | 3 ++-
 mars/services/mutable/tests/test_mutable.py                | 3 ++-
 mars/services/scheduling/supervisor/tests/test_assigner.py | 3 ++-
 .../scheduling/supervisor/tests/test_globalresource.py     | 3 ++-
 mars/services/scheduling/supervisor/tests/test_manager.py  | 3 ++-
 .../scheduling/supervisor/tests/test_queue_balance.py      | 3 ++-
 mars/services/scheduling/supervisor/tests/test_queueing.py | 3 ++-
 .../scheduling/supervisor/tests/test_speculation.py        | 3 ++-
 mars/services/scheduling/tests/test_service.py             | 3 ++-
 mars/services/scheduling/worker/tests/test_execution.py    | 3 ++-
 mars/services/scheduling/worker/tests/test_quota.py        | 3 ++-
 mars/services/scheduling/worker/tests/test_workerslot.py   | 3 ++-
 mars/services/storage/tests/test_service.py                | 5 +++--
 mars/services/storage/tests/test_spill.py                  | 7 ++++---
 mars/services/storage/tests/test_transfer.py               | 5 +++--
 mars/services/subtask/tests/test_service.py                | 3 ++-
 mars/services/subtask/worker/tests/test_subtask.py         | 3 ++-
 mars/services/task/supervisor/tests/test_task_manager.py   | 3 ++-
 mars/services/task/tests/test_service.py                   | 7 ++++---
 mars/services/tests/test_core.py                           | 3 ++-
 mars/services/web/tests/test_core.py                       | 3 ++-
 mars/storage/tests/test_libs.py                            | 3 ++-
 37 files changed, 81 insertions(+), 44 deletions(-)

diff --git a/mars/conftest.py b/mars/conftest.py
index b18722d338..7db4f26c8e 100644
--- a/mars/conftest.py
+++ b/mars/conftest.py
@@ -18,6 +18,7 @@
 
 import psutil
 import pytest
+import pytest_asyncio
 
 from mars.config import option_context
 from mars.core.mode import is_kernel_mode, is_build_mode
@@ -132,7 +133,7 @@ def stop_ray(request):  # pragma: no cover
         ray.shutdown()
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def ray_create_mars_cluster(request):
     from mars.deploy.oscar.ray import new_cluster, _load_config
 
diff --git a/mars/dataframe/contrib/raydataset/tests/test_raydataset.py b/mars/dataframe/contrib/raydataset/tests/test_raydataset.py
index c37e74014e..f867777876 100644
--- a/mars/dataframe/contrib/raydataset/tests/test_raydataset.py
+++ b/mars/dataframe/contrib/raydataset/tests/test_raydataset.py
@@ -16,6 +16,7 @@
 import numpy as np
 import pandas as pd
 import pytest
+import pytest_asyncio
 
 from ..... import dataframe as md
 from .....deploy.oscar.ray import new_cluster
@@ -39,7 +40,7 @@
     sklearn = None
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def create_cluster(request):
     client = await new_cluster(
         supervisor_mem=256 * 1024**2,
diff --git a/mars/deploy/oscar/tests/test_fault_injection.py b/mars/deploy/oscar/tests/test_fault_injection.py
index 3050529f26..94df297ecd 100644
--- a/mars/deploy/oscar/tests/test_fault_injection.py
+++ b/mars/deploy/oscar/tests/test_fault_injection.py
@@ -14,6 +14,7 @@
 
 import os
 import pytest
+import pytest_asyncio
 import traceback
 import numpy as np
 import pandas as pd
@@ -40,7 +41,7 @@
 )
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def fault_cluster(request):
     param = getattr(request, "param", {})
     start_method = os.environ.get("POOL_START_METHOD", None)
diff --git a/mars/deploy/oscar/tests/test_local.py b/mars/deploy/oscar/tests/test_local.py
index 79ec06bd2f..f742ea3334 100644
--- a/mars/deploy/oscar/tests/test_local.py
+++ b/mars/deploy/oscar/tests/test_local.py
@@ -23,6 +23,7 @@
 import numpy as np
 import pandas as pd
 import pytest
+import pytest_asyncio
 
 try:
     import vineyard
@@ -103,7 +104,7 @@
 
 
 @pytest.mark.parametrize(indirect=True)
-@pytest.fixture(params=params)
+@pytest_asyncio.fixture(params=params)
 async def create_cluster(request):
     if request.param == "default":
         config = CONFIG_TEST_FILE
@@ -822,7 +823,7 @@ async def _exec():
 min_task_runtime = 2
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def speculative_cluster():
     config = _load_config()
     config["scheduling"]["speculation"]["enabled"] = True
diff --git a/mars/deploy/oscar/tests/test_ray.py b/mars/deploy/oscar/tests/test_ray.py
index c1056e73fc..74bca575d3 100644
--- a/mars/deploy/oscar/tests/test_ray.py
+++ b/mars/deploy/oscar/tests/test_ray.py
@@ -24,6 +24,7 @@
 import numpy as np
 import pandas as pd
 import pytest
+import pytest_asyncio
 
 import mars
 from .... import oscar as mo
@@ -90,7 +91,7 @@
 EXPECT_PROFILING_STRUCTURE_NO_SLOW["supervisor"]["slow_subtasks"] = {}
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def create_cluster(request):
     param = getattr(request, "param", {})
     ray_config = _load_config(CONFIG_FILE)
diff --git a/mars/deploy/oscar/tests/test_ray_fault_injection.py b/mars/deploy/oscar/tests/test_ray_fault_injection.py
index 499d2a18c9..c627d77ea6 100644
--- a/mars/deploy/oscar/tests/test_ray_fault_injection.py
+++ b/mars/deploy/oscar/tests/test_ray_fault_injection.py
@@ -15,6 +15,7 @@
 import os
 
 import pytest
+import pytest_asyncio
 
 from ....oscar.errors import ServerClosed
 from ....services.tests.fault_injection_manager import (
@@ -45,7 +46,7 @@
 }
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def fault_cluster(request):
     param = getattr(request, "param", {})
     ray_config = _load_config(RAY_CONFIG_FILE)
diff --git a/mars/deploy/oscar/tests/test_ray_scheduling.py b/mars/deploy/oscar/tests/test_ray_scheduling.py
index 96de2bcec8..09bb9a55b8 100644
--- a/mars/deploy/oscar/tests/test_ray_scheduling.py
+++ b/mars/deploy/oscar/tests/test_ray_scheduling.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import pytest
+import pytest_asyncio
 
 from ....tests.core import require_ray
 from ....utils import lazy_import
@@ -26,7 +27,7 @@
 ray = lazy_import("ray")
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def speculative_cluster():
     client = await new_cluster(
         "test_cluster",
diff --git a/mars/oscar/backends/mars/tests/test_debug.py b/mars/oscar/backends/mars/tests/test_debug.py
index c73836e656..14569f07a6 100644
--- a/mars/oscar/backends/mars/tests/test_debug.py
+++ b/mars/oscar/backends/mars/tests/test_debug.py
@@ -21,6 +21,7 @@
 from typing import List
 
 import pytest
+import pytest_asyncio
 
 from ..... import oscar as mo
 from ....debug import reload_debug_opts_from_env, get_debug_options
@@ -62,7 +63,7 @@ async def call_self_ref(self):
         await self.ref().wait(1)
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     start_method = (
         os.environ.get("POOL_START_METHOD", "forkserver")
diff --git a/mars/oscar/backends/mars/tests/test_mars_actor_context.py b/mars/oscar/backends/mars/tests/test_mars_actor_context.py
index 0cd00b762a..dc2133f19a 100644
--- a/mars/oscar/backends/mars/tests/test_mars_actor_context.py
+++ b/mars/oscar/backends/mars/tests/test_mars_actor_context.py
@@ -22,6 +22,7 @@
 
 import pandas as pd
 import pytest
+import pytest_asyncio
 
 from ..... import oscar as mo
 from .....oscar.core import ActorRef, LocalActorRef
@@ -241,7 +242,7 @@ def get_call_log(self):
 
 
 @pytest.mark.parametrize(indirect=True)
-@pytest.fixture(params=[False, True])
+@pytest_asyncio.fixture(params=[False, True])
 async def actor_pool(request):
     start_method = (
         os.environ.get("POOL_START_METHOD", "forkserver")
diff --git a/mars/oscar/backends/ray/tests/test_ray_actor_context.py b/mars/oscar/backends/ray/tests/test_ray_actor_context.py
index 537f70cc17..f7ec075ba3 100644
--- a/mars/oscar/backends/ray/tests/test_ray_actor_context.py
+++ b/mars/oscar/backends/ray/tests/test_ray_actor_context.py
@@ -16,6 +16,7 @@
 import time
 
 import pytest
+import pytest_asyncio
 
 from .....utils import lazy_import
 from .....tests.core import require_ray
@@ -29,7 +30,7 @@
 ray = lazy_import("ray")
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool_context():
     pg_name, n_process = f"ray_cluster_{time.time_ns()}", 2
     from .....serialization.ray import (
diff --git a/mars/oscar/backends/ray/tests/test_ray_actor_driver.py b/mars/oscar/backends/ray/tests/test_ray_actor_driver.py
index 4a508dc341..003440b795 100644
--- a/mars/oscar/backends/ray/tests/test_ray_actor_driver.py
+++ b/mars/oscar/backends/ray/tests/test_ray_actor_driver.py
@@ -16,6 +16,7 @@
 import os
 
 import pytest
+import pytest_asyncio
 
 from ..... import oscar as mo
 from .....tests.core import require_ray
@@ -52,7 +53,7 @@ def index(self):
         return self._index
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def mars_cluster():
     mo.setup_cluster(address_to_resources=TEST_ADDRESS_TO_RESOURCES)
     main_pool_handles = []  # Hold actor_handle to avoid actor being freed.
diff --git a/mars/services/cluster/tests/test_api.py b/mars/services/cluster/tests/test_api.py
index 203d43737d..df5190b36a 100644
--- a/mars/services/cluster/tests/test_api.py
+++ b/mars/services/cluster/tests/test_api.py
@@ -15,6 +15,7 @@
 import asyncio
 
 import pytest
+import pytest_asyncio
 
 from .... import oscar as mo
 from ....utils import get_next_port
@@ -25,7 +26,7 @@
 from ..core import NodeStatus
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     pool = await mo.create_actor_pool("127.0.0.1", n_process=0)
     await pool.start()
diff --git a/mars/services/cluster/tests/test_locator.py b/mars/services/cluster/tests/test_locator.py
index 8bc849d50a..0b64f8cd50 100644
--- a/mars/services/cluster/tests/test_locator.py
+++ b/mars/services/cluster/tests/test_locator.py
@@ -18,6 +18,7 @@
 from typing import List
 
 import pytest
+import pytest_asyncio
 
 from .... import oscar as mo
 from ....utils import Timer
@@ -52,7 +53,7 @@ def put_starting_nodes(self, nodes: List[str], role: NodeRole):
             self._node_infos[node] = NodeStatus.STARTING
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     pool = await mo.create_actor_pool("127.0.0.1", n_process=0)
     await pool.start()
diff --git a/mars/services/cluster/tests/test_procinfo.py b/mars/services/cluster/tests/test_procinfo.py
index b084d328c0..e63daad0e0 100644
--- a/mars/services/cluster/tests/test_procinfo.py
+++ b/mars/services/cluster/tests/test_procinfo.py
@@ -13,12 +13,13 @@
 # limitations under the License.
 
 import pytest
+import pytest_asyncio
 
 from .... import oscar as mo
 from ..procinfo import ProcessInfoManagerActor
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     pool = await mo.create_actor_pool(
         "127.0.0.1", n_process=2, labels=["main", "numa-0", "gpu-0"]
diff --git a/mars/services/cluster/tests/test_service.py b/mars/services/cluster/tests/test_service.py
index 653c6d1e8c..4ff436d454 100644
--- a/mars/services/cluster/tests/test_service.py
+++ b/mars/services/cluster/tests/test_service.py
@@ -16,6 +16,7 @@
 import os
 
 import pytest
+import pytest_asyncio
 
 from .... import oscar as mo
 from ....storage import StorageLevel
@@ -23,7 +24,7 @@
 from .. import ClusterAPI, WorkerSlotInfo, QuotaInfo, StorageInfo, DiskInfo
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pools():
     async def start_pool():
         pool = await mo.create_actor_pool("127.0.0.1", n_process=0)
diff --git a/mars/services/cluster/tests/test_uploader.py b/mars/services/cluster/tests/test_uploader.py
index 20d04b9ef2..36fc3af1ba 100644
--- a/mars/services/cluster/tests/test_uploader.py
+++ b/mars/services/cluster/tests/test_uploader.py
@@ -15,6 +15,7 @@
 import asyncio
 
 import pytest
+import pytest_asyncio
 
 from .... import oscar as mo
 from ... import NodeRole
@@ -23,7 +24,7 @@
 from ..uploader import NodeInfoUploaderActor
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     pool = await mo.create_actor_pool("127.0.0.1", n_process=0)
     await pool.start()
diff --git a/mars/services/mutable/tests/test_mutable.py b/mars/services/mutable/tests/test_mutable.py
index 03e6539ce3..d3c4563c55 100644
--- a/mars/services/mutable/tests/test_mutable.py
+++ b/mars/services/mutable/tests/test_mutable.py
@@ -17,6 +17,7 @@
 import sys
 
 import pytest
+import pytest_asyncio
 import numpy as np
 
 from ....deploy.oscar.local import new_cluster
@@ -28,7 +29,7 @@
 _is_windows = sys.platform.lower().startswith("win")
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def create_cluster():
     client = await new_cluster(n_worker=2, n_cpu=2, web=True)
     async with client:
diff --git a/mars/services/scheduling/supervisor/tests/test_assigner.py b/mars/services/scheduling/supervisor/tests/test_assigner.py
index cfccf9a65f..bb28969fde 100644
--- a/mars/services/scheduling/supervisor/tests/test_assigner.py
+++ b/mars/services/scheduling/supervisor/tests/test_assigner.py
@@ -15,6 +15,7 @@
 import numpy as np
 import asyncio
 import pytest
+import pytest_asyncio
 
 from ..... import oscar as mo
 from .....core import ChunkGraph
@@ -100,7 +101,7 @@ async def create(cls, address: str, **kw):
         return api
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool(request):
     pool = await mo.create_actor_pool("127.0.0.1", n_process=0)
     with_gpu = request.param
diff --git a/mars/services/scheduling/supervisor/tests/test_globalresource.py b/mars/services/scheduling/supervisor/tests/test_globalresource.py
index c78478a15b..64769af4e6 100644
--- a/mars/services/scheduling/supervisor/tests/test_globalresource.py
+++ b/mars/services/scheduling/supervisor/tests/test_globalresource.py
@@ -15,6 +15,7 @@
 import asyncio
 
 import pytest
+import pytest_asyncio
 
 from ..... import oscar as mo
 from .....resource import Resource
@@ -23,7 +24,7 @@
 from ...supervisor import GlobalResourceManagerActor
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     pool = await mo.create_actor_pool("127.0.0.1", n_process=0)
 
diff --git a/mars/services/scheduling/supervisor/tests/test_manager.py b/mars/services/scheduling/supervisor/tests/test_manager.py
index 2bf30d73c9..91d1d7c172 100644
--- a/mars/services/scheduling/supervisor/tests/test_manager.py
+++ b/mars/services/scheduling/supervisor/tests/test_manager.py
@@ -17,6 +17,7 @@
 from typing import List, Tuple, Set
 
 import pytest
+import pytest_asyncio
 
 from ..... import oscar as mo
 from .....typing import BandType
@@ -98,7 +99,7 @@ async def wait_subtask(self, subtask_id: str, band_name: str):
             pass
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     pool = await mo.create_actor_pool("127.0.0.1", n_process=0)
 
diff --git a/mars/services/scheduling/supervisor/tests/test_queue_balance.py b/mars/services/scheduling/supervisor/tests/test_queue_balance.py
index e3a51fbf74..194c1aba23 100644
--- a/mars/services/scheduling/supervisor/tests/test_queue_balance.py
+++ b/mars/services/scheduling/supervisor/tests/test_queue_balance.py
@@ -14,6 +14,7 @@
 
 import asyncio
 import pytest
+import pytest_asyncio
 from collections import defaultdict
 from typing import Tuple, List
 from ..... import oscar as mo
@@ -147,7 +148,7 @@ def dump_data(self):
         return self._submitted_subtask_ids
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     pool = await mo.create_actor_pool("127.0.0.1", n_process=0)
 
diff --git a/mars/services/scheduling/supervisor/tests/test_queueing.py b/mars/services/scheduling/supervisor/tests/test_queueing.py
index 21bcd2d1ac..5ca948b72b 100644
--- a/mars/services/scheduling/supervisor/tests/test_queueing.py
+++ b/mars/services/scheduling/supervisor/tests/test_queueing.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import pytest
+import pytest_asyncio
 from typing import Tuple, List
 
 from ..... import oscar as mo
@@ -70,7 +71,7 @@ def dump_data(self):
         return self._subtask_ids, self._bands
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     pool = await mo.create_actor_pool("127.0.0.1", n_process=0)
 
diff --git a/mars/services/scheduling/supervisor/tests/test_speculation.py b/mars/services/scheduling/supervisor/tests/test_speculation.py
index bde3522821..e62399270f 100644
--- a/mars/services/scheduling/supervisor/tests/test_speculation.py
+++ b/mars/services/scheduling/supervisor/tests/test_speculation.py
@@ -16,6 +16,7 @@
 from typing import List, Tuple, Set
 
 import pytest
+import pytest_asyncio
 
 from ..... import oscar as mo
 from ....cluster import MockClusterAPI
@@ -54,7 +55,7 @@ async def get_exceptions(self):
         return self._exceptions
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     pool = await mo.create_actor_pool("127.0.0.1", n_process=0)
 
diff --git a/mars/services/scheduling/tests/test_service.py b/mars/services/scheduling/tests/test_service.py
index 985a1ea821..0f1f4b3254 100644
--- a/mars/services/scheduling/tests/test_service.py
+++ b/mars/services/scheduling/tests/test_service.py
@@ -18,6 +18,7 @@
 
 import numpy as np
 import pytest
+import pytest_asyncio
 
 from ..api.web import WebSchedulingAPI
 from .... import oscar as mo
@@ -89,7 +90,7 @@ def _approx_resource(actual, expect):
     )
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pools():
     async def start_pool(is_worker: bool):
         if is_worker:
diff --git a/mars/services/scheduling/worker/tests/test_execution.py b/mars/services/scheduling/worker/tests/test_execution.py
index abeb50ba7f..59f7c9a298 100644
--- a/mars/services/scheduling/worker/tests/test_execution.py
+++ b/mars/services/scheduling/worker/tests/test_execution.py
@@ -23,6 +23,7 @@
 import numpy as np
 import pandas as pd
 import pytest
+import pytest_asyncio
 
 from ..... import oscar as mo
 from ..... import remote as mr
@@ -145,7 +146,7 @@ def get_results(self):
         return self._results
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool(request):
     n_slots, enable_kill = request.param
     pool = await mo.create_actor_pool(
diff --git a/mars/services/scheduling/worker/tests/test_quota.py b/mars/services/scheduling/worker/tests/test_quota.py
index b011b77c06..14fb33b22f 100644
--- a/mars/services/scheduling/worker/tests/test_quota.py
+++ b/mars/services/scheduling/worker/tests/test_quota.py
@@ -19,6 +19,7 @@
 from typing import Union
 
 import pytest
+import pytest_asyncio
 
 from ..... import oscar as mo
 from .....tests.core import mock
@@ -36,7 +37,7 @@ def restart_free_slots(self):
         self._restart_record = True
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     start_method = (
         os.environ.get("POOL_START_METHOD", "fork") if sys.platform != "win32" else None
diff --git a/mars/services/scheduling/worker/tests/test_workerslot.py b/mars/services/scheduling/worker/tests/test_workerslot.py
index 61a257bba7..eefeb547c2 100644
--- a/mars/services/scheduling/worker/tests/test_workerslot.py
+++ b/mars/services/scheduling/worker/tests/test_workerslot.py
@@ -20,6 +20,7 @@
 
 import psutil
 import pytest
+import pytest_asyncio
 import pandas as pd
 
 from ..... import oscar as mo
@@ -47,7 +48,7 @@ def get_result(self):
         return self._result
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool(request):
     start_method = (
         os.environ.get("POOL_START_METHOD", "forkserver")
diff --git a/mars/services/storage/tests/test_service.py b/mars/services/storage/tests/test_service.py
index b909f3fb39..1422bf4f8a 100644
--- a/mars/services/storage/tests/test_service.py
+++ b/mars/services/storage/tests/test_service.py
@@ -18,6 +18,7 @@
 import numpy as np
 import pandas as pd
 import pytest
+import pytest_asyncio
 
 from .... import oscar as mo
 from ....resource import Resource
@@ -31,7 +32,7 @@
 _is_windows = sys.platform.lower().startswith("win")
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pools():
     async def start_pool():
         start_method = (
@@ -120,7 +121,7 @@ async def test_storage_service(actor_pools):
     )
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pools_with_gpu():
     async def start_pool():
         start_method = (
diff --git a/mars/services/storage/tests/test_spill.py b/mars/services/storage/tests/test_spill.py
index 52fa669ed0..57b6b10961 100644
--- a/mars/services/storage/tests/test_spill.py
+++ b/mars/services/storage/tests/test_spill.py
@@ -19,6 +19,7 @@
 
 import numpy as np
 import pytest
+import pytest_asyncio
 
 from .... import oscar as mo
 from ....storage import StorageLevel, PlasmaStorage
@@ -34,7 +35,7 @@
 MEMORY_SIZE = 100 * 1024
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     async def start_pool():
         start_method = (
@@ -71,7 +72,7 @@ def _build_storage_config():
     return storage_configs
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def create_actors(actor_pool):
     storage_configs = _build_storage_config()
     manager_ref = await mo.create_actor(
@@ -155,7 +156,7 @@ async def put(
         return data_info
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def create_actors_with_delay(actor_pool):
     storage_configs = _build_storage_config()
     manager_ref = await mo.create_actor(
diff --git a/mars/services/storage/tests/test_transfer.py b/mars/services/storage/tests/test_transfer.py
index aa999cdf77..f5796caa28 100644
--- a/mars/services/storage/tests/test_transfer.py
+++ b/mars/services/storage/tests/test_transfer.py
@@ -19,6 +19,7 @@
 import numpy as np
 import pandas as pd
 import pytest
+import pytest_asyncio
 
 from .... import oscar as mo
 from ....oscar.backends.allocate_strategy import IdleLabel
@@ -31,7 +32,7 @@
 _is_windows = sys.platform.lower().startswith("win")
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pools():
     async def start_pool():
         start_method = (
@@ -56,7 +57,7 @@ async def start_pool():
     await worker_pool_2.stop()
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def create_actors(actor_pools):
     worker_pool_1, worker_pool_2 = actor_pools
 
diff --git a/mars/services/subtask/tests/test_service.py b/mars/services/subtask/tests/test_service.py
index f73323c524..bcf62631e1 100644
--- a/mars/services/subtask/tests/test_service.py
+++ b/mars/services/subtask/tests/test_service.py
@@ -17,6 +17,7 @@
 
 import numpy as np
 import pytest
+import pytest_asyncio
 
 from .... import oscar as mo
 from .... import tensor as mt
@@ -49,7 +50,7 @@ def _gen_subtask(t, session_id):
     return subtask
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pools():
     async def start_pool(is_worker: bool):
         if is_worker:
diff --git a/mars/services/subtask/worker/tests/test_subtask.py b/mars/services/subtask/worker/tests/test_subtask.py
index c0be016fd9..8d922d2fa6 100644
--- a/mars/services/subtask/worker/tests/test_subtask.py
+++ b/mars/services/subtask/worker/tests/test_subtask.py
@@ -19,6 +19,7 @@
 
 import numpy as np
 import pytest
+import pytest_asyncio
 
 from ..... import oscar as mo
 from ..... import tensor as mt
@@ -48,7 +49,7 @@ def set_subtask_result(self, subtask_result: SubtaskResult):
         return
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     start_method = (
         os.environ.get("POOL_START_METHOD", "forkserver")
diff --git a/mars/services/task/supervisor/tests/test_task_manager.py b/mars/services/task/supervisor/tests/test_task_manager.py
index 23dd84af3b..2131f25f12 100644
--- a/mars/services/task/supervisor/tests/test_task_manager.py
+++ b/mars/services/task/supervisor/tests/test_task_manager.py
@@ -22,6 +22,7 @@
 import numpy as np
 import pandas as pd
 import pytest
+import pytest_asyncio
 
 from ..... import dataframe as md
 from ..... import oscar as mo
@@ -46,7 +47,7 @@
 from ..manager import TaskConfigurationActor, TaskManagerActor
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool(request):
     param = getattr(request, "param", {})
     backend = param.get("backend", "mars")
diff --git a/mars/services/task/tests/test_service.py b/mars/services/task/tests/test_service.py
index 5eed76977f..9c79072e25 100644
--- a/mars/services/task/tests/test_service.py
+++ b/mars/services/task/tests/test_service.py
@@ -18,6 +18,7 @@
 import numpy as np
 import pandas as pd
 import pytest
+import pytest_asyncio
 
 from .... import dataframe as md
 from .... import oscar as mo
@@ -40,7 +41,7 @@
 from ..errors import TaskNotExist
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pools():
     async def start_pool(is_worker: bool):
         if is_worker:
@@ -116,7 +117,7 @@ async def _start_services(
 
 
 @pytest.mark.parametrize(indirect=True)
-@pytest.fixture(params=[False, True])
+@pytest_asyncio.fixture(params=[False, True])
 async def start_test_service(actor_pools, request):
     sv_pool, worker_pool = actor_pools
 
@@ -141,7 +142,7 @@ def _get_decref_stage_chunk_keys(cls, stage_processor):
 
 
 @pytest.mark.parametrize(indirect=True)
-@pytest.fixture(params=[True])
+@pytest_asyncio.fixture(params=[True])
 async def start_test_service_with_mock(actor_pools, request):
     sv_pool, worker_pool = actor_pools
 
diff --git a/mars/services/tests/test_core.py b/mars/services/tests/test_core.py
index dffd3954c9..5a674c0765 100644
--- a/mars/services/tests/test_core.py
+++ b/mars/services/tests/test_core.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import pytest
+import pytest_asyncio
 from tornado import httpclient
 
 from ... import oscar as mo
@@ -26,7 +27,7 @@
 )
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool_context():
     pool = await mo.create_actor_pool(f"127.0.0.1:{get_next_port()}", n_process=0)
     await pool.start()
diff --git a/mars/services/web/tests/test_core.py b/mars/services/web/tests/test_core.py
index 72dca5ec94..b7712ac5eb 100644
--- a/mars/services/web/tests/test_core.py
+++ b/mars/services/web/tests/test_core.py
@@ -17,6 +17,7 @@
 import sys
 
 import pytest
+import pytest_asyncio
 from tornado import httpclient
 
 from .... import oscar as mo
@@ -59,7 +60,7 @@ async def get_with_timeout(self, test_id):
         raise ValueError(test_id)
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
 async def actor_pool():
     start_method = (
         os.environ.get("POOL_START_METHOD", "forkserver")
diff --git a/mars/storage/tests/test_libs.py b/mars/storage/tests/test_libs.py
index 8fd5c5c804..eb63d93c0c 100644
--- a/mars/storage/tests/test_libs.py
+++ b/mars/storage/tests/test_libs.py
@@ -19,6 +19,7 @@
 import tempfile
 import pkgutil
 import pytest
+import pytest_asyncio
 
 import numpy as np
 import pandas as pd
@@ -65,7 +66,7 @@
 @pytest.mark.parametrize(
     "ray_start_regular", [{"enable": ray is not None}], indirect=True
 )
-@pytest.fixture(params=params)
+@pytest_asyncio.fixture(params=params)
 async def storage_context(ray_start_regular, request):
     if request.param == "filesystem":
         tempdir = tempfile.mkdtemp()