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 logging
import os
import re
import sys
import unittest

import P4

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 testMkRepCalls(self):
        "Basic mkrep options"
        self.run_cmd("cp -R /sdp /p4depots/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 spec master.1 and service user will be created", 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("ERROR: Protections does not grant access to group ServiceUsers as required", 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])