Add ReuseServerTestCase for class-scoped server reuse

README.md

@@ -67,4 +67,37 @@ class TestExamplePerClassSetup(ExampleTestCaseBase):
         client.execute_command("SET K V")
 ```
 
+**Reusing a Single Server Across All Tests in a Class**
+
+If your tests don't need a fresh server each time (most data-operation tests), use `ReuseServerTestCase` to share one server across the entire class. This avoids the overhead of spawning a new process per test — especially useful when module loading is expensive.
+
+```
+class ExampleModuleTestCase(ReuseServerTestCase):
+    @pytest.fixture(autouse=True)
+    def setup_test(self, setup):
+        server_path = "/path_to_your_valkey_server_binary"
+        args = {"loadmodule": "/path/to/your/module.so"}
+        self.server, self.client = self.create_server(
+            testdir=self.testdir, server_path=server_path, args=args
+        )
+
+class TestExampleReuse(ExampleModuleTestCase):
+    """
+    All tests share the same server. FLUSHALL + CONFIG RESETSTAT
+    runs between tests automatically for isolation.
+    """
+
+    def test_basic1(self):
+        self.client.execute_command("SET K V")
+        assert self.client.execute_command("GET K") == b"V"
+
+    def test_basic2(self):
+        # Previous test's data is flushed — this starts clean
+        assert self.client.execute_command("GET K") is None
+```
+
+`ReuseServerTestCase` inherits `ValkeyTestCase`, so all existing fixtures, `create_server()` calls, and `self.server`/`self.client` assignments work unchanged. To adopt it in your module, just change the base class — no other code changes needed.
+
+Tests run top-to-bottom in definition order (via `pytest-order` with `--order-scope=class`). If a config cannot be restored between tests, the server is torn down and a fresh one starts for the next test.
+
 For more examples, refer to the `tests` directory of this package.

src/valkey_test_case.py

@@ -1,3 +1,4 @@
+import logging
 import subprocess
 import time
 import os
@@ -728,3 +729,108 @@ def waitForReplicaOffsetToSyncUp(self, primary, replica):
             pinfo.get_primary_repl_offset(),
             timeout=TEST_MAX_WAIT_TIME_SECONDS,
         )
+
+
+class ReuseServerTestCase(ValkeyTestCase):
+    """Test case that reuses a single server across all tests in the class.
+
+    Instead of spawning a fresh server per test, one server is started on the
+    first create_server() call and reused for all subsequent tests. FLUSHALL +
+    CONFIG RESETSTAT run between tests to reset state.
+
+    Usage — just change your base class:
+
+        class MyModuleTestCase(ReuseServerTestCase):
+            ...  # keep your existing setup_test exactly as-is
+
+    That's it. self.server, self.client, create_server() all work as before.
+    """
+
+    def create_server(
+        self,
+        testdir=None,
+        bind_ip=None,
+        port=None,
+        server_path=None,
+        args="",
+        skip_teardown=False,
+        conf_file=None,
+        external_server=False,
+        wait_for_ping=True,
+        connect_client=True,
+    ):
+        if hasattr(self.__class__, "_shared_server") and self.__class__._shared_server:
+            return self.__class__._shared_server, self.__class__._shared_client
+
+        if testdir is None:
+            testdir = self.testdir
+        if server_path is None:
+            server_path = self.server_path
+
+        server, client = super().create_server(
+            testdir=testdir,
+            bind_ip=bind_ip,
+            port=port,
+            server_path=server_path,
+            args=args,
+            skip_teardown=skip_teardown,
+            conf_file=conf_file,
+            external_server=external_server,
+            wait_for_ping=wait_for_ping,
+            connect_client=connect_client,
+        )
+        self.__class__._shared_server = server
+        self.__class__._shared_client = client
+        self.__class__._initial_config = client.config_get("*")
+        return server, client
+
+    def teardown(self):
+        if hasattr(self.__class__, "_shared_server") and self.__class__._shared_server:
+            client = self.__class__._shared_client
+            try:
+                client.execute_command("REPLICAOF", "NO", "ONE")
+                client.flushall()
+                client.execute_command("CONFIG", "RESETSTAT")
+                client.execute_command("SCRIPT", "FLUSH")
+                try:
+                    client.execute_command("FUNCTION", "FLUSH")
+                except Exception:
+                    pass
+                users = client.execute_command("ACL", "LIST")
+                for entry in users:
+                    if isinstance(entry, bytes):
+                        entry = entry.decode()
+                    if not entry.startswith("user default "):
+                        username = entry.split(" ")[1]
+                        client.execute_command("ACL", "DELUSER", username)
+                client.execute_command(
+                    "ACL", "SETUSER", "default", "reset", "on", "~*", "&*", "+@all"
+                )
+                if hasattr(self.__class__, "_initial_config"):
+                    current = client.config_get("*")
+                    for key, val in self.__class__._initial_config.items():
+                        if current.get(key) != val:
+                            try:
+                                client.config_set(key, val)
+                            except Exception:
+                                logging.warning(
+                                    f"Could not reset config '{key}' — "
+                                    f"tearing down server for fresh restart"
+                                )
+                                self.__class__._shared_server.exit()
+                                self.__class__._shared_server = None
+                                self.__class__._shared_client = None
+                                return
+            except Exception:
+                logging.warning("Server unreachable during teardown — killing process")
+                self.__class__._shared_server.exit()
+                self.__class__._shared_server = None
+                self.__class__._shared_client = None
+
+    @pytest.fixture(autouse=True, scope="class")
+    def class_teardown(self, request):
+        yield
+        if hasattr(self.__class__, "_shared_server") and self.__class__._shared_server:
+            self.__class__._shared_server.exit()
+            self.__class__._shared_server = None
+            self.__class__._shared_client = None

tests/test_reuse_server.py

@@ -0,0 +1,62 @@
+"""
+Demonstrates ReuseServerTestCase usage.
+
+All tests in this class share ONE server. Between each test, FLUSHALL + CONFIG
+RESETSTAT run automatically to give each test a clean slate without the cost of
+restarting the server.
+
+Tests run top-to-bottom in definition order (via pytest-order with
+--order-scope=class). Some tests verify isolation from the previous test,
+so ordering matters.
+"""
+
+import os
+import pytest
+from conftest import resource_port_tracker
+from valkey_test_case import ReuseServerTestCase
+
+
+class TestReuseServer(ReuseServerTestCase):
+    """Verifies that server reuse works and tests are isolated."""
+
+    @pytest.fixture(autouse=True)
+    def setup_test(self, setup):
+        server_path = f"{os.path.dirname(os.path.realpath(__file__))}/.build/binaries/{os.environ['SERVER_VERSION']}/valkey-server"
+        self.server, self.client = self.create_server(
+            testdir=self.testdir, server_path=server_path
+        )
+
+    def test_write_and_read(self):
+        """Basic write/read on the shared server."""
+        self.client.set("greeting", "hello")
+        assert self.client.get("greeting") == b"hello"
+
+    def test_isolation_from_previous(self):
+        """Proves FLUSHALL cleaned up the previous test's data."""
+        result = self.client.get("greeting")
+        assert result is None, "Key from previous test should not exist"
+
+    def test_server_still_alive(self):
+        """Proves the server survived across tests (no restart)."""
+        assert self.client.ping() is True
+
+    def test_multiple_keys(self):
+        """Write multiple keys, verify they all exist within this test."""
+        for i in range(10):
+            self.client.set(f"key:{i}", f"value:{i}")
+        assert self.client.dbsize() == 10
+
+    def test_previous_keys_gone(self):
+        """Proves the 10 keys from the previous test were flushed."""
+        assert self.client.dbsize() == 0
+
+    def test_config_change_is_restored(self):
+        """Proves configs modified during a test get restored for the next."""
+        original = self.client.config_get("hz")["hz"]
+        self.client.config_set("hz", "50")
+        assert self.client.config_get("hz")["hz"] == "50"
+
+    def test_config_restored_after_previous(self):
+        """Proves the config changed in the previous test was reset."""
+        current = self.client.config_get("hz")["hz"]
+        assert current == "10", f"Expected hz=10 (default), got hz={current}"