admin-command: add verify command

Start using the 'admin' namespace by adding a 'verify' command.
Invocation is 'admin::verify'.
The idea is to progressively add more focused checks than the existing
verify command.
To do so we need an advanced way to express what we want to check.

The first check for admin::verify is 'working-copy.dirstate' which has
no options, because it was an easy first check to implement, which
verifies the integrity of the dirstate.

This changeset was created with the help of Franck Bret.

mercurial/admin/verify.py

@@ -0,0 +1,341 b''
+# admin/verify.py - better repository integrity checking for Mercurial
+#
+# Copyright 2023 Octobus <[e-mail]>
+#
+# 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

tests/test-admin-commands.py

@@ -0,0 +1,399 b''
+# Test admin commands
+
+import functools
+import unittest
+from mercurial.i18n import _
+from mercurial import error, ui as uimod
+from mercurial import registrar
+from mercurial.admin import verify
+
+
+class TestAdminVerifyFindChecks(unittest.TestCase):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.ui = uimod.ui.load()
+        self.repo = b"fake-repo"
+
+        def cleanup_table(self):
+            self.table = {}
+            self.alias_table = {}
+            self.pyramid = {}
+
+        self.addCleanup(cleanup_table, self)
+
+    def setUp(self):
+        self.table = {}
+        self.alias_table = {}
+        self.pyramid = {}
+        check = registrar.verify_check(self.table, self.alias_table)
+
+        # mock some fake check method for tests purpose
+        @check(
+            b"test.dummy",
+            alias=b"dummy",
+            options=[],
+        )
+        def check_dummy(ui, repo, **options):
+            return options
+
+        @check(
+            b"test.fake",
+            alias=b"fake",
+            options=[
+                (b'a', False, _(b'a boolean value (default: False)')),
+                (b'b', True, _(b'a boolean value (default: True)')),
+                (b'c', [], _(b'a list')),
+            ],
+        )
+        def check_fake(ui, repo, **options):
+            return options
+
+        # alias in the middle of a hierarchy
+        check(
+            b"test.noop",
+            alias=b"noop",
+            options=[],
+        )(verify.noop_func)
+
+        @check(
+            b"test.noop.deeper",
+            alias=b"deeper",
+            options=[
+                (b'y', True, _(b'a boolean value (default: True)')),
+                (b'z', [], _(b'a list')),
+            ],
+        )
+        def check_noop_deeper(ui, repo, **options):
+            return options
+
+    # args wrapper utilities
+    def find_checks(self, name):
+        return verify.find_checks(
+            name=name,
+            table=self.table,
+            alias_table=self.alias_table,
+            full_pyramid=self.pyramid,
+        )
+
+    def pass_options(self, checks, options):
+        return verify.pass_options(
+            self.ui,
+            checks,
+            options,
+            table=self.table,
+            alias_table=self.alias_table,
+            full_pyramid=self.pyramid,
+        )
+
+    def get_checks(self, names, options):
+        return verify.get_checks(
+            self.repo,
+            self.ui,
+            names=names,
+            options=options,
+            table=self.table,
+            alias_table=self.alias_table,
+            full_pyramid=self.pyramid,
+        )
+
+    # tests find_checks
+    def test_find_checks_empty_name(self):
+        with self.assertRaises(error.InputError):
+            self.find_checks(name=b"")
+
+    def test_find_checks_wrong_name(self):
+        with self.assertRaises(error.InputError):
+            self.find_checks(name=b"unknown")
+
+    def test_find_checks_dummy(self):
+        name = b"test.dummy"
+        found = self.find_checks(name=name)
+        self.assertEqual(len(found), 1)
+        self.assertIn(name, found)
+        meth = found[name]
+        self.assertTrue(callable(meth))
+        self.assertEqual(len(meth.options), 0)
+
+    def test_find_checks_fake(self):
+        name = b"test.fake"
+        found = self.find_checks(name=name)
+        self.assertEqual(len(found), 1)
+        self.assertIn(name, found)
+        meth = found[name]
+        self.assertTrue(callable(meth))
+        self.assertEqual(len(meth.options), 3)
+
+    def test_find_checks_noop(self):
+        name = b"test.noop.deeper"
+        found = self.find_checks(name=name)
+        self.assertEqual(len(found), 1)
+        self.assertIn(name, found)
+        meth = found[name]
+        self.assertTrue(callable(meth))
+        self.assertEqual(len(meth.options), 2)
+
+    def test_find_checks_from_aliases(self):
+        found = self.find_checks(name=b"dummy")
+        self.assertEqual(len(found), 1)
+        self.assertIn(b"test.dummy", found)
+
+        found = self.find_checks(name=b"fake")
+        self.assertEqual(len(found), 1)
+        self.assertIn(b"test.fake", found)
+
+        found = self.find_checks(name=b"deeper")
+        self.assertEqual(len(found), 1)
+        self.assertIn(b"test.noop.deeper", found)
+
+    def test_find_checks_from_root(self):
+        found = self.find_checks(name=b"test")
+        self.assertEqual(len(found), 3)
+        self.assertIn(b"test.dummy", found)
+        self.assertIn(b"test.fake", found)
+        self.assertIn(b"test.noop.deeper", found)
+
+    def test_find_checks_from_intermediate(self):
+        found = self.find_checks(name=b"test.noop")
+        self.assertEqual(len(found), 1)
+        self.assertIn(b"test.noop.deeper", found)
+
+    def test_find_checks_from_parent_dot_name(self):
+        found = self.find_checks(name=b"noop.deeper")
+        self.assertEqual(len(found), 1)
+        self.assertIn(b"test.noop.deeper", found)
+
+    # tests pass_options
+    def test_pass_options_no_checks_no_options(self):
+        checks = {}
+        options = []
+
+        with self.assertRaises(error.Error):
+            self.pass_options(checks=checks, options=options)
+
+    def test_pass_options_fake_empty_options(self):
+        checks = self.find_checks(name=b"test.fake")
+        funcs = {
+            n: functools.partial(f, self.ui, self.repo)
+            for n, f in checks.items()
+        }
+        options = []
+        # should end with default options
+        expected_options = {"a": False, "b": True, "c": []}
+        func = self.pass_options(checks=funcs, options=options)
+
+        self.assertDictEqual(func[b"test.fake"].keywords, expected_options)
+
+    def test_pass_options_fake_non_existing_options(self):
+        checks = self.find_checks(name=b"test.fake")
+        funcs = {
+            n: functools.partial(f, self.ui, self.repo)
+            for n, f in checks.items()
+        }
+
+        with self.assertRaises(error.InputError):
+            options = [b"test.fake:boom=yes"]
+            self.pass_options(checks=funcs, options=options)
+
+    def test_pass_options_fake_unrelated_options(self):
+        checks = self.find_checks(name=b"test.fake")
+        funcs = {
+            n: functools.partial(f, self.ui, self.repo)
+            for n, f in checks.items()
+        }
+        options = [b"test.noop.deeper:y=yes"]
+
+        with self.assertRaises(error.InputError):
+            self.pass_options(checks=funcs, options=options)
+
+    def test_pass_options_fake_set_option(self):
+        checks = self.find_checks(name=b"test.fake")
+        funcs = {
+            n: functools.partial(f, self.ui, self.repo)
+            for n, f in checks.items()
+        }
+        options = [b"test.fake:a=yes"]
+        expected_options = {"a": True, "b": True, "c": []}
+        func = self.pass_options(checks=funcs, options=options)
+
+        self.assertDictEqual(func[b"test.fake"].keywords, expected_options)
+
+    def test_pass_options_fake_set_option_with_alias(self):
+        checks = self.find_checks(name=b"test.fake")
+        funcs = {
+            n: functools.partial(f, self.ui, self.repo)
+            for n, f in checks.items()
+        }
+        options = [b"fake:a=yes"]
+        expected_options = {"a": True, "b": True, "c": []}
+        func = self.pass_options(checks=funcs, options=options)
+
+        self.assertDictEqual(func[b"test.fake"].keywords, expected_options)
+
+    def test_pass_options_fake_set_all_option(self):
+        checks = self.find_checks(name=b"test.fake")
+        funcs = {
+            n: functools.partial(f, self.ui, self.repo)
+            for n, f in checks.items()
+        }
+        options = [b"test.fake:a=yes", b"test.fake:b=no", b"test.fake:c=0,1,2"]
+        expected_options = {"a": True, "b": False, "c": [b"0", b"1", b"2"]}
+        func = self.pass_options(checks=funcs, options=options)
+
+        self.assertDictEqual(func[b"test.fake"].keywords, expected_options)
+
+    def test_pass_options_fake_set_all_option_plus_unexisting(self):
+        checks = self.find_checks(name=b"test.fake")
+        funcs = {
+            n: functools.partial(f, self.ui, self.repo)
+            for n, f in checks.items()
+        }
+        options = [
+            b"test.fake:a=yes",
+            b"test.fake:b=no",
+            b"test.fake:c=0,1,2",
+            b"test.fake:d=0",
+        ]
+
+        with self.assertRaises(error.InputError):
+            self.pass_options(checks=funcs, options=options)
+
+    def test_pass_options_fake_duplicate_option(self):
+        checks = self.find_checks(name=b"test.fake")
+        funcs = {
+            n: functools.partial(f, self.ui, self.repo)
+            for n, f in checks.items()
+        }
+        options = [
+            b"test.fake:a=yes",
+            b"test.fake:a=no",
+        ]
+
+        with self.assertRaises(error.InputError):
+            self.pass_options(checks=funcs, options=options)
+
+    def test_pass_options_fake_set_malformed_option(self):
+        checks = self.find_checks(name=b"test.fake")
+        funcs = {
+            n: functools.partial(f, self.ui, self.repo)
+            for n, f in checks.items()
+        }
+        options = [
+            b"test.fake:ayes",
+            b"test.fake:b==no",
+            b"test.fake=",
+            b"test.fake:",
+            b"test.fa=ke:d=0",
+            b"test.fa=ke:d=0",
+        ]
+
+        for opt in options:
+            with self.assertRaises(error.InputError):
+                self.pass_options(checks=funcs, options=[opt])
+
+    def test_pass_options_types(self):
+        checks = self.find_checks(name=b"test.fake")
+        funcs = {
+            n: functools.partial(f, self.ui, self.repo)
+            for n, f in checks.items()
+        }
+        # boolean, yes/no
+        options = [b"test.fake:a=yes", b"test.fake:b=no"]
+        expected_options = {"a": True, "b": False, "c": []}
+        func = self.pass_options(checks=funcs, options=options)
+
+        self.assertDictEqual(func[b"test.fake"].keywords, expected_options)
+
+        # boolean, 0/1
+        options = [b"test.fake:a=1", b"test.fake:b=0"]
+        expected_options = {"a": True, "b": False, "c": []}
+        func = self.pass_options(checks=funcs, options=options)
+
+        self.assertDictEqual(func[b"test.fake"].keywords, expected_options)
+
+        # boolean, true/false
+        options = [b"test.fake:a=true", b"test.fake:b=false"]
+        expected_options = {"a": True, "b": False, "c": []}
+        func = self.pass_options(checks=funcs, options=options)
+
+        self.assertDictEqual(func[b"test.fake"].keywords, expected_options)
+
+        # boolean, wrong type
+        options = [b"test.fake:a=si"]
+        with self.assertRaises(error.InputError):
+            self.pass_options(checks=funcs, options=options)
+
+        # lists
+        options = [b"test.fake:c=0,1,2"]
+        expected_options = {"a": False, "b": True, "c": [b"0", b"1", b"2"]}
+        func = self.pass_options(checks=funcs, options=options)
+
+        self.assertDictEqual(func[b"test.fake"].keywords, expected_options)
+
+        options = [b"test.fake:c=x,y,z"]
+        expected_options = {"a": False, "b": True, "c": [b"x", b"y", b"z"]}
+        func = self.pass_options(checks=funcs, options=options)
+
+        self.assertDictEqual(func[b"test.fake"].keywords, expected_options)
+
+    # tests get_checks
+    def test_get_checks_fake(self):
+        funcs = self.get_checks(
+            names=[b"test.fake"], options=[b"test.fake:a=yes"]
+        )
+        options = funcs.get(b"test.fake").keywords
+        expected_options = {"a": True, "b": True, "c": []}
+        self.assertDictEqual(options, expected_options)
+
+    def test_get_checks_multiple_mixed_with_defaults(self):
+        funcs = self.get_checks(
+            names=[b"test.fake", b"test.noop.deeper", b"test.dummy"],
+            options=[
+                b"test.noop.deeper:y=no",
+                b"test.noop.deeper:z=-1,0,1",
+            ],
+        )
+        options = funcs.get(b"test.fake").keywords
+        expected_options = {"a": False, "b": True, "c": []}
+        self.assertDictEqual(options, expected_options)
+
+        options = funcs.get(b"test.noop.deeper").keywords
+        expected_options = {"y": False, "z": [b"-1", b"0", b"1"]}
+        self.assertDictEqual(options, expected_options)
+
+        options = funcs.get(b"test.dummy").keywords
+        expected_options = {}
+        self.assertDictEqual(options, expected_options)
+
+    def test_broken_pyramid(self):
+        """Check that we detect pyramids that can't resolve"""
+        table = {}
+        alias_table = {}
+        pyramid = {}
+        check = registrar.verify_check(table, alias_table)
+
+        # Create two checks that clash
+        @check(b"test.wrong.intermediate")
+        def check_dummy(ui, repo, **options):
+            return options
+
+        @check(b"test.wrong.intermediate.thing")
+        def check_fake(ui, repo, **options):
+            return options
+
+        with self.assertRaises(error.ProgrammingError) as e:
+            verify.get_checks(
+                self.repo,
+                self.ui,
+                names=[b"test.wrong.intermediate"],
+                options=[],
+                table=table,
+                alias_table=alias_table,
+                full_pyramid=pyramid,
+            )
+        assert "`verify.noop_func`" in str(e.exception), str(e.exception)
+
+
+if __name__ == '__main__':
+    import silenttestrunner
+
+    silenttestrunner.main(__name__)

tests/test-admin-commands.t

@@ -0,0 +1,49 b''
+Test admin::verify
+
+  $ hg init admin-verify
+  $ cd admin-verify
+
+Test normal output
+
+  $ hg admin::verify -c dirstate
+  running 1 checks
+  running working-copy.dirstate
+  checking dirstate
+
+Quiet works
+
+  $ hg admin::verify -c dirstate --quiet
+
+Test no check no options
+
+  $ hg admin::verify
+  abort: `checks` required
+  [255]
+
+Test single check without options
+
+  $ hg admin::verify -c working-copy.dirstate
+  running 1 checks
+  running working-copy.dirstate
+  checking dirstate
+
+Test single check (alias) without options
+
+  $ hg admin::verify -c dirstate
+  running 1 checks
+  running working-copy.dirstate
+  checking dirstate
+
+Test wrong check name without options
+
+  $ hg admin::verify -c working-copy.dir
+  abort: unknown check working-copy.dir
+  (did you mean working-copy.dirstate?)
+  [10]
+
+Test wrong alias without options
+
+  $ hg admin::verify -c dir
+  abort: unknown check dir
+  [10]
+

mercurial/admin_commands.py

@@ -5,7 +5,45 b''
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
-from . import registrar
+from .i18n import _
+from .admin import verify
+from . import error, registrar, transaction
+
 
 table = {}
 command = registrar.command(table)
+
+
+@command(
+    b'admin::verify',
+    [
+        (b'c', b'check', [], _(b'add a check'), _(b'CHECK')),
+        (b'o', b'option', [], _(b'pass an option to a check'), _(b'OPTION')),
+    ],
+    helpcategory=command.CATEGORY_MAINTENANCE,
+)
+def admin_verify(ui, repo, **opts):
+    """verify the integrity of the repository
+
+    Alternative UI to `hg verify` with a lot more control over the
+    verification process and better error reporting.
+    """
+
+    if not repo.url().startswith(b'file:'):
+        raise error.Abort(_(b"cannot verify bundle or remote repos"))
+
+    if transaction.has_abandoned_transaction(repo):
+        ui.warn(_(b"abandoned transaction found - run hg recover\n"))
+
+    checks = opts.get("check", [])
+    options = opts.get("option", [])
+
+    funcs = verify.get_checks(repo, ui, names=checks, options=options)
+
+    ui.status(_(b"running %d checks\n") % len(funcs))
+    # Done in two times so the execution is separated from the resolving step
+    for name, func in sorted(funcs.items(), key=lambda x: x[0]):
+        ui.status(_(b"running %s\n") % name)
+        errors = func()
+        if errors:
+            ui.warn(_(b"found %d errors\n") % len(errors))

mercurial/commands.py

@@ -7961,6 +7961,9 b' def verify(ui, repo, **opts):'
     for more information about recovery from corruption of the
     repository.
 
+    For an alternative UI with a lot more control over the verification
+    process and better error reporting, try `hg help admin::verify`.
+
     Returns 0 on success, 1 if errors are encountered.
     """
     level = None

mercurial/registrar.py

@@ -6,6 +6,7 b''
 # GNU General Public License version 2 or any later version.
 
 
+from typing import Any, List, Optional, Tuple
 from . import (
     configitems,
     error,
@@ -533,3 +534,30 b' class internalmerge(_funcregistrarbase):'
 
         # actual capabilities, which this internal merge tool has
         func.capabilities = {b"binary": binarycap, b"symlink": symlinkcap}
+
+
+class verify_check(_funcregistrarbase):
+    """Decorator to register a check for admin::verify
+
+    options is a list of (name, default value, help) to be passed to the check
+    """
+
+    def __init__(self, table=None, alias_table=None):
+        super().__init__(table)
+        if alias_table is None:
+            self._alias_table = {}
+        else:
+            self._alias_table = alias_table
+
+    def _extrasetup(
+        self,
+        name,
+        func,
+        alias: Optional[bytes] = None,
+        options: Optional[List[Tuple[bytes, Any, bytes]]] = None,
+    ):
+        func.alias = alias
+        func.options = options
+
+        if alias:
+            self._alias_table[alias] = name

tests/test-completion.t

@@ -3,6 +3,7 b' Show all commands except debug commands'
   abort
   add
   addremove
+  admin::verify
   annotate
   archive
   backout
@@ -65,6 +66,7 b' Show all commands that start with "a"'
   abort
   add
   addremove
+  admin::verify
   annotate
   archive
 
@@ -257,6 +259,7 b' Show all commands + options'
   abort: dry-run
   add: include, exclude, subrepos, dry-run
   addremove: similarity, subrepos, include, exclude, dry-run
+  admin::verify: check, option
   annotate: rev, follow, no-follow, text, user, file, date, number, changeset, line-number, skip, ignore-all-space, ignore-space-change, ignore-blank-lines, ignore-space-at-eol, include, exclude, template
   archive: no-decode, prefix, rev, type, subrepos, include, exclude
   backout: merge, commit, no-commit, parent, rev, edit, tool, include, exclude, message, logfile, date, user

tests/test-globalopts.t

@@ -378,6 +378,8 b' Testing -h/--help:'
   
   Repository maintenance:
   
+   admin::verify
+                 verify the integrity of the repository
    manifest      output the current or given revision of the project manifest
    recover       roll back an interrupted transaction
    verify        verify the integrity of the repository
@@ -513,6 +515,8 b' Testing -h/--help:'
   
   Repository maintenance:
   
+   admin::verify
+                 verify the integrity of the repository
    manifest      output the current or given revision of the project manifest
    recover       roll back an interrupted transaction
    verify        verify the integrity of the repository

tests/test-help-hide.t

@@ -77,6 +77,8 b' Test hiding some commands (which also ha'
   
   Repository maintenance:
   
+   admin::verify
+                 verify the integrity of the repository
    manifest      output the current or given revision of the project manifest
    recover       roll back an interrupted transaction
    verify        verify the integrity of the repository
@@ -216,6 +218,8 b' Test hiding some topics.'
   
   Repository maintenance:
   
+   admin::verify
+                 verify the integrity of the repository
    manifest      output the current or given revision of the project manifest
    recover       roll back an interrupted transaction
    verify        verify the integrity of the repository

tests/test-help.t

@@ -129,6 +129,8 b' the extension is unknown.'
   
   Repository maintenance:
   
+   admin::verify
+                 verify the integrity of the repository
    manifest      output the current or given revision of the project manifest
    recover       roll back an interrupted transaction
    verify        verify the integrity of the repository
@@ -260,6 +262,8 b' the extension is unknown.'
   
   Repository maintenance:
   
+   admin::verify
+                 verify the integrity of the repository
    manifest      output the current or given revision of the project manifest
    recover       roll back an interrupted transaction
    verify        verify the integrity of the repository
@@ -604,9 +608,16 b' Test ambiguous command help'
   $ hg help ad
   list of commands:
   
+  Working directory management:
+  
    add           add the specified files on the next commit
    addremove     add all new files, delete all missing files
   
+  Repository maintenance:
+  
+   admin::verify
+                 verify the integrity of the repository
+  
   (use 'hg help -v ad' to show built-in aliases and global options)
 
 Test command without options
@@ -626,6 +637,9 b' Test command without options'
       Please see https://mercurial-scm.org/wiki/RepositoryCorruption for more
       information about recovery from corruption of the repository.
   
+      For an alternative UI with a lot more control over the verification
+      process and better error reporting, try 'hg help admin::verify'.
+  
       Returns 0 on success, 1 if errors are encountered.
   
   options:
@@ -2650,6 +2664,13 b' Dish up an empty repo; serve it cold.'
   add all new files, delete all missing files
   </td></tr>
   <tr><td>
+  <a href="/help/admin::verify">
+  admin::verify
+  </a>
+  </td><td>
+  verify the integrity of the repository
+  </td></tr>
+  <tr><td>
   <a href="/help/archive">
   archive
   </a>

tests/test-hgweb-json.t

@@ -2112,6 +2112,10 b' help/ shows help topics'
         "topic": "addremove"
       },
       {
+        "summary": "verify the integrity of the repository",
+        "topic": "admin::verify"
+      },
+      {
         "summary": "create an unversioned archive of a repository revision",
         "topic": "archive"
       },