test_Upgrade.py #8

#!/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 Helix binaries"
        output = self.run_cmd("./get_helix_binaries.sh %s" % ver, "/p4/sdp/helix_binaries", timeout=120)
        self.assertLinePresent("Downloading of Helix binaries 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.run_cmd("cp -R /sdp /hxdepots/sdp")
        self.install_binaries("r%s" % options.from_version)
        self.sudo_cmd("rm -rf /tmp/p4")
        self.run_cmd("python3 /p4/test_SDP.py --instance 1 --setup", timeout=60)
        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", timeout=60)
        self.assertLinePresent("Finished p4_1 Upgrade", output)
        p4.connect()
        # Validate expected server version is running
        info = p4.run_info()[0]
        self.assertLinePresent('P4D/LINUX26X86_64/20%s' % options.to_version, info['serverVersion'])
        self.dailyBackupTest("1")

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

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