test_MultiSDP.py #1

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

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

from __future__ import print_function

import os, sys, re, socket, time, shutil, logging, time, pwd, stat
import P4
import unittest
import fileinput, subprocess
import argparse
import glob

from test_SDP import SDPTest_base

LOGGER_NAME = 'MultiSDPTest'

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)


class MultiSDPTest(SDPTest_base):
    "Tests for multiple SDP instances on different machines"

    def setUp(self):
        self.setup_everything()

    def assertLinePresent(self, line, output):
        "Asserts regex line present in output"
        re_line = re.compile(line, re.MULTILINE)
        self.assertTrue(re_line.search(output), "%s not found in:\n%s" % (line, output))

    def assertLineNotPresent(self, line, output):
        "Asserts regex line NOT present in output"
        re_line = re.compile(line, re.MULTILINE)
        self.assertFalse(re_line.search(output), "%s found in:\n%s" % (line, output))
 
    def testMkRepCalls(self):
        "Basic mkrep options"
        self.run_cmd("cp -R /sdp /hxdepots/sdp")
        self.sudo_cmd("rm -rf /tmp/p4")
        self.run_cmd("python3 /p4/test_SDP.py --instance 1 --setup")
        p4 = P4.P4()
        p4.port = "1667"
        p4.user = "perforce"
        p4.connect()
        for reptype in ["ha"]:
            mkrep_cmd = "/p4/common/bin/mkrep.sh -i 1 -t %s -s bos -r replica1" % reptype
            # Test for missing config file
            output = self.run_cmd(mkrep_cmd)
            self.assertLinePresent("FATAL: Missing site tag configuration file", output)

            # Fix and try again - getting further
            self.run_cmd("cp /sdp/Server/Unix/p4/common/config/SiteTags.cfg /p4/common/config/")
            output = self.run_cmd(mkrep_cmd)
            self.assertLinePresent("Server 'master\.1' doesn't exist", output)
            self.assertLinePresent("FATAL: Failed to load server spec from file", output)

            svr = p4.fetch_server("master.1")
            svr["Services"] = "commit-server"
            p4.save_server(svr)

            output = self.run_cmd(mkrep_cmd)
            self.assertLineNotPresent("FATAL: Failed to load server spec from file", output)
            self.assertLinePresent("FATAL: The Protections table must grant super access to the group: ServiceUsers", output)

            protect = p4.fetch_protect()
            protect['Protections'].append("super group ServiceUsers * //...")
            p4.save_protect(protect)

            output = self.run_cmd(mkrep_cmd)
            self.assertLinePresent("^STEP 12", output)

        # For the other replica types the errors encountered above won't recur
        for reptype in "ham ro rom fr fs frm fsm ffr edge".split():
            mkrep_cmd = "/p4/common/bin/mkrep.sh -i 1 -t %s -s bos -r replica1" % reptype
            # Test for missing config file
            output = self.run_cmd(mkrep_cmd)
            self.assertLinePresent("^STEP 12", output)
            self.assertLineNotPresent("FATAL:", output)
            svrname = "p4d_%s_bos" % reptype
            servers = [x['ServerID'] for x in p4.run_servers()]
            self.assertTrue(svrname in servers)

if __name__ == "__main__":
    init_logging()
    testrunner = None
    unittest.main(testRunner=testrunner, argv=sys.argv[:1])