comparison mercurial/subrepoutil.py @ 51287:f15cb5111a1e

pytype: move some type comment to proper annotation

We support direct type annotations now, while pytype is starting to complains
about them.

51286:81224afd938d

    ]

Substate = Dict[bytes, Tuple[bytes, bytes, bytes]]


def state(ctx, ui):
    # type: (context.changectx, uimod.ui) -> Substate
    """return a state dict, mapping subrepo paths configured in .hgsub
    to tuple: (source from .hgsub, revision from .hgsubstate, kind
    (key in types dict))
    """
    p = config.config()

                    )
                rev[path] = revision
        except FileNotFoundError:
            pass

    def remap(src):
        # type: (bytes) -> bytes
        for pattern, repl in p.items(b'subpaths'):
            # Turn r'C:\foo\bar' into r'C:\\foo\\bar' since re.sub
            # does a string decode.
            repl = stringutil.escapestr(repl)
            # However, we still want to allow back references to go

        state[util.pconvert(path)] = (src.strip(), rev.get(path, b''), kind)

    return state


def writestate(repo, state):
    # type: (localrepo.localrepository, Substate) -> None
    """rewrite .hgsubstate in (outer) repo with these subrepo states"""
    lines = [
        b'%s %s\n' % (state[s][1], s)
        for s in sorted(state)
        if state[s][1] != nullstate[1]
    ]
    repo.wwrite(b'.hgsubstate', b''.join(lines), b'')


def submerge(repo, wctx, mctx, actx, overwrite, labels=None):
    # type: (localrepo.localrepository, context.workingctx, context.changectx, context.changectx, bool, Optional[Any]) -> Substate
    # TODO: type the `labels` arg
    """delegated from merge.applyupdates: merging of .hgsubstate file
    in working context, merging context and ancestor context"""
    if mctx == actx:  # backwards?
        actx = wctx.p1()

    # record merged .hgsubstate
    writestate(repo, sm)
    return sm


def precommit(ui, wctx, status, match, force=False):
    # type: (uimod.ui, context.workingcommitctx, scmutil.status, matchmod.basematcher, bool) -> Tuple[List[bytes], Set[bytes], Substate]
    """Calculate .hgsubstate changes that should be applied before committing

    Returns (subs, commitsubs, newstate) where
    - subs: changed subrepos (including dirty ones)
    - commitsubs: dirty subrepos which the caller needs to commit recursively

    chunks.reverse()
    path = posixpath.join(*chunks)
    return posixpath.normpath(path)


def reporelpath(repo):
    # type: (localrepo.localrepository) -> bytes
    """return path to this (sub)repo as seen from outermost repo"""
    parent = repo
    while hasattr(parent, '_subparent'):
        parent = parent._subparent
    return repo.root[len(pathutil.normasprefix(parent.root)) :]


def subrelpath(sub):
    # type: (subrepo.abstractsubrepo) -> bytes
    """return path to this subrepo as seen from outermost repo"""
    return sub._relpath


def _abssource(repo, push=False, abort=True):
    # type: (localrepo.localrepository, bool, bool) -> Optional[bytes]
    """return pull/push path of repo - either based on parent repo .hgsub info
    or on the top repo config. Abort or return None if no source found."""
    if hasattr(repo, '_subparent'):
        source = urlutil.url(repo._subsource)
        if source.isabs():

    if abort:
        raise error.Abort(_(b"default path for subrepository not found"))


def newcommitphase(ui, ctx):
    # type: (uimod.ui, context.changectx) -> int
    commitphase = phases.newcommitphase(ui)
    substate = getattr(ctx, "substate", None)
    if not substate:
        return commitphase
    check = ui.config(b'phases', b'checksubrepos')

51287:f15cb5111a1e

    ]

Substate = Dict[bytes, Tuple[bytes, bytes, bytes]]


def state(ctx: "context.changectx", ui: "uimod.ui") -> Substate:
    """return a state dict, mapping subrepo paths configured in .hgsub
    to tuple: (source from .hgsub, revision from .hgsubstate, kind
    (key in types dict))
    """
    p = config.config()

                    )
                rev[path] = revision
        except FileNotFoundError:
            pass

    def remap(src: bytes) -> bytes:
        for pattern, repl in p.items(b'subpaths'):
            # Turn r'C:\foo\bar' into r'C:\\foo\\bar' since re.sub
            # does a string decode.
            repl = stringutil.escapestr(repl)
            # However, we still want to allow back references to go

        state[util.pconvert(path)] = (src.strip(), rev.get(path, b''), kind)

    return state


def writestate(repo: "localrepo.localrepository", state: Substate) -> None:
    """rewrite .hgsubstate in (outer) repo with these subrepo states"""
    lines = [
        b'%s %s\n' % (state[s][1], s)
        for s in sorted(state)
        if state[s][1] != nullstate[1]
    ]
    repo.wwrite(b'.hgsubstate', b''.join(lines), b'')


def submerge(
    repo: "localrepo.localrepository",
    wctx: "context.workingctx",
    mctx: "context.changectx",
    actx: "context.changectx",
    overwrite: bool,
    labels: Optional[Any] = None,
) -> Substate:
    # TODO: type the `labels` arg
    """delegated from merge.applyupdates: merging of .hgsubstate file
    in working context, merging context and ancestor context"""
    if mctx == actx:  # backwards?
        actx = wctx.p1()

    # record merged .hgsubstate
    writestate(repo, sm)
    return sm


def precommit(
    ui: "uimod.ui",
    wctx: "context.workingcommitctx",
    status: "scmutil.status",
    match: "matchmod.basematcher",
    force: bool = False,
) -> Tuple[List[bytes], Set[bytes], Substate]:
    """Calculate .hgsubstate changes that should be applied before committing

    Returns (subs, commitsubs, newstate) where
    - subs: changed subrepos (including dirty ones)
    - commitsubs: dirty subrepos which the caller needs to commit recursively

    chunks.reverse()
    path = posixpath.join(*chunks)
    return posixpath.normpath(path)


def reporelpath(repo: "localrepo.localrepository") -> bytes:
    """return path to this (sub)repo as seen from outermost repo"""
    parent = repo
    while hasattr(parent, '_subparent'):
        parent = parent._subparent
    return repo.root[len(pathutil.normasprefix(parent.root)) :]


def subrelpath(sub: "subrepo.abstractsubrepo") -> bytes:
    """return path to this subrepo as seen from outermost repo"""
    return sub._relpath


def _abssource(
    repo: "localrepo.localrepository",
    push: bool = False,
    abort: bool = True,
) -> Optional[bytes]:
    """return pull/push path of repo - either based on parent repo .hgsub info
    or on the top repo config. Abort or return None if no source found."""
    if hasattr(repo, '_subparent'):
        source = urlutil.url(repo._subsource)
        if source.isabs():

    if abort:
        raise error.Abort(_(b"default path for subrepository not found"))


def newcommitphase(ui: "uimod.ui", ctx: "context.changectx") -> int:
    commitphase = phases.newcommitphase(ui)
    substate = getattr(ctx, "substate", None)
    if not substate:
        return commitphase
    check = ui.config(b'phases', b'checksubrepos')