# filemerge.py - file-level merge handling for Mercurial

#

# Copyright 2006, 2007, 2008 Olivia Mackall <olivia@selenic.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 contextlib

import os

import re

import shutil

from .i18n import _

from .node import (

    hex,

    short,

)

from . import (

    encoding,

    error,

    formatter,

    match,

    pycompat,

    registrar,

    scmutil,

    simplemerge,

    tagmerge,

    templatekw,

    templater,

    templateutil,

    util,

)

from .utils import (

    procutil,

    stringutil,

)

def _toolstr(ui, tool, part, *args):

    return ui.config(b"merge-tools", tool + b"." + part, *args)

def _toolbool(ui, tool, part, *args):

    return ui.configbool(b"merge-tools", tool + b"." + part, *args)

def _toollist(ui, tool, part):

    return ui.configlist(b"merge-tools", tool + b"." + part)

internals = {}

# Merge tools to document.

internalsdoc = {}

internaltool = registrar.internalmerge()

# internal tool merge types

nomerge = internaltool.nomerge

mergeonly = internaltool.mergeonly  # just the full merge, no premerge

fullmerge = internaltool.fullmerge  # both premerge and merge

# IMPORTANT: keep the last line of this prompt very short ("What do you want to

# do?") because of issue6158, ideally to <40 English characters (to allow other

# languages that may take more columns to still have a chance to fit in an

# 80-column screen).

_localchangedotherdeletedmsg = _(

    b"file '%(fd)s' was deleted in other%(o)s but was modified in local%(l)s.\n"

    b"You can use (c)hanged version, (d)elete, or leave (u)nresolved.\n"

    b"What do you want to do?"

    b"$$ &Changed $$ &Delete $$ &Unresolved"

)

_otherchangedlocaldeletedmsg = _(

    b"file '%(fd)s' was deleted in local%(l)s but was modified in other%(o)s.\n"

    b"You can use (c)hanged version, leave (d)eleted, or leave (u)nresolved.\n"

    b"What do you want to do?"

    b"$$ &Changed $$ &Deleted $$ &Unresolved"

)

class absentfilectx:

    """Represents a file that's ostensibly in a context but is actually not

    present in it.

    This is here because it's very specific to the filemerge code for now --

    other code is likely going to break with the values this returns."""

    def __init__(self, ctx, f):

        self._ctx = ctx

        self._f = f

    def __bytes__(self):

        return b'absent file %s@%s' % (self._f, self._ctx)

    def path(self):

        return self._f

    def size(self):

        return None

    def data(self):

        return None

    def filenode(self):

        return self._ctx.repo().nullid

    _customcmp = True

    def cmp(self, fctx):

        """compare with other file context

        returns True if different from fctx.

        """

        return not (

            fctx.isabsent()

            and fctx.changectx() == self.changectx()

            and fctx.path() == self.path()

        )

    def flags(self):

        return b''

    def changectx(self):

        return self._ctx

    def isbinary(self):

        return False

    def isabsent(self):

        return True

def _findtool(ui, tool):

    if tool in internals:

        return tool

    cmd = _toolstr(ui, tool, b"executable", tool)

    if cmd.startswith(b'python:'):

        return cmd

    return findexternaltool(ui, tool)

def _quotetoolpath(cmd):

    if cmd.startswith(b'python:'):

        return cmd

    return procutil.shellquote(cmd)

def findexternaltool(ui, tool):

    for kn in (b"regkey", b"regkeyalt"):

        k = _toolstr(ui, tool, kn)

        if not k:

            continue

        p = util.lookupreg(k, _toolstr(ui, tool, b"regname"))

        if p:

            p = procutil.findexe(p + _toolstr(ui, tool, b"regappend"))

            if p:

                return p

    exe = _toolstr(ui, tool, b"executable", tool)

    return procutil.findexe(util.expandpath(exe))

def _picktool(repo, ui, path, binary, symlink, changedelete):

    strictcheck = ui.configbool(b'merge', b'strict-capability-check')

    def hascapability(tool, capability, strict=False):

        if tool in internals:

            return strict and internals[tool].capabilities.get(capability)

        return _toolbool(ui, tool, capability)

    def supportscd(tool):

        return tool in internals and internals[tool].mergetype == nomerge

    def check(tool, pat, symlink, binary, changedelete):

        tmsg = tool

        if pat:

            tmsg = _(b"%s (for pattern %s)") % (tool, pat)

        if not _findtool(ui, tool):

            if pat:  # explicitly requested tool deserves a warning

                ui.warn(_(b"couldn't find merge tool %s\n") % tmsg)

            else:  # configured but non-existing tools are more silent

                ui.note(_(b"couldn't find merge tool %s\n") % tmsg)

        elif symlink and not hascapability(tool, b"symlink", strictcheck):

            ui.warn(_(b"tool %s can't handle symlinks\n") % tmsg)

        elif binary and not hascapability(tool, b"binary", strictcheck):

            ui.warn(_(b"tool %s can't handle binary\n") % tmsg)

        elif changedelete and not supportscd(tool):

            # the nomerge tools are the only tools that support change/delete

            # conflicts

            pass

        elif not procutil.gui() and _toolbool(ui, tool, b"gui"):

            ui.warn(_(b"tool %s requires a GUI\n") % tmsg)

        else:

            return True

        return False

    # internal config: ui.forcemerge

    # forcemerge comes from command line arguments, highest priority

    force = ui.config(b'ui', b'forcemerge')

    if force:

        toolpath = _findtool(ui, force)

        if changedelete and not supportscd(toolpath):

            return b":prompt", None

        else:

            if toolpath:

                return (force, _quotetoolpath(toolpath))

            else:

                # mimic HGMERGE if given tool not found

                return (force, force)

    # HGMERGE takes next precedence

    hgmerge = encoding.environ.get(b"HGMERGE")

    if hgmerge:

        if changedelete and not supportscd(hgmerge):

            return b":prompt", None

        else:

            return (hgmerge, hgmerge)

    # then patterns

    # whether binary capability should be checked strictly

    binarycap = binary and strictcheck

    for pat, tool in ui.configitems(b"merge-patterns"):

        mf = match.match(repo.root, b'', [pat])

        if mf(path) and check(tool, pat, symlink, binarycap, changedelete):

            if binary and not hascapability(tool, b"binary", strict=True):

                ui.warn(

                    _(

                        b"warning: check merge-patterns configurations,"

                        b" if %r for binary file %r is unintentional\n"

                        b"(see 'hg help merge-tools'"

                        b" for binary files capability)\n"

                    )

                    % (pycompat.bytestr(tool), pycompat.bytestr(path))

                )

            toolpath = _findtool(ui, tool)

            return (tool, _quotetoolpath(toolpath))

    # then merge tools

    tools = {}

    disabled = set()

    for k, v in ui.configitems(b"merge-tools"):

        t = k.split(b'.')[0]

        if t not in tools:

            tools[t] = int(_toolstr(ui, t, b"priority"))

        if _toolbool(ui, t, b"disabled"):

            disabled.add(t)

    names = tools.keys()

    tools = sorted(

        [(-p, tool) for tool, p in tools.items() if tool not in disabled]

    )

    uimerge = ui.config(b"ui", b"merge")

    if uimerge:

        # external tools defined in uimerge won't be able to handle

        # change/delete conflicts

        if check(uimerge, path, symlink, binary, changedelete):

            if uimerge not in names and not changedelete:

                return (uimerge, uimerge)

            tools.insert(0, (None, uimerge))  # highest priority

    tools.append((None, b"hgmerge"))  # the old default, if found

    for p, t in tools:

        if check(t, None, symlink, binary, changedelete):

            toolpath = _findtool(ui, t)

            return (t, _quotetoolpath(toolpath))

    # internal merge or prompt as last resort

    if symlink or binary or changedelete:

        if not changedelete and len(tools):

            # any tool is rejected by capability for symlink or binary

            ui.warn(_(b"no tool found to merge %s\n") % path)

        return b":prompt", None

    return b":merge", None

def _eoltype(data):

    """Guess the EOL type of a file"""

    if b'\0' in data:  # binary

        return None

    if b'\r\n' in data:  # Windows

        return b'\r\n'

    if b'\r' in data:  # Old Mac

        return b'\r'

    if b'\n' in data:  # UNIX

        return b'\n'

    return None  # unknown

def _matcheol(file, backup):

    """Convert EOL markers in a file to match origfile"""

    tostyle = _eoltype(backup.data())  # No repo.wread filters?

    if tostyle:

        data = util.readfile(file)

        style = _eoltype(data)

        if style:

            newdata = data.replace(style, tostyle)

            if newdata != data:

                util.writefile(file, newdata)

@internaltool(b'prompt', nomerge)

def _iprompt(repo, mynode, local, other, base, toolconf):

    """Asks the user which of the local `p1()` or the other `p2()` version to

    keep as the merged version."""

    ui = repo.ui

    fd = local.fctx.path()

    uipathfn = scmutil.getuipathfn(repo)

    # Avoid prompting during an in-memory merge since it doesn't support merge

    # conflicts.

    if local.fctx.changectx().isinmemory():

        raise error.InMemoryMergeConflictsError(

            b'in-memory merge does not support file conflicts'

        )

    prompts = partextras([local.label, other.label])

    prompts[b'fd'] = uipathfn(fd)

    try:

        if other.fctx.isabsent():

            index = ui.promptchoice(_localchangedotherdeletedmsg % prompts, 2)

            choice = [b'local', b'other', b'unresolved'][index]

        elif local.fctx.isabsent():

            index = ui.promptchoice(_otherchangedlocaldeletedmsg % prompts, 2)

            choice = [b'other', b'local', b'unresolved'][index]

        else:

            # IMPORTANT: keep the last line of this prompt ("What do you want to

            # do?") very short, see comment next to _localchangedotherdeletedmsg

            # at the top of the file for details.

            index = ui.promptchoice(

                _(

                    b"file '%(fd)s' needs to be resolved.\n"

                    b"You can keep (l)ocal%(l)s, take (o)ther%(o)s, or leave "

                    b"(u)nresolved.\n"

                    b"What do you want to do?"

                    b"$$ &Local $$ &Other $$ &Unresolved"

                )

                % prompts,

                2,

            )

            choice = [b'local', b'other', b'unresolved'][index]

        if choice == b'other':

            return _iother(repo, mynode, local, other, base, toolconf)

        elif choice == b'local':

            return _ilocal(repo, mynode, local, other, base, toolconf)

        elif choice == b'unresolved':

            return _ifail(repo, mynode, local, other, base, toolconf)

    except error.ResponseExpected:

        ui.write(b"\n")

        return _ifail(repo, mynode, local, other, base, toolconf)

@internaltool(b'local', nomerge)

def _ilocal(repo, mynode, local, other, base, toolconf):

    """Uses the local `p1()` version of files as the merged version."""

    return 0, local.fctx.isabsent()

@internaltool(b'other', nomerge)

def _iother(repo, mynode, local, other, base, toolconf):

    """Uses the other `p2()` version of files as the merged version."""

    if other.fctx.isabsent():

        # local changed, remote deleted -- 'deleted' picked

        _underlyingfctxifabsent(local.fctx).remove()

        deleted = True

    else:

        _underlyingfctxifabsent(local.fctx).write(

            other.fctx.data(), other.fctx.flags()

        )

        deleted = False

    return 0, deleted

@internaltool(b'fail', nomerge)

def _ifail(repo, mynode, local, other, base, toolconf):

    """

    Rather than attempting to merge files that were modified on both

    branches, it marks them as unresolved. The resolve command must be

    used to resolve these conflicts."""

    # for change/delete conflicts write out the changed version, then fail

    if local.fctx.isabsent():

        _underlyingfctxifabsent(local.fctx).write(

            other.fctx.data(), other.fctx.flags()

        )

    return 1, False

def _underlyingfctxifabsent(filectx):

    """Sometimes when resolving, our fcd is actually an absentfilectx, but

    we want to write to it (to do the resolve). This helper returns the

    underyling workingfilectx in that case.

    """

    if filectx.isabsent():

        return filectx.changectx()[filectx.path()]

    else:

        return filectx

def _verifytext(input, ui):

    """verifies that text is non-binary"""

    if stringutil.binary(input.text()):

        msg = _(b"%s looks like a binary file.") % input.fctx.path()

        ui.warn(_(b'warning: %s\n') % msg)

        raise error.Abort(msg)

def _premerge(repo, local, other, base, toolconf):

    tool, toolpath, binary, symlink, scriptfn = toolconf

    if symlink or local.fctx.isabsent() or other.fctx.isabsent():

        return 1

    ui = repo.ui

    validkeep = [b'keep', b'keep-merge3', b'keep-mergediff']

    # do we attempt to simplemerge first?

    try:

        premerge = _toolbool(ui, tool, b"premerge", not binary)

    except error.ConfigError:

        premerge = _toolstr(ui, tool, b"premerge", b"").lower()

        if premerge not in validkeep:

            _valid = b', '.join([b"'" + v + b"'" for v in validkeep])

            raise error.ConfigError(

                _(b"%s.premerge not valid ('%s' is neither boolean nor %s)")

                % (tool, premerge, _valid)

            )

    if premerge:

        mode = b'merge'

        if premerge == b'keep-mergediff':

            mode = b'mergediff'

        elif premerge == b'keep-merge3':

            mode = b'merge3'

        if any(

            stringutil.binary(input.text()) for input in (local, base, other)

        ):

            return 1  # continue merging

        merged_text, conflicts = simplemerge.simplemerge(

            local, base, other, mode=mode

        )

        if not conflicts or premerge in validkeep:

            # fcd.flags() already has the merged flags (done in

            # mergestate.resolve())

            local.fctx.write(merged_text, local.fctx.flags())

        if not conflicts:

            ui.debug(b" premerge successful\n")

            return 0

    return 1  # continue merging

def _mergecheck(repo, mynode, fcd, fco, fca, toolconf):

    tool, toolpath, binary, symlink, scriptfn = toolconf

    uipathfn = scmutil.getuipathfn(repo)

    if symlink:

        repo.ui.warn(

            _(b'warning: internal %s cannot merge symlinks for %s\n')

            % (tool, uipathfn(fcd.path()))

        )

        return False

    if fcd.isabsent() or fco.isabsent():

        repo.ui.warn(

            _(

                b'warning: internal %s cannot merge change/delete '

                b'conflict for %s\n'

            )

            % (tool, uipathfn(fcd.path()))

        )

        return False

    return True

def _merge(repo, local, other, base, mode):

    """

    Uses the internal non-interactive simple merge algorithm for merging