test_ccheck_compatibility.py #1

#!/usr/bin/env python3
"""Offline configuration compatibility checks; never connects to a server."""
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
import unittest

ROOT = Path(__file__).resolve().parents[1]
CHECK = ROOT / 'Server/Unix/p4/common/bin/ccheck.sh'
SETUP = ROOT / 'Server/Unix/setup/configure_new_server.sh'
BASH = os.environ.get('SDP_TEST_BASH', shutil.which('bash'))


def function(name):
    text = CHECK.read_text()
    start = re.search(r'^function ' + name + r'\s*\(\)\s*\{', text, re.M)
    return text[start.start():text.index('\n}', start.end()) + 2]


class Compatibility(unittest.TestCase):
    def shell(self, code, **environment):
        return subprocess.run([BASH, '-c', code], env=dict(os.environ, **environment), text=True, capture_output=True)

    def test_numeric_release_boundaries(self):
        helper = function('ccheck_release_at_least')
        for current, minimum, expected in [('2023.2.999', '2024.1', 1), ('2024.1.1', '2024.1', 0), ('2024.2.1', '2025.1', 1), ('2025.1.1', '2025.1', 0), ('2025.1.999', '2025.2', 1), ('2025.2.1', '2025.2', 0), ('2024.10.1', '2024.2', 0), ('invalid', '2024.1', 2)]:
            with self.subTest(current=current, minimum=minimum):
                result = self.shell(helper + '\nccheck_release_at_least "$CURRENT" "$MINIMUM"', CURRENT=current, MINIMUM=minimum)
                self.assertEqual(result.returncode, expected)

    def test_only_documented_new_configurables_are_skipped(self):
        helpers = function('ccheck_release_at_least') + '\n' + function('ccheck_configurable_supported')
        for setting, before, supported in [('dm.user.hideinvalid', '2023.2.9', '2024.1.1'), ('db.journalrotate.warnthresh', '2023.2.9', '2024.1.1'), ('server.startup.autorestart', '2024.2.9', '2025.1.1'), ('lbr.unloaddepot.compress', '2025.1.9', '2025.2.1')]:
            with self.subTest(setting=setting):
                code = 'msg(){ printf "%s\\n" "$*"; };\n' + helpers + '\nccheck_configurable_supported "$SETTING"'
                old = self.shell(code, SETTING=setting, P4D_VERSION=before)
                self.assertEqual(old.returncode, 1)
                self.assertIn('COMPATIBILITY: Skipping ' + setting, old.stdout)
                new = self.shell(code, SETTING=setting, P4D_VERSION=supported)
                self.assertEqual(new.returncode, 0)
                self.assertNotIn('Skipping', new.stdout)
        unknown = self.shell(helpers + '\nccheck_configurable_supported unknown.typo', P4D_VERSION='2023.1.1')
        self.assertEqual(unknown.returncode, 0, 'An unknown name must reach normal query/error handling')

    def test_automation_login_is_optional_but_real_failures_propagate(self):
        helper = function('ccheck_login_automation')
        code = 'msg(){ printf "%s\\n" "$*"; }; run(){ echo "CALLED:$1"; return "$FAILURE"; };\n' + helper + '\nccheck_login_automation'
        absent_env = dict(os.environ)
        absent_env.pop('SDP_AUTOMATION_USERS', None)
        absent = subprocess.run([BASH, '-c', code], env=dict(absent_env, FAILURE='7'), text=True, capture_output=True)
        self.assertEqual(absent.returncode, 0)
        self.assertIn('not required', absent.stdout)
        self.assertNotIn('CALLED:', absent.stdout)
        result = self.shell(code, SDP_AUTOMATION_USERS='builder,trigger', FAILURE='7')
        self.assertEqual(result.returncode, 7)
        self.assertIn('CALLED:p4login -v -automation -L off', result.stdout)

    def test_failed_unknown_configurable_query_is_an_error_not_a_skip(self):
        with tempfile.TemporaryDirectory() as temp:
            folder = Path(temp)
            p4 = folder / 'p4'
            p4.write_text('#!' + sys.executable + '\nimport sys\nprint("permission denied", file=sys.stderr)\nsys.exit(9)\n')
            p4.chmod(0o755)
            config = folder / 'config'
            config.write_text('always|unknown.typo|1|Exact|Required|any|None\n')
            text = CHECK.read_text()
            block = text[text.index('# shellcheck disable=2094'):text.index('done < "$TmpFile"', text.index('# shellcheck disable=2094')) + len('done < "$TmpFile"')]
            code = 'ErrorCount=0; msg(){ echo "$*"; }; dbg(){ :; }; errmsg(){ echo "$*"; ErrorCount=$((ErrorCount+1)); }; declare -A CurrentValue CurrentValueByDefault ExpectedValue CompareStyle Optionality ServerIDType SetNotes Categories;\n' + function('ccheck_release_at_least') + '\n' + function('ccheck_configurable_supported') + '\n' + block + '\nexit "$ErrorCount"'
            result = self.shell(code, TmpFile=str(config), P4=str(p4), P4D_VERSION='2023.1.1', DoSecurityCheck='0', SDPInstance='1', P4ROOT='/unused', P4SERVER='p4_1', KEEPLOGS='7', LOGS='/unused')
            self.assertEqual(result.returncode, 1)
            self.assertIn('Could not query configurable', result.stdout)
            self.assertNotIn('COMPATIBILITY:', result.stdout)

    def test_setup_aborts_after_configuration_application_failure(self):
        with tempfile.TemporaryDirectory() as temp:
            folder = Path(temp)
            check = folder / 'ccheck.sh'
            check.write_text('#!/bin/sh\nexit 7\n')
            check.chmod(0o755)
            text = SETUP.read_text()
            block = text[text.index('if [[ -x "$P4CBIN/ccheck.sh" ]]'):text.index('msg "Restarting server to ensure all configurable changes take effect."')]
            code = 'msg(){ :; }; bail(){ echo "$*"; exit 1; };\n' + block + '\necho UNSAFE_CONTINUE'
            result = self.shell(code, P4CBIN=str(folder), INSTANCE='1')
            self.assertEqual(result.returncode, 1)
            self.assertIn('failed to apply configurables', result.stdout)
            self.assertNotIn('UNSAFE_CONTINUE', result.stdout)

    def disk_helpers(self):
        return '\n'.join(function(name) for name in ('ccheck_release_at_least', 'at_least_size_compare', 'convert_size_values', 'safety_check_for_filesys_min'))

    def test_diskspace_validates_status_and_every_record_before_comparing_minimum(self):
        code = '''ErrorCount=0; ArchiveRootPreview=0; P4=fake_p4
msg(){ echo "$*"; }; dbg(){ :; }; warnmsg(){ echo "$*"; }
errmsg(){ echo "ERROR:$*"; ErrorCount=$((ErrorCount+1)); }
fake_p4(){ printf '%s' "$RECORDS"; return "$QUERY_STATUS"; }
''' + self.disk_helpers() + '''
safety_check_for_filesys_min filesys.depot.min 5G
status=$?; echo "errors=$ErrorCount"; exit "$status"
'''
        cases = [('', '0', 1), ('6000000000', '7', 1), ('invalid', '0', 1), ('6000000000\ninvalid', '0', 1), ('9223372036854775808', '0', 1), ('6000000000\n4000000000', '0', 1), ('4000000000\n6000000000', '0', 1), ('6000000000\n7000000000', '0', 0), ('0006000000000', '0', 0)]
        for records, query_status, expected in cases:
            with self.subTest(records=records, query_status=query_status):
                result = self.shell(code, RECORDS=records, QUERY_STATUS=query_status, P4D_VERSION='2024.1.3006289')
                self.assertEqual(result.returncode, expected, result.stdout + result.stderr)
                self.assertIn('errors=' + str(expected), result.stdout)
                self.assertNotIn('convert_size_values(): Bad Usage', result.stdout)

    def test_unit_only_size_is_rejected(self):
        result = self.shell('errmsg(){ echo ERROR; };\n' + function('at_least_size_compare') + '\nat_least_size_compare B 5G')
        self.assertEqual(result.returncode, 1)
        self.assertIn('ERROR', result.stdout)

    def test_archive_root_dependency_preserves_fix_preview_and_failure_semantics(self):
        code = '''ErrorCount=0; ArchiveRootPreview=0; ConfigFixedCount=0; ArchiveConfigured=0; Queries=0; P4=fake_p4
declare -a EasyFixConfigurables EasyFixValues EasyFixApplied PlannedFixApplied
PlannedFixConfigurables=(server.depot.root); PlannedFixAdvice=(SET_ROOT); PlannedFixAutomated=(1)
msg(){ echo "$*"; }; dbg(){ :; }; warnmsg(){ echo "$*"; }
errmsg(){ echo "ERROR:$*"; ErrorCount=$((ErrorCount+1)); }
run(){ echo "RUN:$1"; [[ "$RUN_STATUS" == 0 ]] || return "$RUN_STATUS"; [[ "$NoOp" == 1 ]] || ArchiveConfigured=1; return 0; }
fake_p4(){ echo QUERY >&2; [[ "$ArchiveConfigured" == 1 ]] && echo 6000000000; return 0; }
if [[ "$CATEGORY" == easy ]]; then
 EasyFixConfigurables=(server.depot.root); EasyFixValues=(/srv/archives); PlannedFixConfigurables=()
fi
''' + self.disk_helpers() + '\n' + function('ccheck_prepare_archive_root') + '''
ccheck_prepare_archive_root || exit "$?"
safety_check_for_filesys_min filesys.depot.min 5G
status=$?
echo "configured=$ArchiveConfigured fixed=$ConfigFixedCount easy=${EasyFixApplied[0]:-0} planned=${PlannedFixApplied[0]:-0}"
exit "$status"
'''
        for category, all_fixes, preview, run_status, expected, configured in [('planned', '1', '0', '0', 0, 1), ('planned', '0', '0', '0', 1, 0), ('planned', '1', '1', '0', 0, 0), ('planned', '1', '0', '7', 1, None), ('easy', '0', '0', '0', 0, 1)]:
            with self.subTest(category=category, all_fixes=all_fixes, preview=preview, run_status=run_status):
                result = self.shell(code, CATEGORY=category, ApplyAllFixes=all_fixes, NoOp=preview, RUN_STATUS=run_status, P4D_VERSION='2024.1.3006289')
                self.assertEqual(result.returncode, expected, result.stdout + result.stderr)
                if configured is not None:
                    self.assertIn('configured=' + str(configured), result.stdout)
                if preview == '1':
                    self.assertNotIn('QUERY', result.stderr)
                    self.assertIn('NO_OP:', result.stdout)
                if all_fixes == '0' and category == 'planned':
                    self.assertNotIn('RUN:', result.stdout)
                if run_status != '0':
                    self.assertNotIn('QUERY', result.stderr)


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