diff --git a/README.md b/README.md index 7fbd8cd4..2c6bf948 100644 --- a/README.md +++ b/README.md @@ -267,3 +267,18 @@ def test_example_3(): env.assertEqual(con2.get('x'), '1') ``` + +### Shutdown grace period + +With the default shutdown policy (`terminateRetries=None`), RLTest retries +SIGTERM once per second for up to 30 seconds, then sends SIGKILL and waits up to +5 seconds for process exit. Valgrind and sanitizer runs get a 300-second grace +period to allow for slower saves and exit-time analysis. Explicit +`terminateRetries`/`terminateRetrySecs` settings retain their existing behavior. +A forced shutdown under the default policy fails the test even without +`--check-exitcode`. + +On macOS, interactive debugger teardown also bounds the inferior-process wait: +30 seconds after SIGTERM, followed by SIGKILL and a 5-second wait. An inferior +left paused at a breakpoint can therefore be killed and reported as a failure; +teardown no longer waits indefinitely for debugger input. diff --git a/RLTest/Enterprise/EnterpriseClusterEnv.py b/RLTest/Enterprise/EnterpriseClusterEnv.py index bd41bc24..08f8e829 100644 --- a/RLTest/Enterprise/EnterpriseClusterEnv.py +++ b/RLTest/Enterprise/EnterpriseClusterEnv.py @@ -139,6 +139,11 @@ def broadcast(self, *cmd): for shard in self.shards: shard.broadcast(*cmd) + def hasShutdownFailure(self, reset=False): + # Visit every shard even if an earlier one failed, to consume all flags. + failures = [shard.hasShutdownFailure(reset=reset) for shard in self.shards] + return any(failures) + def checkExitCode(self): for shard in self.shards: if not shard.checkExitCode(): diff --git a/RLTest/__main__.py b/RLTest/__main__.py index 3367afa7..e6a5a4a2 100644 --- a/RLTest/__main__.py +++ b/RLTest/__main__.py @@ -358,10 +358,18 @@ def __init__(self, runner): self.runner = runner def __enter__(self): - pass + self.runner._pendingResults = [] + self.runner._teardownFailed = False def __exit__(self, type, value, traceback): - self.runner.takeEnvDown() + try: + self.runner.takeEnvDown() + finally: + pending = self.runner._pendingResults + self.runner._pendingResults = None + if type is None and not self.runner._teardownFailed: + for printer, name in pending: + printer(name) class TestTimeLimit(object): """ @@ -630,9 +638,11 @@ def takeEnvDown(self, fullShutDown=False): except: flush_ok = False self.currEnv.stop() - if self.require_clean_exit and self.currEnv and (not self.currEnv.checkExitCode() or not flush_ok): + if self.currEnv.hasShutdownFailure(reset=True) or (self.require_clean_exit and (not self.currEnv.checkExitCode() or not flush_ok)): print(Colors.Bred('\tRedis did not exit cleanly')) self.addFailure(self.currEnv.testName, ['redis process failure']) + self._teardownFailed = True + self.printFail(self.currEnv.testName) if self.args.check_exitcode: raise Exception('Process exited dirty') self.currEnv = None @@ -746,14 +756,16 @@ def _runTest(self, test, numberOfAssertionFailed=0, prefix='', before=lambda x=N hasException = False setup_ok = False + skipped = False try: before_func() setup_ok = True fn() passed = True except unittest.SkipTest: - self.printSkip(testFullName) - return 0 + # Still run teardown and consume shutdown failures for this test. + skipped = True + passed = True except TestAssertionFailure: if self.args.exit_on_failure: self.takeEnvDown(fullShutDown=True) @@ -796,7 +808,14 @@ def _runTest(self, test, numberOfAssertionFailed=0, prefix='', before=lambda x=N self.handleFailure(testFullName=testFullName, prefix=msgPrefix, testname=test.name, env=self.currEnv) passed = False - elif not hasException: + # Attribute a mid-test stop/restart to this test before env reuse + # changes testName. Consume only after reporting, never on restart. + if self.currEnv.hasShutdownFailure(reset=True): + self.addFailure(test.name, ['redis process failure']) + self.printFail(testFullName) + numFailed += 1 + passed = False + elif not hasException and not skipped: self.addFailure(test.name, '') passed = False @@ -808,7 +827,10 @@ def _runTest(self, test, numberOfAssertionFailed=0, prefix='', before=lambda x=N input('press any button to move to the next test') if passed: - self.printPass(testFullName) + if skipped: + self.printSkip(testFullName) + else: + self.printPass(testFullName) if hasException: numFailed += 1 # exception should be counted as failure @@ -827,6 +849,10 @@ def _closeGitHubActionsTestsGroup(self): self.github_actions_group_open = False def printSkip(self, name): + pending = getattr(self, '_pendingResults', None) + if pending is not None: + pending.append((self.printSkip, name)) + return print('%s:\r\n\t%s' % (Colors.Cyan(name), Colors.Green('[SKIP]'))) def printFail(self, name): @@ -836,6 +862,10 @@ def printError(self, name): print('%s:\r\n\t%s' % (Colors.Cyan(name), Colors.Bred('[ERROR]'))) def printPass(self, name): + pending = getattr(self, '_pendingResults', None) + if pending is not None: + pending.append((self.printPass, name)) + return print('%s:\r\n\t%s' % (Colors.Cyan(name), Colors.Green('[PASS]'))) def envScopeGuard(self): @@ -879,12 +909,18 @@ def run_single_test(self, test, on_timeout_func): obj = test.create_instance() except unittest.SkipTest: - self.printSkip(test.name) + if self.currEnv and self.currEnv.hasShutdownFailure(reset=True): + self.addFailure(test.name, ['redis process failure']) + self.printFail(test.name) + else: + self.printSkip(test.name) return 0 except Exception as e: self.printException(e) self.addFailure(test.name + " [__init__]") + if self.currEnv and self.currEnv.hasShutdownFailure(reset=True): + self.addFailure(test.name + " [__init__]", ['redis process failure']) return 0 failures = 0 @@ -892,8 +928,11 @@ def run_single_test(self, test, on_timeout_func): after = getattr(obj, 'tearDown', lambda x=None: None) for subtest in test.get_functions(obj): timeout_handler.reset() + # Only assertion counts belong in the assertion watermark. + # Shutdowns and exceptions contribute to failures separately. + assertions = self.currEnv.getNumberOfFailedAssertion() if self.currEnv else 0 failures += self._runTest(subtest, prefix='\t', - numberOfAssertionFailed=failures, + numberOfAssertionFailed=assertions, before=before, after=after) done += 1 diff --git a/RLTest/env.py b/RLTest/env.py index 47a49156..78798889 100644 --- a/RLTest/env.py +++ b/RLTest/env.py @@ -270,11 +270,17 @@ def __init__(self, testName=None, testDescription=None, module=None, self.startupGraceSecs = startupGraceSecs if startupGraceSecs is not None else Defaults.startup_grace_secs + # Carry unreported failures across both runner replacement and wrapper reuse. + # Leave the old flag intact in case constructing the new environment fails. + previous = Env.RTestInstance.currEnv if Env.RTestInstance else None + self._previousShutdownFailure = previous.hasShutdownFailure() if previous else False + if not freshEnv and Env.RTestInstance and Env.RTestInstance.currEnv and self.compareEnvs(Env.RTestInstance.currEnv): self.envRunner = Env.RTestInstance.currEnv.envRunner else: if Env.RTestInstance and Env.RTestInstance.currEnv: Env.RTestInstance.currEnv.stop() + self._previousShutdownFailure = previous.hasShutdownFailure() self.envRunner = self.getEnvByName() try: @@ -603,6 +609,15 @@ def debugPrint(self, msg, force=False): if Defaults.debug_print or force: print('\t' + Colors.Bold('debug:\t') + Colors.Gray(msg)) + def hasShutdownFailure(self, reset=False): + # External environments do not own Redis processes. + check = getattr(self.envRunner, 'hasShutdownFailure', None) + current = check(reset=reset) if check is not None else False + previous = getattr(self, '_previousShutdownFailure', False) + if reset: + self._previousShutdownFailure = False + return current or previous + def checkExitCode(self): return self.envRunner.checkExitCode() diff --git a/RLTest/redis_cluster.py b/RLTest/redis_cluster.py index ab2e975e..6afe1523 100644 --- a/RLTest/redis_cluster.py +++ b/RLTest/redis_cluster.py @@ -267,6 +267,11 @@ def broadcast(self, *cmd): for shard in self.shards: shard.broadcast(*cmd) + def hasShutdownFailure(self, reset=False): + # Visit every shard even if an earlier one failed, to consume all flags. + failures = [shard.hasShutdownFailure(reset=reset) for shard in self.shards] + return any(failures) + def checkExitCode(self): for shard in self.shards: if not shard.checkExitCode(): diff --git a/RLTest/redis_std.py b/RLTest/redis_std.py index aa5337fb..43fe99f7 100644 --- a/RLTest/redis_std.py +++ b/RLTest/redis_std.py @@ -16,6 +16,10 @@ MASTER = 'master' SLAVE = 'slave' +_TERMINATE_TIMEOUT = 30 +_INSTRUMENTED_TERMINATE_TIMEOUT = 300 +_KILL_TIMEOUT = 5 + class StandardEnv(object): def __init__(self, redisBinaryPath, port=6379, modulePath=None, moduleArgs=None, outputFilesFormat=None, @@ -51,6 +55,7 @@ def __init__(self, redisBinaryPath, port=6379, modulePath=None, moduleArgs=None, self.masterProcess = None self.masterStdout = None self.masterStderr = None + self.shutdownFailed = False self.masterExitCode = None self.slaveProcess = None self.slaveStdout = None @@ -463,27 +468,54 @@ def _stopProcess(self, role): self.verbose_analyse_server_log(role) return try: - if platform.system() == 'Darwin': - # On macOS, with lldb, killing lldb process does not terminate inferior processes - p0 = psutil.Process(pid=process.pid) - pchi = p0.children(recursive=True) - for p in pchi: + if platform.system() == 'Darwin' and self.has_interactive_debugger: + # lldb does not forward termination to its inferiors. Bound the + # whole child group wait, rather than waiting per child forever. + children = psutil.Process(process.pid).children(recursive=True) + for child in children: try: - p.terminate() - p.wait() - except: + child.terminate() + except psutil.NoSuchProcess: pass + _, alive = psutil.wait_procs(children, timeout=_TERMINATE_TIMEOUT) + if alive: + self.shutdownFailed = True + for child in alive: + try: + child.kill() + except psutil.NoSuchProcess: + pass + _, alive = psutil.wait_procs(alive, timeout=_KILL_TIMEOUT) + if alive: + print(Colors.Bred('[TERMINATING] debugger children survived SIGKILL')) if self.terminateRetries is None: - # ask once, then wait for process to exit - process.terminate() - termination_start_time = time.time() - while process.poll() is None: # None returns if the processes is not finished yet, retry until redis exits - time.sleep(0.1) - if time.time() - termination_start_time > 30: - # if process is still running after 30 seconds, try reading its output - process_out, process_err = process.communicate() - print(Colors.Bred(f'\t[TERMINATING] out ({process_out}), error ({process_err})')) + # Wait on the process, not pipe EOF: Redis fork children can + # inherit stdout/stderr and outlive their parent. + grace = (_INSTRUMENTED_TERMINATE_TIMEOUT + if self.sanitizer or (self.debugger and not self.has_interactive_debugger) + else _TERMINATE_TIMEOUT) + deadline = time.monotonic() + grace + while process.poll() is None: + process.terminate() + remaining = deadline - time.monotonic() + if remaining <= 0: + break + try: + process.wait(timeout=min(1, remaining)) + except subprocess.TimeoutExpired: + # Redis may refuse shutdown during initial AOF rewrite; + # retry SIGTERM so it can exit cleanly after the rewrite. + continue + if process.poll() is None: + self.shutdownFailed = True + print(Colors.Bred('[TERMINATING] {0} server id {1} did not exit on SIGTERM; sending SIGKILL'.format(role, serverId))) + process.kill() + try: + process.wait(timeout=_KILL_TIMEOUT) + except subprocess.TimeoutExpired: + print(Colors.Bred('[TERMINATING] {0} server id {1} did not exit after SIGKILL'.format(role, serverId))) + self._print_shutdown_log(role) else: # keep asking every few seconds until process has exited, otherwise kill if self.terminateRetrySecs is None: @@ -508,6 +540,24 @@ def _stopProcess(self, role): 'OSError caught while waiting for {0} process to end: {1}'.format(role, e.__str__()))) pass + def hasShutdownFailure(self, reset=False): + failed = self.shutdownFailed + if reset: + self.shutdownFailed = False + return failed + + def _print_shutdown_log(self, role): + try: + path = os.path.join(self.dbDirPath or '', self._getFileName(role, '.log')) + with open(path, 'rb') as log: + log.seek(0, os.SEEK_END) + log.seek(max(0, log.tell() - 8192)) + print(Colors.Bred('[TERMINATING] last server log bytes ({0}):\n{1}'.format( + path, log.read(8192).decode('utf-8', errors='replace')))) + except Exception as error: + # Diagnostics must never interrupt teardown, including invalid paths. + print(Colors.Bred('[TERMINATING] could not read server log: {0}'.format(error))) + def verbose_analyse_server_log(self, role): path = "{0}".format(self._getFileName(role, '.log')) if self.dbDirPath is not None: diff --git a/tests/unit/test_shutdown.py b/tests/unit/test_shutdown.py new file mode 100644 index 00000000..75fa10d7 --- /dev/null +++ b/tests/unit/test_shutdown.py @@ -0,0 +1,423 @@ +"""Regression coverage for Redis processes that refuse graceful shutdown.""" +import argparse +import os +import time +import signal +import subprocess +import sys +from unittest.mock import Mock, patch + +import pytest + +from RLTest.redis_std import MASTER, SLAVE, StandardEnv +from RLTest.redis_cluster import ClusterEnv +from RLTest.env import Env +from RLTest.__main__ import RLTest + + +def make_env(tmp_path, **kwargs): + with patch.object(StandardEnv, '_getRedisVersion', return_value=80000): + return StandardEnv(redisBinaryPath='redis-server', dbDirPath=str(tmp_path), + outputFilesFormat='%s-shutdown', useSlaves=True, **kwargs) + + +def start_process(ignore_sigterm): + handler = 'signal.SIG_IGN' if ignore_sigterm else 'lambda *_: sys.exit(0)' + code = ('import signal, sys, time\n' + 'signal.signal(signal.SIGTERM, {0})\n' + 'print("ready", flush=True)\n' + 'print("shutdown stderr", file=sys.stderr, flush=True)\n' + 'while True: time.sleep(0.1)\n').format(handler) + process = subprocess.Popen([sys.executable, '-c', code], + stdout=subprocess.PIPE, stderr=subprocess.PIPE) + assert process.stdout.readline() == b'ready\n' + return process + + +@pytest.mark.parametrize('role', [MASTER, SLAVE]) +@pytest.mark.parametrize('ignore_sigterm', [False, True]) +def test_stop_reaps_process_and_reports_forced_shutdown(tmp_path, capsys, role, ignore_sigterm): + env = make_env(tmp_path) + process = start_process(ignore_sigterm) + (tmp_path / env._getFileName(role, '.log')).write_text('Writing initial AOF, can\'t exit.\n') + setattr(env, role + 'Process', process) + try: + with patch('RLTest.redis_std._TERMINATE_TIMEOUT', 0.2): + env.stopEnv(masters=role == MASTER, slaves=role == SLAVE) + expected_code = -signal.SIGKILL if ignore_sigterm else 0 + assert process.returncode == expected_code + assert getattr(env, role + 'ExitCode') == expected_code + assert getattr(env, role + 'Process') is None + output = capsys.readouterr().out + if ignore_sigterm: + assert 'sending SIGKILL' in output + assert role in output + assert 'Writing initial AOF' in output + assert env.hasShutdownFailure() + assert not env.checkExitCode() + else: + assert 'sending SIGKILL' not in output + finally: + if process.poll() is None: + process.kill() + process.communicate(timeout=5) + + +def test_sigkill_wait_timeout_does_not_abort_remaining_shards(tmp_path): + env = make_env(tmp_path) + process = Mock() + process.poll.return_value = None + process.wait.side_effect = subprocess.TimeoutExpired('redis-server', 5) + env.masterProcess = process + cluster = ClusterEnv.__new__(ClusterEnv) + other = Mock() + cluster.shards = [env, other] + with patch('RLTest.redis_std._TERMINATE_TIMEOUT', 0): + cluster.stopEnv() + process.kill.assert_called_once_with() + process.wait.assert_called_once_with(timeout=5) + assert env.masterProcess is None + assert env.masterExitCode is None + assert env.hasShutdownFailure() + other.stopEnv.assert_called_once() + + +@pytest.mark.parametrize('ignore_sigterm', [False, True]) +def test_inherited_pipes_do_not_block_parent_shutdown(tmp_path, ignore_sigterm): + env = make_env(tmp_path) + handler = 'signal.SIG_IGN' if ignore_sigterm else 'lambda *_: sys.exit(0)' + code = ("import os, signal, sys, time\n" + "signal.signal(signal.SIGTERM, {0})\n" + "child = os.fork()\n" + "if child == 0:\n" + " signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" + " time.sleep(60)\n" + " os._exit(0)\n" + "print(child, flush=True)\n" + "while True: time.sleep(.1)\n").format(handler) + process = subprocess.Popen([sys.executable, '-c', code], stdout=subprocess.PIPE, stderr=subprocess.PIPE) + child = int(process.stdout.readline()) + env.slaveProcess = process + try: + start = time.monotonic() + with patch('RLTest.redis_std._TERMINATE_TIMEOUT', .2): + env.stopEnv(masters=False) + assert time.monotonic() - start < 2 + assert env.slaveExitCode == (-signal.SIGKILL if ignore_sigterm else 0) + assert env.slaveProcess is None + assert env.hasShutdownFailure() == ignore_sigterm + os.kill(child, 0) # The inherited pipes are still held open. + finally: + os.kill(child, signal.SIGKILL) + if process.poll() is None: + process.kill() + process.communicate(timeout=5) + + +def test_retries_sigterm_after_initial_refusal(tmp_path): + env = make_env(tmp_path) + code = ("import signal, sys, time\n" + "def refused(*_):\n" + " signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))\n" + "signal.signal(signal.SIGTERM, refused)\n" + "print('ready', flush=True)\n" + "while True: time.sleep(.1)\n") + process = subprocess.Popen([sys.executable, '-c', code], stdout=subprocess.PIPE) + assert process.stdout.readline() == b'ready\n' + env.masterProcess = process + try: + with patch('RLTest.redis_std._TERMINATE_TIMEOUT', 3): + env.stopEnv() + assert env.masterExitCode == 0 + assert not env.hasShutdownFailure() + finally: + if process.poll() is None: + process.kill() + process.communicate(timeout=5) + + +@pytest.mark.parametrize('clustered', [False, True]) +def test_forced_shutdown_fails_runner_without_check_exitcode(tmp_path, clustered): + std = make_env(tmp_path) + process = start_process(True) + std.masterProcess = process + runner = std + if clustered: + runner = ClusterEnv.__new__(ClusterEnv) + runner.shards = [std] + env = Env.__new__(Env) + env.envRunner = runner + env.testName = 'shutdown-regression' + rl = RLTest.__new__(RLTest) + rl.args = argparse.Namespace(env_reuse=False, check_exitcode=False) + rl.require_clean_exit = False + rl.testsFailed = {} + rl.currEnv = env + try: + with patch.object(env, 'isUp', return_value=False): + with patch('RLTest.redis_std._TERMINATE_TIMEOUT', .1): + rl.takeEnvDown() + assert rl.testsFailed == {'shutdown-regression': ['redis process failure']} + assert rl.currEnv is None + assert std.masterProcess is None + finally: + if process.poll() is None: + process.kill() + process.communicate(timeout=5) + + +def test_darwin_debugger_child_wait_is_bounded(tmp_path): + env = make_env(tmp_path) + process = Mock() + process.poll.return_value = 0 + env.masterProcess = process + child = Mock() + with patch.object(env, '_isAlive', return_value=True), \ + patch.object(StandardEnv, 'has_interactive_debugger', True), \ + patch('RLTest.redis_std.platform.system', return_value='Darwin'), \ + patch('RLTest.redis_std.psutil.Process') as parent, \ + patch('RLTest.redis_std.psutil.wait_procs', side_effect=[([], [child]), ([], [child])]) as wait: + parent.return_value.children.return_value = [child] + env.stopEnv() + assert [call[1]['timeout'] for call in wait.call_args_list] == [30, 5] + child.kill.assert_called_once_with() + assert env.masterProcess is None + assert env.hasShutdownFailure() + + +@pytest.mark.parametrize('instrumentation', ['valgrind', 'sanitizer']) +def test_instrumented_shutdown_can_finish_after_thirty_seconds(tmp_path, instrumentation): + from RLTest.debuggers import Valgrind + kwargs = {'debugger': Valgrind('')} if instrumentation == 'valgrind' else {'sanitizer': 'address'} + env = make_env(tmp_path, **kwargs) + process = Mock() + process.poll.side_effect = [None, None, 0, 0, 0] + process.wait.side_effect = [subprocess.TimeoutExpired('redis-server', 1), 0] + env.masterProcess = process + with patch.object(env, '_isAlive', return_value=True), \ + patch('RLTest.redis_std.time.monotonic', side_effect=[0, 0, 31]): + env.stopEnv() + process.kill.assert_not_called() + assert process.wait.call_count == 2 + assert env.masterExitCode == 0 + assert not env.hasShutdownFailure() + + +def test_diagnostic_path_error_cannot_prevent_teardown(tmp_path, capsys): + env = make_env(tmp_path) + process = start_process(True) + env.masterProcess = process + try: + with patch.object(env, '_getFileName', side_effect=TypeError('invalid log format')), \ + patch('RLTest.redis_std._TERMINATE_TIMEOUT', .1): + env.stopEnv() + assert process.returncode == -signal.SIGKILL + assert env.masterProcess is None + assert env.masterExitCode == -signal.SIGKILL + assert env.hasShutdownFailure() + assert 'could not read server log' in capsys.readouterr().out + finally: + if process.poll() is None: + process.kill() + process.communicate(timeout=5) + + +def test_reused_environment_reports_failure_on_original_test(tmp_path): + from RLTest.loader import TestMethod + first, second = make_env(tmp_path), make_env(tmp_path) + cluster = ClusterEnv.__new__(ClusterEnv) + cluster.shards = [first, second] + env = Env.__new__(Env) + env.envRunner = cluster + env.assertionFailedSummary = [] + env.testName = 'original' + rl = RLTest.__new__(RLTest) + rl.currEnv = env + rl.testsFailed = {} + rl.args = argparse.Namespace(env_reuse=True, check_exitcode=False, + exit_on_failure=False, stop_on_failure=False) + rl.require_clean_exit = False + def original(): + # A mid-test stop/restart must retain the failure until the test ends. + first.shutdownFailed = second.shutdownFailed = True + def following(): + pass + with patch.object(rl, 'printFail'), patch.object(rl, 'printPass'): + assert rl._runTest(TestMethod(original, name='original')) == 1 + assert not cluster.hasShutdownFailure() + env.testName = 'following' + assert rl._runTest(TestMethod(following, name='following')) == 0 + with patch.object(env, 'isUp', return_value=False): + rl.takeEnvDown(fullShutDown=True) + assert rl.testsFailed == {'original': ['redis process failure']} + + +def test_class_shutdown_failure_does_not_hide_later_assertions(tmp_path): + from contextlib import nullcontext + from RLTest.env import Defaults + from RLTest.loader import TestMethod + std = make_env(tmp_path) + env = Env.__new__(Env) + env.envRunner = std + env.assertionFailedSummary = [] + rl = RLTest.__new__(RLTest) + rl.currEnv = env + rl.testsFailed = {} + rl.args = argparse.Namespace(test_timeout=0, exit_on_failure=False, stop_on_failure=False) + + def forced_shutdown(): + std.shutdownFailed = True + + def assertion_failure(): + env.assertionFailedSummary.append('later assertion failed') + + test = Mock(is_class=True, env_spec=None) + test.name = 'regression' + test.get_functions.return_value = [TestMethod(forced_shutdown, 'first'), + TestMethod(assertion_failure, 'second'), + TestMethod(assertion_failure, 'third')] + with patch.object(Defaults, 'curr_test_name', None), \ + patch.object(rl, 'envScopeGuard', return_value=nullcontext()), \ + patch.object(rl, 'printFail'), patch.object(rl, 'printPass') as passed, \ + patch.object(Defaults, 'print_verbose_information_on_failure', False): + assert rl.run_single_test(test, lambda: None) == 3 + assert set(rl.testsFailed) == {'first', 'second', 'third'} + passed.assert_not_called() + + +@pytest.mark.parametrize('replacements', [1, 2]) +def test_environment_replacement_retains_shutdown_failure(tmp_path, replacements): + from RLTest.env import Defaults + from RLTest.loader import TestMethod + old = Env.__new__(Env) + old.envRunner = make_env(tmp_path) + old.assertionFailedSummary = [] + rl = RLTest.__new__(RLTest) + rl.currEnv = old + rl.testsFailed = {} + rl.args = argparse.Namespace(exit_on_failure=False, stop_on_failure=False) + + def replacement(): + for _ in range(replacements): + Env(testName='swap', freshEnv=True, logDir=str(tmp_path)) + + def stop(env): + if env is old: + env.envRunner.shutdownFailed = True + + with patch.object(Env, 'RTestInstance', rl), \ + patch.object(Env, 'getEnvByName', side_effect=lambda: make_env(tmp_path)), \ + patch.object(Env, 'start'), patch.object(Env, 'stop', stop), \ + patch.object(Defaults, 'debug_pause', False), \ + patch.object(rl, 'printFail'), patch.object(rl, 'printPass') as passed: + assert rl._runTest(TestMethod(replacement, 'swap')) == 1 + assert not rl.currEnv.hasShutdownFailure() + assert rl.testsFailed == {'swap': ['redis process failure']} + passed.assert_not_called() + + +@pytest.mark.parametrize('forced', [False, True]) +@pytest.mark.parametrize('phase', ['body', 'teardown']) +def test_skipped_test_consumes_its_shutdown_failure(tmp_path, forced, phase): + import unittest + from RLTest.loader import TestMethod + std = make_env(tmp_path) + env = Env.__new__(Env) + env.envRunner = std + env.assertionFailedSummary = [] + rl = RLTest.__new__(RLTest) + rl.currEnv = env + rl.testsFailed = {} + rl.args = argparse.Namespace(exit_on_failure=False, stop_on_failure=False) + + def skipped(): + if phase == 'body': + std.shutdownFailed = forced + raise unittest.SkipTest() + + def teardown(): + if phase == 'teardown': + std.shutdownFailed = forced + + with patch.object(rl, 'printSkip') as skip, patch.object(rl, 'printFail'), \ + patch.object(rl, 'printPass') as passed: + assert rl._runTest(TestMethod(skipped, 'skipped'), after=teardown) == int(forced) + assert not std.hasShutdownFailure() + assert skip.call_count == (0 if forced else 1) + passed.assert_not_called() + assert rl._runTest(TestMethod(lambda: None, 'following')) == 0 + passed.assert_called_once_with('following') + assert rl.testsFailed == ({'skipped': ['redis process failure']} if forced else {}) + + +@pytest.mark.parametrize('forced', [False, True]) +def test_class_constructor_skip_consumes_shutdown_failure(tmp_path, forced): + import unittest + from RLTest.env import Defaults + from RLTest.loader import TestMethod + env = Env.__new__(Env) + env.envRunner = make_env(tmp_path) + env.assertionFailedSummary = [] + rl = RLTest.__new__(RLTest) + rl.currEnv = env + rl.testsFailed = {} + rl.args = argparse.Namespace(test_timeout=0, env_reuse=True, + exit_on_failure=False, stop_on_failure=False) + test = Mock(is_class=True, env_spec=None) + test.name = 'constructor' + + def skipped(): + env.envRunner.shutdownFailed = forced + raise unittest.SkipTest() + + test.create_instance.side_effect = skipped + with patch.object(Defaults, 'curr_test_name', None), \ + patch.object(env, 'flush'), patch.object(rl, 'printSkip') as skip: + assert rl.run_single_test(test, lambda: None) == 0 + assert skip.call_count == (0 if forced else 1) + assert not env.hasShutdownFailure() + assert rl._runTest(TestMethod(lambda: None, 'following')) == 0 + assert rl.testsFailed == ({'constructor': ['redis process failure']} if forced else {}) + + +@pytest.mark.parametrize('forced', [False, True]) +@pytest.mark.parametrize('outcome', ['pass', 'skip', 'constructor_skip']) +def test_result_output_waits_for_environment_teardown(tmp_path, capsys, forced, outcome): + import unittest + from RLTest.env import Defaults + from RLTest.loader import TestMethod + env = Env.__new__(Env) + env.envRunner = make_env(tmp_path) + env.testName = 'teardown' + env.assertionFailedSummary = [] + rl = RLTest.__new__(RLTest) + rl.currEnv = env + rl.testsFailed = {} + rl.require_clean_exit = False + rl.args = argparse.Namespace(test_timeout=0, env_reuse=False, check_exitcode=False, + exit_on_failure=False, stop_on_failure=False) + + def stop(): + output = capsys.readouterr().out + assert '[PASS]' not in output + assert '[SKIP]' not in output + env.envRunner.shutdownFailed = forced + + def target(): + if outcome != 'pass': + raise unittest.SkipTest() + + test = TestMethod(target, 'teardown') + if outcome == 'constructor_skip': + test = Mock(is_class=True, env_spec=None) + test.name = 'teardown' + test.create_instance.side_effect = target + with patch.object(Defaults, 'curr_test_name', None), \ + patch.object(Defaults, 'print_verbose_information_on_failure', False), \ + patch.object(env, 'isUp', return_value=False), patch.object(env, 'stop', stop): + assert rl.run_single_test(test, lambda: None) == (0 if outcome == 'constructor_skip' else 1) + output = capsys.readouterr().out + assert ('[PASS]' in output) == (not forced and outcome == 'pass') + assert ('[SKIP]' in output) == (not forced and outcome != 'pass') + assert ('[FAIL]' in output) == forced + assert rl.testsFailed == ({'teardown': ['redis process failure']} if forced else {})