diff --git a/.github/workflows/bindings_python.yml b/.github/workflows/bindings_python.yml
index a93a72eb3a1a..30d8dd7e1bce 100644
--- a/.github/workflows/bindings_python.yml
+++ b/.github/workflows/bindings_python.yml
@@ -57,7 +57,7 @@ jobs:
         env:
           OPENDAL_MEMORY_TEST: on
         run: |
-          pytest -vk TestMemory
+          pytest --service_type=service_memory -v ./tests
 
   linux:
     runs-on: ubuntu-latest
diff --git a/.github/workflows/service_test_memory.yml b/.github/workflows/service_test_memory.yml
index a955040b1038..efd7dca213fb 100644
--- a/.github/workflows/service_test_memory.yml
+++ b/.github/workflows/service_test_memory.yml
@@ -71,6 +71,6 @@ jobs:
       - name: Test
         shell: bash
         working-directory: bindings/python
-        run: pytest -vk TestMemory
+        run: pytest --service_type=service_memory -v ./tests
         env:
           OPENDAL_MEMORY_TEST: on
diff --git a/.github/workflows/service_test_s3.yml b/.github/workflows/service_test_s3.yml
index 95f703ab7095..95ffa0a963e6 100644
--- a/.github/workflows/service_test_s3.yml
+++ b/.github/workflows/service_test_s3.yml
@@ -281,7 +281,7 @@ jobs:
       - name: Test
         shell: bash
         working-directory: bindings/python
-        run: pytest -vk TestS3
+        run: pytest --service_type=service_s3 -v ./tests
         env:
           OPENDAL_S3_TEST: on
           OPENDAL_S3_BUCKET: test
diff --git a/bindings/python/CONTRIBUTING.md b/bindings/python/CONTRIBUTING.md
index 9b428b10c65b..81e808592bd2 100644
--- a/bindings/python/CONTRIBUTING.md
+++ b/bindings/python/CONTRIBUTING.md
@@ -71,7 +71,7 @@ OpenDAL adopts `pytest` for behavior tests:
 
 ```shell
 maturin develop -E test
-OPENDAL_MEMORY_TEST=on pytest -vk TestMemory
+OPENDAL_MEMORY_TEST=on pytest --service_type=service_memory -v ./tests
 ```
 
 ## Docs
diff --git a/bindings/python/README.md b/bindings/python/README.md
index 871a0744263b..52b508031985 100644
--- a/bindings/python/README.md
+++ b/bindings/python/README.md
@@ -66,7 +66,7 @@ Run some tests:
 
 ```shell
 maturin develop -E test
-OPENDAL_MEMORY_TEST=on pytest -vk TestMemory
+OPENDAL_MEMORY_TEST=on pytest --service_type=service_memory -v ./tests
 ```
 
 Build API docs:
diff --git a/bindings/python/tests/conftest.py b/bindings/python/tests/conftest.py
index 9db7dbe92b8f..54e50552368d 100644
--- a/bindings/python/tests/conftest.py
+++ b/bindings/python/tests/conftest.py
@@ -15,9 +15,63 @@
 # specific language governing permissions and limitations
 # under the License.
 
-from dotenv import load_dotenv
+# from dotenv import load_dotenv
+import pytest
+import os
+
+import opendal
 import pytest
 
 
-load_dotenv()
+# load_dotenv()
 pytest_plugins = ("pytest_asyncio",)
+
+SERVICE_TYPE_INDEX = {
+    "service_memory": "memory",
+    "service_s3": "s3",
+    "service_fs": "fs",
+}
+
+
+def pytest_addoption(parser):
+    parser.addoption("--service_type", action="store", default="")
+
+
+def setup_config(service_name):
+    prefix = f"opendal_{service_name}_"
+    config = {}
+    for key in os.environ.keys():
+        if key.lower().startswith(prefix):
+            config[key[len(prefix) :].lower()] = os.environ.get(key)
+
+    # Check if current test be enabled.
+    test_flag = config.get("test", "")
+    if test_flag != "on" and test_flag != "true":
+        raise ValueError(f"Service {service_name} test is not enabled.")
+    return config
+
+
+def operator(service_name, setup_config):
+    return opendal.Operator(service_name, **setup_config)
+
+
+def async_operator(service_name, setup_config):
+    return opendal.AsyncOperator(service_name, **setup_config)
+
+
+def pytest_generate_tests(metafunc):
+    service_types = metafunc.config.getoption("service_type").split(",")
+    fixtures = []
+    for service_type in service_types:
+        service_name = SERVICE_TYPE_INDEX.get(service_type)
+        if service_name is None:
+            raise ValueError(f"Service {service_type} is not supported.")
+        config = setup_config(service_name)
+        fixtures.append(
+            (
+                service_name,
+                operator(service_name, config),
+                async_operator(service_name, config),
+            )
+        )
+    metafunc.parametrize("service_name, operator, async_operator", fixtures)
diff --git a/bindings/python/tests/test_services.py b/bindings/python/tests/test_services.py
index 836988e7d109..52949d57cb5d 100644
--- a/bindings/python/tests/test_services.py
+++ b/bindings/python/tests/test_services.py
@@ -16,198 +16,178 @@
 # under the License.
 
 import os
-from abc import ABC
 from uuid import uuid4
 from random import randint
 
-import opendal
 import pytest
 
 
-class AbstractTestSuite(ABC):
-    service_name = ""
-
-    def setup_method(self):
-        # Read arguments from envs.
-        prefix = f"opendal_{self.service_name}_"
-        self.config = {}
-        for key in os.environ.keys():
-            if key.lower().startswith(prefix):
-                self.config[key[len(prefix) :].lower()] = os.environ.get(key)
-
-        # Check if current test be enabled.
-        test_flag = self.config.get("test", "")
-        if test_flag != "on" and test_flag != "true":
-            raise ValueError(f"Service {self.service_name} test is not enabled.")
-
-        self.operator = opendal.Operator(self.service_name, **self.config)
-        self.async_operator = opendal.AsyncOperator(self.service_name, **self.config)
-
-    def test_sync_read(self):
-        size = randint(1, 1024)
-        filename = f"random_file_{str(uuid4())}"
-        content = os.urandom(size)
-        self.operator.write(filename, content)
-
-        read_content = self.operator.read(filename)
-        assert read_content is not None
-        assert read_content == content
-
-        self.operator.delete(filename)
-
-    @pytest.mark.asyncio
-    async def test_async_read(self):
-        size = randint(1, 1024)
-        filename = f"random_file_{str(uuid4())}"
-        content = os.urandom(size)
-        await self.async_operator.write(filename, content)
-
-        read_content = await self.async_operator.read(filename)
-        assert read_content is not None
-        assert read_content == content
-
-        await self.async_operator.delete(filename)
-
-    def test_sync_read_stat(self):
-        size = randint(1, 1024)
-        filename = f"random_file_{str(uuid4())}"
-        content = os.urandom(size)
-        self.operator.write(filename, content)
-
-        metadata = self.operator.stat(filename)
-        assert metadata is not None
-        assert metadata.content_length == len(content)
-        assert metadata.mode.is_file()
-
-        self.operator.delete(filename)
-
-    @pytest.mark.asyncio
-    async def test_async_read_stat(self):
-        size = randint(1, 1024)
-        filename = f"random_file_{str(uuid4())}"
-        content = os.urandom(size)
-        await self.async_operator.write(filename, content)
-
-        metadata = await self.async_operator.stat(filename)
-        assert metadata is not None
-        assert metadata.content_length == len(content)
-        assert metadata.mode.is_file()
-
-        await self.async_operator.delete(filename)
-
-        self.operator.delete(filename)
-
-    def test_sync_read_not_exists(self):
-        with pytest.raises(FileNotFoundError):
-            self.operator.read(str(uuid4()))
-
-    @pytest.mark.asyncio
-    async def test_async_read_not_exists(self):
-        with pytest.raises(FileNotFoundError):
-            await self.async_operator.read(str(uuid4()))
-
-    def test_sync_write(self):
-        size = randint(1, 1024)
-        filename = f"test_file_{str(uuid4())}.txt"
-        content = os.urandom(size)
-        size = len(content)
-        self.operator.write(filename, content)
-        metadata = self.operator.stat(filename)
-        assert metadata is not None
-        assert metadata.mode.is_file()
-        assert metadata.content_length == size
-
-        self.operator.delete(filename)
-
-    @pytest.mark.asyncio
-    async def test_async_write(self):
-        size = randint(1, 1024)
-        filename = f"test_file_{str(uuid4())}.txt"
-        content = os.urandom(size)
-        size = len(content)
-        await self.async_operator.write(filename, content)
-        metadata = await self.async_operator.stat(filename)
-        assert metadata is not None
-        assert metadata.mode.is_file()
-        assert metadata.content_length == size
-
-        await self.async_operator.delete(filename)
-
-    def test_sync_write_with_non_ascii_name(self):
-        size = randint(1, 1024)
-        filename = f"❌😱中文_{str(uuid4())}.test"
-        content = os.urandom(size)
-        size = len(content)
-        self.operator.write(filename, content)
-        metadata = self.operator.stat(filename)
-        assert metadata is not None
-        assert metadata.mode.is_file()
-        assert metadata.content_length == size
-
-        self.operator.delete(filename)
-
-    @pytest.mark.asyncio
-    async def test_async_write_with_non_ascii_name(self):
-        size = randint(1, 1024)
-        filename = f"❌😱中文_{str(uuid4())}.test"
-        content = os.urandom(size)
-        size = len(content)
-        await self.async_operator.write(filename, content)
-        metadata = await self.async_operator.stat(filename)
-        assert metadata is not None
-        assert metadata.mode.is_file()
-        assert metadata.content_length == size
-
-        await self.async_operator.delete(filename)
-
-    def test_sync_create_dir(self):
-        path = f"test_dir_{str(uuid4())}/"
-        self.operator.create_dir(path)
-        metadata = self.operator.stat(path)
-        assert metadata is not None
-        assert metadata.mode.is_dir()
-
-        self.operator.delete(path)
-
-    @pytest.mark.asyncio
-    async def test_async_create_dir(self):
-        path = f"test_dir_{str(uuid4())}/"
-        await self.async_operator.create_dir(path)
-        metadata = await self.async_operator.stat(path)
-        assert metadata is not None
-        assert metadata.mode.is_dir()
-
-        await self.async_operator.delete(path)
-
-    def test_sync_delete(self):
-        size = randint(1, 1024)
-        filename = f"test_file_{str(uuid4())}.txt"
-        content = os.urandom(size)
-        size = len(content)
-        self.operator.write(filename, content)
-        self.operator.delete(filename)
-        with pytest.raises(FileNotFoundError):
-            self.operator.stat(filename)
-
-    @pytest.mark.asyncio
-    async def test_async_delete(self):
-        size = randint(1, 1024)
-        filename = f"test_file_{str(uuid4())}.txt"
-        content = os.urandom(size)
-        size = len(content)
-        await self.async_operator.write(filename, content)
-        await self.async_operator.delete(filename)
-        with pytest.raises(FileNotFoundError):
-            await self.operator.stat(filename)
-
-
-class TestS3(AbstractTestSuite):
-    service_name = "s3"
-
-
-class TestFS(AbstractTestSuite):
-    service_name = "fs"
-
-
-class TestMemory(AbstractTestSuite):
-    service_name = "memory"
+def test_sync_read(service_name, operator, async_operator):
+    size = randint(1, 1024)
+    filename = f"random_file_{str(uuid4())}"
+    content = os.urandom(size)
+    operator.write(filename, content)
+
+    read_content = operator.read(filename)
+    assert read_content is not None
+    assert read_content == content
+
+    operator.delete(filename)
+
+
+@pytest.mark.asyncio
+async def test_async_read(service_name, operator, async_operator):
+    size = randint(1, 1024)
+    filename = f"random_file_{str(uuid4())}"
+    content = os.urandom(size)
+    await async_operator.write(filename, content)
+
+    read_content = await async_operator.read(filename)
+    assert read_content is not None
+    assert read_content == content
+
+    await async_operator.delete(filename)
+
+
+def test_sync_read_stat(service_name, operator, async_operator):
+    size = randint(1, 1024)
+    filename = f"random_file_{str(uuid4())}"
+    content = os.urandom(size)
+    operator.write(filename, content)
+
+    metadata = operator.stat(filename)
+    assert metadata is not None
+    assert metadata.content_length == len(content)
+    assert metadata.mode.is_file()
+
+    operator.delete(filename)
+
+
+@pytest.mark.asyncio
+async def test_async_read_stat(service_name, operator, async_operator):
+    size = randint(1, 1024)
+    filename = f"random_file_{str(uuid4())}"
+    content = os.urandom(size)
+    await async_operator.write(filename, content)
+
+    metadata = await async_operator.stat(filename)
+    assert metadata is not None
+    assert metadata.content_length == len(content)
+    assert metadata.mode.is_file()
+
+    await async_operator.delete(filename)
+
+    operator.delete(filename)
+
+
+def test_sync_read_not_exists(service_name, operator, async_operator):
+    with pytest.raises(FileNotFoundError):
+        operator.read(str(uuid4()))
+
+
+@pytest.mark.asyncio
+async def test_async_read_not_exists(service_name, operator, async_operator):
+    with pytest.raises(FileNotFoundError):
+        await async_operator.read(str(uuid4()))
+
+
+def test_sync_write(service_name, operator, async_operator):
+    size = randint(1, 1024)
+    filename = f"test_file_{str(uuid4())}.txt"
+    content = os.urandom(size)
+    size = len(content)
+    operator.write(filename, content)
+    metadata = operator.stat(filename)
+    assert metadata is not None
+    assert metadata.mode.is_file()
+    assert metadata.content_length == size
+
+    operator.delete(filename)
+
+
+@pytest.mark.asyncio
+async def test_async_write(service_name, operator, async_operator):
+    size = randint(1, 1024)
+    filename = f"test_file_{str(uuid4())}.txt"
+    content = os.urandom(size)
+    size = len(content)
+    await async_operator.write(filename, content)
+    metadata = await async_operator.stat(filename)
+    assert metadata is not None
+    assert metadata.mode.is_file()
+    assert metadata.content_length == size
+
+    await async_operator.delete(filename)
+
+
+def test_sync_write_with_non_ascii_name(service_name, operator, async_operator):
+    size = randint(1, 1024)
+    filename = f"❌😱中文_{str(uuid4())}.test"
+    content = os.urandom(size)
+    size = len(content)
+    operator.write(filename, content)
+    metadata = operator.stat(filename)
+    assert metadata is not None
+    assert metadata.mode.is_file()
+    assert metadata.content_length == size
+
+    operator.delete(filename)
+
+
+@pytest.mark.asyncio
+async def test_async_write_with_non_ascii_name(service_name, operator, async_operator):
+    size = randint(1, 1024)
+    filename = f"❌😱中文_{str(uuid4())}.test"
+    content = os.urandom(size)
+    size = len(content)
+    await async_operator.write(filename, content)
+    metadata = await async_operator.stat(filename)
+    assert metadata is not None
+    assert metadata.mode.is_file()
+    assert metadata.content_length == size
+
+    await async_operator.delete(filename)
+
+
+def test_sync_create_dir(service_name, operator, async_operator):
+    path = f"test_dir_{str(uuid4())}/"
+    operator.create_dir(path)
+    metadata = operator.stat(path)
+    assert metadata is not None
+    assert metadata.mode.is_dir()
+
+    operator.delete(path)
+
+
+@pytest.mark.asyncio
+async def test_async_create_dir(service_name, operator, async_operator):
+    path = f"test_dir_{str(uuid4())}/"
+    await async_operator.create_dir(path)
+    metadata = await async_operator.stat(path)
+    assert metadata is not None
+    assert metadata.mode.is_dir()
+
+    await async_operator.delete(path)
+
+
+def test_sync_delete(service_name, operator, async_operator):
+    size = randint(1, 1024)
+    filename = f"test_file_{str(uuid4())}.txt"
+    content = os.urandom(size)
+    size = len(content)
+    operator.write(filename, content)
+    operator.delete(filename)
+    with pytest.raises(FileNotFoundError):
+        operator.stat(filename)
+
+
+@pytest.mark.asyncio
+async def test_async_delete(service_name, operator, async_operator):
+    size = randint(1, 1024)
+    filename = f"test_file_{str(uuid4())}.txt"
+    content = os.urandom(size)
+    size = len(content)
+    await async_operator.write(filename, content)
+    await async_operator.delete(filename)
+    with pytest.raises(FileNotFoundError):
+        await operator.stat(filename)