p4testutils.py #1

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