Add ReuseServerTestCase for class-scoped server reuse

src/valkey_test_case.py

@@ -728,3 +728,83 @@ 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.flushall()
+                client.execute_command("CONFIG", "RESETSTAT")
+                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:
+                                pass
+            except Exception:
+                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,43 @@
+"""
+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.
+"""
+
+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):
+        self.server, self.client = self.create_server(testdir=self.testdir)
+
+    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