p4test.py #2

import glob, sys
pathToBuild = glob.glob('build/lib*')
if len(pathToBuild) > 0:
    sys.path.insert(0, pathToBuild[0])

import P4
import unittest, os, types, shutil
import stat

def rmgeneric(path, __func__):

    try:
        # print 'Removing file ', path
        __func__(path)
        # print 'Removed ', path
    except OSError, (errno, strerror):
        print ERROR_STR % {'path' : path, 'error': strerror }, errno

def removeall(path):

    if not os.path.isdir(path):
        return
    
    files=os.listdir(path)

    for x in files:
        # print 'Removing ', x
        fullpath=os.path.join(path, x)
        if os.path.isfile(fullpath):
            os.chmod(fullpath, stat.S_IWRITE)
            f=os.remove
            rmgeneric(fullpath, f)
        elif os.path.isdir(fullpath):
            removeall(fullpath)
            f=os.rmdir
            rmgeneric(fullpath, f)
            
def del_temp_dir(dir):
    if dir:
        print 'Removing dir', dir
        if os.path.isdir(dir):
            removeall(dir)
            os.rmdir(dir)
            
class TestP4Python(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 tearDown(self):
        if self.p4.connected():
            self.p4.disconnect()
        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):
            del_temp_dir(self.server_root)

    def ensureDirectory(self, directory):
        if not os.path.isdir(directory):
            os.mkdir(directory)
            
class TestP4(TestP4Python):
    
    def testInfo(self):
        self.failUnless(self.p4 != None, "Could not create p4")
        self.p4.connect()
        self.assert_(self.p4.connected(), "Not connected")
        
        info = self.p4.run_info()
        self.failUnless(isinstance(info, list), "run_info() does not return a list")
        info = info.pop()
        self.failUnless(isinstance(info, dict), "run_info().pop() is not a dict")
        self.assertEqual(info['serverRoot'], self.server_root, "Server root incorrect")

    def testEnvironment(self):
        self.assert_(self.p4 != None, "Could not create p4")

        self.p4.charset         = "iso8859-1"    
        self.p4.client          = "myclient"
        self.p4.host            = "myhost"
        self.p4.language        = "german"
        self.p4.maxresults      = 100000
        self.p4.maxscanrows     = 1000000
        self.p4.maxlocktime     = 10000
        self.p4.password        = "mypassword"
        self.p4.port            = "myserver:1666"
        self.p4.prog            = "myprogram"
        self.p4.tagged          = True
        self.p4.ticket_file     = "myticket"
        self.p4.user            = "myuser"

        self.assertEqual( self.p4.charset, "iso8859-1", "charset" )
        self.assertEqual( self.p4.client, "myclient", "client" )
        self.assertEqual( self.p4.host, "myhost", "host" )
        self.assertEqual( self.p4.language, "german", "language" )
        self.assertEqual( self.p4.maxresults, 100000, "maxresults" )
        self.assertEqual( self.p4.maxscanrows, 1000000, "maxscanrows" )
        self.assertEqual( self.p4.maxlocktime, 10000, "maxlocktime" )
        self.assertEqual( self.p4.password, "mypassword", "password" )
        self.assertEqual( self.p4.port, "myserver:1666", "port" )
        self.assertEqual( self.p4.tagged, 1, "tagged" )
        self.assertEqual( self.p4.ticket_file, "myticket", "ticket_file" )
        self.assertEqual( self.p4.user, "myuser", "user" )

    def testClient(self):
        self.p4.connect()
        self.assert_(self.p4.connected(), "Not connected")

        client = self.p4.fetch_client()
        self.assert_( isinstance(client, P4.Spec), "Client is not of type P4.Spec")

        client._root = self.client_root
        client._description = 'Some Test Client\n'

        try:
            self.p4.save_client(client)
        except P4.P4Exception:
            self.fail("Saving client caused exception")

        client2 = self.p4.fetch_client()

        self.assertEqual( client._root, client2._root, "Client root differs")
        self.assertEqual( client._description, client2._description, "Client description differs")
        
    def testFiles(self):
        testDir = 'test_files'
        testAbsoluteDir = os.path.join(self.client_root, testDir)
        os.mkdir(testAbsoluteDir)
        
        self.p4.connect()
        self.assert_(self.p4.connected(), "Not connected")
        self._setClient()
        self.assertEqual(len(self.p4.run_opened()), 0, "Shouldn't have open files")
    
        # 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(testDir + "/" + file)
        
        self.assertEqual(len(self.p4.run_opened()), len(files), "Unexpected number of open files")
        
        change = self.p4.fetch_change()
        self.assert_( isinstance(change, P4.Spec), "Change spec is not of type P4.Spec")
        change._description = "My Add Test"
        
        self._doSubmit("Failed to submit the add", change)
        
        # make sure there are no open files and all files are there
        
        self.assertEqual( len(self.p4.run_opened()), 0, "Still files in the open list")
        self.assertEqual( len(self.p4.run_files('...')), len(files), "Less files than expected")
        
        # edit the files
        
        self.assertEqual( len(self.p4.run_edit('...')), len(files), "Not all files open for edit")
        self.assertEqual( len(self.p4.run_opened()), len(files), "Not enough files open for edit")
        
        change = self.p4.fetch_change()
        change._description = "My Edit Test"
        self._doSubmit("Failed to submit the edit", change)
        self.assertEqual( len(self.p4.run_opened()), 0, "Still files in the open list")
        
        # branch testing
        
        branchDir = 'test_branch'
        try:
            result = self.p4.run_integ(testDir + '/...', branchDir + '/...')
            self.assertEquals(len(result), len(files), "Not all files branched")
        except P4.P4Exception:
            self.fail("Integration failed")
        
        change = self.p4.fetch_change()
        change._description = "My Branch Test"
        self._doSubmit("Failed to submit branch", change)
        
        # branch testing again
        
        branchDir = 'test_branch2'
        try:
            result = self.p4.run_integ(testDir + '/...', branchDir + '/...')
            self.assertEquals(len(result), len(files), "Not all files branched")
        except P4.P4Exception:
            self.fail("Integration failed")
        
        change = self.p4.fetch_change()
        change._description = "My Branch Test"
        self._doSubmit("Failed to submit branch", change)
        
        # filelog checks
        
        filelogs = self.p4.run_filelog( testDir + '/...' )
        self.assertEquals( len(filelogs), len(files) )
        
        df = filelogs[0]
        self.assertEqual( df.depotFile, "//depot/test_files/bar.txt", "Unexpected file in the filelog" )
        self.assertEqual( len(df.revisions), 2, "Unexpected number of revisions" )
        
        rev = df.revisions[0]
        self.assertEqual( rev.rev, 2, "Unexpected revision")
        self.assertEqual( len(rev.integrations), 2, "Unexpected number of integrations")
        self.assertEqual( rev.integrations[ 0 ].how, "branch into", "Unexpected how" )
        self.assertEqual( rev.integrations[ 0 ].file, "//depot/test_branch/bar.txt", "Unexpected target file" )

    def testPasswords(self):
        ticketFile = self.client_root + "/.p4tickets"
        password = "Password"
        self.p4.ticket_file = ticketFile
        self.assertEqual( self.p4.ticket_file, ticketFile, "Ticket file not set correctly")
        
        self.p4.connect()
        client = self.p4.fetch_client()
        client._root = self.client_root
        self.p4.save_client(client)
        
        try:
            self.p4.run_password( "", password )
        except P4.P4Exception:
            self.fail( "Failed to change the password" )
        
        self.p4.password = password
        self.assertEqual( self.p4.password, password, "Could not set password" )
        try:
            self.p4.run_login( )
        except P4.P4Exception:
            self.fail( "Failed to log on")
        
        try:
            self.p4.run_password( password, "" )
        except P4.P4Exception:
            self.fail( "Failed to reset the password" )
        
        self.assert_( os.path.exists(ticketFile), "Ticket file not found")
        
    def testExceptions(self):
        self.assertRaises(P4.P4Exception, self.p4.run_edit, "foo")
        
        self.p4.connect()
        self.assertRaises(P4.P4Exception, self.p4.run_edit, "foo")
        self.assertEqual( len(self.p4.errors), 1, "Did not find any errors")
        
    # father's little helpers
    
    def _setClient(self):
        """Creates a client and makes sure it is set up"""
        self.assert_(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.assert_( 'submittedChange' in result[-1], msg)
        except P4.P4Exception, inst:
            self.fail("submit failed with exception ")
        
if __name__ == '__main__':
    suite = unittest.TestLoader().loadTestsFromTestCase(TestP4)
    unittest.TextTestRunner(verbosity=2).run(suite)