comparison mercurial/hg.py @ 46907:ffd3e823a7e5

urlutil: extract `url` related code from `util` into the new module

The new module is well fitting for this new code. And this will be useful to
make the gathered code collaborate more later.

Differential Revision: https://phab.mercurial-scm.org/D10374

46906:33524c46a092

    vfs as vfsmod,
)
from .utils import (
    hashutil,
    stringutil,
)


release = lock.release

# shared features
sharedbookmarks = b'bookmarks'


def _local(path):
    path = util.expandpath(util.urllocalpath(path))

    try:
        # we use os.stat() directly here instead of os.path.isfile()
        # because the latter started returning `False` on invalid path
        # exceptions starting in 3.8 and we care about handling

def parseurl(path, branches=None):
    '''parse url#branch, returning (url, (branch, branches))'''

    u = util.url(path)
    branch = None
    if u.fragment:
        branch = u.fragment
        u.fragment = None
    return bytes(u), (branch, branches or [])

    b'static-http': statichttprepo,
}


def _peerlookup(path):
    u = util.url(path)
    scheme = u.scheme or b'file'
    thing = schemes.get(scheme) or schemes[b'file']
    try:
        return thing(path)
    except TypeError:

    return repo.local()


def openpath(ui, path, sendaccept=True):
    '''open path with open if local, url.open if remote'''
    pathurl = util.url(path, parsequery=False, parsefragment=False)
    if pathurl.islocal():
        return util.posixfile(pathurl.localpath(), b'rb')
    else:
        return url.open(ui, path, sendaccept=sendaccept)

    >>> defaultdest(b'http://example.org/')
    ''
    >>> defaultdest(b'http://example.org/foo/')
    'foo'
    """
    path = util.url(source).path
    if not path:
        return b''
    return os.path.basename(os.path.normpath(path))

                stream=stream,
            )

    # Resolve the value to put in [paths] section for the source.
    if islocal(source):
        defaultpath = os.path.abspath(util.urllocalpath(source))
    else:
        defaultpath = source

    sharerepo = repository(ui, path=sharepath)
    destrepo = share(

            if dest:
                ui.status(_(b"destination directory: %s\n") % dest)
        else:
            dest = ui.expandpath(dest)

        dest = util.urllocalpath(dest)
        source = util.urllocalpath(source)

        if not dest:
            raise error.InputError(_(b"empty destination path is not valid"))

        destvfs = vfsmod.vfs(dest, expandpath=True)

        srcrepo = srcpeer.local()

        abspath = origsource
        if islocal(origsource):
            abspath = os.path.abspath(util.urllocalpath(origsource))

        if islocal(dest):
            cleandir = dest

        copy = False

                if narrow:
                    with local.wlock(), local.lock():
                        local.setnarrowpats(storeincludepats, storeexcludepats)
                        narrowspec.copytoworkingcopy(local)

                u = util.url(abspath)
                defaulturl = bytes(u)
                local.ui.setconfig(b'paths', b'default', defaulturl, b'clone')
                if not stream:
                    if pull:
                        stream = False

        cleandir = None

        destrepo = destpeer.local()
        if destrepo:
            template = uimod.samplehgrcs[b'cloned']
            u = util.url(abspath)
            u.passwd = None
            defaulturl = bytes(u)
            destrepo.vfs.write(b'hgrc', util.tonativeeol(template % defaulturl))
            destrepo.ui.setconfig(b'paths', b'default', defaulturl, b'clone')

    """
    source, branches = parseurl(ui.expandpath(source), opts.get(b'branch'))
    other = peer(repo, opts, source)
    cleanupfn = other.close
    try:
        ui.status(_(b'comparing with %s\n') % util.hidepassword(source))
        revs, checkout = addbranchrevs(repo, other, branches, opts.get(b'rev'))

        if revs:
            revs = [other.lookup(rev) for rev in revs]
        other, chlist, cleanupfn = bundlerepo.getremotechanges(

            hint=_(b"see 'hg help config.paths'"),
        )
    dest = path.pushloc or path.loc
    branches = path.branch, opts.get(b'branch') or []

    ui.status(_(b'comparing with %s\n') % util.hidepassword(dest))
    revs, checkout = addbranchrevs(repo, repo, branches, opts.get(b'rev'))
    if revs:
        revs = [repo[rev].node() for rev in scmutil.revrange(repo, revs)]

    other = peer(repo, opts, dest)

46907:ffd3e823a7e5

    vfs as vfsmod,
)
from .utils import (
    hashutil,
    stringutil,
    urlutil,
)


release = lock.release

# shared features
sharedbookmarks = b'bookmarks'


def _local(path):
    path = util.expandpath(urlutil.urllocalpath(path))

    try:
        # we use os.stat() directly here instead of os.path.isfile()
        # because the latter started returning `False` on invalid path
        # exceptions starting in 3.8 and we care about handling

def parseurl(path, branches=None):
    '''parse url#branch, returning (url, (branch, branches))'''

    u = urlutil.url(path)
    branch = None
    if u.fragment:
        branch = u.fragment
        u.fragment = None
    return bytes(u), (branch, branches or [])

    b'static-http': statichttprepo,
}


def _peerlookup(path):
    u = urlutil.url(path)
    scheme = u.scheme or b'file'
    thing = schemes.get(scheme) or schemes[b'file']
    try:
        return thing(path)
    except TypeError:

    return repo.local()


def openpath(ui, path, sendaccept=True):
    '''open path with open if local, url.open if remote'''
    pathurl = urlutil.url(path, parsequery=False, parsefragment=False)
    if pathurl.islocal():
        return util.posixfile(pathurl.localpath(), b'rb')
    else:
        return url.open(ui, path, sendaccept=sendaccept)

    >>> defaultdest(b'http://example.org/')
    ''
    >>> defaultdest(b'http://example.org/foo/')
    'foo'
    """
    path = urlutil.url(source).path
    if not path:
        return b''
    return os.path.basename(os.path.normpath(path))

                stream=stream,
            )

    # Resolve the value to put in [paths] section for the source.
    if islocal(source):
        defaultpath = os.path.abspath(urlutil.urllocalpath(source))
    else:
        defaultpath = source

    sharerepo = repository(ui, path=sharepath)
    destrepo = share(

            if dest:
                ui.status(_(b"destination directory: %s\n") % dest)
        else:
            dest = ui.expandpath(dest)

        dest = urlutil.urllocalpath(dest)
        source = urlutil.urllocalpath(source)

        if not dest:
            raise error.InputError(_(b"empty destination path is not valid"))

        destvfs = vfsmod.vfs(dest, expandpath=True)

        srcrepo = srcpeer.local()

        abspath = origsource
        if islocal(origsource):
            abspath = os.path.abspath(urlutil.urllocalpath(origsource))

        if islocal(dest):
            cleandir = dest

        copy = False

                if narrow:
                    with local.wlock(), local.lock():
                        local.setnarrowpats(storeincludepats, storeexcludepats)
                        narrowspec.copytoworkingcopy(local)

                u = urlutil.url(abspath)
                defaulturl = bytes(u)
                local.ui.setconfig(b'paths', b'default', defaulturl, b'clone')
                if not stream:
                    if pull:
                        stream = False

        cleandir = None

        destrepo = destpeer.local()
        if destrepo:
            template = uimod.samplehgrcs[b'cloned']
            u = urlutil.url(abspath)
            u.passwd = None
            defaulturl = bytes(u)
            destrepo.vfs.write(b'hgrc', util.tonativeeol(template % defaulturl))
            destrepo.ui.setconfig(b'paths', b'default', defaulturl, b'clone')

    """
    source, branches = parseurl(ui.expandpath(source), opts.get(b'branch'))
    other = peer(repo, opts, source)
    cleanupfn = other.close
    try:
        ui.status(_(b'comparing with %s\n') % urlutil.hidepassword(source))
        revs, checkout = addbranchrevs(repo, other, branches, opts.get(b'rev'))

        if revs:
            revs = [other.lookup(rev) for rev in revs]
        other, chlist, cleanupfn = bundlerepo.getremotechanges(

            hint=_(b"see 'hg help config.paths'"),
        )
    dest = path.pushloc or path.loc
    branches = path.branch, opts.get(b'branch') or []

    ui.status(_(b'comparing with %s\n') % urlutil.hidepassword(dest))
    revs, checkout = addbranchrevs(repo, repo, branches, opts.get(b'rev'))
    if revs:
        revs = [repo[rev].node() for rev in scmutil.revrange(repo, revs)]

    other = peer(repo, opts, dest)