comparison mercurial/patch.py @ 14348:c1c719103392

patch: extract fs access from patchfile into fsbackend

Most filesystem calls are already isolated in patchfile but this is not enough:
renames are performed before patchfile is available and some chmod calls are
even done outside of the applydiff call. Once all these calls are extracted
into a backend class, we can provide cleaner APIs to write to a working
directory context directly into the repository.

14347:e8debe1eb255

            l = self.readline()
            if not l:
                break
            yield l

# @@ -start,len +start,len @@ or @@ -start +start @@ if len is 1
unidesc = re.compile('@@ -(\d+)(,(\d+))? \+(\d+)(,(\d+))? @@')
contextdesc = re.compile('(---|\*\*\*) (\d+)(,(\d+))? (---|\*\*\*)')
eolmodes = ['strict', 'crlf', 'lf', 'auto']

class patchfile(object):
    def __init__(self, ui, fname, opener, missing=False, eolmode='strict'):
        self.fname = fname
        self.eolmode = eolmode
        self.eol = None
        self.opener = opener
        self.ui = ui
        self.lines = []
        self.exists = False
        self.missing = missing
        if not missing:
            try:
                self.lines = self.readlines(fname)
                self.exists = True
            except IOError:
                pass
        else:
            self.ui.warn(_("unable to find '%s' for patching\n") % self.fname)

        self.hash = {}
        self.dirty = False
        self.offset = 0
        self.skew = 0
        self.rej = []
        self.fileprinted = False
        self.printfile(False)
        self.hunks = 0

    def readlines(self, fname):
        if os.path.islink(fname):
            return [os.readlink(fname)]
        fp = self.opener(fname, 'r')
        try:
            lr = linereader(fp, self.eolmode != 'strict')
            lines = list(lr)
            self.eol = lr.eol
            return lines
        finally:
            fp.close()

    def writelines(self, fname, lines):
        # Ensure supplied data ends in fname, being a regular file or

            except OSError, e:
                if e.errno != errno.ENOENT:
                    raise
            fp = self.opener(fname, 'w')
        try:
            if self.eolmode == 'auto':
                eol = self.eol
            elif self.eolmode == 'crlf':
                eol = '\r\n'
            else:
                eol = '\n'

            if self.eolmode != 'strict' and eol and eol != '\n':
                for l in lines:
                    if l and l[-1] == '\n':
                        l = l[:-1] + eol
                    fp.write(l)
            else:
                fp.writelines(lines)
            if islink:
                self.opener.symlink(fp.getvalue(), fname)
            if st_mode is not None:
                os.chmod(fname, st_mode)
        finally:
            fp.close()

    def unlink(self, fname):
        os.unlink(fname)

    def printfile(self, warn):
        if self.fileprinted:
            return
        if warn or self.ui.verbose:

    def write_rej(self):
        # our rejects are a little different from patch(1).  This always
        # creates rejects in the same form as the original patch.  A file
        # header is inserted so that you can run the reject through patch again
        # without having to type the filename.

        if not self.rej:
            return

        fname = self.fname + ".rej"
        self.ui.warn(
            _("%d out of %d hunks FAILED -- saving rejects to file %s\n") %
            (len(self.rej), self.hunks, fname))

        fp = self.opener(fname, 'w')
        fp.writelines(self.makerejlines(self.fname))
        fp.close()

    def apply(self, h):
        if not h.complete():
            raise PatchError(_("bad hunk #%d %s (%d %d %d %d)") %
                            (h.number, h.desc, len(h.a), h.lena, len(h.b),

            self.rej.append(h)
            return -1

        if isinstance(h, binhunk):
            if h.rmfile():
                self.unlink(self.fname)
            else:
                self.lines[:] = h.new()
                self.offset += len(h.new())
                self.dirty = True
            return 0

        # if there's skew we want to emit the "(offset %d lines)" even
        # when the hunk cleanly applies at start + skew, so skip the
        # fast case code
        if self.skew == 0 and diffhelpers.testhunk(old, self.lines, start) == 0:
            if h.rmfile():
                self.unlink(self.fname)
            else:
                self.lines[start : start + h.lena] = h.new()
                self.offset += h.lenb - h.lena
                self.dirty = True
            return 0

def _applydiff(ui, fp, patcher, copyfn, changed, strip=1, eolmode='strict'):
    rejects = 0
    err = 0
    current_file = None
    cwd = os.getcwd()
    opener = scmutil.opener(cwd)

    for state, values in iterhunks(fp):
        if state == 'hunk':
            if not current_file:
                continue

                rejects += current_file.close()
            afile, bfile, first_hunk = values
            try:
                current_file, missing = selectfile(afile, bfile,
                                                   first_hunk, strip)
                current_file = patcher(ui, current_file, opener,
                                       missing=missing, eolmode=eolmode)
            except PatchError, inst:
                ui.warn(str(inst) + '\n')
                current_file = None
                rejects += 1

14348:c1c719103392

            l = self.readline()
            if not l:
                break
            yield l

class abstractbackend(object):
    def __init__(self, ui):
        self.ui = ui

    def readlines(self, fname):
        """Return target file lines, or its content as a single line
        for symlinks.
        """
        raise NotImplementedError

    def writelines(self, fname, lines):
        """Write lines to target file."""
        raise NotImplementedError

    def unlink(self, fname):
        """Unlink target file."""
        raise NotImplementedError

    def writerej(self, fname, failed, total, lines):
        """Write rejected lines for fname. total is the number of hunks
        which failed to apply and total the total number of hunks for this
        files.
        """
        pass

class fsbackend(abstractbackend):
    def __init__(self, ui, opener):
        super(fsbackend, self).__init__(ui)
        self.opener = opener

    def readlines(self, fname):
        if os.path.islink(fname):
            return [os.readlink(fname)]
        fp = self.opener(fname, 'r')
        try:
            return list(fp)
        finally:
            fp.close()

    def writelines(self, fname, lines):
        # Ensure supplied data ends in fname, being a regular file or

            except OSError, e:
                if e.errno != errno.ENOENT:
                    raise
            fp = self.opener(fname, 'w')
        try:
            fp.writelines(lines)
            if islink:
                self.opener.symlink(fp.getvalue(), fname)
            if st_mode is not None:
                os.chmod(fname, st_mode)
        finally:
            fp.close()

    def unlink(self, fname):
        os.unlink(fname)

    def writerej(self, fname, failed, total, lines):
        fname = fname + ".rej"
        self.ui.warn(
            _("%d out of %d hunks FAILED -- saving rejects to file %s\n") %
            (failed, total, fname))
        fp = self.opener(fname, 'w')
        fp.writelines(lines)
        fp.close()

# @@ -start,len +start,len @@ or @@ -start +start @@ if len is 1
unidesc = re.compile('@@ -(\d+)(,(\d+))? \+(\d+)(,(\d+))? @@')
contextdesc = re.compile('(---|\*\*\*) (\d+)(,(\d+))? (---|\*\*\*)')
eolmodes = ['strict', 'crlf', 'lf', 'auto']

class patchfile(object):
    def __init__(self, ui, fname, backend, missing=False, eolmode='strict'):
        self.fname = fname
        self.eolmode = eolmode
        self.eol = None
        self.backend = backend
        self.ui = ui
        self.lines = []
        self.exists = False
        self.missing = missing
        if not missing:
            try:
                self.lines = self.backend.readlines(fname)
                if self.lines:
                    # Normalize line endings
                    if self.lines[0].endswith('\r\n'):
                        self.eol = '\r\n'
                    elif self.lines[0].endswith('\n'):
                        self.eol = '\n'
                    if eolmode != 'strict':
                        nlines = []
                        for l in self.lines:
                            if l.endswith('\r\n'):
                                l = l[:-2] + '\n'
                            nlines.append(l)
                        self.lines = nlines
                self.exists = True
            except IOError:
                pass
        else:
            self.ui.warn(_("unable to find '%s' for patching\n") % self.fname)

        self.hash = {}
        self.dirty = 0
        self.offset = 0
        self.skew = 0
        self.rej = []
        self.fileprinted = False
        self.printfile(False)
        self.hunks = 0

    def writelines(self, fname, lines):
        if self.eolmode == 'auto':
            eol = self.eol
        elif self.eolmode == 'crlf':
            eol = '\r\n'
        else:
            eol = '\n'

        if self.eolmode != 'strict' and eol and eol != '\n':
            rawlines = []
            for l in lines:
                if l and l[-1] == '\n':
                    l = l[:-1] + eol
                rawlines.append(l)
            lines = rawlines

        self.backend.writelines(fname, lines)

    def printfile(self, warn):
        if self.fileprinted:
            return
        if warn or self.ui.verbose:

    def write_rej(self):
        # our rejects are a little different from patch(1).  This always
        # creates rejects in the same form as the original patch.  A file
        # header is inserted so that you can run the reject through patch again
        # without having to type the filename.
        if not self.rej:
            return
        self.backend.writerej(self.fname, len(self.rej), self.hunks,
                              self.makerejlines(self.fname))

    def apply(self, h):
        if not h.complete():
            raise PatchError(_("bad hunk #%d %s (%d %d %d %d)") %
                            (h.number, h.desc, len(h.a), h.lena, len(h.b),

            self.rej.append(h)
            return -1

        if isinstance(h, binhunk):
            if h.rmfile():
                self.backend.unlink(self.fname)
            else:
                self.lines[:] = h.new()
                self.offset += len(h.new())
                self.dirty = True
            return 0

        # if there's skew we want to emit the "(offset %d lines)" even
        # when the hunk cleanly applies at start + skew, so skip the
        # fast case code
        if self.skew == 0 and diffhelpers.testhunk(old, self.lines, start) == 0:
            if h.rmfile():
                self.backend.unlink(self.fname)
            else:
                self.lines[start : start + h.lena] = h.new()
                self.offset += h.lenb - h.lena
                self.dirty = True
            return 0

def _applydiff(ui, fp, patcher, copyfn, changed, strip=1, eolmode='strict'):
    rejects = 0
    err = 0
    current_file = None
    cwd = os.getcwd()
    backend = fsbackend(ui, scmutil.opener(cwd))

    for state, values in iterhunks(fp):
        if state == 'hunk':
            if not current_file:
                continue

                rejects += current_file.close()
            afile, bfile, first_hunk = values
            try:
                current_file, missing = selectfile(afile, bfile,
                                                   first_hunk, strip)
                current_file = patcher(ui, current_file, backend,
                                       missing=missing, eolmode=eolmode)
            except PatchError, inst:
                ui.warn(str(inst) + '\n')
                current_file = None
                rejects += 1