# merge.py - directory-level update/merge handling for Mercurial

#

# Copyright 2006, 2007 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 collections

import struct

import typing

from typing import Dict, Optional, Tuple

from .i18n import _

from .node import nullrev

from .thirdparty import attr

# Force pytype to use the non-vendored package

if typing.TYPE_CHECKING:

    # noinspection PyPackageRequirements

    import attr

from .utils import stringutil

from .dirstateutils import timestamp

from . import (

    copies,

    encoding,

    error,

    filemerge,

    match as matchmod,

    mergestate as mergestatemod,

    obsutil,

    pathutil,

    policy,

    pycompat,

    scmutil,

    subrepoutil,

    util,

    worker,

)

_pack = struct.pack

_unpack = struct.unpack

def _getcheckunknownconfig(repo, section, name):

    config = repo.ui.config(section, name)

    valid = [b'abort', b'ignore', b'warn']

    if config not in valid:

        validstr = b', '.join([b"'" + v + b"'" for v in valid])

        msg = _(b"%s.%s not valid ('%s' is none of %s)")

        msg %= (section, name, config, validstr)

        raise error.ConfigError(msg)

    return config

def _checkunknownfile(dirstate, wvfs, dircache, wctx, mctx, f, f2=None):

    if wctx.isinmemory():

        # Nothing to do in IMM because nothing in the "working copy" can be an

        # unknown file.

        #

        # Note that we should bail out here, not in ``_checkunknownfiles()``,

        # because that function does other useful work.

        return False

    if f2 is None:

        f2 = f

    return (

        wvfs.isfileorlink_checkdir(dircache, f)

        and dirstate.normalize(f) not in dirstate

        and mctx[f2].cmp(wctx[f])

    )

class _unknowndirschecker:

    """

    Look for any unknown files or directories that may have a path conflict

    with a file.  If any path prefix of the file exists as a file or link,

    then it conflicts.  If the file itself is a directory that contains any

    file that is not tracked, then it conflicts.

    Returns the shortest path at which a conflict occurs, or None if there is

    no conflict.

    """

    def __init__(self):

        # A set of paths known to be good.  This prevents repeated checking of

        # dirs.  It will be updated with any new dirs that are checked and found

        # to be safe.

        self._unknowndircache = set()

        # A set of paths that are known to be absent.  This prevents repeated

        # checking of subdirectories that are known not to exist. It will be

        # updated with any new dirs that are checked and found to be absent.

        self._missingdircache = set()

    def __call__(self, repo, wctx, f):

        if wctx.isinmemory():

            # Nothing to do in IMM for the same reason as ``_checkunknownfile``.

            return False

        # Check for path prefixes that exist as unknown files.

        for p in reversed(list(pathutil.finddirs(f))):

            if p in self._missingdircache:

                return

            if p in self._unknowndircache:

                continue

            if repo.wvfs.audit.check(p):

                if (

                    repo.wvfs.isfileorlink(p)

                    and repo.dirstate.normalize(p) not in repo.dirstate

                ):

                    return p

                if not repo.wvfs.lexists(p):

                    self._missingdircache.add(p)

                    return

                self._unknowndircache.add(p)

        # Check if the file conflicts with a directory containing unknown files.

        if repo.wvfs.audit.check(f) and repo.wvfs.isdir(f):

            # Does the directory contain any files that are not in the dirstate?

            for p, dirs, files in repo.wvfs.walk(f):

                for fn in files:

                    relf = util.pconvert(repo.wvfs.reljoin(p, fn))

                    relf = repo.dirstate.normalize(relf, isknown=True)

                    if relf not in repo.dirstate:

                        return f

        return None

def _checkunknownfiles(repo, wctx, mctx, force, mresult, mergeforce):

    """

    Considers any actions that care about the presence of conflicting unknown

    files. For some actions, the result is to abort; for others, it is to

    choose a different action.

    """

    fileconflicts = set()

    pathconflicts = set()

    warnconflicts = set()

    abortconflicts = set()

    unknownconfig = _getcheckunknownconfig(repo, b'merge', b'checkunknown')

    ignoredconfig = _getcheckunknownconfig(repo, b'merge', b'checkignored')

    pathconfig = repo.ui.configbool(

        b'experimental', b'merge.checkpathconflicts'

    )

    dircache = dict()

    dirstate = repo.dirstate

    wvfs = repo.wvfs

    if not force:

        def collectconflicts(conflicts, config):

            if config == b'abort':

                abortconflicts.update(conflicts)

            elif config == b'warn':

                warnconflicts.update(conflicts)

        checkunknowndirs = _unknowndirschecker()

        for f in mresult.files(

            (

                mergestatemod.ACTION_CREATED,

                mergestatemod.ACTION_DELETED_CHANGED,

            )

        ):

            if _checkunknownfile(dirstate, wvfs, dircache, wctx, mctx, f):

                fileconflicts.add(f)

            elif pathconfig and f not in wctx:

                path = checkunknowndirs(repo, wctx, f)

                if path is not None:

                    pathconflicts.add(path)

        for f, args, msg in mresult.getactions(

            [mergestatemod.ACTION_LOCAL_DIR_RENAME_GET]

        ):

            if _checkunknownfile(

                dirstate, wvfs, dircache, wctx, mctx, f, args[0]

            ):

                fileconflicts.add(f)

        allconflicts = fileconflicts | pathconflicts

        ignoredconflicts = {c for c in allconflicts if repo.dirstate._ignore(c)}

        unknownconflicts = allconflicts - ignoredconflicts

        collectconflicts(ignoredconflicts, ignoredconfig)

        collectconflicts(unknownconflicts, unknownconfig)

    else:

        for f, args, msg in list(

            mresult.getactions([mergestatemod.ACTION_CREATED_MERGE])

        ):

            fl2, anc = args

            different = _checkunknownfile(

                dirstate, wvfs, dircache, wctx, mctx, f

            )

            if repo.dirstate._ignore(f):

                config = ignoredconfig

            else:

                config = unknownconfig

            # The behavior when force is True is described by this table:

            #  config  different  mergeforce  |    action    backup

            #    *         n          *       |      get        n

            #    *         y          y       |     merge       -

            #   abort      y          n       |     merge       -   (1)

            #   warn       y          n       |  warn + get     y

            #  ignore      y          n       |      get        y

            #

            # (1) this is probably the wrong behavior here -- we should

            #     probably abort, but some actions like rebases currently

            #     don't like an abort happening in the middle of

            #     merge.update.

            if not different:

                mresult.addfile(

                    f,

                    mergestatemod.ACTION_GET,

                    (fl2, False),

                    b'remote created',

                )

            elif mergeforce or config == b'abort':

                mresult.addfile(

                    f,

                    mergestatemod.ACTION_MERGE,

                    (f, f, None, False, anc),

                    b'remote differs from untracked local',

                )

            elif config == b'abort':

                abortconflicts.add(f)

            else:

                if config == b'warn':

                    warnconflicts.add(f)

                mresult.addfile(

                    f,

                    mergestatemod.ACTION_GET,

                    (fl2, True),

                    b'remote created',

                )

    for f in sorted(abortconflicts):

        warn = repo.ui.warn

        if f in pathconflicts:

            if repo.wvfs.isfileorlink(f):

                warn(_(b"%s: untracked file conflicts with directory\n") % f)

            else:

                warn(_(b"%s: untracked directory conflicts with file\n") % f)

        else:

            warn(_(b"%s: untracked file differs\n") % f)

    if abortconflicts:

        raise error.StateError(

            _(

                b"untracked files in working directory "

                b"differ from files in requested revision"

            )

        )

    for f in sorted(warnconflicts):

        if repo.wvfs.isfileorlink(f):

            repo.ui.warn(_(b"%s: replacing untracked file\n") % f)

        else:

            repo.ui.warn(_(b"%s: replacing untracked files in directory\n") % f)

    def transformargs(f, args):

        backup = (

            f in fileconflicts

            or pathconflicts

            and (

                f in pathconflicts

                or any(p in pathconflicts for p in pathutil.finddirs(f))

            )

        )

        (flags,) = args

        return (flags, backup)

    mresult.mapaction(

        mergestatemod.ACTION_CREATED, mergestatemod.ACTION_GET, transformargs

    )

def _forgetremoved(wctx, mctx, branchmerge, mresult):

    """

    Forget removed files

    If we're jumping between revisions (as opposed to merging), and if

    neither the working directory nor the target rev has the file,

    then we need to remove it from the dirstate, to prevent the

    dirstate from listing the file when it is no longer in the

    manifest.

    If we're merging, and the other revision has removed a file

    that is not present in the working directory, we need to mark it

    as removed.

    """

    m = mergestatemod.ACTION_FORGET

    if branchmerge:

        m = mergestatemod.ACTION_REMOVE

    for f in wctx.deleted():

        if f not in mctx:

            mresult.addfile(f, m, None, b"forget deleted")

    if not branchmerge:

        for f in wctx.removed():

            if f not in mctx:

                mresult.addfile(

                    f,

                    mergestatemod.ACTION_FORGET,

                    None,

                    b"forget removed",

                )

def _checkcollision(repo, wmf, mresult):

    """

    Check for case-folding collisions.

    """

    # If the repo is narrowed, filter out files outside the narrowspec.

    narrowmatch = repo.narrowmatch()

    if not narrowmatch.always():

        pmmf = set(wmf.walk(narrowmatch))

        if mresult:

            for f in list(mresult.files()):

                if not narrowmatch(f):

                    mresult.removefile(f)

    else:

        # build provisional merged manifest up

        pmmf = set(wmf)

    if mresult:

        # KEEP and EXEC are no-op

        for f in mresult.files(

            (

                mergestatemod.ACTION_ADD,

                mergestatemod.ACTION_ADD_MODIFIED,

                mergestatemod.ACTION_FORGET,

                mergestatemod.ACTION_GET,

                mergestatemod.ACTION_CHANGED_DELETED,

                mergestatemod.ACTION_DELETED_CHANGED,

            )

        ):

            pmmf.add(f)

        for f in mresult.files((mergestatemod.ACTION_REMOVE,)):

            pmmf.discard(f)

        for f, args, msg in mresult.getactions(

            [mergestatemod.ACTION_DIR_RENAME_MOVE_LOCAL]

        ):

            f2, flags = args

            pmmf.discard(f2)

            pmmf.add(f)

        for f in mresult.files((mergestatemod.ACTION_LOCAL_DIR_RENAME_GET,)):

            pmmf.add(f)

        for f, args, msg in mresult.getactions([mergestatemod.ACTION_MERGE]):

            f1, f2, fa, move, anc = args

            if move:

                pmmf.discard(f1)

            pmmf.add(f)

    # check case-folding collision in provisional merged manifest

    foldmap = {}

    for f in pmmf:

        fold = util.normcase(f)

        if fold in foldmap:

            msg = _(b"case-folding collision between %s and %s")

            msg %= (f, foldmap[fold])

            raise error.StateError(msg)

        foldmap[fold] = f

    # check case-folding of directories

    foldprefix = unfoldprefix = lastfull = b''

    for fold, f in sorted(foldmap.items()):

        if fold.startswith(foldprefix) and not f.startswith(unfoldprefix):

            # the folded prefix matches but actual casing is different

            msg = _(b"case-folding collision between %s and directory of %s")

            msg %= (lastfull, f)

            raise error.StateError(msg)

        foldprefix = fold + b'/'

        unfoldprefix = f + b'/'

        lastfull = f

def _filesindirs(repo, manifest, dirs):

    """

    Generator that yields pairs of all the files in the manifest that are found

    inside the directories listed in dirs, and which directory they are found

    in.

    """

    for f in manifest:

        for p in pathutil.finddirs(f):

            if p in dirs:

                yield f, p

                break

def checkpathconflicts(repo, wctx, mctx, mresult):

    """

    Check if any actions introduce path conflicts in the repository, updating

    actions to record or handle the path conflict accordingly.

    """

    mf = wctx.manifest()

    # The set of local files that conflict with a remote directory.

    localconflicts = set()

    # The set of directories that conflict with a remote file, and so may cause

    # conflicts if they still contain any files after the merge.

    remoteconflicts = set()

    # The set of directories that appear as both a file and a directory in the

    # remote manifest.  These indicate an invalid remote manifest, which

    # can't be updated to cleanly.

    invalidconflicts = set()

    # The set of directories that contain files that are being created.

    createdfiledirs = set()

    # The set of files deleted by all the actions.

    deletedfiles = set()

    for f in mresult.files(

        (

            mergestatemod.ACTION_CREATED,

            mergestatemod.ACTION_DELETED_CHANGED,

            mergestatemod.ACTION_MERGE,

            mergestatemod.ACTION_CREATED_MERGE,

        )

    ):

        # This action may create a new local file.

        createdfiledirs.update(pathutil.finddirs(f))

        if mf.hasdir(f):

            # The file aliases a local directory.  This might be ok if all

            # the files in the local directory are being deleted.  This

            # will be checked once we know what all the deleted files are.

            remoteconflicts.add(f)

    # Track the names of all deleted files.

    for f in mresult.files((mergestatemod.ACTION_REMOVE,)):

        deletedfiles.add(f)

    for f, args, msg in mresult.getactions((mergestatemod.ACTION_MERGE,)):

        f1, f2, fa, move, anc = args

        if move:

            deletedfiles.add(f1)

    for f, args, msg in mresult.getactions(

        (mergestatemod.ACTION_DIR_RENAME_MOVE_LOCAL,)

    ):

        f2, flags = args

        deletedfiles.add(f2)

    # Check all directories that contain created files for path conflicts.

    for p in createdfiledirs:

        if p in mf:

            if p in mctx:

                # A file is in a directory which aliases both a local

                # and a remote file.  This is an internal inconsistency

                # within the remote manifest.

                invalidconflicts.add(p)

            else:

                # A file is in a directory which aliases a local file.

                # We will need to rename the local file.

                localconflicts.add(p)

        pd = mresult.getfile(p)

        if pd and pd[0] in (

            mergestatemod.ACTION_CREATED,

            mergestatemod.ACTION_DELETED_CHANGED,

            mergestatemod.ACTION_MERGE,

            mergestatemod.ACTION_CREATED_MERGE,

        ):

            # The file is in a directory which aliases a remote file.

            # This is an internal inconsistency within the remote

            # manifest.

            invalidconflicts.add(p)

    # Rename all local conflicting files that have not been deleted.

    for p in localconflicts:

        if p not in deletedfiles:

            ctxname = bytes(wctx).rstrip(b'+')

            pnew = util.safename(p, ctxname, wctx, set(mresult.files()))

            porig = wctx[p].copysource() or p

            mresult.addfile(

                pnew,

                mergestatemod.ACTION_PATH_CONFLICT_RESOLVE,

                (p, porig),

                b'local path conflict',

            )

            mresult.addfile(

                p,

                mergestatemod.ACTION_PATH_CONFLICT,

                (pnew, b'l'),

                b'path conflict',