#!/usr/bin/env python3 """Offline regression checks for Unix setup and trigger fixes. No Perforce, LDAP, RADIUS or Aspera service is contacted. """ import ast import contextlib import io import os from pathlib import Path import re import shutil import shlex import subprocess import sys import tempfile import types import unittest from unittest import mock SDP = Path(__file__).resolve().parents[1] SETUP = SDP / 'Server/Unix/setup' TRIGGERS = SDP / 'Server/Unix/p4/common/bin/triggers' BASH = shutil.which('bash') or '/bin/bash' def definitions(path, namespace=None): """Load actual functions/classes without running a script's main body.""" tree = ast.parse(path.read_text()) tree.body = [node for node in tree.body if isinstance(node, (ast.FunctionDef, ast.ClassDef))] namespace = dict(namespace or {}) exec(compile(tree, str(path), 'exec'), namespace) return namespace class ShellReviewTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='sdp-setup-tests-') self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) def executable(self, name, text): path = self.root / name path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text) path.chmod(0o755) return path def test_os_tweaks_mixed_correct_and_incorrect_settings_terminates(self): source = (SETUP / 'os_tweaks.sh').read_text() for number, (name, current) in enumerate([ ('khugepaged/defrag', '0'), ('defrag', '[always] madvise never'), ('enabled', 'always madvise [never]')]): fixture = self.root / ('setting%d' % number) fixture.write_text(current) source = source.replace('/sys/kernel/mm/transparent_hugepage/' + name, str(fixture)) whoami = self.executable('bin/whoami', '#!/bin/sh\necho root\n') script = self.root / 'os_tweaks.sh' script.write_text(source) env = dict(os.environ, PATH=str(whoami.parent) + os.pathsep + os.environ['PATH'], LOGFILE=str(self.root / 'os.log')) result = subprocess.run([BASH, str(script)], env=env, capture_output=True, text=True, timeout=5) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertEqual((self.root / 'setting1').read_text().strip(), 'never') self.assertIn('Done. All OS parameters', (self.root / 'os.log').read_text()) def test_installers_preserve_unexpected_failure_exit_codes(self): for name, state in [('install_sdp.sh', 'ErrorCount'), ('gen_sudoers.sh', 'ErrorCount'), ('install_sdp_python.sh', 'OverallReturnStatus')]: with self.subTest(script=name): source = (SETUP / name).read_text() start = source.index('function terminate\n{') function = source[start:source.index('\n}\n', start) + 3] command = ('%s=0; ThisScript=test; THISSCRIPT=test; Log=off; P4U_LOG=off; H1=x; ' 'msg(){ :; }; vvmsg(){ :; }; cleanTrash(){ :; }; stoplog(){ :; };\n' % state) command += function + '\ntrap terminate EXIT\nexit 23\n' result = subprocess.run([BASH, '-c', command], capture_output=True, text=True) self.assertEqual(result.returncode, 23, result.stderr) def trigger_environment(self, fstat_output='', fstat_rc=0, ascp_rc=0): logs = self.root / 'log directory'; logs.mkdir() scratch = self.root / 'scratch directory'; scratch.mkdir() p4 = self.executable('bin/p4 with space', '#!/bin/sh\nprintf \'%s\\n\' ' + shlex.quote(fstat_output) + '\nexit %d\n' % fstat_rc) self.executable('bin/ascp', '#!' + sys.executable + '\nimport os,sys,pathlib\n' 'p=next(a.split("=",1)[1] for a in sys.argv if a.startswith("--file-pair-list="))\n' 'pathlib.Path(os.environ["CAPTURE"]).write_bytes(pathlib.Path(p).read_bytes())\n' 'sys.exit(%d)\n' % ascp_rc) env = dict(os.environ, PATH=str(self.root / 'bin') + os.pathsep + os.environ['PATH'], LOGS=str(logs), P4TMP=str(scratch), P4BIN=str(p4), OSUSER='perforce', P4MASTER='fixture', CAPTURE=str(self.root / 'pairs'), SDP_LOG_FORMAT='off') return env, logs, scratch def test_submit_preserves_archive_spaces_backslashes_and_cleans_temporaries(self): path = '/depot/path with spaces/a\\b*file,v' env, logs, scratch = self.trigger_environment('... lbrPath ' + path) result = subprocess.run([BASH, str(TRIGGERS / 'submit.sh'), '123'], env=env, capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertEqual((self.root / 'pairs').read_text(), path + '\n' + path + '\n') self.assertEqual(list(scratch.iterdir()), []) def test_submit_retains_failure_details_and_removes_temporaries(self): env, logs, scratch = self.trigger_environment('fixture-fstat-error', 1) result = subprocess.run([BASH, str(TRIGGERS / 'submit.sh'), '123'], env=env, capture_output=True, text=True) self.assertNotEqual(result.returncode, 0) self.assertIn('fixture-fstat-error', (logs / 'submit.log').read_text()) self.assertEqual(list(scratch.iterdir()), []) def test_submit_cleans_temporaries_when_no_archive_files_exist(self): env, logs, scratch = self.trigger_environment('') result = subprocess.run([BASH, str(TRIGGERS / 'submit.sh'), '123'], env=env, capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertEqual(list(scratch.iterdir()), []) def test_pull_missing_archive_list_is_logged(self): env, logs, _ = self.trigger_environment() result = subprocess.run([BASH, str(TRIGGERS / 'pull.sh'), str(self.root / 'missing')], env=env, capture_output=True, text=True) self.assertNotEqual(result.returncode, 0) self.assertIn('Archive list is missing', (logs / 'pull.log').read_text()) def test_mkdirs_preview_stays_in_preview_tree(self): # Substitute only the preview root and volume in a private script copy. # Fake binaries merely print versions; no service operation is possible. package = self.root / 'package' common = package / 'Server/Unix/p4/common' for area in ['config', 'etc']: shutil.copytree(SDP / 'Server/Unix/p4/common' / area, common / area) (common / 'bin').mkdir() for name in ['p4', 'p4d']: binary = common / 'bin' / name binary.write_text('#!/bin/sh\necho "Rev. P4/LINUX/2026.1/123456 (fixture)"\n') binary.chmod(0o755) (package / 'Maintenance').mkdir() shutil.copyfile(SDP / 'Maintenance/template.maintenance.cfg', package / 'Maintenance/template.maintenance.cfg') preview_volume = self.root / 'preview' preview = preview_volume / 'p4' script = self.root / 'mkdirs.sh' source = (SETUP / 'mkdirs.sh').read_text() source = source.replace('P4DIR=/tmp/p4', 'P4DIR=' + str(preview)) source = re.sub(r'(?m)^( (?:DD|DB1|DB2|CKP|LG|JR|CN)=)tmp$', lambda m: m[1] + str(preview_volume).lstrip('/'), source) script.write_text(source) config = self.root / 'mkdirs.cfg' config.write_text('\n'.join([ 'SDP_INSTANCE=9', 'MASTER_ID=master', 'REPLICA_ID=replica', 'SERVER_TYPE=p4d_master', 'SDP=' + str(package), 'OSUSER=' + os.environ.get('USER', 'fixture'), 'OSGROUP=staff', 'DD=forbidden_depots', 'DB1=forbidden_db1', 'DB2=forbidden_db2', 'CKP=forbidden_checkpoints', 'LG=forbidden_logs', 'JR=forbidden_journals', 'CN=forbidden_common', 'CASE_SENSITIVE=1', 'P4ADMINPASS=fixture_only', 'ADMINUSER=perforce', 'P4_PORT=1666', 'P4BROKER_PORT=1667', 'SHAREDDATA=FALSE', 'P4DNSNAME=fixture', 'SSL_PREFIX=', 'MAILTO=test@example.invalid', 'MAILFROM=test@example.invalid', 'COMPLAINFROM=test@example.invalid', 'MAILHOST=fixture', 'P4COMMITSERVER=fixture', 'DOMAIN=example.invalid', 'MASTERINSTANCE=9'])) result = subprocess.run([BASH, str(script), str(config), '-test'], env=dict(os.environ, LOGFILE=str(self.root / 'setup.log')), capture_output=True, text=True, timeout=15) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertEqual((preview / '9/root/server.id').read_text(), 'master\n') self.assertTrue((preview / '9/bin/p4_9').is_symlink()) self.assertIn('INSTANCE=9', (preview / 'proxy.crontab').read_text()) self.assertTrue((preview / 'sdp/Maintenance/maintenance.cfg').exists()) self.assertFalse((package / 'Maintenance/maintenance.cfg').exists()) self.assertEqual((common / 'bin/p4d').stat().st_mode & 0o777, 0o755) self.assertIn('Test mode complete', (self.root / 'setup.log').read_text()) class PythonTriggerTests(unittest.TestCase): def test_retired_otp_scripts_are_absent(self): self.assertFalse((TRIGGERS / 'otpauthcheck.py').exists()) self.assertFalse((TRIGGERS / 'otpkeygen.py').exists()) def test_case_trigger_rejects_differently_cased_new_directories(self): ns = definitions(TRIGGERS / 'CheckCaseTrigger.py', {'P4Triggers': types.SimpleNamespace(P4Trigger=object)}) cls = ns['CheckCaseTrigger']; trigger = cls.__new__(cls) trigger.p4 = types.SimpleNamespace(charset='none', run_dirs=lambda path: []) trigger.dirCache = {} self.assertIsNone(trigger.findDirMismatch('//depot', ['New', 'Nested'])) self.assertEqual(trigger.findDirMismatch('//depot', ['new', 'Nested']), '//depot/New') self.assertEqual(trigger.findDirMismatch('//depot', ['New', 'nested']), '//depot/New/Nested') def test_label_options_preserves_other_form_fields(self): ns = definitions(TRIGGERS / 'SetLabelOptions.py', {'re': re}) with tempfile.TemporaryDirectory() as directory: path = Path(directory) / 'label' path.write_text('Label: noautoreload\nDescription:\n\tnoautoreload is in this description\nOptions: unlocked noautoreload\nView:\n\t//noautoreload/...\n') ns['set_label_autoreload'](str(path)) content = path.read_text() self.assertIn('Options: unlocked autoreload', content) self.assertIn('Label: noautoreload', content) self.assertIn('//noautoreload/...', content) self.assertIn('noautoreload is in this description', content) def test_radius_tries_next_server_after_timeout_and_logs_error(self): class Request(dict): def PwCrypt(self, password): return b'fixture' class Client: def __init__(self, server, **kwargs): self.server = server def CreateAuthPacket(self, **kwargs): return Request() def SendPacket(self, request): if self.server == 'unavailable': raise TimeoutError('fixture-timeout') return types.SimpleNamespace(code=2) with tempfile.TemporaryDirectory() as directory: errorlog = Path(directory) / 'radius.log' ns = definitions(TRIGGERS / 'rad_authcheck.py', { 'Client': Client, 'Dictionary': lambda value: value, 'pyrad': types.SimpleNamespace(packet=types.SimpleNamespace(AccessAccept=2, AccessRequest=1)), 'RAD_SERVERS': ['unavailable', 'working'], 'RAD_SHARED_SECRET': b'fixture', 'RAD_DICTIONARY': 'fixture', 'ERRORLOG': str(errorlog), 'serverid': 'fixture', 'sys': sys}) self.assertEqual(ns['checkRadiusPassword']('fixture-user', 'fixture-password'), 0) self.assertIn('fixture-timeout\n', errorlog.read_text()) self.assertNotIn('fixture-password', errorlog.read_text()) def test_radius_local_password_can_contain_colons(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / 'local.passwd'; path.write_text('user:a:b:c\n') ns = definitions(TRIGGERS / 'rad_authcheck.py', {'LOCAL_PASSWD_FILE': str(path), 'os': os}) self.assertEqual(ns['getLocalPassword']('user'), 'a:b:c') def test_group_query_failure_rejects_form_and_reports_diagnostic(self): ns = definitions(TRIGGERS / 'submit_form_1.py', { 'sys': types.SimpleNamespace(argv=['trigger', 'unused', 'user;echo unexpected'], exit=sys.exit), 'os': os, 'subprocess': subprocess}) with mock.patch.object(subprocess, 'check_output', side_effect=subprocess.CalledProcessError(1, 'p4', output='fixture outage')) as command: output = io.StringIO() with contextlib.redirect_stdout(output), self.assertRaises(SystemExit) as result: ns['main']() self.assertEqual(result.exception.code, 1) self.assertIn('fixture outage', output.getvalue()) self.assertEqual(command.call_args.args[0][-1], 'user;echo unexpected') self.assertIsInstance(command.call_args.args[0], list) if __name__ == '__main__': unittest.main()