p4testutils.py #1

# -*- encoding: UTF8 -*-
# Test module for triggers.

from __future__ import print_function

import sys
import P4
import logging
import unittest, os, shutil, stat
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 io import StringIO
else:
    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 TestCase(unittest.TestCase):
    """Common structure"""

    def __init__(self, logger_name, log_file, methodName='runTest'):
        super(TestCase, self).__init__(methodName=methodName)
        if logger_name and log_file:
            self.logger = logging.getLogger(logger_name)
            self.logger.setLevel(logging.DEBUG)
            logformat = '%(levelname)s [%(asctime)s] [%(filename)s : %(lineno)d] - %(message)s'
            logging.basicConfig(format=logformat, filename=log_file, level=logging.DEBUG)

    # Python compatibility
    def assertRegex(self, *args, **kwargs):
        if python3:
            return super(TestCase, self).assertRegex(*args, **kwargs)
        else:
            return super(TestCase, self).assertRegexpMatches(*args, **kwargs)

class P4Server:
    def __init__(self, logger=None):
        self.startdir = os.getcwd()
        self.root = os.path.join(self.startdir, 'testrun')
        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.p4.run('depots') # triggers creation of the user
        self.p4.run('configure', 'set', 'dm.integ.engine=%d' % INTEG_ENGINE)

        self.p4.disconnect() # required to pick up the configure changes
        self.p4.connect()

        # Override standard environment incase test dir is not checked in due to parent p4ignore file
        self.p4.ignore_file = ".p4ignore-tests"

        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 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)