# patch.py - patch file parsing routines

#

# Copyright 2006 Brendan Cully <brendan@kublai.com>

# Copyright 2007 Chris Mason <chris.mason@oracle.com>

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

from __future__ import annotations

import collections

import contextlib

import copy

import os

import re

import shutil

import typing

import zlib

from .i18n import _

from .node import (

    hex,

    sha1nodeconstants,

    short,

)

from . import (

    copies,

    diffhelper,

    diffutil,

    encoding,

    error,

    mail,

    mdiff,

    pathutil,

    pycompat,

    scmutil,

    similar,

    util,

    vfs as vfsmod,

)

from .utils import (

    dateutil,

    hashutil,

    procutil,

    stringutil,

)

if typing.TYPE_CHECKING:

    from typing import (

        Any,

        Iterator,

    )

stringio = util.stringio

gitre = re.compile(br'diff --git a/(.*) b/(.*)')

tabsplitter = re.compile(br'(\t+|[^\t]+)')

wordsplitter = re.compile(

    br'(\t+| +|[a-zA-Z0-9_\x80-\xff]+|[^ \ta-zA-Z0-9_\x80-\xff])'

)

PatchError = error.PatchError

PatchParseError = error.PatchParseError

PatchApplicationError = error.PatchApplicationError

# public functions

def split(stream):

    '''return an iterator of individual patches from a stream'''

    def isheader(line, inheader):

        if inheader and line.startswith((b' ', b'\t')):

            # continuation

            return True

        if line.startswith((b' ', b'-', b'+')):

            # diff line - don't check for header pattern in there

            return False

        l = line.split(b': ', 1)

        return len(l) == 2 and b' ' not in l[0]

    def chunk(lines):

        return stringio(b''.join(lines))

    def hgsplit(stream, cur):

        inheader = True

        for line in stream:

            if not line.strip():

                inheader = False

            if not inheader and line.startswith(b'# HG changeset patch'):

                yield chunk(cur)

                cur = []

                inheader = True

            cur.append(line)

        if cur:

            yield chunk(cur)

    def mboxsplit(stream, cur):

        for line in stream:

            if line.startswith(b'From '):

                yield from split(chunk(cur[1:]))

                cur = []

            cur.append(line)

        if cur:

            yield from split(chunk(cur[1:]))

    def mimesplit(stream, cur):

        def msgfp(m):

            fp = stringio()

            # pytype: disable=wrong-arg-types

            g = mail.Generator(fp, mangle_from_=False)

            # pytype: enable=wrong-arg-types

            g.flatten(m)

            fp.seek(0)

            return fp

        for line in stream:

            cur.append(line)

        c = chunk(cur)

        m = mail.parse(c)

        if not m.is_multipart():

            yield msgfp(m)

        else:

            ok_types = ('text/plain', 'text/x-diff', 'text/x-patch')

            for part in m.walk():

                ct = part.get_content_type()

                if ct not in ok_types:

                    continue

                yield msgfp(part)

    def headersplit(stream, cur):

        inheader = False

        for line in stream:

            if not inheader and isheader(line, inheader):

                yield chunk(cur)

                cur = []

                inheader = True

            if inheader and not isheader(line, inheader):

                inheader = False

            cur.append(line)

        if cur:

            yield chunk(cur)

    def remainder(cur):

        yield chunk(cur)

    class fiter:

        def __init__(self, fp):

            self.fp = fp

        def __iter__(self):

            return self

        def next(self):

            l = self.fp.readline()

            if not l:

                raise StopIteration

            return l

        __next__ = next

    inheader = False

    cur = []

    mimeheaders = [b'content-type']

    if not hasattr(stream, 'next'):

        # http responses, for example, have readline but not next

        stream = fiter(stream)

    for line in stream:

        cur.append(line)

        if line.startswith(b'# HG changeset patch'):

            return hgsplit(stream, cur)

        elif line.startswith(b'From '):

            return mboxsplit(stream, cur)

        elif isheader(line, inheader):

            inheader = True

            if line.split(b':', 1)[0].lower() in mimeheaders:

                # let email parser handle this

                return mimesplit(stream, cur)

        elif line.startswith(b'--- ') and inheader:

            # No evil headers seen by diff start, split by hand

            return headersplit(stream, cur)

        # Not enough info, keep reading

    # if we are here, we have a very plain patch

    return remainder(cur)

## Some facility for extensible patch parsing:

# list of pairs ("header to match", "data key")

patchheadermap = [

    (b'Date', b'date'),

    (b'Branch', b'branch'),

    (b'Node ID', b'nodeid'),

]

@contextlib.contextmanager

def extract(ui, fileobj):

    """extract patch from data read from fileobj.

    patch can be a normal patch or contained in an email message.

    return a dictionary. Standard keys are:

      - filename,

      - message,

      - user,

      - date,

      - branch,

      - node,

      - p1,

      - p2.

    Any item can be missing from the dictionary. If filename is missing,

    fileobj did not contain a patch. Caller must unlink filename when done."""

    fd, tmpname = pycompat.mkstemp(prefix=b'hg-patch-')

    tmpfp = os.fdopen(fd, 'wb')

    try:

        yield _extract(ui, fileobj, tmpname, tmpfp)

    finally:

        tmpfp.close()

        os.unlink(tmpname)

def _extract(ui, fileobj, tmpname, tmpfp):

    # attempt to detect the start of a patch

    # (this heuristic is borrowed from quilt)

    diffre = re.compile(

        br'^(?:Index:[ \t]|diff[ \t]-|RCS file: |'

        br'retrieving revision [0-9]+(\.[0-9]+)*$|'

        br'---[ \t].*?^\+\+\+[ \t]|'

        br'\*\*\*[ \t].*?^---[ \t])',

        re.MULTILINE | re.DOTALL,

    )

    data = {}

    msg = mail.parse(fileobj)

    subject = msg['Subject'] and mail.headdecode(msg['Subject'])

    data[b'user'] = msg['From'] and mail.headdecode(msg['From'])

    if not subject and not data[b'user']:

        # Not an email, restore parsed headers if any

        subject = (

            b'\n'.join(

                b': '.join(map(encoding.strtolocal, h)) for h in msg.items()

            )

            + b'\n'

        )

    # should try to parse msg['Date']

    parents = []

    nodeid = msg['X-Mercurial-Node']

    if nodeid:

        data[b'nodeid'] = nodeid = mail.headdecode(nodeid)

        ui.debug(b'Node ID: %s\n' % nodeid)

    if subject:

        if subject.startswith(b'[PATCH'):

            pend = subject.find(b']')

            if pend >= 0:

                subject = subject[pend + 1 :].lstrip()

        subject = re.sub(br'\n[ \t]+', b' ', subject)

        ui.debug(b'Subject: %s\n' % subject)

    if data[b'user']:

        ui.debug(b'From: %s\n' % data[b'user'])

    diffs_seen = 0

    ok_types = (b'text/plain', b'text/x-diff', b'text/x-patch')

    message = b''

    for part in msg.walk():

        content_type = pycompat.bytestr(part.get_content_type())

        ui.debug(b'Content-Type: %s\n' % content_type)

        if content_type not in ok_types:

            continue

        m = diffre.search(payload)

        if m:

            hgpatch = False

            hgpatchheader = False

            ignoretext = False

            ui.debug(b'found patch at byte %d\n' % m.start(0))

            diffs_seen += 1

            cfp = stringio()

            for line in payload[: m.start(0)].splitlines():

                if line.startswith(b'# HG changeset patch') and not hgpatch:

                    ui.debug(b'patch generated by hg export\n')

                    hgpatch = True

                    hgpatchheader = True

                    # drop earlier commit message content

                    cfp.seek(0)

                    cfp.truncate()

                    subject = None

                elif hgpatchheader:

                    if line.startswith(b'# User '):

                        data[b'user'] = line[7:]

                        ui.debug(b'From: %s\n' % data[b'user'])

                    elif line.startswith(b"# Parent "):

                        parents.append(line[9:].lstrip())

                    elif line.startswith(b"# "):

                        for header, key in patchheadermap:

                            prefix = b'# %s ' % header

                            if line.startswith(prefix):

                                data[key] = line[len(prefix) :]

                                ui.debug(b'%s: %s\n' % (header, data[key]))

                    else:

                        hgpatchheader = False

                elif line == b'---':

                    ignoretext = True

                if not hgpatchheader and not ignoretext:

                    cfp.write(line)

                    cfp.write(b'\n')

            message = cfp.getvalue()

            if tmpfp:

                tmpfp.write(payload)

                if not payload.endswith(b'\n'):

                    tmpfp.write(b'\n')

        elif not diffs_seen and message and content_type == b'text/plain':

            message += b'\n' + payload

    if subject and not message.startswith(subject):

        message = b'%s\n%s' % (subject, message)

    data[b'message'] = message

    tmpfp.close()

    if parents:

        data[b'p1'] = parents.pop(0)

        if parents:

            data[b'p2'] = parents.pop(0)

    if diffs_seen:

        data[b'filename'] = tmpname

    return data

class patchmeta:

    """Patched file metadata

    'op' is the performed operation within ADD, DELETE, RENAME, MODIFY

    or COPY.  'path' is patched file path. 'oldpath' is set to the

    origin file when 'op' is either COPY or RENAME, None otherwise. If

    file mode is changed, 'mode' is a tuple (islink, isexec) where

    'islink' is True if the file is a symlink and 'isexec' is True if

    the file is executable. Otherwise, 'mode' is None.

    """

    def __init__(self, path):

        self.path = path

        self.oldpath = None

        self.mode = None

        self.op = b'MODIFY'

        self.binary = False

    def setmode(self, mode):

        islink = mode & 0o20000

        isexec = mode & 0o100

        self.mode = (islink, isexec)

    def copy(self):

        other = patchmeta(self.path)

        other.oldpath = self.oldpath

        other.mode = self.mode

        other.op = self.op

        other.binary = self.binary

        return other

    def _ispatchinga(self, afile):

        if afile == b'/dev/null':

            return self.op == b'ADD'

        return afile == b'a/' + (self.oldpath or self.path)

    def _ispatchingb(self, bfile):

        if bfile == b'/dev/null':

            return self.op == b'DELETE'

        return bfile == b'b/' + self.path

    def ispatching(self, afile, bfile):

        return self._ispatchinga(afile) and self._ispatchingb(bfile)

    def __repr__(self):

        return "<patchmeta %s %r>" % (self.op, self.path)

def readgitpatch(lr):

    """extract git-style metadata about patches from <patchname>"""

    # Filter patch for git information

    gp = None

    gitpatches = []

    for line in lr:

        line = line.rstrip(b'\r\n')

        if line.startswith(b'diff --git a/'):

            m = gitre.match(line)

            if m:

                if gp:

                    gitpatches.append(gp)

                dst = m.group(2)

                gp = patchmeta(dst)

        elif gp:

            if line.startswith(b'--- '):

                gitpatches.append(gp)

                gp = None

                continue

            if line.startswith(b'rename from '):

                gp.op = b'RENAME'

                gp.oldpath = line[12:]

            elif line.startswith(b'rename to '):

                gp.path = line[10:]

            elif line.startswith(b'copy from '):

                gp.op = b'COPY'

                gp.oldpath = line[10:]

            elif line.startswith(b'copy to '):

                gp.path = line[8:]

            elif line.startswith(b'deleted file'):

                gp.op = b'DELETE'

            elif line.startswith(b'new file mode '):

                gp.op = b'ADD'

                gp.setmode(int(line[-6:], 8))

            elif line.startswith(b'new mode '):

                gp.setmode(int(line[-6:], 8))

            elif line.startswith(b'GIT binary patch'):

                gp.binary = True

    if gp:

        gitpatches.append(gp)

    return gitpatches

class linereader:

    # simple class to allow pushing lines back into the input stream

    def __init__(self, fp):

        self.fp = fp

        self.buf = []

    def push(self, line):

        if line is not None:

            self.buf.append(line)

    def readline(self):

        if self.buf:

            l = self.buf[0]

            del self.buf[0]

            return l

        return self.fp.readline()

    def __iter__(self):

        return iter(self.readline, b'')

class abstractbackend:

    def __init__(self, ui):

        self.ui = ui

    def getfile(self, fname):

        """Return target file data and flags as a (data, (islink,

        isexec)) tuple. Data is None if file is missing/deleted.

        """

        raise NotImplementedError

    def setfile(self, fname, data, mode, copysource):

        """Write data to target file fname and set its mode. mode is a