Add ReuseServerTestCase for class-scoped server reuse

README.md

@@ -67,4 +67,42 @@ 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. Server state is reset between
+    tests automatically (FLUSHALL, config restore, ACL reset, etc.)
+    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.
+
+`create_server()` only starts the server on the first call — subsequent calls return the cached instance. Between tests, the overridden `teardown()` resets state instead of killing the server: it issues `RESET` on the shared connection, kills any client connections a test spawned, flushes data/scripts/functions, resets ACL users and the slowlog/latency/ACL logs, and restores any modified config values. If the server becomes unreachable or a config cannot be restored, it is torn down and a fresh one starts for the next test.
+
+If a test creates additional servers (e.g. a server without a module loaded for RDB testing), those are tracked in `server_list` and automatically cleaned up at the end of that test. Only the shared server persists across tests.
+
+Tests run top-to-bottom in definition order (via `pytest-order` with `--order-scope=class`).
+
 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,154 @@ 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. Between
+    tests, _reset_server_state() restores isolation by running: RESET on the
+    shared connection, CLIENT KILL for connections a test spawned, REPLICAOF NO
+    ONE, FLUSHALL, CONFIG RESETSTAT, SCRIPT FLUSH, FUNCTION FLUSH, SLOWLOG /
+    LATENCY / ACL LOG resets, ACL user reset, and full config restore.
+
+    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,
+    ):
+        # Return cached server if already running — no new server is created.
+        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):
+        # Reset shared server state between tests instead of shutting it down.
+        if hasattr(self.__class__, "_shared_server") and self.__class__._shared_server:
+            self._reset_server_state()
+        # Clean up any additional servers created during this test.
+        for server in self.server_list:
+            if server and server is not self.__class__._shared_server:
+                server.exit()
+        self.server_list = []
+
+    def _reset_server_state(self):
+        """Reset the shared server to a clean state between tests.
+
+        Resets the shared connection, kills any client connections a test
+        spawned, clears data, scripts, functions, server-side logs (slowlog,
+        latency, ACL log), and ACL users, unwinds replication, and restores all
+        config values to their initial state. If the server is unreachable or a
+        config cannot be restored, the server is killed so the next test gets a
+        fresh instance.
+        """
+        client = self.__class__._shared_client
+        try:
+            # RESET the shared connection first to clear any per-connection
+            # state a test left behind (MULTI/WATCH, CLIENT TRACKING, RESP
+            # version, selected DB, MONITOR/pubsub). Doing this first ensures
+            # the following commands aren't silently queued inside a MULTI.
+            client.execute_command("RESET")
+            # Kill any client connections a test spawned. CLIENT KILL defaults
+            # to SKIPME yes, so the shared client issuing this is not killed.
+            client.execute_command("CLIENT", "KILL", "TYPE", "normal")
+            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
+            # Clear server-side logs so per-test log checks start clean.
+            client.execute_command("SLOWLOG", "RESET")
+            client.execute_command("LATENCY", "RESET")
+            client.execute_command("ACL", "LOG", "RESET")
+            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",
+                "nopass",
+                "~*",
+                "&*",
+                "+@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
+        for server in getattr(self, "server_list", []):
+            if server:
+                server.exit()
+        self.server_list = []

tests/test_reuse_server.py

@@ -0,0 +1,103 @@
+"""
+Demonstrates ReuseServerTestCase usage.
+
+All tests in this class share ONE server. Between each test, the server state
+is reset automatically (connection RESET, spawned clients killed, FLUSHALL,
+config restore, ACL reset, log resets, etc.) 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):
+        version = os.environ.get("SERVER_VERSION", "unstable")
+        server_path = f"{os.path.dirname(os.path.realpath(__file__))}/.build/binaries/{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.__class__._original_hz = original
+        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."""
+        expected = self.__class__._original_hz
+        current = self.client.config_get("hz")["hz"]
+        assert current == expected, f"Expected hz={expected}, got hz={current}"
+
+    def test_spawn_extra_connection_and_acl_user(self):
+        """Leave an extra connection and an ACL user behind for teardown."""
+        extra = self.server.get_new_client()
+        self.__class__._extra_client_id = extra.execute_command("CLIENT", "ID")
+        self.client.execute_command(
+            "ACL", "SETUSER", "leaked", "on", ">pw", "~*", "+@all"
+        )
+        assert self._acl_user_exists("leaked")
+
+    def test_extra_connection_and_acl_user_gone(self):
+        """Proves teardown killed the spare connection and deleted the ACL user."""
+        # The connection spawned in the previous test should no longer exist.
+        live_ids = {
+            int(line.split("id=")[1].split(" ")[0])
+            for line in self._client_list().splitlines()
+            if "id=" in line
+        }
+        assert (
+            self.__class__._extra_client_id not in live_ids
+        ), "Spawned connection should have been killed by CLIENT KILL"
+        # The ACL user created in the previous test should be gone.
+        assert not self._acl_user_exists(
+            "leaked"
+        ), "ACL user from previous test should have been deleted"
+
+    def _client_list(self):
+        result = self.client.execute_command("CLIENT", "LIST")
+        return result.decode() if isinstance(result, bytes) else result
+
+    def _acl_user_exists(self, name):
+        for entry in self.client.execute_command("ACL", "LIST"):
+            if isinstance(entry, bytes):
+                entry = entry.decode()
+            if entry.startswith(f"user {name} "):
+                return True
+        return False