test_SDPEnv.py #1

# -*- encoding: UTF8 -*-

# test_create_env.py
# Tests create_env.py in SDP (Server Deployment Package)

import os
import sys
import copy
import imp
import re
import socket
import string
import time
import types
import shutil
import tempfile
import getopt
import fileinput
import traceback

# import P4

import unittest

# Add the nearby "code" directory to the module loading path to pick up p4
sys.path.insert(0, os.path.join('..', 'setup'))

from SDPEnv import SDPException, SDPConfigException, SDPInstance, SDPConfig

#---Code
verbose = 0
slow = 1

# The default temporary file prefix starts with an '@'.  But
# that would mean that temporary files will look like revision
# specifications to Perforce.  So use a prefix that's acceptable
# to Perforce.
tempfile.gettempprefix = lambda: '%d.' % os.getpid()

# We need to log to a log file.  The SDPTest log will get redirected to
# this file, as will the output of various commands.

log_filename = (time.strftime('SDPTest.%Y%m%dT%H%M%S.log',
                                 time.gmtime(time.time())))
log_filename = os.path.abspath(log_filename)
log_file = open(log_filename, "a")

def log_exception():
    type, val, tb = sys.exc_info()
    log_message(string.join(traceback.format_exception(type, val, tb), ''))
    del type, val, tb

def log_message(msg):
    date = time.strftime('%Y-%m-%d %H:%M:%S UTC',
                         time.gmtime(time.time()))
    log_file.write("%s  %s\n" % (date, msg))
    log_file.flush()
    if verbose:
        print("%s  %s\n" % (date, msg))

def sorted_ini(config_lines):
    "Sorts the config lines based on name of sections - for reliable comparison"
    result = []
    sections = {}
    sname = ""
    for line in [l.strip() for l in config_lines]:
        if not line or (len(line) > 0 and line[0] == '#'):
            continue
        if line and line[0] == '[':
            sname = line
            sections[sname] = []
        else:
            sections[sname].append(line)
    for sname in sorted(sections.keys()):
        result.append(sname)
        result.extend(sections[sname])
    return result

sys.stdout.write("SDPTest test suite, logging to %s\n" % log_filename)
sys.stdout.flush()

# Perforce
#
# This class supplies test methods.
# It also provides "system", which is like os.system,
# but captures output, checks for errors, and writes to the log.

class Perforce:

    # Temporary directory for Perforce server and associated files.
    p4dir = "test"

    # Run command and check results
    #
    # Calls an external program, raising an exception upon any error
    # and returning standard output and standard error from the program,
    # as well as writing them to the log.

    def system(self, command, ignore_failure = 0, input_lines = None):
        log_file.write('Executing %s\n' % repr(command))
        (child_stdin, child_stdout) = os.popen4(command)
        if input_lines:
            child_stdin.writelines(input_lines)
            child_stdin.close()
        output = child_stdout.read()
        result = child_stdout.close()
        log_file.write(output)
        if not ignore_failure and result:
            message = ('Command "%s" failed with result code %d.' %
                       (command, result))
            log_file.write(message + '\n')
            raise message
        return output

# The SDPTest_base class is a generic test case.
# Other test cases will inherit from this class.
#
# When a class implements several test cases, the methods that implement
# test cases (in the PyUnit sense) should have names starting "test_".
# When a class implements a single test case, the method should be
# called "runTest".

class SDPTest_base(unittest.TestCase):

    # Set up everything so that test cases can run
    #
    p4d = Perforce()
    p4api_ver = "2005.1"

    def setup_everything(self, config_changes = {}):
        pass
        # for key, value in config_changes.items():
            # setattr(config, key, value)

    def log_result(self, msg="", output=[]):
        type, val, tb = sys.exc_info()
        log_message(msg + '\n' + string.join(traceback.format_stack(limit=2), ''))
        if len(output) > 0:
            import pprint
            pp = pprint.PrettyPrinter(indent=4)
            log_message(pp.pformat(output))
        # if len(self.p4.warnings) > 0:
            # log_message("Warnings: " + "\n".join(self.p4.warnings))
        # if len(self.p4.errors) > 0:
            # log_message("Errors: " + "\n".join(self.p4.errors))

    def write_to_file(self, fname, results):
        temp_file = open(fname, "w")
        for l in results:
            temp_file.write(l)
        temp_file.close()

    def setUp(self):
        self.setup_everything()

    def run_test(self):
        pass

#--- Test Cases

class config_validation_base(SDPTest_base):
    "Basic config validation tests"

    def setUp(self):
        super(config_validation_base, self).setUp()
        self.required_options  = [x.lower() for x in "P4NAME SDP_SERVICE_TYPE SDP_HOSTNAME SDP_INSTANCE SDP_P4PORT_NUMBER SDP_OS_USERNAME".split()]
        self.required_options.extend([x.lower() for x in "METADATA_ROOT DEPOTDATA_ROOT LOGDATA_ROOT".split()])

    def create_data(self, data):
        return "\n".join([x.strip() for x in data.split()])

    def configNotValid(self, data, reqd_msg):
        "Test and catch exception"
        sc = SDPConfig(config_data=self.create_data(data))
        with self.assertRaisesRegexp(SDPConfigException, reqd_msg) as cm:
            sc.isvalid_config()
        return str(cm.exception)

class required_options_missing(config_validation_base):
    def runTest(self):
        "Test required options must be present"

        # First we test for reqd options
        reqd_options_missing_msg = "The following required options are missing"

        # No required options specified
        data = """[Master]
        some_var=some_val
        """
        errmsg = self.configNotValid(data, reqd_options_missing_msg)
        for opt in self.required_options:
            self.assertRegex(errmsg, opt)

        # p4name still missing as it is blank
        data = """[Master]
        p4name=
        """
        errmsg = self.configNotValid(data, reqd_options_missing_msg)
        self.assertRegex(errmsg, "p4name")

        # Now we specify the value so make sure it isn't present in the output
        data = """[Master]
        p4name=some_val
        """
        errmsg = self.configNotValid(data, reqd_options_missing_msg)
        self.assertNotRegex(errmsg, "p4name")

        # Now provide all values
        data = """[Master]
                P4NAME=Master
                SDP_SERVICE_TYPE=standard
                SDP_HOSTNAME=WIN-B7UQ3E1TN83
                SDP_P4PORT_NUMBER=1777
                SDP_INSTANCE=1
                SDP_OS_USERNAME=p4admin
                SDP_P4SUPERUSER=admin
                METADATA_ROOT=E:
                DEPOTDATA_ROOT=F:
                LOGDATA_ROOT=G:
                """
        sc = SDPConfig(config_data=self.create_data(data))
        self.assertTrue(sc.isvalid_config())

class config_field_validation(config_validation_base):
    def runTest(self):
        "Test config fields must be valid"

        data = """[Master]
                P4NAME=Master
                SDP_SERVICE_TYPE=standard
                SDP_HOSTNAME=WIN-B7UQ3E1TN83
                SDP_P4PORT_NUMBER=a1213
                SDP_INSTANCE=1
                SDP_OS_USERNAME=p4admin
                SDP_P4SUPERUSER=admin
                METADATA_ROOT=E:
                DEPOTDATA_ROOT=F:
                LOGDATA_ROOT=G:
                """
        errmsg = self.configNotValid(data, "SDP_P4PORT_NUMBER must be numeric")
        self.assertRegex(errmsg, "Master")

        data = """[Master]
                P4NAME=Master
                SDP_SERVICE_TYPE=non-standard
                SDP_HOSTNAME=WIN-B7UQ3E1TN83
                SDP_P4PORT_NUMBER=1213
                SDP_INSTANCE=1
                SDP_OS_USERNAME=p4admin
                SDP_P4SUPERUSER=admin
                METADATA_ROOT=E:
                DEPOTDATA_ROOT=F:
                LOGDATA_ROOT=G:
                """
        errmsg = self.configNotValid(data, "SDP_SERVICE_TYPE must be one of "
            "'standard, replica, forwarding-replica, build-server'")
        self.assertRegex(errmsg, "Master")

class output_file_contents(config_validation_base):
    def runTest(self):
        "Test output scripts are correct"

        self.maxDiff = None # Show all differences

        # Provide all values
        default_data = """[DEFAULT]
                maillist=rcowham@perforce.com
                mailfrom=sdp@test.com
                mailhost=unknownserver@perforce.com
                python=c:\python27
                ADMIN_PASS_FILENAME=adminpass.txt
                KEEPCKPS=7
                KEEPLOGS=7
                LIMIT_ONE_DAILY_CHECKPOINT=false
                """

        hostname = socket.gethostname().lower()
        master_data = """[Master]
                P4NAME=Master
                SDP_SERVICE_TYPE=standard
                SDP_HOSTNAME=%s
                SDP_P4PORT_NUMBER=1777
                SDP_INSTANCE=1
                SDP_OS_USERNAME=p4admin
                SDP_P4SUPERUSER=admin
                SDP_P4SERVICEUSER=fred
                METADATA_ROOT=E:
                DEPOTDATA_ROOT=F:
                LOGDATA_ROOT=G:
                """ % (hostname)
        sc = SDPConfig(config_data=self.create_data(default_data + master_data))
        self.assertTrue(sc.isvalid_config())

        sc.write_master_config_ini()

        expected_dirs = [r'E:\p4\1\root',
                        r'G:\p4\1',
                        r'F:\p4\1\bin',
                        r'F:\p4\1\tmp',
                        r'F:\p4\1\depots',
                        r'F:\p4\1\checkpoints',
                        r'F:\p4\1\ssl',
                        r'F:\p4\common\bin',
                        r'F:\p4\common\bin\triggers',
                        r'E:\p4\1\root\save',
                        r'E:\p4\1\offline_db',
                        r'G:\p4\1\logs',
                        r'F:\p4\config']
        dirs = sc.get_instance_dirs()
        self.assertEquals(expected_dirs, dirs)
        expected_cmds = {hostname: [
                        r'F:\p4\common\bin\instsrv.exe p4_1 "F:\p4\1\bin\p4s.exe"',
                        r'F:\p4\1\bin\p4.exe set -S p4_1 P4ROOT=E:\p4\1\root',
                        r'F:\p4\1\bin\p4.exe set -S p4_1 P4JOURNAL=G:\p4\1\logs\journal',
                        r'F:\p4\1\bin\p4.exe set -S p4_1 P4NAME=Master',
                        r'F:\p4\1\bin\p4.exe set -S p4_1 P4PORT=1777',
                        r'F:\p4\1\bin\p4.exe set -S p4_1 P4LOG=G:\p4\1\logs\Master.log']}
        cmds = sc.get_service_install_cmds()
        self.assertEquals(expected_cmds, cmds)

        master_data = r"""[Master]
                P4NAME=Master
                SDP_SERVICE_TYPE=standard
                SDP_HOSTNAME=%s
                SDP_P4PORT_NUMBER=1777
                SDP_INSTANCE=1
                SDP_OS_USERNAME=p4admin
                SDP_P4SUPERUSER=admin
                SDP_P4SERVICEUSER=fred
                METADATA_ROOT=E:\master
                DEPOTDATA_ROOT=F:\master
                LOGDATA_ROOT=G:\master
                """ % (hostname)
        replica_data = r"""[Replica1]
                P4NAME=Replica1
                SDP_SERVICE_TYPE=replica
                SDP_HOSTNAME=%s
                SDP_P4PORT_NUMBER=1778
                SDP_INSTANCE=2
                SDP_OS_USERNAME=p4admin
                SDP_P4SUPERUSER=admin
                SDP_P4SERVICEUSER=fred
                METADATA_ROOT=E:\replica
                DEPOTDATA_ROOT=F:\replica
                LOGDATA_ROOT=G:\replica
                """ % (hostname)
        sc = SDPConfig(config_data=self.create_data(default_data + master_data + replica_data))
        self.assertTrue(sc.isvalid_config())

        # Make sure config written is correct
        expected_config = r"""# Global sdp_config.ini


                [1:win-b7uq3e1tn83]
                p4port=win-b7uq3e1tn83:1777
                p4name=Master
                sdp_os_username=p4admin
                sdp_p4serviceuser=fred
                metadata_root=E:\master
                depotdata_root=F:\master
                logdata_root=G:\master
                sdp_p4superuser=admin
                admin_pass_filename=adminpass.txt
                mailfrom=sdp@test.com
                maillist=rcowham@perforce.com
                mailhost=unknownserver@perforce.com
                python=c:\python27
                keepckps=7
                keeplogs=7
                limit_one_daily_checkpoint=false

                [2:win-b7uq3e1tn83]
                p4port=win-b7uq3e1tn83:1778
                p4name=Replica1
                sdp_os_username=p4admin
                sdp_p4serviceuser=fred
                metadata_root=E:\replica
                depotdata_root=F:\replica
                logdata_root=G:\replica
                sdp_p4superuser=admin
                admin_pass_filename=adminpass.txt
                mailfrom=sdp@test.com
                maillist=rcowham@perforce.com
                mailhost=unknownserver@perforce.com
                python=c:\python27
                keepckps=7
                keeplogs=7
                limit_one_daily_checkpoint=false""".split("\n")
        expected_config_lines = sorted_ini([l.strip() for l in expected_config])
        sc.write_master_config_ini()
        with open("sdp_config.ini", "r") as fh:
            lines = sorted_ini([l.strip() for l in fh.readlines()])
            self.assertListEqual(expected_config_lines, lines)

        expected_dirs = [r'E:\master\p4\1\root',
                        r'G:\master\p4\1',
                        r'F:\master\p4\1\bin',
                        r'F:\master\p4\1\tmp',
                        r'F:\master\p4\1\depots',
                        r'F:\master\p4\1\checkpoints',
                        r'F:\master\p4\1\ssl',
                        r'F:\master\p4\common\bin',
                        r'F:\master\p4\common\bin\triggers',
                        r'E:\master\p4\1\root\save',
                        r'E:\master\p4\1\offline_db',
                        r'G:\master\p4\1\logs',
                        r'F:\master\p4\config',
                        r'E:\replica\p4\2\root',
                        r'G:\replica\p4\2',
                        r'F:\replica\p4\2\bin',
                        r'F:\replica\p4\2\tmp',
                        r'F:\replica\p4\2\depots',
                        r'F:\replica\p4\2\checkpoints',
                        r'F:\replica\p4\2\ssl',
                        r'F:\replica\p4\common\bin',
                        r'F:\replica\p4\common\bin\triggers',
                        r'E:\replica\p4\2\root\save',
                        r'E:\replica\p4\2\offline_db',
                        r'G:\replica\p4\2\logs',
                        r'F:\replica\p4\config']
        dirs = sc.get_instance_dirs()
        self.assertEquals(expected_dirs, dirs)
        expected_cmds = {hostname: [
                        r'F:\master\p4\common\bin\instsrv.exe p4_1 "F:\master\p4\1\bin\p4s.exe"',
                        r'F:\master\p4\1\bin\p4.exe set -S p4_1 P4ROOT=E:\master\p4\1\root',
                        r'F:\master\p4\1\bin\p4.exe set -S p4_1 P4JOURNAL=G:\master\p4\1\logs\journal',
                        r'F:\master\p4\1\bin\p4.exe set -S p4_1 P4NAME=Master',
                        r'F:\master\p4\1\bin\p4.exe set -S p4_1 P4PORT=1777',
                        r'F:\master\p4\1\bin\p4.exe set -S p4_1 P4LOG=G:\master\p4\1\logs\Master.log',
                        r'F:\replica\p4\common\bin\instsrv.exe p4_2 "F:\replica\p4\2\bin\p4s.exe"',
                        r'F:\replica\p4\2\bin\p4.exe set -S p4_2 P4ROOT=E:\replica\p4\2\root',
                        r'F:\replica\p4\2\bin\p4.exe set -S p4_2 P4JOURNAL=G:\replica\p4\2\logs\journal',
                        r'F:\replica\p4\2\bin\p4.exe set -S p4_2 P4NAME=Replica1',
                        r'F:\replica\p4\2\bin\p4.exe set -S p4_2 P4PORT=1778',
                        r'F:\replica\p4\2\bin\p4.exe set -S p4_2 P4LOG=G:\replica\p4\2\logs\Replica1.log']}
        cmds = sc.get_service_install_cmds()
        self.assertEquals(expected_cmds, cmds)

        bat_prefix = 'p4 -p %s:1777 -u admin ' % (hostname)
        expected_bat_contents = {'Master': [
                bat_prefix + r'configure set Master#journalPrefix=F:\master\p4\1\checkpoints\p4_1',
                bat_prefix + r'serverid Master',
                bat_prefix + r'configure set Master#serverlog.file.8=G:\master\p4\1\logs\integrity.csv',
                bat_prefix + r'configure set Master#serverlog.retain.8=7',
                bat_prefix + r'configure set Master#serverlog.file.3=G:\master\p4\1\logs\errors.csv',
                bat_prefix + r'configure set Master#serverlog.retain.3=7',
                bat_prefix + r'configure set Master#serverlog.file.7=G:\master\p4\1\logs\events.csv',
                bat_prefix + r'configure set Master#serverlog.retain.7=7',
                bat_prefix + r'configure set db.peeking=2',
                bat_prefix + r'configure set dm.user.noautocreate=2',
                bat_prefix + r'configure set dm.user.resetpassword=1',
                bat_prefix + r'configure set filesys.P4ROOT.min=1G',
                bat_prefix + r'configure set filesys.depot.min=1G',
                bat_prefix + r'configure set filesys.P4JOURNAL.min=1G',
                bat_prefix + r'configure set spec.hashbuckets=99',
                bat_prefix + r'configure set monitor=1',
                bat_prefix + r'configure set server=3',
                bat_prefix + r'configure set net.tcpsize=128k',
                bat_prefix + r'configure set server.commandlimits=2',
                bat_prefix + r'configure set Replica1#journalPrefix=F:\replica\p4\2\checkpoints\p4_2',
                bat_prefix + r'configure set Replica1#P4PORT=1778',
                bat_prefix + r'configure set Replica1#P4TARGET=%s:1777' % (hostname),
                bat_prefix + r'configure set Replica1#P4TICKETS=F:\replica\p4\2\p4tickets.txt',
                bat_prefix + r'configure set Replica1#P4LOG=G:\replica\p4\2\logs\Replica1.log',
                bat_prefix + r'configure set Replica1#server=3',
                bat_prefix + r'configure set "Replica1#startup.1=pull -i 1"',
                bat_prefix + r'configure set "Replica1#startup.2=pull -u -i 1"',
                bat_prefix + r'configure set "Replica1#startup.3=pull -u -i 1"',
                bat_prefix + r'configure set "Replica1#startup.4=pull -u -i 1"',
                bat_prefix + r'configure set "Replica1#startup.5=pull -u -i 1"',
                bat_prefix + r'configure set "Replica1#lbr.replication=readonly"',
                bat_prefix + r'configure set "Replica1#db.replication=readonly"',
                bat_prefix + r'configure set "Replica1#serviceUser=Replica1"'],
                 'Replica1': [
                            r'p4 -p %s:1778 -u admin serverid Replica1' % (hostname)]}
        bat_contents = sc.get_configure_bat_contents()
        self.assertEquals(expected_bat_contents, bat_contents)

# RUNNING THE TESTS
def tests():
    if len(sys.argv) == 1:
        suite = unittest.TestSuite()
        # Decide which tests to run

        tests = [required_options_missing, config_field_validation, output_file_contents]
        # tests = [login]
        for t in tests:
            suite.addTest(t())
        return suite
    else:
        # Unfortunately the following doesn't work with pdb, but is OK otherwise
        loader = unittest.defaultTestLoader
        module = __import__('__main__')
        suite = loader.loadTestsFromName(sys.argv[1], module)
        return suite

def main():
    # Filter out our arguments before handing unknown ones on
    argv = []
    argv.append(sys.argv[0])
    try:
        opts, args = getopt.getopt(sys.argv[1:], "hv",
                        ["help", "verbose"])
    except getopt.GetoptError:
        print("Usage: -v/verbose")
        sys.exit(2)

    for opt, arg in opts:
        if opt in ("-v", "--verbose"):
            verbose = 1
            argv.append("-v")
        elif opt in ("-h", "--help"):
            argv.append("-h")   # pass it through
        else:   # If unknown pass them on
            argv.append(opt)
            argv.append(arg)
    # unittest.main(defaultTest="tests", argv=argv)
    runner = unittest.TextTestRunner(verbosity=2)
    runner.run(tests())

if __name__ == "__main__":
    main()