# Subversion 1.4/1.5 Python API backend

#

# Copyright(C) 2007 Daniel Holth et al

import codecs

import locale

import os

import pickle

import re

import xml.dom.minidom

from mercurial.i18n import _

from mercurial.pycompat import open

from mercurial import (

    encoding,

    error,

    pycompat,

    util,

    vfs as vfsmod,

)

from mercurial.utils import (

    dateutil,

    procutil,

    stringutil,

)

from . import common

stringio = util.stringio

propertycache = util.propertycache

urlerr = util.urlerr

urlreq = util.urlreq

commandline = common.commandline

commit = common.commit

converter_sink = common.converter_sink

converter_source = common.converter_source

decodeargs = common.decodeargs

encodeargs = common.encodeargs

makedatetimestamp = common.makedatetimestamp

mapfile = common.mapfile

MissingTool = common.MissingTool

NoRepo = common.NoRepo

# Subversion stuff. Works best with very recent Python SVN bindings

# e.g. SVN 1.5 or backports. Thanks to the bzr folks for enhancing

# these bindings.

try:

    import svn

    import svn.client

    import svn.core

    import svn.ra

    import svn.delta

    from . import transport

    import warnings

    warnings.filterwarnings(

        'ignore', module='svn.core', category=DeprecationWarning

    )

    svn.core.SubversionException  # trigger import to catch error

except ImportError:

    svn = None

# In Subversion, paths and URLs are Unicode (encoded as UTF-8), which

# Subversion converts from / to native strings when interfacing with the OS.

# When passing paths and URLs to Subversion, we have to recode them such that

# it roundstrips with what Subversion is doing.

fsencoding = None

def init_fsencoding():

    global fsencoding, fsencoding_is_utf8

    if fsencoding is not None:

        return

    if pycompat.iswindows:

        # On Windows, filenames are Unicode, but we store them using the MBCS

        # encoding.

        fsencoding = 'mbcs'

    else:

        # This is the encoding used to convert UTF-8 back to natively-encoded

        # strings in Subversion 1.14.0 or earlier with APR 1.7.0 or earlier.

        with util.with_lc_ctype():

            fsencoding = locale.nl_langinfo(locale.CODESET) or 'ISO-8859-1'

    fsencoding = codecs.lookup(fsencoding).name

    fsencoding_is_utf8 = fsencoding == codecs.lookup('utf-8').name

def fs2svn(s):

    if fsencoding_is_utf8:

        return s

    else:

        return s.decode(fsencoding).encode('utf-8')

def formatsvndate(date):

    return dateutil.datestr(date, b'%Y-%m-%dT%H:%M:%S.000000Z')

def parsesvndate(s):

    # Example SVN datetime. Includes microseconds.

    # ISO-8601 conformant

    # '2007-01-04T17:35:00.902377Z'

    return dateutil.parsedate(s[:19] + b' UTC', [b'%Y-%m-%dT%H:%M:%S'])

class SvnPathNotFound(Exception):

    pass

def revsplit(rev):

    """Parse a revision string and return (uuid, path, revnum).

    >>> revsplit(b'svn:a2147622-4a9f-4db4-a8d3-13562ff547b2'

    ...          b'/proj%20B/mytrunk/mytrunk@1')

    ('a2147622-4a9f-4db4-a8d3-13562ff547b2', '/proj%20B/mytrunk/mytrunk', 1)

    >>> revsplit(b'svn:8af66a51-67f5-4354-b62c-98d67cc7be1d@1')

    ('', '', 1)

    >>> revsplit(b'@7')

    ('', '', 7)

    >>> revsplit(b'7')

    ('', '', 0)

    >>> revsplit(b'bad')

    ('', '', 0)

    """

    parts = rev.rsplit(b'@', 1)

    revnum = 0

    if len(parts) > 1:

        revnum = int(parts[1])

    parts = parts[0].split(b'/', 1)

    uuid = b''

    mod = b''

    if len(parts) > 1 and parts[0].startswith(b'svn:'):

        uuid = parts[0][4:]

        mod = b'/' + parts[1]

    return uuid, mod, revnum

def quote(s):

    # As of svn 1.7, many svn calls expect "canonical" paths. In

    # theory, we should call svn.core.*canonicalize() on all paths

    # before passing them to the API.  Instead, we assume the base url

    # is canonical and copy the behaviour of svn URL encoding function

    # so we can extend it safely with new components. The "safe"

    # characters were taken from the "svn_uri__char_validity" table in

    # libsvn_subr/path.c.

    return urlreq.quote(s, b"!$&'()*+,-./:=@_~")

def geturl(path):

    """Convert path or URL to a SVN URL, encoded in UTF-8.

    This can raise UnicodeDecodeError if the path or URL can't be converted to

    unicode using `fsencoding`.

    """

    try:

        return svn.client.url_from_path(

            svn.core.svn_path_canonicalize(fs2svn(path))

        )

    except svn.core.SubversionException:

        # svn.client.url_from_path() fails with local repositories

        pass

    if os.path.isdir(path):

        path = os.path.normpath(util.abspath(path))

        if pycompat.iswindows:

            path = b'/' + util.normpath(path)

        # Module URL is later compared with the repository URL returned

        # by svn API, which is UTF-8.

        path = fs2svn(path)

        path = b'file://%s' % quote(path)

    return svn.core.svn_path_canonicalize(path)

def optrev(number):

    optrev = svn.core.svn_opt_revision_t()

    optrev.kind = svn.core.svn_opt_revision_number

    optrev.value.number = number

    return optrev

class changedpath(object):

    def __init__(self, p):

        self.copyfrom_path = p.copyfrom_path

        self.copyfrom_rev = p.copyfrom_rev

        self.action = p.action

def get_log_child(

    fp,

    url,

    paths,

    start,

    end,

    limit=0,

    discover_changed_paths=True,

    strict_node_history=False,

):

    protocol = -1

    def receiver(orig_paths, revnum, author, date, message, pool):

        paths = {}

        if orig_paths is not None:

                paths[k] = changedpath(v)

        pickle.dump((paths, revnum, author, date, message), fp, protocol)

    try:

        # Use an ra of our own so that our parent can consume

        # our results without confusing the server.

        t = transport.SvnRaTransport(url=url)

        svn.ra.get_log(

            t.ra,

            paths,

            start,

            end,

            limit,

            discover_changed_paths,

            strict_node_history,

            receiver,

        )

    except IOError:

        # Caller may interrupt the iteration

        pickle.dump(None, fp, protocol)

    except Exception as inst:

        pickle.dump(stringutil.forcebytestr(inst), fp, protocol)

    else:

        pickle.dump(None, fp, protocol)

    fp.flush()

    # With large history, cleanup process goes crazy and suddenly

    # consumes *huge* amount of memory. The output file being closed,

    # there is no need for clean termination.

    os._exit(0)

def debugsvnlog(ui, **opts):

    """Fetch SVN log in a subprocess and channel them back to parent to

    avoid memory collection issues.

    """

    with util.with_lc_ctype():

        if svn is None:

            raise error.Abort(

                _(b'debugsvnlog could not load Subversion python bindings')

            )

        args = decodeargs(ui.fin.read())

        get_log_child(ui.fout, *args)

class logstream(object):

    """Interruptible revision log iterator."""

    def __init__(self, stdout):

        self._stdout = stdout

    def __iter__(self):

        while True:

            try:

                entry = pickle.load(self._stdout)

            except EOFError:

                raise error.Abort(

                    _(

                        b'Mercurial failed to run itself, check'

                        b' hg executable is in PATH'

                    )

                )

            try:

                orig_paths, revnum, author, date, message = entry

            except (TypeError, ValueError):

                if entry is None:

                    break

                raise error.Abort(_(b"log stream exception '%s'") % entry)

            yield entry

    def close(self):

        if self._stdout:

            self._stdout.close()

            self._stdout = None

class directlogstream(list):

    """Direct revision log iterator.

    This can be used for debugging and development but it will probably leak

    memory and is not suitable for real conversions."""

    def __init__(

        self,

        url,

        paths,

        start,

        end,

        limit=0,

        discover_changed_paths=True,

        strict_node_history=False,

    ):

        def receiver(orig_paths, revnum, author, date, message, pool):

            paths = {}

            if orig_paths is not None:

                    paths[k] = changedpath(v)

            self.append((paths, revnum, author, date, message))

        # Use an ra of our own so that our parent can consume

        # our results without confusing the server.

        t = transport.SvnRaTransport(url=url)

        svn.ra.get_log(

            t.ra,

            paths,

            start,

            end,

            limit,

            discover_changed_paths,

            strict_node_history,

            receiver,

        )

    def close(self):

        pass

# Check to see if the given path is a local Subversion repo. Verify this by

# looking for several svn-specific files and directories in the given

# directory.

def filecheck(ui, path, proto):

    for x in (b'locks', b'hooks', b'format', b'db'):

        if not os.path.exists(os.path.join(path, x)):

            return False

    return True

# Check to see if a given path is the root of an svn repo over http. We verify

# this by requesting a version-controlled URL we know can't exist and looking

# for the svn-specific "not found" XML.

def httpcheck(ui, path, proto):

    try:

        opener = urlreq.buildopener()

        rsp = opener.open(

            pycompat.strurl(b'%s://%s/!svn/ver/0/.svn' % (proto, path)), b'rb'

        )

        data = rsp.read()

    except urlerr.httperror as inst:

        if inst.code != 404:

            # Except for 404 we cannot know for sure this is not an svn repo

            ui.warn(

                _(

                    b'svn: cannot probe remote repository, assume it could '

                    b'be a subversion repository. Use --source-type if you '

                    b'know better.\n'

                )

            )

            return True

        data = inst.fp.read()

    except Exception:

        # Could be urlerr.urlerror if the URL is invalid or anything else.

        return False

    return b'<m:human-readable errcode="160013">' in data

protomap = {

    b'http': httpcheck,

    b'https': httpcheck,

    b'file': filecheck,

}

class NonUtf8PercentEncodedBytes(Exception):

    pass

# Subversion paths are Unicode. Since the percent-decoding is done on

# UTF-8-encoded strings, percent-encoded bytes are interpreted as UTF-8.

def url2pathname_like_subversion(unicodepath):

    if pycompat.ispy3:

        # On Python 3, we have to pass unicode to urlreq.url2pathname().

        # Percent-decoded bytes get decoded using UTF-8 and the 'replace' error

        # handler.

        unicodepath = urlreq.url2pathname(unicodepath)

        if u'\N{REPLACEMENT CHARACTER}' in unicodepath:

            raise NonUtf8PercentEncodedBytes

        else:

            return unicodepath

    else:

        # If we passed unicode on Python 2, it would be converted using the

        # latin-1 encoding. Therefore, we pass UTF-8-encoded bytes.

        unicodepath = urlreq.url2pathname(unicodepath.encode('utf-8'))

        try:

            return unicodepath.decode('utf-8')

        except UnicodeDecodeError:

            raise NonUtf8PercentEncodedBytes

def issvnurl(ui, url):

    try:

        proto, path = url.split(b'://', 1)

        if proto == b'file':

            if (

                pycompat.iswindows

                and path[:1] == b'/'