# -*- 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()
# | Change | User | Description | Committed | |
---|---|---|---|---|---|
#3 | 16784 | C. Thomas Tyler |
Routine Merge Down to dev from main using: p4 -s merge -n -b perforce_software-sdp-dev |
||
#2 | 16373 | C. Thomas Tyler |
Routine Merge Down to dev from main using: p4 merge -b perforce_software-sdp-dev |
||
#1 | 10638 | C. Thomas Tyler | Populate perforce_software-sdp-dev. | ||
//guest/perforce_software/sdp/main/Server/test/test_SDPEnv.py | |||||
#1 | 10148 | C. Thomas Tyler | Promoted the Perforce Server Deployment Package to The Workshop. |