import glob, sys, time, stat
pathToBuild = glob.glob('build/lib*')
if len(pathToBuild) > 0:
sys.path.insert(0, pathToBuild[0])
import P4
import unittest, os, types, shutil, stat
def onRmTreeError( function, path, exc_info ):
os.chmod( path, stat.S_IWRITE)
os.remove( path )
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()
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)
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 ")
def testResolve(self):
testDir = 'test_resolve'
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 the file for testing resolve
file = "foo"
fname = os.path.join(testAbsoluteDir, file)
f = open(fname, "w")
f.write("First Line")
f.close()
theFile = testDir + "/" + file
self.p4.run_add(theFile)
change = self.p4.fetch_change()
change._description = "Initial"
self._doSubmit("Failed to submit initial", change)
# create a second revision
self.p4.run_edit(theFile)
f = open(fname, "a")
f.write("Second Line")
f.close()
change = self.p4.fetch_change()
change._description = "Second"
self._doSubmit("Failed to submit second", change)
# now sync back to first revision
self.p4.run_sync(theFile + "#1")
# edit the first revision, thus setting up the conflict
self.p4.run_edit(theFile)
# sync back the head revision, this will schedule the resolve
self.p4.run_sync(theFile)
class MyResolver(P4.Resolver):
def __init__(self, testObject):
self.t = testObject
def resolve(self, mergeData):
self.t.assertEqual(mergeData.your_name, "//TestClient/test_resolve/foo",
"Unexpected your_name: %s" % mergeData.your_name)
self.t.assertEqual(mergeData.their_name, "//depot/test_resolve/foo#2",
"Unexpected their_name: %s" % mergeData.their_name)
self.t.assertEqual(mergeData.base_name, "//depot/test_resolve/foo#1",
"Unexpected base_name: %s" % mergeData.base_name)
self.t.assertEqual(mergeData.merge_hint, "at", "Unexpected merge hint: %s" % mergeData.merge_hint)
return "at"
self.p4.run_resolve(resolver = MyResolver(self))
def testMap(self):
# don't need connection, simply test all the Map features
map = P4.Map()
self.assertEqual(map.count(), 0, "Map does not have count == 0")
self.assertEqual(map.is_empty(), True, "Map is not empty")
map.insert("//depot/main/... //ws/...")
self.assertEqual(map.count(), 1, "Map does not have 1 entry")
self.assertEqual(map.is_empty(), False, "Map is still empty")
self.assertEqual(map.includes("//depot/main/foo"), True, "Map does not map //depot/main/foo")
self.assertEqual(map.includes("//ws/foo", False), True, "Map does not map //ws/foo")
map.insert("-//depot/main/exclude/... //ws/exclude/...")
self.assertEqual(map.count(), 2, "Map does not have 2 entries")
self.assertEqual(map.includes("//depot/main/foo"), True, "Map does not map foo anymore")
self.assertEqual(map.includes("//depot/main/exclude/foo"), False, "Map still maps foo")
self.assertEqual(map.includes("//ws/foo", False), True, "Map does not map foo anymore (reverse)")
self.assertEqual(map.includes("//ws/exclude/foo"), False, "Map still maps foo (reverse)")
map.clear()
self.assertEqual(map.count(), 0, "Map has elements after clearing")
self.assertEqual(map.is_empty(), True, "Map is still not empty after clearing")
a = [ "//depot/main/... //ws/main/..." ,
"//depot/main/doc/... //ws/doc/..."]
map = P4.Map(a)
self.assertEqual(map.count(), 3, "Map does not contain 3 elements")
map2 = P4.Map("//ws/...", "C:\Work\...")
self.assertEqual(map2.count(), 1, "Map2 does not contain any elements")
map3 = P4.Map.join(map, map2)
self.assertEqual(map3.count(), 3, "Join did not produce three entries")
map.clear()
map.insert( '"//depot/dir with spaces/..." "//ws/dir with spaces/..."' )
self.assertEqual( map.includes("//depot/dir with spaces/foo"), True, "Quotes not handled correctly" )
def testThreads( self ):
import threading
class AsyncInfo( threading.Thread ):
def __init__( self, port ):
threading.Thread.__init__( self )
self.p4 = P4.P4()
self.p4.port = port
def run( self ):
self.p4.connect()
info = self.p4.run_info()
self.p4.disconnect()
threads = []
for i in range(1,10):
threads.append( AsyncInfo(self.port) )
for thread in threads:
thread.start()
for thread in threads:
thread.join()
if __name__ == '__main__':
suite = unittest.TestLoader().loadTestsFromTestCase(TestP4)
unittest.TextTestRunner(verbosity=2).run(suite)