comparison hgext/convert/common.py @ 51685:0eb515c7bec8

typing: add trivial type hints to the convert extension's common modules

This started as ensuring that the `encoding` and `orig_encoding` attributes has
a type other than `Any`, so pytype can catch problems where it needs to be str
for stdlib encoding and decoding.  It turns out that adding the hint in
`mercurial.encoding` is what was needed, but I picked a bunch of low hanging
fruit while here.  There's definitely more to do, and I see a problem where
`shlex.shlex` is being fed bytes instead of str, but there are not enough type
hints yet to make pytype notice.

51684:20e2a20674dc

import os
import pickle
import re
import shlex
import subprocess

from mercurial.i18n import _
from mercurial.pycompat import open
from mercurial import (
    encoding,

from mercurial.utils import (
    dateutil,
    procutil,
)

propertycache = util.propertycache


def _encodeornone(d):
    if d is None:
        return
    return d.encode('latin1')


class _shlexpy3proxy:
    def __init__(self, l):
        self._l = l

    def __iter__(self):
        return (_encodeornone(v) for v in self._l)

    @property
    def infile(self):
        return self._l.infile or b'<unknown>'

    @property
    def lineno(self):
        return self._l.lineno


def shlexer(data=None, filepath=None, wordchars=None, whitespace=None):
    if data is None:
        data = open(filepath, b'r', encoding='latin1')
    else:
        if filepath is not None:
            raise error.ProgrammingError(

    if wordchars is not None:
        l.wordchars += wordchars.decode('latin1')
    return _shlexpy3proxy(l)


def encodeargs(args):
    def encodearg(s):
        lines = base64.encodebytes(s)
        lines = [l.splitlines()[0] for l in pycompat.iterbytestr(lines)]
        return b''.join(lines)

    s = pickle.dumps(args)
    return encodearg(s)


def decodeargs(s):
    s = base64.decodebytes(s)
    return pickle.loads(s)


class MissingTool(Exception):
    pass


def checktool(exe, name=None, abort=True):
    name = name or exe
    if not procutil.findexe(exe):
        if abort:
            exc = error.Abort
        else:

class NoRepo(Exception):
    pass


SKIPREV = b'SKIP'


class commit:
    def __init__(
        self,
        author,
        date,
        desc,
        parents,
        branch=None,
        rev=None,
        extra=None,
        sortkey=None,
        saverev=True,
        phase=phases.draft,
        optparents=None,
        ctx=None,
    ):
        self.author = author or b'unknown'
        self.date = date or b'0 0'
        self.desc = desc
        self.parents = parents  # will be converted and used as parents
        self.optparents = optparents or []  # will be used if already converted

class converter_source:
    """Conversion source interface"""

    def __init__(self, ui, repotype, path=None, revs=None):
        """Initialize conversion source (or raise NoRepo("message")
        exception if path is not a valid repository)"""
        self.ui = ui
        self.path = path
        self.revs = revs
        self.repotype = repotype

        self.encoding = b'utf-8'

    def checkhexformat(self, revstr, mapname=b'splicemap'):
        """fails if revstr is not a 40 byte hex. mercurial and git both uses
        such format for their revision numbering
        """
        if not re.match(br'[0-9a-fA-F]{40,40}$', revstr):
            raise error.Abort(
                _(b'%s entry %s is not a valid revision identifier')
                % (mapname, revstr)
            )

    def before(self):
        pass

    def after(self):
        pass

    def targetfilebelongstosource(self, targetfilename):
        """Returns true if the given targetfile belongs to the source repo. This
        is useful when only a subdirectory of the target belongs to the source

        Tag names must be UTF-8 strings.
        """
        raise NotImplementedError

    def recode(self, s, encoding=None):
        if not encoding:
            encoding = self.encoding or b'utf-8'

        if isinstance(s, str):
            return s.encode("utf-8")

        This function is only needed to support --filemap
        """
        raise NotImplementedError

    def converted(self, rev, sinkrev):
        '''Notify the source that a revision has been converted.'''

    def hasnativeorder(self):
        """Return true if this source has a meaningful, native revision
        order. For instance, Mercurial revisions are store sequentially
        while there is no such global ordering with Darcs.
        """
        return False

    def hasnativeclose(self):
        """Return true if this source has ability to close branch."""
        return False

    def lookuprev(self, rev):
        """If rev is a meaningful revision reference in source, return

        Bookmark names are to be UTF-8 strings.
        """
        return {}

    def checkrevformat(self, revstr, mapname=b'splicemap'):
        """revstr is a string that describes a revision in the given
        source control system.  Return true if revstr has correct
        format.
        """
        return True


class converter_sink:
    """Conversion sink (target) interface"""

    def __init__(self, ui, repotype, path):
        """Initialize conversion sink (or raise NoRepo("message")
        exception if path is not a valid repository)

        created is a list of paths to remove if a fatal error occurs
        later"""

        was changed in a revision, even if there was no change.  This method
        tells the destination that we're using a filemap and that it should
        filter empty revisions.
        """

    def before(self):
        pass

    def after(self):
        pass

    def putbookmarks(self, bookmarks):
        """Put bookmarks into sink.

        special cases."""
        raise NotImplementedError


class commandline:
    def __init__(self, ui, command):
        self.ui = ui
        self.command = command

    def prerun(self):
        pass

    def postrun(self):
        pass

    def _cmdline(self, cmd, *args, **kwargs):
        kwargs = pycompat.byteskwargs(kwargs)
        cmdline = [self.command, cmd] + list(args)
        for k, v in kwargs.items():
            if len(k) == 1:
                cmdline.append(b'-' + k)

        if not self.ui.debugflag:
            cmdline += [b'2>', pycompat.bytestr(os.devnull)]
        cmdline = b' '.join(cmdline)
        return cmdline

    def _run(self, cmd, *args, **kwargs):
        def popen(cmdline):
            p = subprocess.Popen(
                procutil.tonativestr(cmdline),
                shell=True,
                bufsize=-1,

            )
            return p

        return self._dorun(popen, cmd, *args, **kwargs)

    def _run2(self, cmd, *args, **kwargs):
        return self._dorun(procutil.popen2, cmd, *args, **kwargs)

    def _run3(self, cmd, *args, **kwargs):
        return self._dorun(procutil.popen3, cmd, *args, **kwargs)

    def _dorun(self, openfunc, cmd, *args, **kwargs):
        cmdline = self._cmdline(cmd, *args, **kwargs)
        self.ui.debug(b'running: %s\n' % (cmdline,))
        self.prerun()
        try:
            return openfunc(cmdline)
        finally:
            self.postrun()

    def run(self, cmd, *args, **kwargs):
        p = self._run(cmd, *args, **kwargs)
        output = p.communicate()[0]
        self.ui.debug(output)
        return output, p.returncode

    def runlines(self, cmd, *args, **kwargs):
        p = self._run(cmd, *args, **kwargs)
        output = p.stdout.readlines()
        p.wait()
        self.ui.debug(b''.join(output))
        return output, p.returncode

    def checkexit(self, status, output=b''):
        if status:
            if output:
                self.ui.warn(_(b'%s error:\n') % self.command)
                self.ui.warn(output)
            msg = procutil.explainexit(status)
            raise error.Abort(b'%s %s' % (self.command, msg))

    def run0(self, cmd, *args, **kwargs):
        output, status = self.run(cmd, *args, **kwargs)
        self.checkexit(status, output)
        return output

    def runlines0(self, cmd, *args, **kwargs):
        output, status = self.runlines(cmd, *args, **kwargs)
        self.checkexit(status, b''.join(output))
        return output

    @propertycache

        # Since ARG_MAX is for command line _and_ environment, lower our limit
        # (and make happy Windows shells while doing this).
        return argmax // 2 - 1

    def _limit_arglist(self, arglist, cmd, *args, **kwargs):
        cmdlen = len(self._cmdline(cmd, *args, **kwargs))
        limit = self.argmax - cmdlen
        numbytes = 0
        fl = []
        for fn in arglist:

                fl = [fn]
                numbytes = b
        if fl:
            yield fl

    def xargs(self, arglist, cmd, *args, **kwargs):
        for l in self._limit_arglist(arglist, cmd, *args, **kwargs):
            self.run0(cmd, *(list(args) + l), **kwargs)


class mapfile(dict):
    def __init__(self, ui, path):
        super(mapfile, self).__init__()
        self.ui = ui
        self.path = path
        self.fp = None
        self.order = []
        self._read()

    def _read(self):
        if not self.path:
            return
        try:
            fp = open(self.path, b'rb')
        except FileNotFoundError:

            if key not in self:
                self.order.append(key)
            super(mapfile, self).__setitem__(key, value)
        fp.close()

    def __setitem__(self, key, value):
        if self.fp is None:
            try:
                self.fp = open(self.path, b'ab')
            except IOError as err:
                raise error.Abort(

                )
        self.fp.write(util.tonativeeol(b'%s %s\n' % (key, value)))
        self.fp.flush()
        super(mapfile, self).__setitem__(key, value)

    def close(self):
        if self.fp:
            self.fp.close()
            self.fp = None

51685:0eb515c7bec8

import os
import pickle
import re
import shlex
import subprocess
import typing

from typing import (
    Any,
    AnyStr,
    Optional,
)

from mercurial.i18n import _
from mercurial.pycompat import open
from mercurial import (
    encoding,

from mercurial.utils import (
    dateutil,
    procutil,
)

if typing.TYPE_CHECKING:
    from typing import (
        overload,
    )
    from mercurial import (
        ui as uimod,
    )

propertycache = util.propertycache


if typing.TYPE_CHECKING:

    @overload
    def _encodeornone(d: str) -> bytes:
        pass

    @overload
    def _encodeornone(d: None) -> None:
        pass


def _encodeornone(d):
    if d is None:
        return
    return d.encode('latin1')


class _shlexpy3proxy:
    def __init__(self, l: shlex.shlex) -> None:
        self._l = l

    def __iter__(self):
        return (_encodeornone(v) for v in self._l)

    @property
    def infile(self):
        return self._l.infile or b'<unknown>'

    @property
    def lineno(self) -> int:
        return self._l.lineno


def shlexer(
    data=None,
    filepath: Optional[str] = None,
    wordchars: Optional[bytes] = None,
    whitespace: Optional[bytes] = None,
):
    if data is None:
        data = open(filepath, b'r', encoding='latin1')
    else:
        if filepath is not None:
            raise error.ProgrammingError(

    if wordchars is not None:
        l.wordchars += wordchars.decode('latin1')
    return _shlexpy3proxy(l)


def encodeargs(args: Any) -> bytes:
    def encodearg(s: bytes) -> bytes:
        lines = base64.encodebytes(s)
        lines = [l.splitlines()[0] for l in pycompat.iterbytestr(lines)]
        return b''.join(lines)

    s = pickle.dumps(args)
    return encodearg(s)


def decodeargs(s: bytes) -> Any:
    s = base64.decodebytes(s)
    return pickle.loads(s)


class MissingTool(Exception):
    pass


def checktool(
    exe: bytes, name: Optional[bytes] = None, abort: bool = True
) -> None:
    name = name or exe
    if not procutil.findexe(exe):
        if abort:
            exc = error.Abort
        else:

class NoRepo(Exception):
    pass


SKIPREV: bytes = b'SKIP'


class commit:
    def __init__(
        self,
        author: bytes,
        date: bytes,
        desc: bytes,
        parents,
        branch: Optional[bytes] = None,
        rev=None,
        extra=None,
        sortkey=None,
        saverev=True,
        phase: int = phases.draft,
        optparents=None,
        ctx=None,
    ) -> None:
        self.author = author or b'unknown'
        self.date = date or b'0 0'
        self.desc = desc
        self.parents = parents  # will be converted and used as parents
        self.optparents = optparents or []  # will be used if already converted

class converter_source:
    """Conversion source interface"""

    def __init__(
        self,
        ui: "uimod.ui",
        repotype: bytes,
        path: Optional[bytes] = None,
        revs=None,
    ) -> None:
        """Initialize conversion source (or raise NoRepo("message")
        exception if path is not a valid repository)"""
        self.ui = ui
        self.path = path
        self.revs = revs
        self.repotype = repotype

        self.encoding = b'utf-8'

    def checkhexformat(
        self, revstr: bytes, mapname: bytes = b'splicemap'
    ) -> None:
        """fails if revstr is not a 40 byte hex. mercurial and git both uses
        such format for their revision numbering
        """
        if not re.match(br'[0-9a-fA-F]{40,40}$', revstr):
            raise error.Abort(
                _(b'%s entry %s is not a valid revision identifier')
                % (mapname, revstr)
            )

    def before(self) -> None:
        pass

    def after(self) -> None:
        pass

    def targetfilebelongstosource(self, targetfilename):
        """Returns true if the given targetfile belongs to the source repo. This
        is useful when only a subdirectory of the target belongs to the source

        Tag names must be UTF-8 strings.
        """
        raise NotImplementedError

    def recode(self, s: AnyStr, encoding: Optional[bytes] = None) -> bytes:
        if not encoding:
            encoding = self.encoding or b'utf-8'

        if isinstance(s, str):
            return s.encode("utf-8")

        This function is only needed to support --filemap
        """
        raise NotImplementedError

    def converted(self, rev, sinkrev) -> None:
        '''Notify the source that a revision has been converted.'''

    def hasnativeorder(self) -> bool:
        """Return true if this source has a meaningful, native revision
        order. For instance, Mercurial revisions are store sequentially
        while there is no such global ordering with Darcs.
        """
        return False

    def hasnativeclose(self) -> bool:
        """Return true if this source has ability to close branch."""
        return False

    def lookuprev(self, rev):
        """If rev is a meaningful revision reference in source, return

        Bookmark names are to be UTF-8 strings.
        """
        return {}

    def checkrevformat(self, revstr, mapname: bytes = b'splicemap') -> bool:
        """revstr is a string that describes a revision in the given
        source control system.  Return true if revstr has correct
        format.
        """
        return True


class converter_sink:
    """Conversion sink (target) interface"""

    def __init__(self, ui: "uimod.ui", repotype: bytes, path: bytes) -> None:
        """Initialize conversion sink (or raise NoRepo("message")
        exception if path is not a valid repository)

        created is a list of paths to remove if a fatal error occurs
        later"""

        was changed in a revision, even if there was no change.  This method
        tells the destination that we're using a filemap and that it should
        filter empty revisions.
        """

    def before(self) -> None:
        pass

    def after(self) -> None:
        pass

    def putbookmarks(self, bookmarks):
        """Put bookmarks into sink.

        special cases."""
        raise NotImplementedError


class commandline:
    def __init__(self, ui: "uimod.ui", command: bytes) -> None:
        self.ui = ui
        self.command = command

    def prerun(self) -> None:
        pass

    def postrun(self) -> None:
        pass

    def _cmdline(self, cmd: bytes, *args: bytes, **kwargs) -> bytes:
        kwargs = pycompat.byteskwargs(kwargs)
        cmdline = [self.command, cmd] + list(args)
        for k, v in kwargs.items():
            if len(k) == 1:
                cmdline.append(b'-' + k)

        if not self.ui.debugflag:
            cmdline += [b'2>', pycompat.bytestr(os.devnull)]
        cmdline = b' '.join(cmdline)
        return cmdline

    def _run(self, cmd: bytes, *args: bytes, **kwargs):
        def popen(cmdline):
            p = subprocess.Popen(
                procutil.tonativestr(cmdline),
                shell=True,
                bufsize=-1,

            )
            return p

        return self._dorun(popen, cmd, *args, **kwargs)

    def _run2(self, cmd: bytes, *args: bytes, **kwargs):
        return self._dorun(procutil.popen2, cmd, *args, **kwargs)

    def _run3(self, cmd: bytes, *args: bytes, **kwargs):
        return self._dorun(procutil.popen3, cmd, *args, **kwargs)

    def _dorun(self, openfunc, cmd: bytes, *args: bytes, **kwargs):
        cmdline = self._cmdline(cmd, *args, **kwargs)
        self.ui.debug(b'running: %s\n' % (cmdline,))
        self.prerun()
        try:
            return openfunc(cmdline)
        finally:
            self.postrun()

    def run(self, cmd: bytes, *args: bytes, **kwargs):
        p = self._run(cmd, *args, **kwargs)
        output = p.communicate()[0]
        self.ui.debug(output)
        return output, p.returncode

    def runlines(self, cmd: bytes, *args: bytes, **kwargs):
        p = self._run(cmd, *args, **kwargs)
        output = p.stdout.readlines()
        p.wait()
        self.ui.debug(b''.join(output))
        return output, p.returncode

    def checkexit(self, status, output: bytes = b'') -> None:
        if status:
            if output:
                self.ui.warn(_(b'%s error:\n') % self.command)
                self.ui.warn(output)
            msg = procutil.explainexit(status)
            raise error.Abort(b'%s %s' % (self.command, msg))

    def run0(self, cmd: bytes, *args: bytes, **kwargs):
        output, status = self.run(cmd, *args, **kwargs)
        self.checkexit(status, output)
        return output

    def runlines0(self, cmd: bytes, *args: bytes, **kwargs):
        output, status = self.runlines(cmd, *args, **kwargs)
        self.checkexit(status, b''.join(output))
        return output

    @propertycache

        # Since ARG_MAX is for command line _and_ environment, lower our limit
        # (and make happy Windows shells while doing this).
        return argmax // 2 - 1

    def _limit_arglist(self, arglist, cmd: bytes, *args: bytes, **kwargs):
        cmdlen = len(self._cmdline(cmd, *args, **kwargs))
        limit = self.argmax - cmdlen
        numbytes = 0
        fl = []
        for fn in arglist:

                fl = [fn]
                numbytes = b
        if fl:
            yield fl

    def xargs(self, arglist, cmd: bytes, *args: bytes, **kwargs):
        for l in self._limit_arglist(arglist, cmd, *args, **kwargs):
            self.run0(cmd, *(list(args) + l), **kwargs)


class mapfile(dict):
    def __init__(self, ui: "uimod.ui", path: bytes) -> None:
        super(mapfile, self).__init__()
        self.ui = ui
        self.path = path
        self.fp = None
        self.order = []
        self._read()

    def _read(self) -> None:
        if not self.path:
            return
        try:
            fp = open(self.path, b'rb')
        except FileNotFoundError:

            if key not in self:
                self.order.append(key)
            super(mapfile, self).__setitem__(key, value)
        fp.close()

    def __setitem__(self, key, value) -> None:
        if self.fp is None:
            try:
                self.fp = open(self.path, b'ab')
            except IOError as err:
                raise error.Abort(

                )
        self.fp.write(util.tonativeeol(b'%s %s\n' % (key, value)))
        self.fp.flush()
        super(mapfile, self).__setitem__(key, value)

    def close(self) -> None:
        if self.fp:
            self.fp.close()
            self.fp = None