test_maintenance_review.py #2

#!/usr/bin/env python3
"""Offline regression tests for maintenance review fixes; no P4/SMTP operations."""
import contextlib
import importlib.util
import io
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import types
import unittest
from unittest.mock import Mock, patch

MAINT = Path(__file__).resolve().parents[1] / 'Maintenance'
BASH = os.environ.get('SDP_TEST_BASH', shutil.which('bash'))
sys.path.insert(0, str(MAINT))
import sdputils


class P4Exception(Exception):
    pass


@contextlib.contextmanager
def module(name, args=()):
    fake_p4 = types.SimpleNamespace(P4=Mock, P4Exception=P4Exception)
    spec = importlib.util.spec_from_file_location('review_' + name, MAINT / (name + '.py'))
    result = importlib.util.module_from_spec(spec)
    utils = Mock(server='test:1666', p4user='tester', p4='/mock/p4')
    utils.get.side_effect = lambda key, *a, **k: {'weeks': '4', 'userweeks': '8', 'deleteweeks': '12', 'mailport': '25', 'mailsecure': '0'}.get(key, '')
    with patch.dict(sys.modules, {'P4': fake_p4}), patch.object(sys, 'argv', [name, *args]), patch.object(sdputils, 'SDPUtils', return_value=utils):
        spec.loader.exec_module(result)
        yield result


class Helpers(unittest.TestCase):
    def test_checked_command_keeps_failure_details_without_arguments_or_input(self):
        error = subprocess.CalledProcessError(7, ['p4', 'secret-argument'], stderr='server diagnostic')
        with patch.object(sdputils.subprocess, 'run', side_effect=error), self.assertLogs(level='ERROR') as logs:
            with self.assertRaises(subprocess.CalledProcessError):
                sdputils.run_checked(['p4', 'secret-argument'], input='secret-input', capture_output=True)
        self.assertIn('server diagnostic', '\n'.join(logs.output))
        self.assertNotIn('secret-', '\n'.join(logs.output))

    def test_text_log_retains_failure(self):
        with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {'LOGS': tmp}):
            with self.assertRaisesRegex(RuntimeError, 'specific failure'):
                with sdputils.maintenance_run('example.py'):
                    raise RuntimeError('specific failure')
            text = (Path(tmp) / 'example.log').read_text()
            self.assertIn('Starting example.py', text)
            self.assertIn('specific failure', text)

    def test_text_log_nonzero_exit_not_completed(self):
        with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {'LOGS': tmp}):
            with self.assertRaises(SystemExit):
                with sdputils.maintenance_run('example.py'):
                    sys.exit(3)
            text = (Path(tmp) / 'example.log').read_text()
            self.assertIn('failed (exit 3)', text)
            self.assertNotIn('Completed', text)

    def test_unwritable_log_stops_before_operation(self):
        with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {'LOGS': tmp}):
            (Path(tmp) / 'example.log').mkdir()
            executed = False
            # The installed shared logger rejects directories before opening;
            # the append-only fallback rejects them at FileHandler creation.
            with self.assertRaises((IsADirectoryError, ValueError)):
                with sdputils.maintenance_run('example.py'):
                    executed = True
            self.assertFalse(executed)

    def test_python_prints_are_teed_to_log_and_console(self):
        out, err = io.StringIO(), io.StringIO()
        with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {'LOGS': tmp}), contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
            with self.assertRaises(SystemExit):
                with sdputils.maintenance_run('example.py'):
                    print('operation output')
                    print('specific failure detail', file=sys.stderr)
                    sys.exit(1)
            text = (Path(tmp) / 'example.log').read_text()
            self.assertIn('operation output', text)
            self.assertIn('specific failure detail', text)
        self.assertIn('operation output', out.getvalue())
        self.assertIn('specific failure detail', err.getvalue())

    def test_command_exception_never_logs_sensitive_arguments(self):
        with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {'LOGS': tmp}):
            with self.assertRaises(SystemExit) as exit_result:
                with sdputils.maintenance_run('example.py'):
                    raise subprocess.CalledProcessError(7, ['p4', '-P', 'secret-password'], stderr='server diagnostic')
            text = (Path(tmp) / 'example.log').read_text()
            self.assertNotIn('secret-password', text)
            self.assertIn('server diagnostic', text)
            self.assertEqual(exit_result.exception.code, 7)

    def test_windows_without_logs_uses_working_directory(self):
        with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {}, clear=True), patch.object(sdputils.platform, 'system', return_value='Windows'):
            old = os.getcwd()
            try:
                os.chdir(tmp)
                with sdputils.maintenance_run('example.py', instance='1'):
                    print('windows output')
                self.assertTrue(Path('example.log').exists())
            finally:
                os.chdir(old)

    def test_stdout_only_child_failure_reaches_log_and_console(self):
        out = io.StringIO()
        with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {'LOGS': tmp}), contextlib.redirect_stdout(out):
            with self.assertRaises(SystemExit) as exit_result:
                with sdputils.maintenance_run('child.py'):
                    sdputils.run_checked([sys.executable, '-c', 'print("child stdout diagnostic"); raise SystemExit(9)'])
            self.assertEqual(exit_result.exception.code, 9)
            self.assertIn('child stdout diagnostic', (Path(tmp) / 'child.log').read_text())
        self.assertEqual(out.getvalue(), 'child stdout diagnostic\n')

    def test_depot_compatibility_and_config_independent_of_cwd(self):
        with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {}, clear=True):
            old = os.getcwd()
            try:
                os.chdir(tmp)
                utils = sdputils.SDPUtils('1')
            finally:
                os.chdir(old)
            self.assertEqual(utils.p4, utils.p4binary)
            self.assertIn('swarm', sdputils.SKIP_USERS)
            self.assertIsNotNone(utils.get('weeks'))


class APIFailures(unittest.TestCase):
    def test_failed_queries_raise_instead_of_empty_success(self):
        cases = [
            ('accessdates', 'get_inactive_specs', ('clients', 'Access', 4)),
            ('unloadaccessdates', 'get_inactive_unloaded', ('clients', 4)),
            ('unload_clients', 'get_inactive_clients', (4,)),
            ('unload_clients_with_delete', 'get_inactive_clients', (4,)),
            ('unload_clients_with_delete', 'get_clients_with_exclusive_locks', ()),
            ('delete_unload_clients', 'get_inactive_unloaded_clients', (4,)),
            ('email_pending_client_deletes', '_get_inactive_clients', (4,)),
        ]
        for name, function, args in cases:
            with self.subTest(name=name, function=function), module(name) as mod:
                api = Mock()
                api.run.side_effect = P4Exception('permission denied')
                with self.assertRaises(P4Exception):
                    getattr(mod, function)(api, *args)

    def test_api_batch_failures_return_failure_and_continue_remaining_items(self):
        for name, function, extra in [
            ('unload_clients', 'unload_clients', ()),
            ('delete_unload_clients', 'delete_clients', ()),
            ('unload_clients_with_delete', 'process_clients', (set(),)),
        ]:
            with self.subTest(name=name), module(name) as mod:
                api = Mock()
                api.run.side_effect = [P4Exception('failed first'), []]
                self.assertFalse(getattr(mod, function)(api, ['first', 'second'], *extra))
                self.assertEqual(api.run.call_count, 2)

    def test_batch_main_exits_nonzero(self):
        for name in ['unload_clients', 'delete_unload_clients', 'unload_clients_with_delete']:
            with self.subTest(name=name), module(name) as mod:
                api = Mock()
                api.run.side_effect = lambda *args: [{'client': 'old', 'Access': '1'}] if args[0] == 'clients' else [] if args[0] == 'opened' else (_ for _ in ()).throw(P4Exception('mutation failed'))
                with patch.object(mod, 'connect_p4', return_value=api):
                    self.assertEqual(mod.main(['1']), 1)
                api.disconnect.assert_called_once()

    def test_branch_report_uses_branch_key(self):
        with module('accessdates') as mod:
            api = Mock()
            api.run.return_value = [{'branch': 'old-branch', 'Access': '1'}]
            self.assertEqual(mod.get_inactive_specs(api, 'branches', 'Access', 1), ['old-branch'])

    def test_exclusive_modifier_can_follow_other_modifiers(self):
        with module('unload_clients_with_delete') as mod:
            api = Mock()
            api.run.return_value = [{'client': 'locked', 'type': 'binary+Fl'}, {'client': 'plain', 'type': 'binary+F'}]
            self.assertEqual(mod.get_clients_with_exclusive_locks(api), {'locked'})

    def test_user_delete_failure_is_returned(self):
        with module('p4deleteuser') as mod:
            remover = object.__new__(mod.P4UserRemoverAPI)
            remover.p4 = Mock()
            remover.p4.run.side_effect = P4Exception('delete failed')
            self.assertFalse(remover.delete_user('ordinary'))
            self.assertTrue(remover.delete_user('p4admin'))
            self.assertEqual(remover.p4.run.call_count, 1)

    def test_failed_user_report_does_not_publish_partial_csv(self):
        with module('totalusers') as mod:
            api = Mock()
            api.run.side_effect = P4Exception('query failed')
            with patch.object(mod, 'P4', return_value=api), self.assertRaises(P4Exception):
                mod.gather_users_for_server(('test:1666', 'test', '1'), set(), False)
            api.disconnect.assert_called_once()

    def test_failed_unload_report_disconnects(self):
        with module('unloadaccessdates') as mod:
            api = Mock()
            api.run.side_effect = P4Exception('query failed')
            with patch.object(mod, 'connect_p4', return_value=api), patch.object(mod, 'write_list') as write, self.assertRaises(P4Exception):
                mod.main(['1'])
            write.assert_not_called()
            api.disconnect.assert_called_once()

    def test_failed_email_is_not_success(self):
        with module('email_pending_client_deletes') as mod:
            smtp = Mock()
            smtp.send_message.side_effect = RuntimeError('SMTP rejected message')
            with self.assertRaises(RuntimeError):
                mod._send_warning(smtp, '[email protected]', '[email protected]', None, 'client', 'one week', 'test')
        with module('email_pending_user_deletes') as mod:
            api = Mock()
            api.run.side_effect = P4Exception('users denied')
            with self.assertRaises(P4Exception):
                mod.warn_inactive_users(api, 4, 'test', '', '', None, 'test')


class Entrypoints(unittest.TestCase):
    def test_optional_instance_group_arguments(self):
        for args, instance in [(['alice', 'team'], '1'), (['2', 'alice', 'team'], '2')]:
            with self.subTest(args=args), module('addusertogroup', args) as mod:
                with patch.object(sdputils, 'run_checked', return_value=subprocess.CompletedProcess([], 0, 'Group: team\nUsers:\n')) as run:
                    mod.main()
                self.assertEqual(mod.SDP_INSTANCE, instance)
                self.assertEqual(run.call_args_list[0].args[0][-1], 'team')
                self.assertIn('\talice\n', run.call_args.kwargs['input'])

    def test_label_name_is_not_an_instance(self):
        with patch.dict(os.environ, {'SDP_INSTANCE': '2'}), module('convert_label_to_autoreload', ['release-label']) as mod:
            self.assertEqual(mod.SDP_INSTANCE, '2')

    def test_conversion_stops_before_delete_when_read_fails(self):
        with module('convert_label_to_autoreload', ['label']) as mod:
            with patch.object(sdputils, 'run_checked', side_effect=subprocess.CalledProcessError(1, ['p4'])) as run:
                with self.assertRaises(subprocess.CalledProcessError):
                    mod.convertlabel('label')
            self.assertEqual(run.call_count, 1)

    def test_pymail_flags_are_not_instance(self):
        with module('pymail', ['-t', '[email protected]', '-s', 'Subject', '-i', 'message.txt']) as mod:
            self.assertEqual(mod.SDP_INSTANCE, '1')
        with module('pymail', ['2', '-t', '[email protected]']) as mod:
            self.assertEqual(mod.SDP_INSTANCE, '2')
            self.assertEqual(sys.argv[1], '-t')

    def test_missing_label_is_nonzero(self):
        with module('isitalabel', ['absent']) as mod, patch.object(mod, 'check', return_value=0):
            with self.assertRaises(SystemExit) as exc:
                mod.main()
            self.assertEqual(exc.exception.code, 1)

    def test_group_owner_email_preserves_initial_characters(self):
        with module('group_audit', ['1']) as mod:
            smtp = Mock()
            with patch.object(sdputils, 'run_checked', return_value=subprocess.CompletedProcess([], 0, '... Email [email protected]\n')), patch.object(mod.smtplib, 'SMTP', return_value=smtp):
                mod.email_owner('team', 'alan')
            self.assertEqual(smtp.sendmail.call_args.args[1], '[email protected]')

    def test_shelf_cleanup_uses_current_query_and_preserves_unrelated_list(self):
        with module('del_shelve', ['1']) as mod, tempfile.TemporaryDirectory() as tmp:
            old = os.getcwd()
            try:
                os.chdir(tmp)
                Path('clients.txt').write_text('stale-client\n')
                api = Mock()
                with patch.object(mod.accessdates, 'connect_p4', return_value=api), patch.object(mod.accessdates, 'get_inactive_specs', return_value=['current-client']), patch.object(sdputils, 'run_checked', return_value=subprocess.CompletedProcess([], 0, '123\n')) as run:
                    mod.archive('clients', 'client', 4)
                self.assertEqual(run.call_args.args[0][-1], 'current-client')
                self.assertEqual(Path('clients.txt').read_text(), 'stale-client\n')
                api.disconnect.assert_called_once()
            finally:
                os.chdir(old)

    def test_user_cleanup_uses_current_inactivity_query(self):
        with module('delusers', ['1']) as mod:
            api, remover = Mock(), Mock()
            remover.delete_user.return_value = True
            with patch.object(mod.accessdates, 'connect_p4', return_value=api), patch.object(mod.accessdates, 'get_inactive_specs', return_value=['inactive']), patch.object(mod.p4deleteuser, 'P4UserRemoverAPI', return_value=remover):
                self.assertEqual(mod.main(), 0)
            remover.delete_user.assert_called_once_with('inactive')
            remover.disconnect.assert_called_once()

    def test_failed_describe_is_logged_and_not_deleted(self):
        with module('remove_empty_pending_changes', ['1']) as mod:
            error = subprocess.CalledProcessError(1, ['p4'], stderr='describe denied')
            with patch.object(mod.subprocess, 'run', side_effect=[subprocess.CompletedProcess([], 0, '123,user\n'), error]) as run, self.assertLogs(level='ERROR') as logs:
                self.assertEqual(mod.main(), 1)
            self.assertIn('describe denied', '\n'.join(logs.output))
            self.assertEqual(run.call_count, 2)

    def test_import_does_not_login(self):
        with module('addusertogroup', ['alice', 'team']) as mod:
            mod.utils.login.assert_not_called()


class RollbackAndFilesystem(unittest.TestCase):
    def test_failed_sync_stops_before_edit_or_print(self):
        with module('rollback') as mod:
            with patch.object(mod, 'get_client_file', return_value='/mock/file'), patch.object(mod, 'run_p4', return_value=('', 1)) as run, patch.object(mod.subprocess, 'run') as process:
                self.assertEqual(mod.rollback_file({'depotFile': '//depot/file', 'rev': 2, 'action': 'edit'}, 123, False), 1)
            self.assertEqual(run.call_count, 1)
            process.assert_not_called()

    def test_failed_edit_stops_before_print(self):
        with module('rollback') as mod:
            with patch.object(mod, 'get_client_file', return_value='/mock/file'), patch.object(mod, 'run_p4', side_effect=[('', 0), ('', 1)]), patch.object(mod.subprocess, 'run') as process:
                self.assertEqual(mod.rollback_file({'depotFile': '//depot/file', 'rev': 2, 'action': 'edit'}, 123, False), 1)
            process.assert_not_called()

    def test_incomplete_rollback_does_not_submit(self):
        with module('rollback', ['123', '--submit']) as mod:
            with patch.object(mod, 'check_submitted'), patch.object(mod, 'describe_changelist', return_value=[{}]), patch.object(mod, 'create_pending_cl', return_value=456), patch.object(mod, 'rollback_file', return_value=1), patch.object(mod, 'run_p4') as run:
                with self.assertRaises(SystemExit) as exc:
                    mod.main()
            self.assertEqual(exc.exception.code, 1)
            run.assert_not_called()

    def test_lowercase_collision_preserves_both_files(self):
        with module('lowertree') as mod, tempfile.TemporaryDirectory(prefix='sdp-review-') as tmp:
            root = Path(tmp)
            (root / 'File').write_text('upper')
            (root / 'file').write_text('lower')
            if os.path.samefile(root / 'File', root / 'file'):
                self.skipTest('Case-insensitive filesystem')
            with self.assertRaises(FileExistsError):
                mod.rename_dir(tmp)
            self.assertEqual((root / 'File').read_text(), 'upper')
            self.assertEqual((root / 'file').read_text(), 'lower')

    def test_lowercase_does_not_rename_parent_directory(self):
        with module('lowertree') as mod, tempfile.TemporaryDirectory(prefix='sdp-review-') as tmp:
            root = Path(tmp) / 'Parent'
            root.mkdir()
            (root / 'Child').mkdir()
            (root / 'Child/File').write_text('data')
            mod.rename_dir(str(root / 'Child'))
            self.assertIn('Parent', os.listdir(tmp))
            self.assertIn('child', os.listdir(root))
            self.assertIn('file', os.listdir(root / 'child'))

    def test_collision_guard_on_case_insensitive_test_host(self):
        with module('lowertree') as mod, patch.object(mod.os.path, 'exists', return_value=True), patch.object(mod.os.path, 'samefile', return_value=False), patch.object(mod.os, 'rename') as rename:
            with self.assertRaises(FileExistsError):
                mod.rename_dir('Conflict')
            rename.assert_not_called()

    def test_lowercase_does_not_follow_directory_symlink(self):
        with module('lowertree') as mod, tempfile.TemporaryDirectory(prefix='sdp-review-') as tmp:
            root = Path(tmp)
            (root / 'tree').mkdir()
            (root / 'outside').mkdir()
            (root / 'outside/Preserve').write_text('data')
            (root / 'tree/link').symlink_to(root / 'outside', target_is_directory=True)
            mod.rename_dir(str(root / 'tree'))
            self.assertTrue((root / 'outside/Preserve').exists())


class Wrappers(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory(prefix='sdp-review-')
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name)
        self.p4 = self.root / 'p4'
        self.common = self.p4 / 'common/bin'
        self.maint = self.p4 / 'sdp/Maintenance'
        self.mockbin = self.root / 'mockbin'
        for path in [self.common, self.maint, self.mockbin]:
            path.mkdir(parents=True)
        self.env = dict(os.environ, HOME=str(self.root), PATH=str(self.mockbin)+':/usr/bin:/bin', HOSTNAME='test', TEST_ROOT=str(self.root))
        self.write(self.common / 'p4_vars', 'export P4SERVER=p4_1 P4PORT=test:1666 P4TMP="$TEST_ROOT/p4/$1/tmp"\n')
        self.write(self.common / 'p4login', 'echo login-diagnostic\nexit "${FAIL_LOGIN:-0}"\n')
        self.write(self.mockbin / 'pyenv', 'exit 0\n')
        self.write(self.mockbin / 'rm', 'case "$*" in *server.locks*) echo lock-cleanup >> "$TEST_ROOT/actions";; *) exec /bin/rm "$@";; esac\n')
        self.write(self.mockbin / 'accessdates.py', 'echo report-diagnostic\nif [[ ${FAIL_REPORT:-0} == 0 ]]; then echo old-client > clients.txt; fi\nexit "${FAIL_REPORT:-0}"\n')
        self.write(self.mockbin / 'unload_clients.py', 'echo unloading >> "$TEST_ROOT/actions"\nexit "${FAIL_UNLOAD:-0}"\n')
        self.write(self.mockbin / 'remove_empty_pending_changes.py', 'exit 0\n')
        for name in ['maintenance', 'edge_maintenance']:
            self.write(self.maint / name, (MAINT / name).read_text().replace('/p4/', str(self.p4) + '/'))
        shutil.copyfile(MAINT / 'maintenance_workspace.sh', self.maint / 'maintenance_workspace.sh')
        (self.maint / 'clients.txt').write_text('stale-client\n')

    def write(self, path, text):
        path.write_text('#!' + BASH + '\n' + text)
        path.chmod(0o755)

    def run_wrapper(self, name, **env):
        return subprocess.run([BASH, str(self.maint / name), '1'], env=dict(self.env, **env), cwd=self.root, capture_output=True, text=True, timeout=10)

    def test_login_failure_stops_before_cleanup_and_is_logged(self):
        for name in ['maintenance', 'edge_maintenance']:
            with self.subTest(name=name):
                result = self.run_wrapper(name, FAIL_LOGIN='7')
                self.assertNotEqual(result.returncode, 0, result.stderr)
                self.assertFalse((self.root / 'actions').exists())
                self.assertIn('login-diagnostic', (self.maint / 'test_p4_1_log.txt').read_text())

    def test_report_failure_does_not_unload_stale_clients(self):
        for name in ['maintenance', 'edge_maintenance']:
            with self.subTest(name=name):
                result = self.run_wrapper(name, FAIL_REPORT='7')
                self.assertEqual(result.returncode, 1, result.stderr)
                self.assertNotIn('unloading', (self.root / 'actions').read_text())

    def test_unload_failure_has_nonzero_exit(self):
        for name in ['maintenance', 'edge_maintenance']:
            with self.subTest(name=name):
                result = self.run_wrapper(name, FAIL_UNLOAD='7')
                self.assertEqual(result.returncode, 1, result.stderr)

    def test_unwritable_wrapper_log_stops_before_cleanup(self):
        (self.maint / 'test_p4_1_log.txt').mkdir()
        result = self.run_wrapper('maintenance')
        self.assertNotEqual(result.returncode, 0)
        self.assertFalse((self.root / 'actions').exists())

    def test_file_inventory_query_failures_are_logged_and_nonzero(self):
        self.write(self.common / 'p4_vars', 'export P4BIN="$TEST_ROOT/mockbin/p4" LOGS="$TEST_ROOT"\n')
        self.write(self.common / 'sdp_structured_log.sh', 'sdp_run_log_init() { exec >> "$LOGS/filelist.log" 2>&1; trap \'_sdp_on_exit "$?"\' EXIT; }\n_sdp_on_exit() { echo "$1" > "$LOGS/filelist.exit"; }\n')
        fake_p4 = self.mockbin / 'p4'
        fake_p4.write_text('#!' + sys.executable + '\n' + '''
import marshal, os, sys
stage = sys.argv[3]
if stage == os.environ['FAIL_STAGE']:
    print(stage + ' specific failure', file=sys.stderr)
    sys.exit(7)
record = {'code': 'stat'}
record.update({'depots': {'name': 'depot'}, 'dirs': {'dir': '//depot/dir'},
               'files': {'depotFile': '//depot/dir/file'}}[stage])
marshal.dump(record, sys.stdout.buffer)
''')
        fake_p4.chmod(0o755)
        script = self.maint / 'create_p4_filelist.sh'
        self.write(script, (MAINT / script.name).read_text().replace('/p4/', str(self.p4) + '/'))
        shutil.copyfile(MAINT / 'file_inventory.py', self.maint / 'file_inventory.py')
        # Concatenation is now replaced by an atomic publish; its write-failure
        # coverage lives in test_maintenance_improvements.Inventory.
        for stage in ['depots', 'dirs', 'files']:
            with self.subTest(stage=stage):
                (self.root / 'filelist.log').write_text('')
                result = subprocess.run([BASH, str(script), '1'], env=dict(self.env, FAIL_STAGE=stage), cwd=self.maint, capture_output=True, text=True, timeout=10)
                self.assertNotEqual(result.returncode, 0)
                text = (self.root / 'filelist.log').read_text()
                self.assertIn(stage + ' specific failure', text)
                self.assertNotIn('Done.', text)
                self.assertEqual((self.root / 'filelist.exit').read_text().strip(), '7')

    def test_email_file_operation_failures_preserve_list_and_never_send(self):
        self.write(self.mockbin / 'p4', 'echo [email protected]\n')
        self.write(self.mockbin / 'python3', 'echo sent >> "$TEST_ROOT/actions"\n')
        for utility in ['cp', 'mktemp', 'mv']:
            original = shutil.which(utility)
            self.write(self.mockbin / utility,
                       'if [[ "$FAIL_UTILITY" == ' + utility + ' ]]; then echo "' + utility + ' specific failure" >&2; exit 7; fi\nexec "' + original + '" "$@"\n')
        script = self.maint / 'email.sh'
        self.write(script, (MAINT / script.name).read_text().replace('/p4/', str(self.p4) + '/'))
        for utility in ['cp', 'mktemp', 'mv']:
            with self.subTest(utility=utility):
                (self.maint / 'emaillist.txt').write_text('[email protected]\n')
                result = subprocess.run([BASH, str(script), 'Subject'], env=dict(self.env, FAIL_UTILITY=utility), cwd=self.maint, capture_output=True, text=True, timeout=10)
                self.assertEqual(result.returncode, 7)
                self.assertIn(utility + ' specific failure', result.stderr)
                self.assertEqual((self.maint / 'emaillist.txt').read_text(), '[email protected]\n')
                if (self.root / 'actions').exists():
                    self.assertNotIn('sent', (self.root / 'actions').read_text())

    def test_failed_email_list_query_preserves_list_and_never_sends(self):
        self.write(self.mockbin / 'p4', 'echo denied >&2\nexit 1\n')
        self.write(self.mockbin / 'python3', 'echo sent >> "$TEST_ROOT/actions"\n')
        (self.maint / 'emaillist.txt').write_text('[email protected]\n')
        script = self.maint / 'email.sh'
        self.write(script, (MAINT / 'email.sh').read_text().replace('/p4/', str(self.p4) + '/'))
        result = subprocess.run([BASH, str(script), 'Subject'], env=self.env, cwd=self.maint, capture_output=True, text=True, timeout=10)
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual((self.maint / 'emaillist.txt').read_text(), '[email protected]\n')
        actions = (self.root / 'actions').read_text() if (self.root / 'actions').exists() else ''
        self.assertNotIn('sent', actions)


if __name__ == '__main__':
    unittest.main()