TestTriggers.py #1

# -*- encoding: UTF8 -*-

from __future__ import print_function

import glob, sys, time, stat, platform

import P4
from P4 import P4Exception
import unittest, types, shutil
from subprocess import Popen, PIPE
import sys
import os, os.path

def onRmTreeError( function, path, exc_info ):
    os.chmod( path, stat.S_IWRITE)
    os.remove( path )

class TestTriggers(unittest.TestCase):

    def setUp(self):
        self.setDirectories()
        self.p4d = "p4d"
        self.port = "rsh:%s -r \"%s\" -L log -vserver=3 -i" % ( self.p4d, self.server_root )
        self.p4 = P4.P4()
        self.p4.port = self.port

    def enableUnicode(self):
        cmd = [self.p4d, "-r", self.server_root, "-L", "log", "-vserver=3", "-xi"]
        p = Popen(cmd, stdout=PIPE)
        f = p.stdout
        for s in f.readlines():
            pass
        p.wait()
        f.close()

    def tearDown(self):
        if self.p4.connected():
            self.p4.disconnect()
        time.sleep( 1 )
        self.cleanupTestTree()

    def setDirectories(self):
        self.startdir = os.getcwd()
        self.server_root = os.path.join(self.startdir, 'testroot')
        self.client_root = os.path.join(self.server_root, 'client')

        self.cleanupTestTree()
        self.ensureDirectory(self.server_root)
        self.ensureDirectory(self.client_root)

    def cleanupTestTree(self):
        os.chdir(self.startdir)
        if os.path.isdir(self.server_root):
            shutil.rmtree(self.server_root, False, onRmTreeError)

    def ensureDirectory(self, directory):
        if not os.path.isdir(directory):
            os.mkdir(directory)

    def getServerPatchLevel(self, info):
        c = re.compile("[^/]*/[^/]*/[^/]*/([^/]*)\s\(\d+/\d+/\d+\)")
        
        serverVersion = info[0]["serverVersion"]
        m = c.match(serverVersion)
        if m:
            serverPatch = m.group(1)
            return int(serverPatch)
        else:
            print("Cannot extract patch level from {0}".format(serverVersion))
            sys.exit(1)

    def _setClient(self):
        """Creates a client and makes sure it is set up"""
        self.assertTrue(self.p4.connected(), "Not connected")
        self.p4.cwd = self.client_root
        self.p4.client = "TestClient"
        client = self.p4.fetch_client()
        client._root = self.client_root
        self.p4.save_client(client)

    def _doSubmit(self, msg, *args):
        """Submits the changes"""
        try:
            result = self.p4.run_submit(*args)
            self.assertTrue( 'submittedChange' in result[-1], msg)
        except P4.P4Exception as inst:
            self.fail("submit failed with exception ")