test_Upgrade.py #1

#!/usr/bin/env python3
# -*- encoding: UTF8 -*-

# test_Upgrade.py
# Tests SDP (Server Deployment Package) UPGRADES on Docker Compose images
# See documentation and run_tests.sh in /sdp/main/test/README.md

from __future__ import print_function

import argparse
import logging
import os
import re
import sys
import unittest
import P4

import test_SDP
from test_SDP import SDPTest_base

LOGGER_NAME = 'UpgradeSDPTest'

logger = logging.getLogger(LOGGER_NAME)

options = None

def init_logging():
    global logger
    logger.setLevel(logging.DEBUG)
    formatter = logging.Formatter('%(asctime)s:%(name)s:%(levelname)s: %(message)s')
    fh = logging.FileHandler('/tmp/%s.log' % LOGGER_NAME, mode='w')
    fh.setLevel(logging.DEBUG)
    fh.setFormatter(formatter)
    logger.addHandler(fh)
    test_SDP.logger = logger


class UpgradeSDPTest(SDPTest_base):
    "Tests for SDP upgrades"

    def setUp(self):
        self.setup_everything()
        self.p4 = None

    def install_binaries(self, ver):
        "Acquire P4 binaries from Perforce FTP server."
        output = self.run_cmd("./get_p4_binaries.sh -r %s" % ver, "/p4depots/sdp/p4_binaries", timeout=120, stop_on_error=True)
        self.assertLinePresent("Downloading of Perforce P4 files completed OK", output)

    def p4run(self, *args):
        "Run the command logging"
        logger.debug('p4 cmd: %s' % ",".join([str(x) for x in args]))
        result = self.p4.run(args)
        logger.debug('result: %s' % str(result))
        return result

    def dailyBackupTest(self, instance):
        "Test daily backup script"
        jnl_counter = self.p4run('counter', 'journal')[0]['value']
        logger.debug("dailyBackupTest")
        self.run_cmd('/p4/common/bin/daily_checkpoint.sh %s' % instance)
        # Quick check on log file contents
        log_contents = self.readLog('checkpoint.log', instance)
        self.assertRegex(log_contents, "Dumping to /p4/%s/checkpoints/p4_%s.ckp" % (instance, instance))
        self.assertRegex(log_contents, "journal")
        new_jnl_counter = self.p4run('counter', 'journal')[0]['value']
        self.assertEqual(int(new_jnl_counter), int(jnl_counter) + 1)

    def testUpgrade(self):
        "Basic upgrade tests"
        self.sudo_cmd("rm -rf /tmp/p4/")
        # Note that setup will download binaries as appropriate
        self.run_cmd("python3 /p4/test_SDP.py --instance 1 --setup --version %s" % options.from_version, timeout=120)
        p4 = P4.P4()
        self.p4 = p4
        p4.port = "1667"
        p4.user = "perforce"
        p4.connect()
        p4.disconnect()
        self.install_binaries("r%s" % options.to_version)
        output = self.run_cmd("/p4/common/bin/upgrade.sh 1 -y -Ov", timeout=120)
        self.assertLinePresent("Could not verify the p4broker server is online", output)
        self.assertLineNotPresent("Exiting because no upgrades are needed", output)
        p4.connect()
        # Validate expected server version is running
        info = p4.run_info()[0]
        self.assertLinePresent('P4D/LINUX26.*/20%s' % options.to_version, info['serverVersion'])
        self.dailyBackupTest("1")

    def testUpgradeSBDLowSecurityAborts(self):
        "SDP-1350: upgrading p4d to or thru P4D 2026.1 with a too-low 'security' value aborts the entire upgrade before anything is touched."
        self.sudo_cmd("rm -rf /tmp/p4/")
        self.run_cmd("python3 /p4/test_SDP.py --instance 1 --setup --version 25.2", timeout=60)
        p4 = P4.P4()
        self.p4 = p4
        p4.port = "1667"
        p4.user = "perforce"
        p4.connect()
        # Standard test setup already configures security=3 (see configureServer());
        # force it below this new check's safe threshold to exercise the abort path.
        self.p4run('configure', 'set', 'security=0')
        p4.disconnect()
        self.install_binaries("r26.1")
        output = self.run_cmd("/p4/common/bin/upgrade.sh 1 -y", timeout=60)
        self.assertLinePresent("WARNING: This upgrade crosses the P4D 2026.1 Secure By Default threshold", output)
        self.assertLinePresent(r"Aborting p4d upgrade: current 'security' value \(0\)", output)
        self.assertLinePresent("Aborting due to failed preflight checks", output)
        # Confirm the all-or-nothing guarantee: nothing was upgraded, not just p4d.
        p4.connect()
        info = p4.run_info()[0]
        self.assertLinePresent('P4D/LINUX26.*/2025.2', info['serverVersion'])
        self.assertLineNotPresent('P4D/LINUX26.*/2026.1', info['serverVersion'])

if __name__ == "__main__":
    init_logging()
    parser = argparse.ArgumentParser(add_help=False)
    parser.add_argument('--from-version', default="25.1")
    parser.add_argument('--to-version', default="26.1")
    options, args = parser.parse_known_args()

    testrunner = None
    unittest.main(testRunner=testrunner, argv=sys.argv[:1] + args)