mercurial-scm/hg: mercurial/merge.py comparison

comparison mercurial/merge.py @ 52697:f3762eafed66

typing: add some type annotations to the `merge.mergeresult` class The generated type annotations around `filemap()` and `files()` were slightly modified by the pyupgrade series culminating in 70a75d379daf. This module is way more complicated than the other changes, but these weren't too bad to figure out. The typing caught a trivial issue in `sparse`- it was passing an empty data list to `addfile()` for the `ACTION_REMOVE` case, instead of a tuple or None. `merge.manifestmerge()` calls this function with None for the data, so 1) it has to be typed as optional, and 2) is safe to pass None in the sparse code.

author	Matt Harbison <matt_harbison@yahoo.com>
date	Mon, 13 Jan 2025 12:24:33 -0500
parents	45dc0f874b8c
children	ea9846b8e539

comparison

equal deleted inserted replaced

-:10e7adbffa8c
+:f3762eafed66
 import collections
 import os
 import struct
 import typing
-from typing import Dict, Iterator, Optional, Tuple
+from typing import Dict, Iterable, Iterator, Optional, Tuple
 from .i18n import _
 from .node import nullrev
 from .thirdparty import attr
 subrepoutil,
 util,
 worker,
 )
+if typing.TYPE_CHECKING:
+# TODO: figure out what exactly is in this tuple
+MergeResultData = tuple
+MergeResultAction = tuple[bytes, Optional[MergeResultData], bytes]
+"""The filename, data about the merge, and message about the merge."""
+FileMappingValue = tuple[
+mergestatemod.MergeAction, Optional[MergeResultData], bytes
+]
+"""The merge action, data about the merge, and message about the merge, for
+the keyed file."""
 rust_update_mod = policy.importrust("update")
 _pack = struct.pack
 _unpack = struct.unpack
 if relf not in repo.dirstate:
 return f
 return None
-def _checkunknownfiles(repo, wctx, mctx, force, mresult, mergeforce):
+def _checkunknownfiles(
+repo, wctx, mctx, force, mresult: mergeresult, 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.
 """
 mresult.mapaction(
 mergestatemod.ACTION_CREATED, mergestatemod.ACTION_GET, transformargs
 )
-def _forgetremoved(wctx, mctx, branchmerge, mresult):
+def _forgetremoved(wctx, mctx, branchmerge, mresult: mergeresult) -> None:
 """
 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,
 None,
 b"forget removed",
 )
-def _checkcollision(repo, wmf, mresult):
+def _checkcollision(repo, wmf, mresult: mergeresult | None) -> None:
 """
 Check for case-folding collisions.
 """
 # If the repo is narrowed, filter out files outside the narrowspec.
 narrowmatch = repo.narrowmatch()
 foldprefix = fold + b'/'
 unfoldprefix = f + b'/'
 lastfull = f
-def _filesindirs(repo, manifest, dirs):
+def _filesindirs(repo, manifest, dirs) -> Iterator[tuple[bytes, bytes]]:
 """
 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.
 """
 if p in dirs:
 yield f, p
 break
-def checkpathconflicts(repo, wctx, mctx, mresult):
+def checkpathconflicts(repo, wctx, mctx, mresult: mergeresult) -> None:
 """
 Check if any actions introduce path conflicts in the repository, updating
 actions to record or handle the path conflict accordingly.
 """
 mf = wctx.manifest()
 if remoteconflicts:
 # Check if all files in the conflicting directories have been removed.
 ctxname = bytes(mctx).rstrip(b'+')
 for f, p in _filesindirs(repo, mf, remoteconflicts):
 if f not in deletedfiles:
-m, args, msg = mresult.getfile(p)
+mapping_value = mresult.getfile(p)
+# Help pytype- in theory, this could be None since no default
+# value is passed to getfile() above.
+assert mapping_value is not None
+m, args, msg = mapping_value
 pnew = util.safename(p, ctxname, wctx, set(mresult.files()))
 if m in (
 mergestatemod.ACTION_DELETED_CHANGED,
 mergestatemod.ACTION_MERGE,
 ):
 raise error.StateError(
 _(b"destination manifest contains path conflicts")
 )
-def _filternarrowactions(narrowmatch, branchmerge, mresult):
+def _filternarrowactions(
+narrowmatch, branchmerge, mresult: mergeresult
+) -> None:
 """
 Filters out actions that can ignored because the repo is narrowed.
 Raise an exception if the merge cannot be completed because the repo is
 narrowed.
 """An object representing result of merging manifests.
 It has information about what actions need to be performed on dirstate
 mapping of divergent renames and other such cases."""
-def __init__(self):
+_filemapping: dict[bytes, FileMappingValue]
+_actionmapping: dict[
+mergestatemod.MergeAction, dict[bytes, tuple[MergeResultData, bytes]]
+]
+def __init__(self) -> None:
 """
 filemapping: dict of filename as keys and action related info as values
 diverge: mapping of source name -> list of dest name for
 divergent renames
 renamedelete: mapping of source name -> list of destinations for files
 def updatevalues(self, diverge, renamedelete):
 self._diverge = diverge
 self._renamedelete = renamedelete
-def addfile(self, filename, action, data, message):
+def addfile(
+self,
+filename: bytes,
+action: mergestatemod.MergeAction,
+data: MergeResultData | None,
+message,
+) -> None:
 """adds a new file to the mergeresult object
 filename: file which we are adding
 action: one of mergestatemod.ACTION_*
 data: a tuple of information like fctx and ctx related to this merge
 del self._actionmapping[a][filename]
 self._filemapping[filename] = (action, data, message)
 self._actionmapping[action][filename] = (data, message)
-def mapaction(self, actionfrom, actionto, transform):
+def mapaction(
+self,
+actionfrom: mergestatemod.MergeAction,
+actionto: mergestatemod.MergeAction,
+transform,
+):
 """changes all occurrences of action `actionfrom` into `actionto`,
 transforming its args with the function `transform`.
 """
 orig = self._actionmapping[actionfrom]
 del self._actionmapping[actionfrom]
 for f, (data, msg) in orig.items():
 data = transform(f, data)
 self._filemapping[f] = (actionto, data, msg)
 dest[f] = (data, msg)
-def getfile(self, filename, default_return=None):
+def getfile(
+self, filename: bytes, default_return: FileMappingValue | None = None
+) -> FileMappingValue | None:
 """returns (action, args, msg) about this file
 returns default_return if the file is not present"""
 if filename in self._filemapping:
 return self._filemapping[filename]
 return default_return
-def files(self, actions=None):
+def files(
+self, actions: Iterable[mergestatemod.MergeAction] | None = None
+) -> Iterator[bytes]:
 """returns files on which provided action needs to perfromed
 If actions is None, all files are returned
 """
 # TODO: think whether we should return renamedelete and
 else:
 for a in actions:
 yield from self._actionmapping[a]
-def removefile(self, filename):
+def removefile(self, filename: bytes) -> None:
 """removes a file from the mergeresult object as the file might
 not merging anymore"""
 action, data, message = self._filemapping[filename]
 del self._filemapping[filename]
 del self._actionmapping[action][filename]
-def getactions(self, actions, sort=False):
+def getactions(
+self, actions: Iterable[mergestatemod.MergeAction], sort: bool = False
+) -> Iterator[MergeResultAction]:
 """get list of files which are marked with these actions
 if sort is true, files for each action is sorted and then added
 Returns a list of tuple of form (filename, data, message)
 """
 yield f, args, msg
 else:
 for f, (args, msg) in self._actionmapping[a].items():
 yield f, args, msg
-def len(self, actions=None):
+def len(
+self, actions: Iterable[mergestatemod.MergeAction] | None = None
+) -> int:
 """returns number of files which needs actions
 if actions is passed, total of number of files in that action
 only is returned"""
 if actions is None:
 return len(self._filemapping)
 return sum(len(self._actionmapping[a]) for a in actions)
-def filemap(self, sort=False) -> Iterator[tuple]:  # TODO: fill out tuple
+def filemap(
+self, sort: bool = False
+) -> Iterator[tuple[bytes, MergeResultData]]:
 if sort:
 yield from sorted(self._filemapping.items())
 else:
 yield from self._filemapping.items()
-def addcommitinfo(self, filename, key, value):
+def addcommitinfo(self, filename: bytes, key, value) -> None:
 """adds key-value information about filename which will be required
 while committing this merge"""
 self._commitinfo[filename][key] = value
 @property
 @property
 def commitinfo(self):
 return self._commitinfo
 @property
-def actionsdict(self):
+def actionsdict(
+self,
+) -> dict[mergestatemod.MergeAction, list[MergeResultAction]]:
 """returns a dictionary of actions to be perfomed with action as key
 and a list of files and related arguments as values"""
 res = collections.defaultdict(list)
 for a, d in self._actionmapping.items():
 for f, (args, msg) in d.items():
 res[a].append((f, args, msg))
 return res
-def setactions(self, actions):
+def setactions(self, actions) -> None:
 self._filemapping = actions
 self._actionmapping = collections.defaultdict(dict)
 for f, (act, data, msg) in self._filemapping.items():
 self._actionmapping[act][f] = data, msg
-def hasconflicts(self):
+def hasconflicts(self) -> bool:
 """tells whether this merge resulted in some actions which can
 result in conflicts or not"""
 for a in self._actionmapping.keys():
 if (
 a
 force,
 matcher,
 acceptremote,
 followcopies,
 forcefulldiff=False,
-):
+) -> mergeresult:
 """
 Merge wctx and p2 with ancestor pa and generate merge action list
 branchmerge and force are as passed in to update
 matcher = matcher to filter file lists
 mresult.updatevalues(diverge, renamedelete)
 return mresult
-def _resolvetrivial(repo, wctx, mctx, ancestor, mresult):
+def _resolvetrivial(repo, wctx, mctx, ancestor, mresult: mergeresult) -> None:
 """Resolves false conflicts where the nodeid changed but the content
 remained the same."""
 # We force a copy of actions.items() because we're going to mutate
 # actions as we resolve trivial conflicts.
 for f in list(mresult.files((mergestatemod.ACTION_CHANGED_DELETED,))):
 force,
 acceptremote,
 followcopies,
 matcher=None,
 mergeforce=False,
-):
+) -> mergeresult:
 """
 Calculate the actions needed to merge mctx into wctx using ancestors
 Uses manifestmerge() to merge manifest and get list of actions required to
 perform for merging two manifests. If there are multiple ancestors, uses bid
 if i > 0:
 yield False, (i, f)
 yield True, filedata
-def _prefetchfiles(repo, ctx, mresult):
+def _prefetchfiles(repo, ctx, mresult: mergeresult) -> None:
 """Invoke ``scmutil.prefetchfiles()`` for the files relevant to the dict
 of merge actions.  ``ctx`` is the context being merged in."""
 # Skipping 'a', 'am', 'f', 'r', 'dm', 'e', 'k', 'p' and 'pr', because they
 # don't touch the context to be merged in.  'cd' is skipped, because
 )
 def applyupdates(
 repo,
-mresult,
+mresult: mergeresult,
 wctx,
 mctx,
 overwrite,
 wantfiledata,
 labels=None,

Mercurial > public > mercurial-scm > hg

comparison mercurial/merge.py @ 52697:f3762eafed66