annotate mercurial/upgrade_utils/auto_upgrade.py @ 52095:3e7b9357bbb8

tests: add coverage to for `HGCB_BUNDLE_BASENAME` with special characters

Per request on IRC, to show the behavior of dropping the quoting of
`HGCB_BUNDLE_BASENAME` in the next commit.  This current failure is basically
the same error and output that currently happens on Windows with any path (even
without the embedded quote).  The only difference is Windows doesn't print the
`cp: cannot stat ...` line.

# upgrade.py - functions for automatic upgrade of Mercurial repository
#
# Copyright (c) 2022-present, Pierre-Yves David
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.

from __future__ import annotations

from ..i18n import _

from .. import (
    error,
    requirements as requirementsmod,
    scmutil,
)

from . import (
    actions,
    engine,
)


class AutoUpgradeOperation(actions.BaseOperation):
    """A limited Upgrade Operation used to run simple auto upgrade task

    (Expand it as needed in the future)
    """

    def __init__(self, req):
        super().__init__(
            new_requirements=req,
            backup_store=False,
        )


def get_share_safe_action(repo):
    """return an automatic-upgrade action for `share-safe` if applicable

    If no action is needed, return None, otherwise return a callback to upgrade
    or downgrade the repository according the configuration and repository
    format.
    """
    ui = repo.ui
    requirements = repo.requirements
    auto_upgrade_share_source = ui.configbool(
        b'format',
        b'use-share-safe.automatic-upgrade-of-mismatching-repositories',
    )
    auto_upgrade_quiet = ui.configbool(
        b'format',
        b'use-share-safe.automatic-upgrade-of-mismatching-repositories:quiet',
    )

    action = None

    if (
        auto_upgrade_share_source
        and requirementsmod.SHARED_REQUIREMENT not in requirements
    ):
        sf_config = ui.configbool(b'format', b'use-share-safe')
        sf_local = requirementsmod.SHARESAFE_REQUIREMENT in requirements
        if sf_config and not sf_local:
            msg = _(
                b"automatically upgrading repository to the `share-safe`"
                b" feature\n"
            )
            hint = b"(see `hg help config.format.use-share-safe` for details)\n"

            def action():
                if not (ui.quiet or auto_upgrade_quiet):
                    ui.write_err(msg)
                    ui.write_err(hint)
                requirements.add(requirementsmod.SHARESAFE_REQUIREMENT)
                scmutil.writereporequirements(repo, requirements)

        elif sf_local and not sf_config:
            msg = _(
                b"automatically downgrading repository from the `share-safe`"
                b" feature\n"
            )
            hint = b"(see `hg help config.format.use-share-safe` for details)\n"

            def action():
                if not (ui.quiet or auto_upgrade_quiet):
                    ui.write_err(msg)
                    ui.write_err(hint)
                requirements.discard(requirementsmod.SHARESAFE_REQUIREMENT)
                scmutil.writereporequirements(repo, requirements)

    return action


def get_tracked_hint_action(repo):
    """return an automatic-upgrade action for `tracked-hint` if applicable

    If no action is needed, return None, otherwise return a callback to upgrade
    or downgrade the repository according the configuration and repository
    format.
    """
    ui = repo.ui
    requirements = set(repo.requirements)
    auto_upgrade_tracked_hint = ui.configbool(
        b'format',
        b'use-dirstate-tracked-hint.automatic-upgrade-of-mismatching-repositories',
    )
    auto_upgrade_quiet = ui.configbool(
        b'format',
        b'use-dirstate-tracked-hint.automatic-upgrade-of-mismatching-repositories:quiet',
    )

    action = None

    if auto_upgrade_tracked_hint:
        th_config = ui.configbool(b'format', b'use-dirstate-tracked-hint')
        th_local = requirementsmod.DIRSTATE_TRACKED_HINT_V1 in requirements
        if th_config and not th_local:
            msg = _(
                b"automatically upgrading repository to the `tracked-hint`"
                b" feature\n"
            )
            hint = b"(see `hg help config.format.use-dirstate-tracked-hint` for details)\n"

            def action():
                if not (ui.quiet or auto_upgrade_quiet):
                    ui.write_err(msg)
                    ui.write_err(hint)
                requirements.add(requirementsmod.DIRSTATE_TRACKED_HINT_V1)
                op = AutoUpgradeOperation(requirements)
                engine.upgrade_tracked_hint(ui, repo, op, add=True)

        elif th_local and not th_config:
            msg = _(
                b"automatically downgrading repository from the `tracked-hint`"
                b" feature\n"
            )
            hint = b"(see `hg help config.format.use-dirstate-tracked-hint` for details)\n"

            def action():
                if not (ui.quiet or auto_upgrade_quiet):
                    ui.write_err(msg)
                    ui.write_err(hint)
                requirements.discard(requirementsmod.DIRSTATE_TRACKED_HINT_V1)
                op = AutoUpgradeOperation(requirements)
                engine.upgrade_tracked_hint(ui, repo, op, add=False)

    return action


def get_dirstate_v2_action(repo):
    """return an automatic-upgrade action for `dirstate-v2` if applicable

    If no action is needed, return None, otherwise return a callback to upgrade
    or downgrade the repository according the configuration and repository
    format.
    """
    ui = repo.ui
    requirements = set(repo.requirements)
    auto_upgrade_dv2 = ui.configbool(
        b'format',
        b'use-dirstate-v2.automatic-upgrade-of-mismatching-repositories',
    )
    auto_upgrade_dv2_quiet = ui.configbool(
        b'format',
        b'use-dirstate-v2.automatic-upgrade-of-mismatching-repositories:quiet',
    )

    action = None

    if auto_upgrade_dv2:
        d2_config = ui.configbool(b'format', b'use-dirstate-v2')
        d2_local = requirementsmod.DIRSTATE_V2_REQUIREMENT in requirements
        if d2_config and not d2_local:
            msg = _(
                b"automatically upgrading repository to the `dirstate-v2`"
                b" feature\n"
            )
            hint = (
                b"(see `hg help config.format.use-dirstate-v2` for details)\n"
            )

            def action():
                if not (ui.quiet or auto_upgrade_dv2_quiet):
                    ui.write_err(msg)
                    ui.write_err(hint)
                requirements.add(requirementsmod.DIRSTATE_V2_REQUIREMENT)
                fake_op = AutoUpgradeOperation(requirements)
                engine.upgrade_dirstate(repo.ui, repo, fake_op, b'v1', b'v2')

        elif d2_local and not d2_config:
            msg = _(
                b"automatically downgrading repository from the `dirstate-v2`"
                b" feature\n"
            )
            hint = (
                b"(see `hg help config.format.use-dirstate-v2` for details)\n"
            )

            def action():
                if not (ui.quiet or auto_upgrade_dv2_quiet):
                    ui.write_err(msg)
                    ui.write_err(hint)
                requirements.discard(requirementsmod.DIRSTATE_V2_REQUIREMENT)
                fake_op = AutoUpgradeOperation(requirements)
                engine.upgrade_dirstate(repo.ui, repo, fake_op, b'v2', b'v1')

    return action


AUTO_UPGRADE_ACTIONS = [
    get_dirstate_v2_action,
    get_share_safe_action,
    get_tracked_hint_action,
]


def may_auto_upgrade(repo, maker_func):
    """potentially perform auto-upgrade and return the final repository to use

    Auto-upgrade are "quick" repository upgrade that might automatically be run
    by "any" repository access. See `hg help config.format` for automatic
    upgrade documentation.

    note: each relevant upgrades are done one after the other for simplicity.
    This avoid having repository is partially inconsistent state while
    upgrading.

    repo: the current repository instance
    maker_func: a factory function that can recreate a repository after an upgrade
    """
    clear = False

    loop = 0

    try:
        while not clear:
            loop += 1
            if loop > 100:
                # XXX basic protection against infinite loop, make it better.
                raise error.ProgrammingError("Too many auto upgrade loops")
            clear = True
            for get_action in AUTO_UPGRADE_ACTIONS:
                action = get_action(repo)
                if action is not None:
                    clear = False
                    with repo.wlock(wait=False), repo.lock(wait=False):
                        action = get_action(repo)
                        if action is not None:
                            action()
                        repo = maker_func()
    except error.LockError:
        # if we cannot get the lock, ignore the auto-upgrade attemps and
        # proceed. We might want to make this behavior configurable in the
        # future.
        pass

    return repo