TestCheckCaseTrigger.py #3

from __future__ import print_function

import unittest
import time, os, sys
import P4

pythonVer = "python"
if sys.version_info[0] >= 3:
    pythonVer = "python3"

parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, parent_dir)

from p4testutils import TestCase, P4Server, localDirectory, create_file, append_to_file

# from TestTriggers import TestTriggers

os.environ["LOGS"] = "."
LOGGER_NAME = "TestCheckCaseTrigger"
LOG_FILE = "log-TestCheckCaseTrigger.log"

# Handle Python2 version of os.makedirs
def makedirs(folder, *args, **kwargs):
    if not os.path.exists(folder):
        os.makedirs(folder)

def connect(func):
    def func_wrapper(self):
        self.p4.connect()
        self.assertTrue(self.p4.connected(), "Not connected")
        func(self)
        self.p4.disconnect()
        os.chdir(self.server.client_root)

    return func_wrapper

class TestCheckCaseTrigger(TestCase):

    def __init__(self, methodName='runTest'):
        super(TestCheckCaseTrigger, self).__init__(LOGGER_NAME, LOG_FILE, methodName=methodName)

    def setUp(self):
        self.server = P4Server()
        trigpath = os.path.join(parent_dir, "CheckCaseTrigger.py")
        p4 = self.server.p4
        self.client_root = self.server.client_root
        self.p4 = p4
        p4.logger = self.logger
        # This works if no spaces in server root pathname!
        port = p4.port.replace('"', '')
        self.logger.debug("port: |%s|" % port)
        triggers = p4.fetch_triggers()
        triggers['Triggers'] = ['CheckCase change-submit //... " {} {} -p %quote%{}%quote% '
                                '-u {} %change% "'.format(pythonVer, trigpath, port, p4.user)]
        self.logger.debug(triggers)
        p4.save_triggers(triggers)
        # Reconnect to pick up changes
        p4.disconnect()
        p4.connect()

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

    def localFile(self, name):
        return os.path.join(self.client_root, name)

    def addFile(self, name):
        makedirs(self.client_root, exist_ok=True)
        
        path, filename = os.path.split(name)
        if path:
            pname = os.path.join(self.client_root, path)
            makedirs(pname, exist_ok=True)
        
        fname = os.path.join(self.client_root, name)
        with open(fname, "w") as f:
            f.write("Content")
        self.p4.run_add(fname)
    
    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 ")

    def createFiles(self, testDir):
        testAbsoluteDir = os.path.join(self.client_root, testDir)
        os.mkdir(testAbsoluteDir)

        # create a bunch of files
        files = ('foo.txt', 'bar.txt', 'baz.txt')
        for file in files:
            fname = os.path.join(testAbsoluteDir, file)
            f = open(fname, "w")
            f.write("Test Text")
            f.close()
            self.p4.run_add(fname)

        self.assertEqual(len(self.p4.run_opened()), len(files), "Unexpected number of open files")
        return files

    def expectFailSubmit(self, msg, desc):
        """Submits the changes, but expects to fail the trigger"""
        try:
            result = self.p4.run_submit("-d", desc)
            self.fail(msg)
        except P4.P4Exception as inst:
            pass # expected

    def expectSucceedSubmit(self, msg, desc):
        """Submits the change, but expects to succeed"""
        change = self.p4.fetch_change()
        change._description = desc
        self._doSubmit(msg, change)

    @connect
    def testSimpleSubmit(self):
        self.assertEqual(len(self.p4.run_opened()), 0, "Shouldn't have open files")

        testDir = 'test_files'
        files = self.createFiles(testDir)

        self.expectSucceedSubmit("Failed to submit the add", "My Add Test")

    @connect
    def testSimpleFail(self):
        self.addFile("foo")
        
        self.expectSucceedSubmit("Adding one file should not fail", "Simple Add")

        self.p4.run_sync("@0")
        self.addFile("Foo")

        self.expectFailSubmit("Did not catch foo != Foo", "Foo")
       
    @connect
    def testDirectoryFail(self):
        self.addFile("foo/bar")
  
        self.expectSucceedSubmit("Adding one file should not fail", "Simple Add")
        
        self.p4.run_sync("@0")
        self.addFile("FOO/bar")
        
        self.expectFailSubmit("Did not catch foo != FOO/bar", "bar")
   
    @connect
    def testRenameFile(self):
        self.addFile("foo")

        self.expectSucceedSubmit("Adding one file should not fail", "Simple Add")

        self.p4.run_edit(self.localFile("foo"))
        self.p4.run_move(self.localFile("foo"), self.localFile("bar"))

        self.expectSucceedSubmit("Renaming a file should not fail", "Simple Move")

    @connect
    @unittest.skipUnless(sys.platform.startswith("linux"), "requires linux")
    def testRenameFail(self):
        self.addFile("foo")
        self.addFile("bar")

        self.expectSucceedSubmit("Adding two files should not fail", "Double Add")

        self.p4.run_sync("%s@0" % self.localFile("bar"))
        self.p4.run_edit(self.localFile("foo"))
        self.p4.run_move(self.localFile("foo"), self.localFile("BAR"))

        self.expectFailSubmit("Moving into a file with existing naming conflict should not succeed", "Illegal move")

    @connect
    def testDirectoryConflictFail(self):
        self.addFile("foo")

        self.expectSucceedSubmit("Adding single file should not fail", "Simple Add")

        self.p4.run_sync("@0")
        self.addFile("FOO/bar")

        self.expectFailSubmit("Adding a directory conflicting with an existing file should not succeed", "Illegal add")

    @connect
    def testFileWithDirectoryConflictFail(self):
        self.addFile("FOO/bar")
        self.expectSucceedSubmit("Adding single file should not fail", "Simple Add")

        self.p4.run_sync("@0")
        os.rmdir(os.path.join(self.client_root, "FOO"))
        self.addFile("foo")

        self.expectFailSubmit("Adding a file conflicting with an existing directory should not succeed", "Illegal add")

    @connect
    @unittest.skipUnless(sys.platform.startswith("linux"), "requires linux")
    def testFileWithDirectorySameChangeFail(self):
        self.addFile("FOO/bar")
        self.addFile("foo")

        self.expectFailSubmit("Adding a file conflicting with an existing directory should not succeed", "Illegal add")

    @connect
    @unittest.skipUnless(sys.platform.startswith("linux"), "requires linux")
    def testMultipleFilesFail(self):
        self.addFile("foo")
        self.addFile("FOO")

        self.expectFailSubmit("Adding two conflicting files should not succeed", "Illegal add")

    @connect
    @unittest.skipUnless(sys.platform.startswith("linux"), "requires linux")
    def testMultipleDirectoriesFail(self):
        self.addFile("foo/foo")
        self.addFile("FOO/foo")

        self.expectFailSubmit("Adding two conflicting directories should not succeed", "Illegal add")

    @connect
    @unittest.skipUnless(sys.platform.startswith("linux"), "requires linux")
    def testMessageFail(self):
        self.addFile("foo")
        self.addFile("Foo")

        try:
            self.p4.run_submit('-d','Expect to fail')
            self.fail("Illegal add did not cause exception")
        except P4.P4Exception as exc:
            error = exc.errors[0]
            self.assertTrue('//depot/foo' in error)
            self.assertTrue('//depot/Foo' in error)


if __name__ == '__main__':
    unittest.main()