version_compatibility.py #2

#!/usr/bin/env python3
"""Clean P4 release compatibility; no upgrade or existing-install migration.

Host mode runs seven published server release lines on two immutable UBI images.
Worker mode only operates inside a marked disposable container as its SDP user.
Every required cell/scenario must pass; missing binaries or capabilities fail.
"""
import argparse
from concurrent.futures import ThreadPoolExecutor, as_completed
import hashlib
import json
import marshal
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import time

from baseline_comparison import verify_manifest

RELEASES = ('2023.1', '2023.2', '2024.1', '2024.2', '2025.1', '2025.2', '2026.1')
PLATFORMS = ('rhel8', 'rhel9')
SCENARIOS = ('clean_install', 'service_user_authentication', 'exact_fixture', 'restart_preserves_data', 'checkpoint_restore',
             'journal_rotation_restore', 'maintenance_queries', 'sendq_cleanup')
LABEL = 'com.perforce.sdp.version-disposable'
SYSTEMD_START_TIMEOUT = 180
IMAGES = {'rhel8': 'sdp-version-rhel8:8.10', 'rhel9': 'sdp-version-rhel9:9.8'}
LEGACY_IMAGES = {'rhel8': 'sdp-version-rhel8-amd64:8.10', 'rhel9': 'sdp-version-rhel9-amd64:9.8'}


def capability_gaps(provision_text):
    count = provision_text.count("Skipping safety check; p4d version is too old to support needed 'p4 diskspace' features.")
    gaps = ([str(count) + ' installer disk-space safety checks are unavailable in this server release; see provision.log.']
            if count else [])
    gaps.extend(dict.fromkeys(line.strip() for line in provision_text.splitlines()
                              if line.strip().startswith('COMPATIBILITY: Skipping ')))
    return gaps


def configuration_errors(provision_text):
    return list(dict.fromkeys(line.strip() for line in provision_text.splitlines()
                              if re.match(r'\s*Error:', line) or re.match(r'\s*ExitCode:\s*[1-9][0-9]*\s*$', line)))


def clone_process_alive(pid):
    # Worker mode runs inside Linux; admin stop acknowledges before exit.
    return Path('/proc/' + str(pid)).exists()


def stop_clone(clone, stop, timeout=30):
    """Delete scratch only after confirmed shutdown; preserve uncertain recovery data."""
    try:
        pid = int((clone / 'pid').read_text().strip())
        if pid <= 1:
            raise ValueError('Invalid clone PID')
        stop()
        deadline = time.monotonic() + timeout
        while clone_process_alive(pid):
            if time.monotonic() >= deadline:
                raise RuntimeError('Clone did not exit after admin stop')
            time.sleep(.05)
    except Exception as error:
        raise RuntimeError('Clone shutdown is unconfirmed; scratch retained at ' + str(clone)) from error
    shutil.rmtree(clone)


def matrix_errors(rows, releases=RELEASES, platforms=PLATFORMS, source_hash=None):
    errors = []
    expected = {(platform, release) for platform in platforms for release in releases}
    actual = [(row.get('platform'), row.get('release')) for row in rows]
    if len(set(actual)) != len(actual):
        errors.append('Duplicate platform/release cell')
    for key in sorted(expected - set(actual)):
        errors.append('Missing required cell: ' + '/'.join(key))
    for key in sorted(set(actual) - expected, key=str):
        errors.append('Unexpected cell: ' + '/'.join(str(value) for value in key))
    for row in rows:
        key = str(row.get('platform')) + '/' + str(row.get('release'))
        if row.get('status') != 'passed':
            errors.append(key + ': did not pass')
        scenarios = row.get('scenarios', [])
        names = [scenario.get('name') for scenario in scenarios]
        if len(set(names)) != len(names) or set(SCENARIOS) != set(names):
            errors.append(key + ': incomplete/duplicate scenarios')
        if any(scenario.get('status') != 'passed' for scenario in scenarios):
            errors.append(key + ': failed scenario')
        for field in ('image_id', 'architecture', 'python', 'source_manifest_sha256'):
            if not row.get(field):
                errors.append(key + ': missing ' + field)
        if source_hash is not None and row.get('source_manifest_sha256') != source_hash:
            errors.append(key + ': source manifest differs from this run')
        if row.get('journal') != '/p4/1/journals/journal':
            errors.append(key + ': unexpected journal path')
        builds = row.get('binary_builds', {})
        if set(builds) != {'p4', 'p4d', 'p4broker', 'p4p'} or any('/' + str(row.get('release')) + '/' not in build for build in builds.values()):
            errors.append(key + ': missing or mismatched binary release')
        checksums = row.get('checksums', {})
        if len(checksums) != 4 or any(not re.fullmatch(r'[a-fA-F0-9]{64}', value) for value in checksums.values()):
            errors.append(key + ': incomplete vendor checksums')
    return errors


def worker(args):
    # Import only after mode selection: the host may not have pwd/P4 installed.
    from runtime_integration import RuntimeIntegration

    class VersionIntegration(RuntimeIntegration):
        def service_user(self):
            user = 'sdp-version-native-service'
            policy = self.fixture.p4('configure', 'show', 'dm.user.resetpassword')
            values = {row['Value'] for row in policy if 'Value' in row}
            if len(values) != 1 or not values.issubset({'0', '1'}):
                raise RuntimeError('Cannot establish the server password-reset policy')
            self.service_password_policy = next(iter(values))
            fields = {'User': user, 'Type': 'service', 'AuthMethod': 'perforce',
                      'Email': user + '@example.invalid',
                      'FullName': 'Disposable version compatibility service'}
            self.fixture.p4('user', '-f', '-i', data=marshal.dumps({key.encode(): value.encode()
                                                                   for key, value in fields.items()}, 0))
            # The password is an isolated fixture value sent only via stdin.
            shell = '''source /p4/common/bin/p4_vars 1 || exit 1
source /p4/common/bin/sdp_service_user.sh || exit 1
IFS= read -r password || exit 1
sdp_initialize_service_password "$1" "$password" || exit 1
printf '%s\\n' "$password" | "$P4BIN" -u "$1" login || exit 1
"$P4BIN" -u "$1" login -s || exit 1
"$P4BIN" -u "$1" -ztag info || exit 1
'''
            result = subprocess.run(['bash', '-c', shell, 'version-service', user], env=self.env,
                                    input='VersionFixture2026!\n', capture_output=True, text=True, timeout=120)
            (self.logs / 'version-service-user.log').write_text(result.stdout + result.stderr)
            if result.returncode or 'serverVersion' not in result.stdout:
                raise RuntimeError('Native service user password/authentication/read failed: ' + (result.stdout + result.stderr)[-3000:])
            if self.fixture.p4('configure', 'show', 'dm.user.resetpassword') != policy:
                raise RuntimeError('Service user initialization changed the server password-reset policy')

        def exact_fixture(self):
            self.fixture.create()

        def restart(self):
            self.command(['sudo', 'systemctl', 'restart', 'p4d_1.service'])
            self.command(['/p4/common/bin/p4login', self.args.instance])
            self.fixture.assert_unchanged()

        def restored_inventory(self, counter=None):
            offline = self.home / 'offline_db'
            if not (offline / 'offline_db_usable.txt').is_file():
                raise RuntimeError('Restored offline DB is missing its usability marker')
            clone = Path(tempfile.mkdtemp(prefix='sdp-version-clone-', dir=self.home / 'tmp'))
            for table in offline.glob('db.*'):
                shutil.copy2(table, clone / table.name)
            (clone / 'server.id').write_text('sdp-test-version-restore\n')
            port = '127.0.0.1:20001'
            self.command([self.env['P4DBIN'], '-r', clone, '-p', port, '-J', 'off',
                          '-L', clone / 'log', '--daemonsafe', '--pid-file=' + str(clone / 'pid')])
            try:
                password = Path(self.env['SDP_ADMIN_PASSWORD_FILE']).read_bytes()
                self.fixture.p4('login', '-a', data=password, port=port)
                self.fixture.assert_unchanged(port)
                if counter:
                    rows = self.fixture.p4('counter', counter, port=port)
                    if not any(row.get('value') == '24680' for row in rows):
                        raise RuntimeError('Journal-only counter was lost during replay')
            finally:
                stop_clone(clone, lambda: self.fixture.p4('admin', 'stop', port=port))

        def checkpoint(self):
            self.script('live_checkpoint.sh')
            self.script('daily_checkpoint.sh')
            self.script('recreate_offline_db.sh')
            self.restored_inventory()

        def journals(self):
            self.fixture.post_recovery_write()
            self.fixture.expected = self.fixture.snapshot()
            counter = 'sdp.version.journal'
            self.fixture.p4('counter', counter, '24680')
            before = int(self.command([self.env['P4BIN'], 'counter', 'journal']).stdout.strip())
            self.script('rotate_journal.sh')
            after = int(self.command([self.env['P4BIN'], 'counter', 'journal']).stdout.strip())
            if after != before + 1:
                raise RuntimeError('Journal rotation did not advance exactly once')
            self.script('recreate_offline_db.sh')
            self.restored_inventory(counter)

        def maintenance(self):
            work = self.fixture.work
            for script, arguments in [('accessdates.py', ['1']), ('create_p4_filelist.sh', ['1']),
                                      ('unload_clients.py', ['1', '--limit', '0'])]:
                path = Path('/p4/sdp/Maintenance') / script
                command = ['bash' if script.endswith('.sh') else sys.executable, str(path), *arguments]
                result = subprocess.run(command, cwd=work, env=self.env, capture_output=True, text=True, timeout=90)
                (work / (script + '.output')).write_text(result.stdout + result.stderr)
                if result.returncode:
                    raise RuntimeError(script + ': ' + (result.stdout + result.stderr)[-3000:])
            inventory = (work / 'p4_file_list.txt').read_text().splitlines()
            expected = {row['depotFile'] for row in self.fixture.expected['revisions']}
            actual = {path for path in inventory if path.startswith('//' + self.fixture.name + '/')}
            if actual != expected:
                raise RuntimeError('Maintenance inventory differs from all fixture depot paths')
            self.fixture.assert_unchanged()

    run = VersionIntegration(args)
    details = {
        'service_user_authentication': 'Create a native service account, initialize its password using sdp_service_user.sh, authenticate and read actual server info.',
        'exact_fixture': 'Real edited/deleted/moved/text/binary archive bytes, shelf and label/client/group specifications.',
        'restart_preserves_data': 'Installed systemd restart and exact fixture re-read.',
        'checkpoint_restore': 'SDP checkpoint and offline restore served on a second loopback endpoint; exact fixture equality.',
        'journal_rotation_restore': 'Post-checkpoint submit and counter are replayed from rotated journal into offline DB; exact fixture equality.',
        'maintenance_queries': 'Real inactivity query, full inventory and limit-zero client selection preserve fixture state.',
        'sendq_cleanup': 'Actual local configuration scope/path selection and stopped cleanup preserve directories, symlink targets and exact fixture data; running-server cleanup is refused.'}
    for name, action in [('service_user_authentication', run.service_user), ('exact_fixture', run.exact_fixture), ('restart_preserves_data', run.restart),
                         ('checkpoint_restore', run.checkpoint), ('journal_rotation_restore', run.journals),
                         ('maintenance_queries', run.maintenance), ('sendq_cleanup', run.sendq_cleanup)]:
        if run.records and run.records[-1]['status'] != 'passed':
            run.records.append({'name': name, 'status': 'blocked', 'detail': 'A prerequisite scenario failed.'})
        else:
            run.scenario(name, [], details[name], action)
    data = {'scenarios': run.records, 'journal': run.env['P4JOURNAL'], 'python': sys.version,
            'settings': {'P4PORT': run.env['P4PORT'], 'SERVERID': run.env['SERVERID'],
                         'shared_storage': False, 'read_only_restore_probe_journal': 'off',
                         'dm.user.resetpassword': getattr(run, 'service_password_policy', None)}, 'fixture': run.fixture.expected}
    Path('/p4/1/logs/version-compatibility.json').write_text(json.dumps(data, indent=2) + '\n')
    print(json.dumps(data, indent=2))
    return int(any(row['status'] != 'passed' for row in run.records))


class Matrix:
    def __init__(self, args):
        self.args = args
        self.source_hash = verify_manifest(args.source)
        self.results = args.results.resolve()
        self.results.mkdir(parents=True, exist_ok=True)
        self.lifecycle_lock = threading.Lock()

    def command(self, arguments, logfile=None, check=True, timeout=1200):
        if logfile:
            with logfile.open('w') as output:
                result = subprocess.run([self.args.docker, *arguments], stdout=output, stderr=subprocess.STDOUT, timeout=timeout)
        else:
            result = subprocess.run([self.args.docker, *arguments], capture_output=True, text=True, timeout=timeout)
        if check and result.returncode:
            raise RuntimeError('Container operation failed: ' + (str(logfile) if logfile else result.stderr[-3000:]))
        return result

    def start_container(self, name, image, on_created):
        # Privileged systemd lifecycle operations share host infrastructure.
        # Provisioning and test execution remain concurrent after readiness.
        with self.lifecycle_lock:
            self.command(['run', '--detach', '--name', name, '--label', LABEL + '=true', '--privileged',
                          '--platform', 'linux/' + image['Architecture'], '--cgroupns=private',
                          '--tmpfs', '/run', '--tmpfs', '/run/lock', '--tmpfs', '/tmp', image['Id']])
            on_created()
            deadline = time.monotonic() + SYSTEMD_START_TIMEOUT
            while time.monotonic() < deadline:
                try:
                    state = self.command(['exec', name, 'systemctl', 'is-system-running'], check=False,
                                         timeout=min(15, max(.1, deadline - time.monotonic()))).stdout.strip()
                except subprocess.TimeoutExpired:
                    continue
                if state in ('running', 'degraded'):
                    return
                time.sleep(1)
            raise RuntimeError('Systemd startup timed out')

    def stop_container(self, name, logfile):
        with self.lifecycle_lock:
            return self.command(['stop', '--time', '30', name], logfile, check=False)

    def cell(self, platform, release):
        folder = self.results / (platform + '-' + release)
        folder.mkdir(exist_ok=True)
        name = self.args.name_prefix + '-' + platform + '-' + release.replace('.', '-')
        row = {'platform': platform, 'release': release, 'container': name, 'status': 'failed', 'scenarios': [],
               'source_manifest_sha256': self.source_hash, 'journal': '/p4/1/journals/journal'}
        created = False
        def mark_created():
            nonlocal created
            created = True
        try:
            if self.command(['inspect', name], check=False).returncode == 0:
                raise RuntimeError('Refusing an existing container; use a fresh name prefix (no populated database reuse)')
            image_name = (LEGACY_IMAGES if release.startswith('2023.') else IMAGES)[platform]
            image = json.loads(self.command(['image', 'inspect', image_name]).stdout)[0]
            host_arch = self.command(['info', '--format', '{{.Architecture}}']).stdout.strip()
            host_arch = {'x86_64': 'amd64', 'aarch64': 'arm64'}.get(host_arch, host_arch)
            row.update(image_id=image['Id'], architecture=image['Architecture'],
                       host_architecture=host_arch, emulated=image['Architecture'] != host_arch, image=image_name)
            (folder / 'image-inspect.json').write_text(json.dumps(image, indent=2) + '\n')
            self.start_container(name, image, mark_created)
            self.command(['exec', name, 'mkdir', '-p', '/inputs/sdp', '/results'])
            self.command(['exec', name, 'touch', '/run/sdp-container-test'])
            self.command(['cp', str(self.args.source.resolve()) + '/.', name + ':/inputs/sdp'])
            result = self.command(['exec', name, 'bash', '/inputs/sdp/test/containers/version_bootstrap.sh', 'r' + release[2:]],
                                  folder / 'bootstrap.log', check=False)
            row['scenarios'].append({'name': 'clean_install', 'status': 'passed' if result.returncode == 0 else 'failed',
                                     'detail': 'Fresh actual mkdirs.sh/mkdirs.cfg install, matching release binaries and systemd service.'})
            if result.returncode:
                raise RuntimeError('Clean installation failed; see bootstrap/provision logs')
            worker_log = folder / 'worker.json'
            result = self.command(['exec', '--user', 'perforce', '--workdir', '/p4/1', name,
                                   '/p4/common/python/bin/python3', '/inputs/sdp/test/containers/version_compatibility.py',
                                   '--worker', '--root', '/srv/sdp-test-integration', '--instance', '1', '--port', '19999',
                                   '--os-user', 'perforce', '--confirm-disposable'], worker_log, check=False)
            data = json.loads(worker_log.read_text())
            row.update({key: data[key] for key in ('python', 'settings', 'journal')})
            row['scenarios'].extend(data['scenarios'])
            row['status'] = 'passed' if result.returncode == 0 else 'failed'
        except (OSError, ValueError, KeyError, RuntimeError, subprocess.SubprocessError) as error:
            row['error'] = str(error)
            if created and str(error) == 'Systemd startup timed out':
                diagnostics = [(['inspect', name], 'startup-inspect.json'),
                               (['logs', name], 'startup-console.log'),
                               (['exec', name, 'systemctl', 'is-system-running'], 'startup-state.log'),
                               (['exec', name, 'systemctl', 'list-jobs', '--no-pager'], 'startup-jobs.log'),
                               (['exec', name, 'systemctl', 'list-units', '--state=activating,failed', '--no-pager'],
                                'startup-units.log')]
                for arguments, filename in diagnostics:
                    try:
                        self.command(arguments, folder / filename, check=False, timeout=15)
                    except (OSError, subprocess.SubprocessError) as diagnostic_error:
                        row.setdefault('startup_diagnostic_errors', []).append(filename + ': ' + str(diagnostic_error))
        finally:
            if created:
                self.command(['cp', name + ':/results/.', str(folder)], check=False)
                self.command(['cp', name + ':/p4/1/logs/.', str(folder / 'server-logs')], check=False)
                for binary in ('p4', 'p4d', 'p4broker', 'p4p'):
                    path = folder / (binary + '-version.txt')
                    if path.is_file(): row.setdefault('binary_builds', {})[binary] = path.read_text().strip()
                sources = folder / 'binary-sources.txt'
                if sources.is_file():
                    row['checksums'] = dict(line.rsplit(' ', 1) for line in sources.read_text().splitlines())
                provision = folder / 'provision.log'
                provision_text = provision.read_text(errors='replace') if provision.is_file() else ''
                row['capability_gaps'] = capability_gaps(provision_text) if provision.is_file() else ['Installation evidence unavailable.']
                errors = configuration_errors(provision_text)
                if errors:
                    row.update(status='failed', configuration_errors=errors)
                    for scenario in row['scenarios']:
                        if scenario['name'] == 'clean_install':
                            scenario.update(status='failed', error='Installer reported configuration errors; see configuration_errors and provision.log.')
                if self.stop_container(name, folder / 'stop.log').returncode:
                    row.update(status='failed', cleanup_error='Could not stop the disposable version container')
        present = {scenario['name'] for scenario in row['scenarios']}
        row['scenarios'].extend({'name': scenario, 'status': 'blocked', 'detail': 'Required preceding setup did not complete.'}
                                for scenario in SCENARIOS if scenario not in present)
        (folder / 'case.json').write_text(json.dumps(row, indent=2) + '\n')
        return row

    def run(self):
        releases, platforms = self.args.release or RELEASES, self.args.platform or PLATFORMS
        cells = [(platform, release) for release in releases for platform in platforms]
        completed = {}
        with ThreadPoolExecutor(max_workers=self.args.jobs) as pool:
            pending = {pool.submit(self.cell, platform, release): index
                       for index, (platform, release) in enumerate(cells)}
            for future in as_completed(pending):
                index = pending[future]
                platform, release = cells[index]
                try:
                    completed[index] = future.result()
                except Exception as error:
                    # A tooling/collection failure still produces a failed cell,
                    # and must not discard independent completed cell evidence.
                    completed[index] = {
                        'platform': platform, 'release': release, 'status': 'failed',
                        'source_manifest_sha256': self.source_hash, 'error': str(error),
                        'scenarios': [{'name': name, 'status': 'blocked',
                                       'detail': 'Cell tooling did not complete.'} for name in SCENARIOS]}
                rows = [completed[key] for key in sorted(completed)]
                self.write(rows, releases, platforms)
        return self.write(rows, releases, platforms)

    def write(self, rows, releases, platforms):
        errors = matrix_errors(rows, releases, platforms, self.source_hash)
        data = {'schema_version': 1, 'status': 'failed' if errors else 'passed', 'errors': errors,
                'source_manifest_sha256': self.source_hash, 'required_releases': list(releases),
                'required_platforms': list(platforms), 'jobs': self.args.jobs, 'cases': rows,
                'scope': 'Clean commit-server release compatibility on declared distro Python; no upgrade, all-role or fleet-scale claim.'}
        (self.results / 'version-compatibility.json').write_text(json.dumps(data, indent=2) + '\n')
        return int(bool(errors))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--worker', action='store_true')
    parser.add_argument('--root'); parser.add_argument('--instance'); parser.add_argument('--port', type=int)
    parser.add_argument('--os-user'); parser.add_argument('--confirm-disposable', action='store_true')
    parser.add_argument('--source', type=Path); parser.add_argument('--results', type=Path)
    parser.add_argument('--docker', default='docker')
    parser.add_argument('--jobs', type=int, choices=(1, 2), default=1,
                        help='Maximum independent disposable cells running at once (default: 1)')
    parser.add_argument('--name-prefix', default='sdp-version-' + str(int(time.time())))
    parser.add_argument('--platform', action='append', choices=PLATFORMS)
    parser.add_argument('--release', action='append', choices=RELEASES)
    args = parser.parse_args()
    if args.worker:
        return worker(args)
    if not args.source or not args.results or not re.fullmatch(r'sdp-version-[a-z0-9-]+', args.name_prefix):
        parser.error('--source, --results and a safe sdp-version-* prefix are required')
    if any(values is not None and len(values) != len(set(values)) for values in (args.release, args.platform)):
        parser.error('Duplicate release/platform selections would reuse a disposable container')
    return Matrix(args).run()


if __name__ == '__main__':
    raise SystemExit(main())