# -*- encoding: UTF8 -*- # tests missing # - maximum # - preflight check from __future__ import print_function import sys import time import P4 import logutils import subprocess import inspect import unittest, os, shutil, stat, re from subprocess import Popen, PIPE python3 = sys.version_info[0] >= 3 if sys.hexversion < 0x02070000 or (0x0300000 < sys.hexversion < 0x0303000): sys.exit("Python 2.7 or 3.3 or newer is required to run this program.") if python3: from configparser import ConfigParser from io import StringIO else: from ConfigParser import ConfigParser from StringIO import StringIO import P4Transfer P4D = "p4d" P4USER = "testuser" P4CLIENT = "test_ws" TRANSFER_CLIENT = "transfer" TRANSFER_CONFIG = "transfer.cfg" TEST_COUNTER_NAME = "P4Transfer" INTEG_ENGINE = 3 def onRmTreeError(function, path, exc_info): os.chmod(path, stat.S_IWRITE) os.remove(path) 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) def append_to_file(file_name, contents): "Append contents to file" with open(file_name, 'a+') as f: f.write(contents) class P4Server: def __init__(self, root, logger): self.root = root self.logger = logger self.server_root = os.path.join(root, "server") self.client_root = os.path.join(root, "client") ensureDirectory(self.root) ensureDirectory(self.server_root) ensureDirectory(self.client_root) self.p4d = P4D self.port = "rsh:%s -r \"%s\" -L log -i" % (self.p4d, self.server_root) self.p4 = P4.P4() self.p4.port = self.port self.p4.user = P4USER self.p4.client = P4CLIENT self.p4.connect() self.p4cmd('depots') # triggers creation of the user self.p4cmd('configure', 'set', 'dm.integ.engine=%d' % INTEG_ENGINE) self.p4.disconnect() # required to pick up the configure changes self.p4.connect() self.client_name = P4CLIENT client = self.p4.fetch_client(self.client_name) client._root = self.client_root self.p4.save_client(client) def shutDown(self): if self.p4.connected(): self.p4.disconnect() def createTransferClient(self, name, root): pass def enableUnicode(self): cmd = [self.p4d, "-r", self.server_root, "-L", "log", "-vserver=3", "-xi"] f = Popen(cmd, stdout=PIPE).stdout for s in f.readlines(): pass f.close() def getCounter(self): "Returns value of counter as integer" result = self.p4.run('counter', TEST_COUNTER_NAME) if result and 'counter' in result[0]: return int(result[0]['value']) return 0 def p4cmd(self, *args): "Execute p4 cmd while logging arguments and results" if not self.logger: self.logger = logutils.getLogger(P4Transfer.LOGGER_NAME) self.logger.debug('testp4:', args) output = self.p4.run(args) self.logger.debug('testp4r:', output) return output class TestP4Transfer(unittest.TestCase): def __init__(self, methodName='runTest'): self.stdoutput = StringIO() logutils.getLogger(P4Transfer.LOGGER_NAME, self.stdoutput) self.logger = logutils.getLogger(P4Transfer.LOGGER_NAME) super(TestP4Transfer, self).__init__(methodName=methodName) def assertRegex(self, *args, **kwargs): if python3: return super(TestP4Transfer, self).assertRegex(*args, **kwargs) else: return super(TestP4Transfer, self).assertRegexpMatches(*args, **kwargs) def setUp(self): self.setDirectories() def tearDown(self): self.source.shutDown() self.target.shutDown() time.sleep(1) #self.cleanupTestTree() def setDirectories(self): self.startdir = os.getcwd() self.transfer_root = os.path.join(self.startdir, 'transfer') self.cleanupTestTree() ensureDirectory(self.transfer_root) self.source = P4Server(os.path.join(self.transfer_root, 'source'), self.logger) self.target = P4Server(os.path.join(self.transfer_root, 'target'), self.logger) self.transfer_client_root = localDirectory(self.transfer_root, 'transfer_client') def cleanupTestTree(self): os.chdir(self.startdir) if os.path.isdir(self.transfer_root): shutil.rmtree(self.transfer_root, False, onRmTreeError) def setupTransfer(self): """Creates client workspaces on source and target and a config file""" msg = "Test: %s ======================" % inspect.stack()[1][3] self.logger.debug(msg) source_client = self.source.p4.fetch_client(TRANSFER_CLIENT) source_client._root = self.transfer_client_root source_client._view = ['//depot/inside/... //%s/...' % TRANSFER_CLIENT] self.source.p4.save_client(source_client) target_client = self.target.p4.fetch_client('transfer') target_client._root = self.transfer_client_root target_client._view = ['//depot/import/... //%s/...' % TRANSFER_CLIENT] self.target.p4.save_client(target_client) self.createConfigFile() def createConfigFile(self, srcOptions=None, targOptions=None, options=None): "Creates config file with extras if appropriate" if options is None: options = {} if srcOptions is None: srcOptions = {} if targOptions is None: targOptions = {} self.parser = ConfigParser() self.parser.add_section('source') self.parser.set('source', 'p4port', self.source.port) self.parser.set('source', 'p4user', P4USER) self.parser.set('source', 'p4client', TRANSFER_CLIENT) for opt in srcOptions.keys(): self.parser.set('source', opt, srcOptions[opt]) self.parser.add_section('target') self.parser.set('target', 'p4port', self.target.port) self.parser.set('target', 'p4user', P4USER) self.parser.set('target', 'p4client', TRANSFER_CLIENT) for opt in targOptions.keys(): self.parser.set('target', opt, targOptions[opt]) self.parser.add_section('general') self.parser.set('general', 'logfile', os.path.join(self.transfer_root, 'temp', 'test.log')) if not os.path.exists(os.path.join(self.transfer_root, 'temp')): os.mkdir(os.path.join(self.transfer_root, 'temp')) self.parser.set('general', 'counter_name', TEST_COUNTER_NAME) for opt in options.keys(): self.parser.set('general', opt, options[opt]) # write the config file self.transfer_cfg = os.path.join(self.transfer_root, TRANSFER_CONFIG) with open(self.transfer_cfg, 'w') as f: self.parser.write(f) def run_P4Transfer(self, *args): base_args = ['-c', self.transfer_cfg, '-s'] if args: base_args.extend(args) pt = P4Transfer.P4Transfer(*base_args) result = pt.replicate() return result def assertCounters(self, sourceValue, targetValue): sourceCounter = self.target.getCounter() targetCounter = len(self.target.p4.run("changes")) self.assertEqual(sourceCounter, sourceValue, "Source counter is not {} but {}".format(sourceValue, sourceCounter)) self.assertEqual(targetCounter, targetValue, "Target counter is not {} but {}".format(targetValue, targetCounter)) def testArgParsing(self): "Basic argparsing for the module" self.setupTransfer() args = ['-c', self.transfer_cfg, '-s'] pt = P4Transfer.P4Transfer(*args) self.assertEqual(pt.options.config, self.transfer_cfg) self.assertTrue(pt.options.stoponerror) self.assertFalse(pt.options.preflight) args = ['-c', self.transfer_cfg] pt = P4Transfer.P4Transfer(*args) def testClientsMatch(self): "Make sure clients match the right hand side" msg = "Test: %s ======================" % inspect.stack()[0][3] self.logger.debug(msg) source_name = "src-%s" % TRANSFER_CLIENT source_client = self.source.p4.fetch_client(source_name) source_client._root = self.transfer_client_root source_client._view = ['//depot/inside/... //%s/source/...' % source_name] self.source.p4.save_client(source_client) target_name = "targ-%s" % TRANSFER_CLIENT target_client = self.target.p4.fetch_client(target_name) target_client._root = self.transfer_client_root target_client._view = ['//depot/import/... //%s/target/...' % target_name] self.target.p4.save_client(target_client) srcOptions = {"p4client": source_name} targOptions = {"p4client": target_name} self.createConfigFile(srcOptions=srcOptions, targOptions=targOptions) msg = "" try: base_args = ['-c', self.transfer_cfg, '-s'] pt = P4Transfer.P4Transfer(*base_args) result = pt.setupReplicate() except Exception as e: msg = str(e) self.assertRegex(msg, "workspace mappings have different right hand sides") source_client = self.source.p4.fetch_client(source_name) source_client._root = self.transfer_client_root source_client._view = ['//depot/inside/... //%s/root/...' % source_name] self.source.p4.save_client(source_client) target_client = self.target.p4.fetch_client(target_name) target_client._root = "%s-tmp" % self.transfer_client_root target_client._view = ['//depot/import/... //%s/root/...' % target_name] self.target.p4.save_client(target_client) msg = "" try: base_args = ['-c', self.transfer_cfg, '-s'] pt = P4Transfer.P4Transfer(*base_args) result = pt.setupReplicate() except Exception as e: msg = str(e) self.assertRegex(msg, "server workspace root directories must be the same") target_client = self.target.p4.fetch_client(target_name) target_client._root = self.transfer_client_root target_client._view = ['//depot/import/... //%s/root/...' % target_name] self.target.p4.save_client(target_client) msg = "" pt = P4Transfer.P4Transfer(*base_args) result = pt.setupReplicate() self.assertCounters(0, 0) def testChangeFormatting(self): "Formatting options for change descriptions" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, 'Test content') self.source.p4cmd('add', inside_file1) desc = 'inside_file1 added' self.source.p4cmd('submit', '-d', desc) self.run_P4Transfer() self.assertCounters(1, 1) changes = self.target.p4cmd('changes', '-l', '-m1') self.assertRegex(changes[0]['desc'], "%s\n\nTransferred from p4://rsh:.*@1\n$" % desc) options = {"change_description_format": "Originally $sourceChange by $sourceUser"} self.createConfigFile(options=options) self.source.p4cmd('edit', inside_file1) desc = 'inside_file1 edited' self.source.p4cmd('submit', '-d', desc) self.run_P4Transfer() self.assertCounters(2, 2) changes = self.target.p4cmd('changes', '-l', '-m1') self.assertRegex(changes[0]['desc'], "Originally 2 by %s" % P4USER) options = {"change_description_format": "Was $sourceChange by $sourceUser $fred\n$sourceDescription"} self.createConfigFile(options=options) self.source.p4cmd('edit', inside_file1) desc = 'inside_file1 edited again' self.source.p4cmd('submit', '-d', desc) self.run_P4Transfer() self.assertCounters(3, 3) changes = self.target.p4cmd('changes', '-l', '-m1') self.assertEqual(changes[0]['desc'], "Was 3 by %s $fred\n%s\n" % (P4USER, desc)) def testChangeMapFile(self): "How a change map file is written" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, 'Test content') self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'file added') self.run_P4Transfer() self.assertCounters(1, 1) changes = self.target.p4cmd('changes', '-l', '-m1') options = {"change_map_file": "change_map.csv"} change_map_file = '//depot/import/change_map.csv' self.createConfigFile(options=options) self.source.p4cmd('edit', inside_file1) self.source.p4cmd('submit', '-d', 'edited') self.run_P4Transfer() self.assertCounters(2, 3) change = self.target.p4.run_describe('3')[0] self.assertEqual(len(change['depotFile']), 1) self.assertEqual(change['depotFile'][0], change_map_file) content = self.target.p4.run_print('-q', change_map_file)[1].split("\n") self.logger.debug("content:", content) self.assertRegex(content[0], "sourceP4Port,sourceChangeNo,targetChangeNo") self.assertRegex(content[1], "rsh.*,2,2") self.source.p4cmd('edit', inside_file1) self.source.p4cmd('submit', '-d', 'edited again') self.source.p4cmd('edit', inside_file1) self.source.p4cmd('submit', '-d', 'and again') self.run_P4Transfer() self.assertCounters(4, 6) change = self.target.p4.run_describe('6')[0] self.assertEqual(len(change['depotFile']), 1) self.assertEqual(change['depotFile'][0], change_map_file) content = self.target.p4.run_print('-q', change_map_file)[1].split("\n") self.logger.debug("content:", content) self.assertRegex(content[1], "rsh.*,2,2") self.assertRegex(content[2], "rsh.*,3,4") def testAdd(self): "Basic file add" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, 'Test content') self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') self.run_P4Transfer() changes = self.target.p4cmd('changes') self.assertEqual(len(changes), 1, "Target does not have exactly one change") self.assertEqual(changes[0]['change'], "1", "Target change is not 1") files = self.target.p4cmd('files', '//depot/...') self.assertEqual(len(files), 1, "Target does not have exactly one file") self.assertEqual(files[0]['depotFile'], '//depot/import/inside_file1', "File not transferred to correct place") self.assertCounters(1, 1) @unittest.skipIf(python3, "Unicode not supported in Python3 yet...") def testUnicode(self): "Adding of files with Unicode filenames" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") if python3: inside_file1 = "inside_file1uåäö" else: inside_file1 = u"inside_file1uåäö".encode(sys.getfilesystemencoding()) inside_file2 = "Am\xE8lioration.txt" localinside_file1 = os.path.join(inside, inside_file1) localinside_file2 = os.path.join(inside, inside_file2) create_file(localinside_file1, 'Test content') create_file(localinside_file2, 'Some Test content') self.source.p4cmd('add', '-f', localinside_file1) self.source.p4cmd('add', '-f', localinside_file2) self.source.p4cmd('submit', '-d', 'inside_file1 added') self.run_P4Transfer() changes = self.target.p4cmd('changes') self.assertEqual(len(changes), 1) self.assertEqual(changes[0]['change'], "1") files = self.target.p4cmd('files', '//depot/...') self.assertEqual(len(files), 2) self.assertEqual(files[1]['depotFile'], '//depot/import/%s' % inside_file1) self.assertCounters(1, 1) def testAddWildcardChars(self): "Add where filenames contain Perforce wildcards" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "@inside_file1") inside_file2 = os.path.join(inside, "%25inside_file2") inside_file3 = os.path.join(inside, "#inside_file3") create_file(inside_file1, 'Test content') create_file(inside_file2, 'Test content') create_file(inside_file3, 'Test content') self.source.p4cmd('add', '-f', inside_file1) self.source.p4cmd('add', '-f', inside_file2) self.source.p4cmd('add', '-f', inside_file3) self.source.p4cmd('submit', '-d', 'files added') self.run_P4Transfer() self.assertCounters(1, 1) changes = self.target.p4cmd('changes', ) self.assertEqual(len(changes), 1) self.assertEqual(changes[0]['change'], "1") files = self.target.p4cmd('files', '//depot/...') self.assertEqual(len(files), 3) self.assertEqual(files[0]['depotFile'], '//depot/import/%23inside_file3') self.assertEqual(files[1]['depotFile'], '//depot/import/%2525inside_file2') self.assertEqual(files[2]['depotFile'], '//depot/import/%40inside_file1') inside_file1Fixed = inside_file1.replace("@", "%40") inside_file2Fixed = inside_file2.replace("%", "%25") inside_file3Fixed = inside_file3.replace("#", "%23") self.source.p4cmd('edit', inside_file1Fixed) self.source.p4cmd('edit', inside_file2Fixed) self.source.p4cmd('edit', inside_file3Fixed) append_to_file(inside_file1, 'Different stuff') append_to_file(inside_file2, 'Different stuff') append_to_file(inside_file3, 'Different stuff') self.source.p4cmd('submit', '-d', 'files modified') self.run_P4Transfer() self.assertCounters(2, 2) changes = self.target.p4cmd('changes') self.assertEqual(len(changes), 2) self.assertEqual(changes[0]['change'], "2") self.source.p4cmd('integrate', "//depot/inside/*", "//depot/inside/new/*") self.source.p4cmd('submit', '-d', 'files branched') self.run_P4Transfer() self.assertCounters(3, 3) def testEditAndDelete(self): "Edits and Deletes" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, 'Test content') self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', "inside_file1 added") self.source.p4cmd('edit', inside_file1) append_to_file(inside_file1, 'More content') self.source.p4cmd('submit', '-d', "inside_file1 edited") self.run_P4Transfer() changes = self.target.p4cmd('changes', ) self.assertEqual(len(changes), 2) self.assertEqual(changes[0]['change'], "2") self.assertCounters(2, 2) self.source.p4cmd('delete', inside_file1) self.source.p4cmd('submit', '-d', "inside_file1 deleted") self.run_P4Transfer() self.assertCounters(3, 3) changes = self.target.p4cmd('changes', ) self.assertEqual(len(changes), 3, "Target does not have exactly three changes") filelog = self.target.p4.run_filelog('//depot/import/inside_file1') self.assertEqual(filelog[0].revisions[0].action, 'delete', "Target has not been deleted") create_file(inside_file1, 'New content') self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', "Re-added") self.run_P4Transfer() self.assertCounters(4, 4) filelog = self.target.p4.run_filelog('//depot/import/inside_file1') self.assertEqual(filelog[0].revisions[0].action, 'add', "Target has not been re-added") def testFileTypes(self): "File types are transferred appropriately" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, "Test content") self.source.p4cmd('add', '-tbinary', inside_file1) self.source.p4cmd('submit', '-d', "inside_file1 added") self.run_P4Transfer() self.assertCounters(1, 1) filelog = self.target.p4.run_filelog('//depot/import/inside_file1') self.assertEqual(filelog[0].revisions[0].type, 'binary', "File type is not binary, but %s" % filelog[0].revisions[0].type) self.source.p4cmd('edit', '-t+x', inside_file1) append_to_file(inside_file1, "More content") self.source.p4cmd('submit', '-d', "Type changed") self.run_P4Transfer() self.assertCounters(2, 2) filelog = self.target.p4.run_filelog('//depot/import/inside_file1') self.assertEqual(filelog[0].revisions[0].type, 'xbinary', "File type is not xbinary, but %s" % filelog[0].revisions[0].type) inside_file2 = os.path.join(inside, "inside_file2") create_file(inside_file2, "$Id$\n$DateTime$") self.source.p4cmd('add', '-t+k', inside_file2) self.source.p4cmd('submit', '-d', "Ktext added") self.run_P4Transfer() self.assertCounters(3, 3) filelog = self.target.p4.run_filelog('//depot/import/inside_file2') self.assertEqual(filelog[0].revisions[0].type, 'ktext', "File type is not ktext, but %s" % filelog[0].revisions[0].type) verifyResult = self.target.p4.run_verify('-q', '//depot/import/inside_file2') self.assertEqual(len(verifyResult), 0) # just to see that ktext gets transferred properly content = self.target.p4.run_print('//depot/import/inside_file2')[1] lines = content.split("\n") self.assertEqual(lines[0], '$Id: //depot/import/inside_file2#1 $', "Content does not match : %s" % lines[0]) def testSimpleIntegrate(self): "Simple integration options - inside client workspace view" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, "Test content") self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') inside_file2 = os.path.join(inside, "inside_file2") self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('submit', '-d', 'inside_file1 -> inside_file2') self.run_P4Transfer() self.assertCounters(2, 2) changes = self.target.p4cmd('changes', ) self.assertEqual(len(changes), 2, "Target does not have exactly two changes") self.source.p4cmd('edit', inside_file1) append_to_file(inside_file1, "More content") self.source.p4cmd('submit', '-d', 'inside_file1 edited') self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4.run_resolve('-at') self.source.p4cmd('submit', '-d', 'inside_file1 -> inside_file2 (copy)') self.run_P4Transfer() self.assertCounters(4, 4) filelog = self.target.p4.run_filelog('//depot/import/inside_file2') self.assertEqual(len(filelog[0].revisions), 2, "Not exactly 2 target revisions") self.assertEqual(len(filelog[0].revisions[1].integrations), 1, "Not exactly 1 integration into target") self.assertEqual(filelog[0].revisions[0].integrations[0].how, "copy from", "'How' is not copy from") # Now make 2 changes and integrate them one at a time. self.source.p4cmd('edit', inside_file1) append_to_file(inside_file1, "More content2") self.source.p4cmd('submit', '-d', 'inside_file1 edited') self.source.p4cmd('edit', inside_file1) append_to_file(inside_file1, "More content3") self.source.p4cmd('submit', '-d', 'inside_file1 edited') self.source.p4cmd('integrate', inside_file1 + "#3", inside_file2) self.source.p4.run_resolve('-at') self.source.p4cmd('submit', '-d', 'inside_file1 -> inside_file2 (copy)') self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4.run_resolve('-at') self.source.p4cmd('submit', '-d', 'inside_file1 -> inside_file2 (copy)') self.run_P4Transfer() self.assertCounters(8, 8) filelog = self.target.p4.run_filelog('//depot/import/inside_file2') self.logger.debug(filelog) self.assertEqual(len(filelog[0].revisions), 4) self.assertEqual(len(filelog[0].revisions[1].integrations), 1, "Not exactly 1 integration into target") self.assertEqual(filelog[0].revisions[0].integrations[0].how, "copy from", "'How' is not copy from") def testComplexIntegrate(self): "More complex integrations with various resolve options" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") content1 = """ Line 1 Line 2 - changed Line 3 """ create_file(inside_file1, """ Line 1 Line 2 Line 3 """) self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') inside_file2 = os.path.join(inside, "inside_file2") self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('submit', '-d', 'inside_file1 -> inside_file2') # Prepare merge self.source.p4cmd('edit', inside_file1, inside_file2) create_file(inside_file1, content1) create_file(inside_file2, """ Line 1 Line 2 Line 3 - changed """) self.source.p4cmd('submit', '-d', "Changed both contents") # Integrate with merge self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4.run_resolve('-am') self.source.p4cmd('submit', '-d', "Merged contents") contentMerged = self.source.p4.run_print(inside_file2)[1] sourceCounter = 4 targetCounter = 4 self.run_P4Transfer() self.assertCounters(sourceCounter, targetCounter) filelog = self.target.p4.run_filelog('//depot/import/inside_file2') self.logger.debug('test:', filelog) self.assertEqual(filelog[0].revisions[0].integrations[0].how, 'edit from') self.assertEqual(self.target.p4.run_print('//depot/import/inside_file2')[1], contentMerged) # Prepare integrate with edit self.source.p4cmd('edit', inside_file1, inside_file2) create_file(inside_file1, content1) self.source.p4cmd('submit', '-d', "Created a conflict") # Integrate with edit self.source.p4cmd('integrate', inside_file1, inside_file2) class EditResolve(P4.Resolver): def resolve(self, mergeData): create_file(mergeData.result_path, """ Line 1 Line 2 - changed Line 3 - edited """) return 'ae' self.source.p4.run_resolve(resolver=EditResolve()) self.source.p4cmd('submit', '-d', "Merge with edit") sourceCounter += 2 targetCounter += 2 self.run_P4Transfer() self.assertCounters(sourceCounter, targetCounter) # Prepare ignore self.source.p4cmd('edit', inside_file1) append_to_file(inside_file1, "For your eyes only") self.source.p4cmd('submit', '-d', "Edit source again") self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4.run_resolve('-ay') # ignore self.source.p4cmd('submit', '-d', "Ignored change in inside_file1") sourceCounter += 2 targetCounter += 2 self.run_P4Transfer() self.assertCounters(sourceCounter, targetCounter) filelog = self.target.p4.run_filelog('//depot/import/inside_file2') self.assertEqual(filelog[0].revisions[0].integrations[0].how, 'ignored', "How is not ignored") content = self.target.p4.run_print('-a', '//depot/import/inside_file2') self.assertEqual(content[1], content[3], "Content of #1 not equal to #2") # Prepare delete self.source.p4cmd('delete', inside_file1) self.source.p4cmd('submit', '-d', "Delete file 1") self.source.p4.run_merge(inside_file1, inside_file2) # to trigger resolve self.source.p4.run_resolve('-at') self.source.p4cmd('submit', '-d', "Propagated delete") sourceCounter += 2 targetCounter += 2 self.run_P4Transfer() self.assertCounters(sourceCounter, targetCounter) filelog = self.target.p4.run_filelog('//depot/import/inside_file2') self.assertEqual(len(filelog[0].revisions[0].integrations), 1, "No integration for delete") self.assertEqual(filelog[0].revisions[0].integrations[0].how, 'delete from', "How is not delete from") # Prepare re-add create_file(inside_file1, content1) self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 re-added') self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('submit', '-d', "inside_file2 re-added") sourceCounter += 2 targetCounter += 2 self.run_P4Transfer() self.assertCounters(sourceCounter, targetCounter) filelog = self.target.p4.run_filelog('//depot/import/inside_file2') self.assertEqual(filelog[0].revisions[0].integrations[0].how, 'branch from', "How is not branch from") # def testIntegrateOfDeletes(self): # "Integration requiring flags to propagate a delete after adding a file again" # self.setupTransfer() # self.source.p4cmd("configure", "set", "dm.integ.engine=2") # self.target.p4cmd("configure", "set", "dm.integ.engine=2") # # inside = localDirectory(self.source.client_root, "inside") # inside_file1 = os.path.join(inside, "inside_file1") # create_file(inside_file1, "Test content") # self.source.p4cmd('add', inside_file1) # self.source.p4cmd('submit', '-d', 'inside_file1 added') # # inside_file2 = os.path.join(inside, "inside_file2") # # self.source.p4cmd('integrate', inside_file1, inside_file2) # self.source.p4cmd('submit', '-d', 'inside_file1 -> inside_file2') # # self.source.p4cmd('delete', inside_file1) # self.source.p4cmd('submit', '-d', 'deleted inside_file1') # # self.source.p4cmd('sync', '%s#1' % inside_file1) # self.source.p4cmd('add', '-df', inside_file1) # self.source.p4cmd('submit', '-d', 're-added inside_file1') # # self.source.p4cmd('edit', inside_file1) # append_to_file(inside_file1, "more content") # self.source.p4cmd('submit', '-d', 'added content') # # self.source.p4cmd('integrate', inside_file1, inside_file2) # self.source.p4cmd('submit', '-d', 'inside_file1 -> inside_file2 again') # # self.run_P4Transfer() # self.assertCounters(2, 2) def testForcedIntegrate(self): "Integration requiring -f" self.setupTransfer() self.source.p4cmd("configure", "set", "dm.integ.engine=2") self.target.p4cmd("configure", "set", "dm.integ.engine=2") inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") inside_file2 = os.path.join(inside, "inside_file2") create_file(inside_file1, "Test content") self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('submit', '-d', 'inside_file1 -> inside_file2') self.source.p4cmd('edit', inside_file1) append_to_file(inside_file1, "more content") self.source.p4cmd('submit', '-d', 'added content') self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('resolve', '-as') self.source.p4cmd('submit', '-d', 'inside_file1 -> inside_file2 again') # Backout the change which was integrated in, and then force integrate self.source.p4cmd('sync', "%s#1" % inside_file2) self.source.p4cmd('edit', inside_file2) self.source.p4cmd('sync', inside_file2) self.source.p4cmd('resolve', '-at', inside_file2) self.source.p4cmd('submit', '-d', 'backed out inside_file2') self.source.p4cmd('integrate', '-f', inside_file1, inside_file2) self.source.p4cmd('resolve', '-at') self.source.p4cmd('submit', '-d', 'Force integrate') self.run_P4Transfer() self.assertCounters(6, 6) def testDirtyMerge(self): """A merge which is supposedly clean but in reality is a dirty one - this is possible when transferring with old servers""" self.setupTransfer() content = """ Line 1 Line 2 Line 3 """ content2 = """ Line 1 Line 2 Line 3 - changed """ content3 = """ Line 1 Line 2 - changed Line 3 - edited """ content4 = """ Line 1 Line 2 - changed Line 3 - edited and changed """ inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, content) self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') inside_file2 = os.path.join(inside, "inside_file2") self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('submit', '-d', 'inside_file1 -> inside_file2') # Prepare merge with edit self.source.p4cmd('edit', inside_file1, inside_file2) create_file(inside_file1, content2) create_file(inside_file2, content3) self.source.p4cmd('submit', '-d', "Changed both contents") self.source.p4cmd('integrate', inside_file1, inside_file2) class EditResolve(P4.Resolver): def resolve(self, mergeData): create_file(mergeData.result_path, content4) return 'ae' self.source.p4.run_resolve(resolver=EditResolve()) self.source.p4cmd('submit', '-d', "Merge with edit") filelog = self.source.p4.run_filelog(inside_file2) self.assertEqual(len(filelog[0].revisions[0].integrations), 1) self.assertEqual(filelog[0].revisions[0].integrations[0].how, 'edit from') # Convert dirty merge to pretend clean merge. # # Dirty merge (fields 12/10) # @pv@ 0 @db.integed@ @//depot/inside/inside_file2@ @//depot/inside/inside_file1@ 1 2 2 3 12 4 # @pv@ 0 @db.integed@ @//depot/inside/inside_file1@ @//depot/inside/inside_file2@ 2 3 1 2 10 4 # # Clean merge (fields 0/1) # @pv@ 0 @db.integed@ @//depot/inside/inside_file2@ @//depot/inside/inside_ file1@ 1 2 2 3 0 4 # @pv@ 0 @db.integed@ @//depot/inside/inside_file1@ @//depot/inside/inside_file2@ 2 3 1 2 1 4 jnl_rec = "@rv@ 0 @db.integed@ @//depot/inside/inside_file2@ @//depot/inside/inside_file1@ 1 2 2 3 0 4\n" + \ "@rv@ 0 @db.integed@ @//depot/inside/inside_file1@ @//depot/inside/inside_file2@ 2 3 1 2 1 4\n" self.applyJournalPatch(jnl_rec) self.run_P4Transfer() self.assertCounters(4, 4) def testDodgyMerge(self): """Like testDirtyMerge but user has cherry picked and then hand edited - clean merge is different on disk""" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, """ Line 1 Line 2 Line 3 """) self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') inside_file2 = os.path.join(inside, "inside_file2") self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('submit', '-d', 'inside_file1 -> inside_file2') self.source.p4cmd('edit', inside_file1) create_file(inside_file1, """ Line 1 - changed file1 Line 2 Line 3 """) self.source.p4cmd('submit', '-d', "Changed inside_file1") self.source.p4cmd('edit', inside_file1) create_file(inside_file1, """ Line 1 - changed file1 Line 2 Line 3 - changed file1 """) self.source.p4cmd('submit', '-d', "Changed inside_file1") self.source.p4cmd('edit', inside_file2) create_file(inside_file2, """ Line 1 Line 2 - changed file2 Line 3 """) self.source.p4cmd('submit', '-d', "Changed inside_file1") # Merge with edit - but cherry picked self.source.p4cmd('integrate', "%s#3,3" % inside_file1, inside_file2) class EditResolve(P4.Resolver): def resolve(self, mergeData): create_file(mergeData.result_path, """ Line 1 - edited Line 2 - changed file2 Line 3 - changed file1 """) return 'ae' self.source.p4.run_resolve(resolver=EditResolve()) self.source.p4cmd('submit', '-d', "Merge with edit") filelog = self.source.p4.run_filelog(inside_file2) self.assertEqual(len(filelog[0].revisions[0].integrations), 1) self.assertEqual(filelog[0].revisions[0].integrations[0].how, 'edit from') self.logger.debug("print:", self.source.p4.run_print(inside_file2)) # Convert dirty merge to clean merge # # Dirty merge (fields 12/10) # @pv@ 0 @db.integed@ @//depot/inside/inside_file2@ @//depot/inside/inside_file1@ 2 3 2 3 12 6 # @pv@ 0 @db.integed@ @//depot/inside/inside_file1@ @//depot/inside/inside_file2@ 2 3 2 3 10 6 # # Clean merge (fields 0/1) # @pv@ 0 @db.integed@ @//depot/inside/inside_file2@ @//depot/inside/inside_ file1@ 1 2 2 3 0 4 # @pv@ 0 @db.integed@ @//depot/inside/inside_file1@ @//depot/inside/inside_file2@ 2 3 1 2 1 4 jnl_rec = "@rv@ 0 @db.integed@ @//depot/inside/inside_file2@ @//depot/inside/inside_file1@ 2 3 2 3 0 6\n" + \ "@pv@ 0 @db.integed@ @//depot/inside/inside_file1@ @//depot/inside/inside_file2@ 2 3 2 3 1 6" self.applyJournalPatch(jnl_rec) filelog = self.source.p4.run_filelog(inside_file2) self.assertEqual(len(filelog[0].revisions[0].integrations), 1) self.assertEqual(filelog[0].revisions[0].integrations[0].how, 'merge from') self.logger.debug("print:", self.source.p4.run_print(inside_file2)) self.run_P4Transfer() self.assertCounters(6, 6) self.logger.debug("print:", self.target.p4.run_print("//depot/import/inside_file2")) def testRemoteIntegs(self): """An integrate from a remote depot - gives the 'import' action """ self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, "Some content") self.source.p4cmd("add", inside_file1) self.source.p4cmd("submit", '-d', 'inside_file1 added') filelog = self.source.p4.run_filelog(inside_file1) self.assertEqual(filelog[0].revisions[0].action, 'add') recs = self.dumpDBFiles("db.rev,db.revcx,db.revhx") self.logger.debug(recs) # '@pv@ 9 @db.rev@ @//depot/inside/inside_file1@ 1 0 0 1 1420649505 1420649505 581AB2D89F05C294D4FE69C623BDEF83 13 0 0 @//depot/inside/inside_file1@ @1.1@ 0 ', # @pv@ 0 @db.revcx@ 1 @//depot/inside/inside_file1@ 1 0 ', # '@pv@ 9 @db.revhx@ @//depot/inside/inside_file1@ 1 0 0 1 1420649505 1420649505 581AB2D89F05C294D4FE69C623BDEF83 13 0 0 @//depot/inside/inside_file1@ @1.1@ 0 ' recs[0] = recs[0].replace("@ 1 0 0 1 ", "@ 1 0 5 1 ") recs[1] = recs[1].replace("@ 1 0", "@ 1 5") recs[2] = recs[2].replace("@ 1 0 0 1 ", "@ 1 0 5 1 ") recs[0] = recs[0].replace("@pv@", "@rv@") recs[1] = recs[1].replace("@pv@", "@rv@") recs[2] = recs[2].replace("@pv@", "@rv@") self.applyJournalPatch("\n".join(recs)) filelog = self.source.p4.run_filelog(inside_file1) self.assertEqual(filelog[0].revisions[0].action, 'import') self.run_P4Transfer() self.assertCounters(1, 1) filelog = self.target.p4.run_filelog('//depot/import/inside_file1') self.assertEqual(filelog[0].revisions[0].action, 'add') def testDirtyBranch(self): """A copy which is supposedly clean but in reality has been edited""" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, "Some content") self.source.p4cmd("add", inside_file1) self.source.p4cmd("submit", '-d', 'inside_file1 added') inside_file2 = os.path.join(inside, "inside_file2") self.source.p4cmd("integrate", inside_file1, inside_file2) self.source.p4cmd("edit", inside_file2) append_to_file(inside_file2, "New content") self.source.p4cmd("submit", '-d', 'inside_file1 -> inside_file2 with edit') filelog = self.source.p4.run_filelog(inside_file1) self.assertEqual(len(filelog[0].revisions[0].integrations), 1) self.assertEqual(filelog[0].revisions[0].integrations[0].how, 'add into') # Needs to be created via journal patch # # Branch as edit (fields 2/11) # @pv@ 0 @db.integed@ @//depot/inside/inside_file2@ @//depot/inside/inside_file1@ 0 1 0 1 2 2 # @pv@ 0 @db.integed@ @//depot/inside/inside_file1@ @//depot/inside/inside_file2@ 0 1 0 1 11 2 # # Clean branch (fields 2/3) # @pv@ 0 @db.integed@ @//depot/inside/inside_file2@ @//depot/inside/inside_file1@ 0 1 0 1 2 2 # @pv@ 0 @db.integed@ @//depot/inside/inside_file1@ @//depot/inside/inside_file2@ 0 1 0 1 3 2 jnl_rec = "@rv@ 0 @db.integed@ @//depot/inside/inside_file2@ @//depot/inside/inside_file1@ 0 1 0 1 2 2\n" + \ "@rv@ 0 @db.integed@ @//depot/inside/inside_file1@ @//depot/inside/inside_file2@ 0 1 0 1 3 2\n" self.applyJournalPatch(jnl_rec) filelog = self.source.p4.run_filelog(inside_file1) self.assertEqual(len(filelog[0].revisions[0].integrations), 1) self.assertEqual(filelog[0].revisions[0].integrations[0].how, 'branch into') filelog = self.source.p4.run_filelog(inside_file2) self.assertEqual(len(filelog[0].revisions[0].integrations), 1) self.assertEqual(filelog[0].revisions[0].integrations[0].how, 'branch from') self.run_P4Transfer() self.assertCounters(2, 2) def applyJournalPatch(self, jnl_rec): "Apply journal patch" jnl_fix = os.path.join(self.source.server_root, "jnl_fix") create_file(jnl_fix, jnl_rec) cmd = '%s -r "%s" -jr "%s"' % (self.source.p4d, self.source.server_root, jnl_fix) self.logger.debug("Cmd: %s" % cmd) output = subprocess.check_output(cmd, shell=True) def dumpDBFiles(self, tables): "Extract journal records" cmd = '%s -r "%s" -k %s -jd -' % (self.source.p4d, self.source.server_root, tables) self.logger.debug("Cmd: %s" % cmd) output = subprocess.check_output(cmd, shell=True, universal_newlines=True) self.logger.debug("Output: %s" % output) results = [r for r in output.split("\n") if re.search("^@pv@", r)] return results def testMultipleIntegrate(self): "Multipled integrations transferred in one go" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, "Some content") self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') inside_file2 = os.path.join(inside, "inside_file2") self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('submit', '-d', 'inside_file1 -> inside_file2') file3 = os.path.join(inside, "file3") self.source.p4cmd('integrate', inside_file2, file3) self.source.p4cmd('submit', '-d', 'inside_file2 -> File3') self.run_P4Transfer() self.assertCounters(3, 3) filelog1 = self.target.p4.run_filelog('//depot/import/inside_file1') filelog2 = self.target.p4.run_filelog('//depot/import/inside_file2') filelog3 = self.target.p4.run_filelog('//depot/import/file3') self.assertEqual(len(filelog1[0].revisions), 1, "Not exactly one inside_file1 revision") self.assertEqual(len(filelog2[0].revisions), 1, "Not exactly one inside_file2 revision") self.assertEqual(len(filelog3[0].revisions), 1, "Not exactly one file3 revision") self.assertEqual(len(filelog1[0].revisions[0].integrations), 1, "Not exactly one inside_file1 integ record") self.assertEqual(len(filelog2[0].revisions[0].integrations), 2, "Not exactly two inside_file2 integ records") self.assertEqual(len(filelog3[0].revisions[0].integrations), 1, "Not exactly one file3 integ record") def testInsideOutside(self): "Test integrations between the inside<->outside where only one element is thus transferred" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") outside = localDirectory(self.source.client_root, "outside") # add from outside, integrate in outside_file = os.path.join(outside, 'outside_file') create_file(outside_file, "Some content") self.source.p4cmd('add', outside_file) self.source.p4cmd('submit', '-d', "Outside outside_file") inside_file2 = os.path.join(inside, 'inside_file2') self.source.p4cmd('integrate', outside_file, inside_file2) self.source.p4cmd('submit', '-d', "Integrated from outside to inside") self.run_P4Transfer() self.assertCounters(2, 1) changes = self.target.p4cmd('changes', ) self.assertEqual(len(changes), 1, "Not exactly one change on target") filelog = self.target.p4.run_filelog('//depot/import/inside_file2') self.assertEqual(filelog[0].revisions[0].action, "add", "inside_file2 action is not add") # edit from outside, integrated in self.source.p4cmd('edit', outside_file) append_to_file(outside_file, "More content") self.source.p4cmd('submit', '-d', "Outside outside_file edited") self.run_P4Transfer() self.assertCounters(2, 1) # counters will not move, no change within the client workspace's scope self.source.p4cmd('integrate', outside_file, inside_file2) self.source.p4.run_resolve('-at') self.source.p4cmd('submit', '-d', "Copied outside_file -> inside_file2") self.run_P4Transfer() self.assertCounters(4, 2) changes = self.target.p4cmd('changes', ) self.assertEqual(len(changes), 2, "Not exactly two changes on target") filelog = self.target.p4.run_filelog('//depot/import/inside_file2') self.assertEqual(filelog[0].revisions[0].action, "edit", "inside_file2 action is not edit") # delete from outside, integrate in self.source.p4cmd('delete', outside_file) self.source.p4cmd('submit', '-d', "outside_file deleted") self.source.p4cmd('integrate', outside_file, inside_file2) self.source.p4cmd('submit', '-d', "inside_file2 deleted from outside_file") self.run_P4Transfer() self.assertCounters(6, 3) changes = self.target.p4cmd('changes', ) self.assertEqual(len(changes), 3, "Not exactly three changes on target") filelog = self.target.p4.run_filelog('//depot/import/inside_file2') self.assertEqual(filelog[0].revisions[0].action, "delete", "inside_file2 action is not delete") # Add to both inside and outside in the same changelist - check only inside transferred create_file(inside_file2, "Different content") create_file(outside_file, "Different content") self.source.p4cmd('add', inside_file2, outside_file) self.source.p4cmd('submit', '-d', "adding inside and outside") self.run_P4Transfer() self.assertCounters(7, 4) change = self.target.p4.run_describe('4')[0] self.assertEqual(len(change['depotFile']), 1) self.assertEqual(change['depotFile'][0], '//depot/import/inside_file2') def testOutsideInsideDirtyCopy(self): "Test integrations between the inside<->outside where copy is not actually clean" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") outside = localDirectory(self.source.client_root, "outside") # Add and delete inside file inside_file = os.path.join(inside, 'inside_file') create_file(inside_file, "Inside content") self.source.p4cmd('add', inside_file) self.source.p4cmd('submit', '-d', "Added inside_file") self.source.p4cmd('delete', inside_file) self.source.p4cmd('submit', '-d', "Deleted inside_file") # add from outside, integrate in outside_file = os.path.join(outside, 'outside_file') create_file(outside_file, "Outside content") self.source.p4cmd('add', outside_file) self.source.p4cmd('submit', '-d', "Outside outside_file") self.source.p4cmd('integrate', outside_file, inside_file) self.source.p4cmd('edit', inside_file) append_to_file(inside_file, "extra stuff") self.source.p4cmd('submit', '-d', "Integrated from outside to inside") filelog = self.source.p4.run_filelog(outside_file) self.assertEqual(len(filelog[0].revisions[0].integrations), 1) self.assertEqual(filelog[0].revisions[0].integrations[0].how, 'add into') # Branch as edit (fields 2/11) # @pv@ 0 @db.integed@ @//depot/inside/inside_file@ @//depot/outside/outside_file@ 0 1 2 3 2 4 # @pv@ 0 @db.integed@ @//depot/outside/outside_file@ @//depot/inside/inside_file@ 2 3 0 1 11 4 # # Clean branch (fields 2/3) # @pv@ 0 @db.integed@ @//depot/inside/inside_file@ @//depot/outside/outside_file@ 0 1 2 3 2 4 # @pv@ 0 @db.integed@ @//depot/outside/outside_file@ @//depot/inside/inside_file@ 2 3 0 1 3 4 jnl_rec = "@rv@ 0 @db.integed@ @//depot/inside/inside_file@ @//depot/outside/outside_file@ 0 1 2 3 2 4\n" + \ "@rv@ 0 @db.integed@ @//depot/outside/outside_file@ @//depot/inside/inside_file@ 2 3 0 1 3 4\n" self.applyJournalPatch(jnl_rec) filelog = self.source.p4.run_filelog(inside_file) self.assertEqual(len(filelog[0].revisions[0].integrations), 1) self.assertEqual(filelog[0].revisions[0].integrations[0].how, 'branch from') self.run_P4Transfer() self.assertCounters(4, 3) def testFailedSubmit(self): """Test what happens if submits fail, e.g. due to trigger""" self.setupTransfer() protect = self.source.p4.fetch_protect() self.logger.debug('protect:', protect) self.logger.debug(self.target.p4.save_protect(protect)) self.target.p4cmd('admin', 'restart') self.target.p4.disconnect() self.target.p4.connect() triggers = self.target.p4.fetch_triggers() triggers['Triggers'] = ['test-trigger change-submit //depot/... "fail No submits allowed at this time"'] self.target.p4.save_triggers(triggers) self.target.p4.disconnect() self.target.p4.connect() inside = localDirectory(self.source.client_root, "inside") inside_file = os.path.join(inside, 'inside_file') create_file(inside_file, "Some content") self.source.p4cmd('add', inside_file) self.source.p4cmd('submit', '-d', "adding inside, hidden") result = self.run_P4Transfer() self.assertEqual(result, 1) self.assertCounters(0, 1) @unittest.skip("Not properly working test yet - issues around fetching protect...") def testHiddenFiles(self): """Test for adding and integrating from hidden files - not visible to transfer user""" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file = os.path.join(inside, 'inside_file') create_file(inside_file, "Some content") hidden_file = os.path.join(inside, 'hidden_file') create_file(hidden_file, "Some content") self.source.p4cmd('add', inside_file, hidden_file) self.source.p4cmd('submit', '-d', "adding inside, hidden") # Now change permissions self.source.p4cmd('edit', inside_file, hidden_file) append_to_file(inside_file, 'More content') append_to_file(hidden_file, 'More content') self.source.p4cmd('submit', '-d', "changing inside, hidden") p4superuser = 'p4newadmin' self.source.p4.user = p4superuser protect = self.source.p4.fetch_protect() protect['Protections'].append("write user %s * -//depot/...hidden_file" % P4USER) self.source.p4.save_protect(protect) self.logger.debug("protect:", self.source.p4.run_protect("-o")) self.source.p4.user = P4USER self.source.p4.disconnect() self.source.p4.connect() p = self.source.p4.run_protects('//depot/...') self.logger.debug('protects:', p) self.run_P4Transfer() self.assertCounters(2, 2) change = self.target.p4.run_describe('1')[0] self.assertEqual(len(change['depotFile']), 1) self.assertEqual(change['depotFile'][0], '//depot/import/inside_file') change = self.target.p4.run_describe('2')[0] self.assertEqual(len(change['depotFile']), 1) self.assertEqual(change['depotFile'][0], '//depot/import/inside_file') # Edit and integrate in self.source.p4.user = p4superuser self.source.p4cmd('edit', hidden_file) append_to_file(hidden_file, 'Yet More content') self.source.p4cmd('submit', '-d', "Edited") self.source.p4.user = P4USER self.run_P4Transfer() self.assertCounters(2, 2) self.source.p4.user = p4superuser inside_file2 = os.path.join(inside, 'inside_file2') self.source.p4cmd('integrate', hidden_file, inside_file2) self.source.p4cmd('submit', '-d', "Copied outside_file -> inside_file2") self.logger.debug(self.source.p4.run_print(inside_file2)) self.source.p4.user = P4USER self.run_P4Transfer() self.assertCounters(4, 3) change = self.target.p4.run_describe('3')[0] self.assertEqual(len(change['depotFile']), 1) self.assertEqual(change['depotFile'][0], '//depot/import/inside_file2') self.assertEqual(change['action'][0], 'branch') def testMoves(self): """Test for Move and then a file being moved back, also move inside<->outside""" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") outside = localDirectory(self.source.client_root, "outside") original_file = os.path.join(inside, 'original', 'original_file') renamed_file = os.path.join(inside, 'new', 'new_file') create_file(original_file, "Some content") self.source.p4cmd('add', original_file) self.source.p4cmd('submit', '-d', "adding original file") self.source.p4cmd('edit', original_file) self.source.p4.run_move(original_file, renamed_file) self.source.p4cmd('submit', '-d', "renaming file") self.run_P4Transfer() self.assertCounters(2, 2) change = self.target.p4.run_describe('1')[0] self.assertEqual(len(change['depotFile']), 1) self.assertEqual(change['depotFile'][0], '//depot/import/original/original_file') change = self.target.p4.run_describe('2')[0] self.assertEqual(len(change['depotFile']), 2) self.assertEqual(change['depotFile'][0], '//depot/import/new/new_file') self.assertEqual(change['depotFile'][1], '//depot/import/original/original_file') self.assertEqual(change['action'][0], 'move/add') self.assertEqual(change['action'][1], 'move/delete') self.source.p4cmd('edit', renamed_file) self.source.p4.run_move(renamed_file, original_file) self.source.p4cmd('submit', '-d', "renaming file back") self.run_P4Transfer() self.assertCounters(3, 3) change = self.target.p4.run_describe('3')[0] self.assertEqual(len(change['depotFile']), 2) self.assertEqual(change['depotFile'][0], '//depot/import/new/new_file') self.assertEqual(change['depotFile'][1], '//depot/import/original/original_file') self.assertEqual(change['action'][0], 'move/delete') self.assertEqual(change['action'][1], 'move/add') # Now move inside to outside outside_file = os.path.join(outside, 'outside_file') self.source.p4cmd('edit', original_file) self.source.p4.run_move(original_file, outside_file) self.source.p4cmd('submit', '-d', "moving file outside") self.run_P4Transfer() self.assertCounters(4, 4) change = self.target.p4.run_describe('4')[0] self.assertEqual(len(change['depotFile']), 1) self.assertEqual(change['depotFile'][0], '//depot/import/original/original_file') self.assertEqual(change['action'][0], 'delete') # Now move outside to inside self.source.p4cmd('edit', outside_file) self.source.p4.run_move(outside_file, original_file) self.source.p4cmd('submit', '-d', "moving file from outside back to inside") self.run_P4Transfer() self.assertCounters(5, 5) change = self.target.p4.run_describe('5')[0] self.assertEqual(len(change['depotFile']), 1) self.assertEqual(change['depotFile'][0], '//depot/import/original/original_file') self.assertEqual(change['action'][0], 'add') def testMoveAfterDelete(self): """Test for Move after a Delete""" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") file1 = os.path.join(inside, 'file1') file2 = os.path.join(inside, 'file2') create_file(file1, "Some content") self.source.p4cmd("add", file1) self.source.p4cmd("submit", '-d', "adding original file") self.source.p4cmd("delete", file1) self.source.p4cmd("submit", '-d', "deleting original file") self.source.p4cmd("sync", "%s#1" % file1) self.source.p4cmd("edit", file1) self.source.p4cmd("move", file1, file2) try: self.source.p4cmd("resolve") except: pass try: self.source.p4cmd("submit", '-d', "renaming old version of original file") except: pass self.source.p4cmd("sync") self.source.p4cmd("submit", "-c3") self.run_P4Transfer() self.assertCounters(3, 3) change = self.target.p4.run_describe('3')[0] self.assertEqual(len(change['depotFile']), 2) self.assertEqual(change['depotFile'][0], '//depot/import/file1') self.assertEqual(change['depotFile'][1], '//depot/import/file2') self.assertEqual(change['action'][0], 'move/delete') self.assertEqual(change['action'][1], 'move/add') def testMoveAfterDeleteAndEdit(self): """Test for Move after a Delete when file content is also changed""" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") file1 = os.path.join(inside, 'file1') file2 = os.path.join(inside, 'file2') create_file(file1, "Some content") self.source.p4cmd("add", file1) self.source.p4cmd("submit", '-d', "adding original file") self.source.p4cmd("delete", file1) self.source.p4cmd("submit", '-d', "deleting original file") self.source.p4cmd("sync", "%s#1" % file1) self.source.p4cmd("edit", file1) self.source.p4cmd("move", file1, file2) try: self.source.p4cmd("resolve") except: pass try: self.source.p4cmd("submit", '-d', "renaming old version of original file") except: pass self.source.p4cmd("sync") append_to_file(file2, "A change") self.source.p4cmd("submit", "-c3") self.run_P4Transfer() self.assertCounters(3, 3) change = self.target.p4.run_describe('3')[0] self.assertEqual(len(change['depotFile']), 2) self.assertEqual(change['depotFile'][0], '//depot/import/file1') self.assertEqual(change['depotFile'][1], '//depot/import/file2') self.assertEqual(change['action'][0], 'move/delete') self.assertEqual(change['action'][1], 'move/add') def testMoveAndCopy(self): """Test for Move with subsequent copy of a file""" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") original_file = os.path.join(inside, 'original', 'original_file') renamed_file = os.path.join(inside, 'new', 'new_file') branched_file = os.path.join(inside, 'branch', 'new_file') create_file(original_file, "Some content") self.source.p4cmd('add', original_file) self.source.p4cmd('submit', '-d', "adding original file") self.source.p4cmd('edit', original_file) self.source.p4cmd('move', original_file, renamed_file) self.source.p4cmd('submit', '-d', "renaming file") self.source.p4cmd('integrate', '-Di', renamed_file, branched_file) self.source.p4cmd('submit', '-d', "copying files") self.run_P4Transfer() self.assertCounters(3, 3) change = self.target.p4.run_describe('2')[0] self.assertEqual(len(change['depotFile']), 2) self.assertEqual(change['depotFile'][0], '//depot/import/new/new_file') self.assertEqual(change['depotFile'][1], '//depot/import/original/original_file') self.assertEqual(change['action'][0], 'move/add') self.assertEqual(change['action'][1], 'move/delete') change = self.target.p4.run_describe('3')[0] self.assertEqual(len(change['depotFile']), 1) self.assertEqual(change['depotFile'][0], '//depot/import/branch/new_file') self.assertEqual(change['action'][0], 'branch') def testIntegDt(self): """Test for integ -Dt being required - only necessary with older integ.engine""" self.setupTransfer() self.target.p4cmd("configure", "set", "dm.integ.engine=2") self.source.p4cmd("configure", "set", "dm.integ.engine=2") inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") inside_file2 = os.path.join(inside, "inside_file2") create_file(inside_file1, "Test content") self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('submit', '-d', 'inside_file2 added') self.source.p4cmd('delete', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 deleted') self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('submit', '-d', 'inside_file2 deleted') create_file(inside_file1, "Test content again") self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('submit', '-d', 'inside_file2 added') self.run_P4Transfer() self.assertCounters(6, 6) def testInteg2to1(self): """Test for integrating 2 versions into single target.""" self.setupTransfer() # self.target.p4cmd("configure", "set", "dm.integ.engine=2") # self.source.p4cmd("configure", "set", "dm.integ.engine=2") inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") inside_file2 = os.path.join(inside, "inside_file2") create_file(inside_file1, "Test content") self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') self.source.p4cmd('edit', inside_file1) append_to_file(inside_file1, "more stuff") self.source.p4cmd('submit', '-d', 'inside_file1 edited') class EditResolve(P4.Resolver): def resolve(self, mergeData): create_file(mergeData.result_path, "new contents\nsome more") return 'ae' self.source.p4cmd('integrate', "%s#1" % inside_file1, inside_file2) self.source.p4cmd('add', inside_file2) self.source.p4cmd('integrate', "%s#2,2" % inside_file1, inside_file2) self.source.p4cmd('edit', inside_file2) self.source.p4.run_resolve(resolver=EditResolve()) self.source.p4cmd('submit', '-d', 'inside_file2 added with multiple integrates') self.source.p4cmd('filelog', inside_file2) self.run_P4Transfer() self.assertCounters(3, 3) filelog = self.target.p4.run_filelog('//depot/import/inside_file2')[0] self.logger.debug(filelog) self.assertEqual(len(filelog.revisions[0].integrations), 2) self.assertEqual(filelog.revisions[0].integrations[0].how, "edit from") self.assertEqual(filelog.revisions[0].integrations[1].how, "add from") def testAddFrom(self): """Test for adding a file which has in itself then branched.""" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") inside_file2 = os.path.join(inside, "inside_file2") inside_file3 = os.path.join(inside, "inside_file3") create_file(inside_file1, "Test content") self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') self.source.p4cmd('delete', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 deleted') self.source.p4cmd('sync', "%s#1" % inside_file1) self.source.p4cmd('add', inside_file1) self.source.p4cmd('move', inside_file1, inside_file2) append_to_file(inside_file2, "more stuff") self.source.p4cmd('submit', '-d', 'inside_file2 created by branching with add') filelog = self.source.p4.run_filelog(inside_file2)[0] self.logger.debug(filelog) self.assertEqual(len(filelog.revisions[0].integrations), 1) self.assertEqual(filelog.revisions[0].integrations[0].how, "add from") self.source.p4cmd('integrate', inside_file2, inside_file3) self.source.p4cmd('submit', '-d', 'inside_file3 created as copy') self.run_P4Transfer() self.assertCounters(4, 4) filelog = self.target.p4.run_filelog('//depot/import/inside_file2')[0] self.logger.debug(filelog) self.assertEqual(len(filelog.revisions[0].integrations), 2) self.assertEqual(filelog.revisions[0].integrations[0].how, "add from") def testDeleteDelete(self): """Test for a delete on top of a delete.""" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, "Test content") self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') self.source.p4cmd('delete', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 deleted') with self.source.p4.at_exception_level(P4.P4.RAISE_ERRORS): self.source.p4cmd('sync', "%s#1" % '//depot/inside/inside_file1') self.source.p4cmd('delete', '//depot/inside/inside_file1') self.source.p4cmd('opened') try: self.source.p4cmd('submit', '-d', 'inside_file1 deleted again') except Exception as e: self.logger.info(str(e)) err = self.source.p4.errors[0] if re.search("Submit failed -- fix problems above then", err): m = re.search("p4 submit -c (\d+)", err) if m: self.source.p4cmd('submit', '-c', m.group(1)) filelog = self.source.p4.run_filelog('//depot/inside/inside_file1')[0] self.logger.debug(filelog) self.assertEqual(filelog.revisions[0].action, 'delete') self.assertEqual(filelog.revisions[1].action, 'delete') self.run_P4Transfer() self.assertCounters(3, 3) filelog = self.target.p4.run_filelog('//depot/import/inside_file1')[0] self.logger.debug(filelog) self.assertEqual(filelog.revisions[0].action, 'delete') self.assertEqual(filelog.revisions[1].action, 'delete') def testIntegDeleteIgnore(self): """Test for an integrated delete with ignore.""" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") inside_file2 = os.path.join(inside, "inside_file2") create_file(inside_file1, "Test content") self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') self.source.p4cmd('integ', '-Rb', inside_file1, inside_file2) self.source.p4cmd('resolve', '-ay') self.source.p4cmd('submit', '-d', 'inside_file1 deleted') filelog = self.source.p4.run_filelog('//depot/inside/inside_file2')[0] self.logger.debug(filelog) self.assertEqual(filelog.revisions[0].action, 'delete') self.run_P4Transfer() self.assertCounters(2, 2) filelog = self.target.p4.run_filelog('//depot/import/inside_file2')[0] self.logger.debug(filelog) self.assertEqual(filelog.revisions[0].action, 'delete') def testMultipleIntegrates(self): """Test for more than one integration into same target revision""" self.setupTransfer() self.target.p4cmd("configure", "set", "dm.integ.engine=2") self.source.p4cmd("configure", "set", "dm.integ.engine=2") inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") inside_file2 = os.path.join(inside, "inside_file2") create_file(inside_file1, "Test content") self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('submit', '-d', 'inside_file2 added') self.source.p4cmd('edit', inside_file1) append_to_file(inside_file1, "more stuff") self.source.p4cmd('submit', '-d', 'inside_file1 edited') self.source.p4cmd('edit', inside_file1) append_to_file(inside_file1, "/nyet more stuff") self.source.p4cmd('submit', '-d', 'inside_file1 edited again') self.source.p4cmd('integrate', "%s#2" % inside_file1, inside_file2) self.source.p4cmd('resolve', '-as') self.source.p4cmd('integrate', "%s#3,3" % inside_file1, inside_file2) self.source.p4cmd('resolve', '-ay') # Ignore self.source.p4cmd('submit', '-d', 'integrated twice seperately into file2') self.run_P4Transfer() self.assertCounters(5, 5) filelog = self.target.p4.run_filelog('//depot/import/inside_file2')[0] self.logger.debug(filelog) self.assertEqual(len(filelog.revisions[0].integrations), 2) self.assertEqual(filelog.revisions[0].integrations[0].how, "ignored") self.assertEqual(filelog.revisions[0].integrations[1].how, "copy from") def testIntegI(self): """Test for integ -i required - only necessary with older integ.engine""" self.setupTransfer() self.target.p4cmd("configure", "set", "dm.integ.engine=2") inside = localDirectory(self.source.client_root, "inside") original_file = os.path.join(inside, 'original', 'original_file') new_file = os.path.join(inside, 'branch', 'new_file') create_file(original_file, "Some content") create_file(new_file, "Some new content") self.source.p4cmd('add', original_file) self.source.p4cmd('submit', '-d', "adding file1") self.source.p4cmd('add', new_file) self.source.p4cmd('submit', '-d', "adding file2") self.source.p4cmd('edit', original_file) create_file(original_file, "Some content addition") self.source.p4cmd('submit', '-d', "adding file1") self.source.p4cmd('integ', original_file, new_file) self.source.p4cmd('resolve', '-at') self.source.p4cmd('submit', '-d', "adding file2") self.run_P4Transfer() self.assertCounters(4, 4) change = self.target.p4.run_describe('4')[0] self.logger.debug(change) self.assertEqual(len(change['depotFile']), 1) self.assertEqual(change['depotFile'][0], '//depot/import/branch/new_file') self.assertEqual(change['action'][0], 'integrate') def testObliteratedSource(self): "File has been integrated and then source obliterated" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") outside = localDirectory(self.source.client_root, "outside") # add from outside, integrate in outside_file1 = os.path.join(outside, 'outside_file1') create_file(outside_file1, "Some content") self.source.p4cmd('add', outside_file1) self.source.p4cmd('submit', '-d', "Outside outside_file1") inside_file2 = os.path.join(inside, 'inside_file2') self.source.p4cmd('integrate', outside_file1, inside_file2) self.source.p4cmd('submit', '-d', "Integrated from outside to inside") self.source.p4.run_obliterate('-y', outside_file1) self.run_P4Transfer('--stoponerror') self.assertCounters(2, 1) def testKeywords(self): "Look for files with keyword expansion" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") files = [] for f in range(1, 5): fname = "file{}".format(f) files.append(os.path.join(inside, fname)) for fname in files: create_file(fname, 'Test content') self.source.p4cmd('add', fname) self.source.p4.run_reopen("-t", "ktext", files[0]) self.source.p4.run_reopen("-t", "kxtext", files[1]) self.source.p4.run_reopen("-t", "text+kmx", files[2]) self.source.p4.run_reopen("-t", "ktext+xkm", files[3]) self.source.p4cmd('submit', '-d', 'File(s) added') self.run_P4Transfer("--nokeywords") newFiles = self.target.p4cmd('files', "//...") self.assertEqual(len(newFiles), 4, "Target does not have exactly 4 files") self.assertEqual(newFiles[0]['type'], "text", "File1 does not have type text, but {}".format(newFiles[0]['type'])) self.assertEqual(newFiles[1]['type'], "xtext", "File2 does not have type xtext, but {}".format(newFiles[1]['type'])) self.assertEqual(newFiles[2]['type'], "text+mx", "File3 does not have type text+mx, but {}".format(newFiles[2]['type'])) self.assertEqual(newFiles[3]['type'], "text+mx", "File4 does not have type text+xm, but {}".format(newFiles[3]['type'])) fname = files[0] self.source.p4cmd('edit', "-t", "+k", fname) append_to_file(fname, "More stuff") self.source.p4cmd('submit', '-d', 'File edited') self.run_P4Transfer("--nokeywords") files = self.target.p4cmd('files', "//depot/import/file1") self.assertEqual(files[0]['type'], "text", "File1 does not have type text, but {}".format(files[0]['type'])) def testTempobjFiletype(self): """Tests for files with no content""" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") inside_file2 = os.path.join(inside, "inside_file2") create_file(inside_file1, "Test content") create_file(inside_file2, "Test content") self.source.p4cmd('add', '-t', 'text+S2', inside_file1) self.source.p4cmd('add', '-t', 'binary+S', inside_file2) self.source.p4cmd('submit', '-d', 'files added') self.source.p4cmd('edit', inside_file1) append_to_file(inside_file1, 'New text') self.source.p4cmd('delete', inside_file2) self.source.p4cmd('submit', '-d', 'version 2') self.source.p4cmd('edit', inside_file1) append_to_file(inside_file1, 'More text') self.source.p4cmd('submit', '-d', 'version 3') self.run_P4Transfer() self.assertCounters(3, 3) filelog = self.target.p4.run_filelog('//depot/import/inside_file1') revisions = filelog[0].revisions self.logger.debug('test:', revisions) self.assertEqual(len(revisions), 3) for rev in revisions: self.logger.debug('test:', rev.rev, rev.action, rev.digest) self.logger.debug(self.target.p4.run_print('//depot/import/inside_file1#%s' % rev.rev)) def testBackoutMove(self): """In this test we move a file and then rollback to the previous changelist way that P4V does - this does an 'add -d'""" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, "Test content") self.source.p4cmd('add', inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') inside_file2 = os.path.join(inside, "inside_file2") self.source.p4cmd('edit', inside_file1) self.source.p4.run_move(inside_file1, inside_file2) self.source.p4cmd('submit', '-d', 'moved inside_file1 to inside_file2') self.source.p4.run_sync('-f', '%s#1' % inside_file1) self.source.p4cmd('add', '-d', inside_file1) self.source.p4cmd('submit', '-d', 'new inside_file1') self.source.p4cmd('edit', inside_file1) create_file(inside_file1, "Different test content") self.source.p4cmd('submit', '-d', 'changed inside_file1 again') self.run_P4Transfer() self.assertCounters(4, 4) filelog = self.target.p4.run_filelog('//depot/import/inside_file1') revisions = filelog[0].revisions self.logger.debug('test:', revisions) self.assertEqual(len(revisions), 4) for rev in revisions: self.logger.debug('test:', rev.rev, rev.action, rev.digest) self.logger.debug(self.target.p4.run_print('//depot/import/inside_file1#%s' % rev.rev)) for rev in self.source.p4.run_filelog('//depot/inside/inside_file1')[0].revisions: self.logger.debug('test-src:', rev.rev, rev.action, rev.digest) self.logger.debug(self.source.p4.run_print('//depot/inside/inside_file1#%s' % rev.rev)) self.assertEqual(revisions[3].action, "add") self.assertEqual(revisions[1].action, "add") self.assertEqual(revisions[0].action, "edit") # 2 latest revisions should not be the same, but the revisions before and after back out # should be. self.assertEqual(revisions[1].digest, revisions[3].digest) self.assertNotEqual(revisions[0].digest, revisions[1].digest) def testAddProblem(self): "Trying to reproduce problem reported where a branch required a resolve" self.setupTransfer() inside = localDirectory(self.source.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1") create_file(inside_file1, 'Test content') self.source.p4cmd('add', "-t", "xtext", inside_file1) self.source.p4cmd('submit', '-d', 'inside_file1 added') self.source.p4cmd('edit', inside_file1) with open(inside_file1, 'a') as f: print("more content", file=f) self.source.p4cmd('submit', '-d', 'inside_file1 updated') self.source.p4cmd('edit', inside_file1) with open(inside_file1, 'a') as f: print("even more content", file=f) self.source.p4cmd('submit', '-d', 'inside_file1 updated (again)') inside_file2 = os.path.join(inside, "inside_file2") self.source.p4cmd('integrate', inside_file1, inside_file2) self.source.p4cmd('submit', '-d', 'branched into inside_file2') self.run_P4Transfer() changes = self.target.p4cmd('changes', ) self.assertEqual(len(changes), 4, "Target does not have exactly four changes") files = self.target.p4cmd('files', '//depot/...') self.assertEqual(len(files), 2, "Target does not have exactly two files") self.assertCounters(4, 4) if __name__ == '__main__': unittest.main()