# admin/verify.py - better repository integrity checking for Mercurial

#

# Copyright 2023 Octobus <contact@octobus.net>

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

import collections

import copy

import functools

from ..i18n import _

from .. import error, pycompat, registrar, requirements

from ..utils import stringutil

verify_table = {}

verify_alias_table = {}

check = registrar.verify_check(verify_table, verify_alias_table)

# Use this to declare options/aliases in the middle of the hierarchy.

# Checks like these are not run themselves and cannot have a body.

# For an example, see the `revlogs` check.

def noop_func(*args, **kwargs):

    return

@check(b"working-copy.dirstate", alias=b"dirstate")

def check_dirstate(ui, repo, **options):

    ui.status(_(b"checking dirstate\n"))

    parent1, parent2 = repo.dirstate.parents()

    m1 = repo[parent1].manifest()

    m2 = repo[parent2].manifest()

    errors = 0

    is_narrow = requirements.NARROW_REQUIREMENT in repo.requirements

    narrow_matcher = repo.narrowmatch() if is_narrow else None

    for err in repo.dirstate.verify(m1, m2, narrow_matcher):

        ui.warn(err[0] % err[1:])

        errors += 1

    return errors

# Tree of all checks and their associated function

pyramid = {}

def build_pyramid(table, full_pyramid):

    """Create a pyramid of checks of the registered checks.

    It is a name-based hierarchy that can be arbitrarily nested."""

    for entry, func in sorted(table.items(), key=lambda x: x[0], reverse=True):

        cursor = full_pyramid

        levels = entry.split(b".")

        for level in levels[:-1]:

            current_node = cursor.setdefault(level, {})

            cursor = current_node

        if cursor.get(levels[-1]) is None:

            cursor[levels[-1]] = (entry, func)

        elif func is not noop_func:

            m = b"intermediate checks need to use `verify.noop_func`"

            raise error.ProgrammingError(m)

def find_checks(name, table=None, alias_table=None, full_pyramid=None):

    """Find all checks for a given name and returns a dict of

    (qualified_check_name, check_function)

    # Examples

    Using a full qualified name:

    "working-copy.dirstate" -> {

        "working-copy.dirstate": CF,

    }

    Using a *prefix* of a qualified name:

    "store.revlogs" -> {

        "store.revlogs.changelog": CF,

        "store.revlogs.manifestlog": CF,

        "store.revlogs.filelog": CF,

    }

    Using a defined alias:

    "revlogs" -> {

        "store.revlogs.changelog": CF,

        "store.revlogs.manifestlog": CF,

        "store.revlogs.filelog": CF,

    }

    Using something that is none of the above will be an error.

    """

    if table is None:

        table = verify_table

    if alias_table is None:

        alias_table = verify_alias_table

    if name == b"full":

        return table

    checks = {}

    # is it a full name?

    check = table.get(name)

    if check is None:

        # is it an alias?

        qualified_name = alias_table.get(name)

        if qualified_name is not None:

            name = qualified_name

            check = table.get(name)

        else:

            split = name.split(b".", 1)

            if len(split) == 2:

                # split[0] can be an alias

                qualified_name = alias_table.get(split[0])

                if qualified_name is not None:

                    name = b"%s.%s" % (qualified_name, split[1])

                    check = table.get(name)

    else:

        qualified_name = name

    # Maybe it's a subtree in the check hierarchy that does not

    # have an explicit alias.

    levels = name.split(b".")

    if full_pyramid is not None:

        if not full_pyramid:

            build_pyramid(table, full_pyramid)

        pyramid.clear()

        pyramid.update(full_pyramid.items())

    else:

        build_pyramid(table, pyramid)

    subtree = pyramid

    # Find subtree

    for level in levels:

        subtree = subtree.get(level)

        if subtree is None:

            hint = error.getsimilar(list(alias_table) + list(table), name)

            hint = error.similarity_hint(hint)

            raise error.InputError(_(b"unknown check %s" % name), hint=hint)

    # Get all checks in that subtree

    if isinstance(subtree, dict):

        stack = list(subtree.items())

        while stack:

            current_name, entry = stack.pop()

            if isinstance(entry, dict):

                stack.extend(entry.items())

            else:

                # (qualified_name, func)

                checks[entry[0]] = entry[1]

    else:

        checks[name] = check

    return checks

def pass_options(

    ui,

    checks,

    options,

    table=None,

    alias_table=None,

    full_pyramid=None,

):

    """Given a dict of checks (fully qualified name to function), and a list

    of options as given by the user, pass each option down to the right check

    function."""

    ui.debug(b"passing options to check functions\n")

    to_modify = collections.defaultdict(dict)

    if not checks:

        raise error.Error(_(b"`checks` required"))

    for option in sorted(options):

        split = option.split(b":")

        hint = _(

            b"syntax is 'check:option=value', "

            b"eg. revlogs.changelog:copies=yes"

        )

        option_error = error.InputError(

            _(b"invalid option '%s'") % option, hint=hint

        )

        if len(split) != 2:

            raise option_error

        check_name, option_value = split

        if not option_value:

            raise option_error

        split = option_value.split(b"=")

        if len(split) != 2:

            raise option_error

        option_name, value = split

        if not value:

            raise option_error

        path = b"%s:%s" % (check_name, option_name)

        matching_checks = find_checks(

            check_name,

            table=table,

            alias_table=alias_table,

            full_pyramid=full_pyramid,

        )

        for name in matching_checks:

            check = checks.get(name)

            if check is None:

                msg = _(b"specified option '%s' for unselected check '%s'\n")

                raise error.InputError(msg % (name, option_name))

            assert hasattr(check, "func")  # help Pytype

            if not hasattr(check.func, "options"):

                raise error.InputError(

                    _(b"check '%s' has no option '%s'") % (name, option_name)

                )

            try:

                matching_option = next(

                    (o for o in check.func.options if o[0] == option_name)

                )

            except StopIteration:

                raise error.InputError(

                    _(b"check '%s' has no option '%s'") % (name, option_name)

                )

            # transform the argument from cli string to the expected Python type

            _name, typ, _docstring = matching_option

            as_typed = None

            if isinstance(typ, bool):

                as_bool = stringutil.parsebool(value)

                if as_bool is None:

                    raise error.InputError(

                        _(b"'%s' is not a boolean ('%s')") % (path, value)

                    )

                as_typed = as_bool

            elif isinstance(typ, list):

                as_list = stringutil.parselist(value)

                if as_list is None:

                    raise error.InputError(

                        _(b"'%s' is not a list ('%s')") % (path, value)

                    )

                as_typed = as_list

            else:

                raise error.ProgrammingError(b"unsupported type %s", type(typ))

            if option_name in to_modify[name]:

                raise error.InputError(

                    _(b"duplicated option '%s' for '%s'") % (option_name, name)

                )

            else:

                assert as_typed is not None

                to_modify[name][option_name] = as_typed

    # Manage case where a check is set but without command line options

    # it will later be set with default check options values

    for name, f in checks.items():

        if name not in to_modify:

            to_modify[name] = {}

    # Merge default options with command line options

    for check_name, cmd_options in to_modify.items():

        check = checks.get(check_name)

        func = checks[check_name]

        merged_options = {}

        # help Pytype

        assert check is not None

        assert check.func is not None

        assert hasattr(check.func, "options")

        if check.func.options:

            # copy the default value in case it's mutable (list, etc.)

            merged_options = {

                o[0]: copy.deepcopy(o[1]) for o in check.func.options

            }

            if cmd_options:

                for k, v in cmd_options.items():

                    merged_options[k] = v

        options = pycompat.strkwargs(merged_options)

        checks[check_name] = functools.partial(func, **options)

        ui.debug(b"merged options for '%s': '%r'\n" % (check_name, options))

    return checks

def get_checks(

    repo,

    ui,

    names=None,

    options=None,

    table=None,

    alias_table=None,

    full_pyramid=None,

):

    """Given a list of function names and optionally a list of

    options, return matched checks with merged options (command line options

    values take precedence on default ones)

    It runs find checks, then resolve options and returns a dict of matched

    functions with resolved options.

    """

    funcs = {}

    if names is None:

        names = []

    if options is None:

        options = []

    # find checks

    for name in names:

        matched = find_checks(

            name,

            table=table,

            alias_table=alias_table,

            full_pyramid=full_pyramid,

        )

        matched_names = b", ".join(matched)

        ui.debug(b"found checks '%s' for name '%s'\n" % (matched_names, name))

        funcs.update(matched)

    funcs = {n: functools.partial(f, ui, repo) for n, f in funcs.items()}

    # resolve options

    checks = pass_options(

        ui,

        funcs,

        options,

        table=table,

        alias_table=alias_table,

        full_pyramid=full_pyramid,

    )

    return checks