# Copyright (C) 2005 Aaron Bentley # # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA import sys, os from subprocess import Popen, PIPE from bzrlib.transport import get_transport from urlparse import urlsplit, urlunsplit from bzrlib.workingtree import WorkingTree import bzrlib.add import bzrlib.transform import bzrlib.branch import bzrlib.bzrdir class BzrPatchProc: """This class process the bzr patch TODO: error handling execute attribute support """ def __init__(self, tt, tree, bzrdir): self.link = None self.changes = [] self.mode = None self.command = None self.target_id = None self.tt = tt self.tree = tree self.skipchange = True self.bzrdir = bzrdir self.branch = self.bzrdir.open_branch( ) self.repo = self.bzrdir.find_repository( ) h = self.branch.revision_history( ) self.inv = self.repo.get_inventory(h[-1]) def get_transid(self, path): return self.tt.trans_id_tree_path(path) def _extractname(self, s): x = s[0] i = 1 ls = len(s) while i < ls: if s[i] == '\\': assert(i+1 < ls ) i += 2 continue if s[i] == x: return s[1:i],i+1 i += 1 assert(False) def extractname(self, s): space=-1 for i in range(0,3): space = s.find(" ",space+1) # find the 2nd space assert(space>=0) return self._extractname(s[space+1:])[0] def extractnames(self, s): space=-1 for i in range(0,3): space = s.find(" ",space+1) # find the 2nd space assert(space>=0) s=s[space+1:] source, pos = self._extractname(s) assert( pos +4 < len(s) ) dest, dummy = self._extractname(s[pos+4:]) return source,dest def flush(self): self.process( ) def do_patch(self, changes, targetid): # apply change to target_id if 1: cmd = ['patch', '-o', self.tt._limbo_name(targetid), '--directory', self.branch.base, '-p1'] #sys.stdout.write("# %r\n"%cmd) child_proc = Popen(cmd, stdin=PIPE) for line in changes: #sys.stdout.write("# %s\n"%line) child_proc.stdin.write(line+'\n') child_proc.stdin.close() r = child_proc.wait() self.changes = [] def process(self, cmd = None): while cmd and len(cmd) and cmd[-1] in "\n\r": cmd = cmd[:-1] if ( cmd == None or cmd.startswith("===") ) and self.changes: assert(self.target_id) #print "********** mode=",self.mode if self.mode == "mv" or self.mode == "mod": self.tt.delete_contents(self.target_id) if self.mode == "add" or self.mode == "mod" or self.mode == "mv": self.tt.create_file([],self.target_id) self.do_patch(self.changes, self.target_id) self.target_id = None self.changes = [] self.mode = None if cmd == None: return if not cmd.startswith("==="): if self.skipchange: return assert(self.target_id) assert(self.mode) self.changes.append(cmd) return # thr target_id is set during a rename; but there is possibility # that after a rename, is not to do a change self.target_id = None # skip change self.skipchange = False self.mode = None if ( cmd.startswith("=== removed file") or cmd.startswith("=== removed directory") or cmd.startswith("=== removed symlink") ): target = self.extractname(cmd)[2:] tid = self.get_transid(target) self.tt.delete_versioned(tid) print "removing '%s'"%target self.skipchange = True self.mode = "rm" elif cmd.startswith("=== added file"): target = self.extractname(cmd)[2:] self.target_id = self.get_transid(target) #self.tt.create_file([],self.target_id) print "adding '%s'"%target self.mode = "add" elif cmd.startswith("=== modified file"): target = self.extractname(cmd)[2:] self.target_id = self.get_transid(target) self.mode = "mod" print "patching '%s'"%target elif cmd.startswith("=== added directory"): target = self.extractname(cmd)[2:] target_id = self.get_transid(target) self.tt.create_directory(target_id) print "adding directory '%s'"%target elif cmd.startswith("=== added symlink"): assert(not self.link) self.link = self.extractname(cmd)[2:] elif cmd.startswith("=== target is"): assert(self.link) target = self.extractname(cmd) link_id = self.get_transid(self.link) print "symlinking '%s' => '%s'\n"%(target, link_id) self.tt.create_symlink(target, link_id) self.link = None elif ( cmd.startswith("=== renamed symlink") or cmd.startswith("=== renamed file") or cmd.startswith("=== renamed directory") ): source,dest = self.extractnames(cmd) source = source[2:] dest = dest[2:] sourceid = self.get_transid(source) pdir,pname = os.path.split(dest) pdir_id = self.get_transid(pdir) self.tt.adjust_path(pname, pdir_id, sourceid) if cmd.startswith("=== renamed file"): self.target_id = sourceid self.mode = "mv" print "renaming '%s' => '%s'"%(source,dest) else: sys.stderr.write("Unsupported tag: '%s'\n"%cmd) def apply(self): self.tt.apply( ) def do_patch(p): tree = WorkingTree.open_containing(u'.')[0] tt = bzrlib.transform.TreeTransform(tree) bzrdir = bzrlib.bzrdir.BzrDir.open(".") bzr_tags_proc = BzrPatchProc(tt, tree, bzrdir) for l in p: # we have to remove the \n\r bzr_tags_proc.process(l) bzr_tags_proc.flush( ) bzr_tags_proc.apply( ) def patch(tree, location, strip, legacy): """Apply a patch to a branch, using patch(1). URLs may be used.""" my_file = None if location is None: my_file = sys.stdin else: for prefix in ('http://', 'sftp://', 'file://'): if not location.startswith(prefix): continue (scheme, loc, path, query, fragment) = urlsplit(location) loc_start = urlunsplit((scheme, loc, '/', '', '')) my_file = get_transport(loc_start).get(path[1:]) if my_file is None: my_file = file(location, 'rb') cmd = ['patch', '--directory', tree.basedir, '--strip', str(strip)] r = 0 if legacy: child_proc = Popen(cmd, stdin=PIPE) for line in my_file: child_proc.stdin.write(line) child_proc.stdin.close() r = child_proc.wait() else: do_patch(sys.stdin.readlines( )) return r