# -*- encoding: UTF8 -*- # Test module for triggers. from __future__ import print_function import sys import time import P4 import subprocess import inspect import platform import logging from textwrap import dedent import unittest, os, shutil, stat, re from subprocess import Popen, PIPE python3 = sys.version_info[0] >= 3 if sys.hexversion < 0x02070000 or (0x0300000 < sys.hexversion < 0x0303000): sys.exit("Python 2.7 or 3.3 or newer is required to run this program.") if python3: from configparser import ConfigParser from io import StringIO else: from ConfigParser import ConfigParser from StringIO import StringIO P4D = "p4d" P4USER = "testuser" P4CLIENT = "test_ws" INTEG_ENGINE = 3 saved_stdoutput = StringIO() test_logger = None def onRmTreeError(function, path, exc_info): os.chmod(path, stat.S_IWRITE) os.remove(path) def ensureDirectory(directory): if not os.path.isdir(directory): os.makedirs(directory) def localDirectory(root, *dirs): "Create and ensure it exists" dir_path = os.path.join(root, *dirs) ensureDirectory(dir_path) return dir_path def create_file(file_name, contents): "Create file with specified contents" ensureDirectory(os.path.dirname(file_name)) if python3: contents = bytes(contents.encode()) with open(file_name, 'wb') as f: f.write(contents) def append_to_file(file_name, contents): "Append contents to file" if python3: contents = bytes(contents.encode()) with open(file_name, 'ab+') as f: f.write(contents) def getP4ConfigFilename(): "Returns os specific filename" if 'P4CONFIG' in os.environ: return os.environ['P4CONFIG'] if os.name == "nt": return "p4config.txt" return ".p4config" class P4Server: def __init__(self, logger=None): self.startdir = os.getcwd() self.root = os.path.join(self.startdir, 'test') if logger: self.logger = logger else: self.logger = logging.getLogger() self.server_root = os.path.join(self.root, "server") self.client_root = os.path.join(self.root, "client") self.cleanupTestTree() ensureDirectory(self.root) ensureDirectory(self.server_root) ensureDirectory(self.client_root) self.p4d = P4D self.port = "rsh:%s -r \"%s\" -L log -i" % (self.p4d, self.server_root) self.p4 = P4.P4() self.p4.port = self.port self.p4.user = P4USER self.p4.client = P4CLIENT self.p4.connect() self.p4cmd('depots') # triggers creation of the user self.p4cmd('configure', 'set', 'dm.integ.engine=%d' % INTEG_ENGINE) self.p4.disconnect() # required to pick up the configure changes self.p4.connect() self.client_name = P4CLIENT client = self.p4.fetch_client(self.client_name) client._root = self.client_root client._lineend = 'unix' self.p4.save_client(client) self.writeP4Config() def shutDown(self): if self.p4.connected(): self.p4.disconnect() def enableUnicode(self): cmd = [self.p4d, "-r", self.server_root, "-L", "log", "-vserver=3", "-xi"] f = Popen(cmd, stdout=PIPE).stdout for s in f.readlines(): pass f.close() def p4cmd(self, *args): "Execute p4 cmd while logging arguments and results" if not self.logger: self.logger = logging.getLogger() self.logger.debug('testp4: %s' % str(args)) output = self.p4.run(args) self.logger.debug('testp4r: %s' % str(output)) return output def cleanupTestTree(self): os.chdir(self.startdir) if os.path.isdir(self.root): shutil.rmtree(self.root, False, onRmTreeError) def writeP4Config(self): "Write appropriate files - useful for occasional manual debugging" p4config_filename = getP4ConfigFilename() config = os.path.join(self.root, p4config_filename) with open(config, "w") as fh: fh.write('P4PORT=%s\n' % self.port) fh.write('P4USER=%s\n' % self.p4.user) fh.write('P4CLIENT=%s\n' % self.p4.client)
# | Change | User | Description | Committed | |
---|---|---|---|---|---|
#3 | 23150 | Robert Cowham | Refactor to move tests into subdir | ||
#2 | 22847 | Robert Cowham |
Fixed problems with trigger table. Save before tidying up. |
||
#1 | 18687 | Russell C. Jackson (Rusty) | Merge main to dev. | ||
//guest/perforce_software/sdp/main/Server/Unix/p4/common/bin/triggers/testp4.py | |||||
#1 | 18675 | Robert Cowham | Better test harness - installs trigger and tests via p4api. |