diff --git a/src/valkey_test_case.py b/src/valkey_test_case.py index 90a3470..a924582 100644 --- a/src/valkey_test_case.py +++ b/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 diff --git a/tests/test_reuse_server.py b/tests/test_reuse_server.py new file mode 100644 index 0000000..f8f303e --- /dev/null +++ b/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