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)