testP4NetContentTrigger.py #1

# Test harness for P4NetContentTrigger

from __future__ import print_function

import sys
import unittest
import os, shutil, stat, time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import testp4
import P4

python3 = sys.version_info[0] >= 3

def onRmTreeError(function, path, exc_info):
    os.chmod(path, stat.S_IWRITE)
    os.remove(path)
    
def rmdir(dirname):
    shutil.rmtree(dirname, False, onRmTreeError)

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))
    with open(file_name, 'w') as f:
        print(contents, file=f)

class TestP4NetContentTrigger(unittest.TestCase):

    def __init__(self, methodName='runTest'):
        super(TestP4NetContentTrigger, self).__init__(methodName=methodName)

    def assertRegex(self, *args, **kwargs):
        if python3:
            return super(TestP4NetContentTrigger, self).assertRegex(*args, **kwargs)
        else:
            return super(TestP4NetContentTrigger, self).assertRegexpMatches(*args, **kwargs)

    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_basic(self):
        "check triggers works OK"
        self.server = testp4.P4Server()
        trigdir = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "bin", "debug"))
        trigger_exe = os.path.join(trigdir, "P4NetContentTrigger.exe")
        trigger_log = os.path.join(trigdir, "contenttrigger.log")
        try:
            os.remove(trigger_log)
        except:
            pass
        p4 = self.server.p4
        triggers = p4.fetch_triggers()
        triggers['Triggers'] = [
            'c-trigger change-content //... "{0} -v -t -c %changelist%"'.format(trigger_exe)]
        p4.save_triggers(triggers)
        p4.disconnect()     # Needed to pick up triggers
        p4.connect()

        client = p4.fetch_client()
        client._view = ['//depot/... //%s/...' % client._client]
        p4.save_client(client)
        
        trigger_errmsg = "validation failed: Testing: Trigger failed"
        
        fname1 = os.path.join(self.server.client_root, "test.txt")
        create_file(fname1, 'Test content - should pass')

        p4.run_reconcile("%s/..." % self.server.client_root)
        chg = p4.fetch_change()
        chg._description = "test change - should work"
        try:
            p4.save_submit(chg)
        except P4.P4Exception as e:
            self.assertTrue(False, "Unexpected exception")

        p4.run_edit(fname1)
        create_file(fname1, 'Test content - should fail')
        chg = p4.fetch_change()
        chg._description = "test change - should fail"
        try:
            print(p4.save_submit(chg))
            self.assertTrue(False, "Expected exception not encountered")
        except P4.P4Exception as e:
            self.assertRegex(str(e), trigger_errmsg)

        p4.run_revert("//...")

        #-----------------------
        # Test for various errors - first that bad parameters cause a problem
        triggers = p4.fetch_triggers()
        triggers['Triggers'] = [
            'c-trigger change-content //... "{0} -v -t --fail -c %changelist%"'.format(
            trigger_exe)]
        p4.save_triggers(triggers)
        p4.disconnect()     # Needed to pick up triggers
        p4.connect()
        
        trigger_errmsg = "validation failed: P4NetContentTrigger.*\-\-help"
        
        p4.run_edit(fname1)
        create_file(fname1, 'Test content - should pass again')
        chg = p4.fetch_change()
        chg._description = "test change - should work"
        try:
            print(p4.save_submit(chg))
            self.assertTrue(False, "Expected exception not encountered")
        except P4.P4Exception as e:
            self.assertRegex(str(e), trigger_errmsg)
            
        p4.run_revert("//...")

        #-----------------------
        # Test for password problems
        p4.run_password("", "my_password")
        p4.run('configure', 'set', 'security=3')
        p4.password = "my_password"
        p4.run_login()

        trigger_user = "trigger_user"
        user = p4.fetch_user("-f", trigger_user)
        p4.save_user(user, "-f")
        
        # Seperate interface
        p4trig = P4.P4()
        p4trig.port = p4.port
        p4trig.user = trigger_user
        p4trig.connect()
        p4trig.run_password("", "trigger_password")

        # p4.user = testp4.P4USER
        # print(p4.run("login", "-s"))
                
        triggers = p4.fetch_triggers()
        triggers['Triggers'] = [
            'check-content-trigger change-content //... "{0} -v -t -u {1} -c %changelist%"'.format(
            trigger_exe, trigger_user)]
        p4.save_triggers(triggers)
        p4.disconnect()     # Needed to pick up triggers
        p4.connect()
        
        p4.password = "my_password"
        p4.run_login()
        trigger_errmsg = "validation failed: .*Password error during trigger execution"

        p4.run_edit(fname1)
        create_file(fname1, 'Test content - should pass again')
        chg = p4.fetch_change()
        chg._description = "test change - should work"
        try:
            p4.save_submit(chg)
            self.assertTrue(False, "Expected exception not encountered")
        except P4.P4Exception as e:
            self.assertRegex(str(e), trigger_errmsg)

        # Repeat test and this time it should work with trigger user logged in
        p4.run_revert("//...")
        p4trig.password = "trigger_password"
        p4trig.run_login()
        
        p4.run_edit(fname1)
        create_file(fname1, 'Test content - should pass again')
        chg = p4.fetch_change()
        chg._description = "test change - should work"
        try:
            p4.save_submit(chg)
        except P4.P4Exception as e:
            self.assertTrue(False, "Unexpected exception")
            
if __name__ == '__main__':
    unittest.main()