comparison mercurial/revlogutils/flagutil.py @ 43076:2372284d9457

formatting: blacken the codebase

This is using my patch to black
(https://github.com/psf/black/pull/826) so we don't un-wrap collection
literals.

Done with:

  hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**"' | xargs black -S

# skip-blame mass-reformatting only

# no-check-commit reformats foo_bar functions

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

43075:57875cf423c9

    REVIDX_ISCENSORED,
    REVIDX_RAWTEXT_CHANGING_FLAGS,
    REVIDX_SIDEDATA,
)

from .. import (
    error,
    util
)

# blanked usage of all the name to prevent pyflakes constraints
# We need these name available in the module for extensions.
REVIDX_ISCENSORED
REVIDX_ELLIPSIS

# Store flag processors (cf. 'addflagprocessor()' to register)
flagprocessors = {
    REVIDX_ISCENSORED: None,
}

def addflagprocessor(flag, processor):
    """Register a flag processor on a revision data flag.

    Invariant:

      debug commands. In this case the transform only indicates whether the
      contents can be used for hash integrity checks.
    """
    insertflagprocessor(flag, processor, flagprocessors)

def insertflagprocessor(flag, processor, flagprocessors):
    if not flag & REVIDX_KNOWN_FLAGS:
        msg = _("cannot register processor on unknown flag '%#x'.") % (flag)
        raise error.ProgrammingError(msg)
    if flag not in REVIDX_FLAGS_ORDER:
        msg = _("flag '%#x' undefined in REVIDX_FLAGS_ORDER.") % (flag)
        raise error.ProgrammingError(msg)
    if flag in flagprocessors:
        msg = _("cannot register multiple processors on flag '%#x'.") % (flag)
        raise error.Abort(msg)
    flagprocessors[flag] = processor

def processflagswrite(revlog, text, flags, sidedata):
    """Inspect revision data flags and applies write transformations defined
    by registered flag processors.

    Returns a 2-tuple of ``(text, validatehash)`` where ``text`` is the
    processed text and ``validatehash`` is a bool indicating whether the
    returned text should be checked for hash integrity.
    """
    return _processflagsfunc(revlog, text, flags, 'write',
                             sidedata=sidedata)[:2]

def processflagsread(revlog, text, flags):
    """Inspect revision data flags and applies read transformations defined
    by registered flag processors.

    processed text and ``validatehash`` is a bool indicating whether the
    returned text should be checked for hash integrity.
    """
    return _processflagsfunc(revlog, text, flags, 'read')

def processflagsraw(revlog, text, flags):
    """Inspect revision data flags to check is the content hash should be
    validated.

    ``text`` - the revision data to process

    processed text and ``validatehash`` is a bool indicating whether the
    returned text should be checked for hash integrity.
    """
    return _processflagsfunc(revlog, text, flags, 'raw')[1]

def _processflagsfunc(revlog, text, flags, operation, sidedata=None):
    """internal function to process flag on a revlog

    This function is private to this module, code should never needs to call it
    directly."""
    # fast path: no flag processors will run
    if flags == 0:
        return text, True, {}
    if operation not in ('read', 'write', 'raw'):
        raise error.ProgrammingError(_("invalid '%s' operation") %
                                     operation)
    # Check all flags are known.
    if flags & ~REVIDX_KNOWN_FLAGS:
        raise revlog._flagserrorclass(_("incompatible revision flag '%#x'") %
                                      (flags & ~REVIDX_KNOWN_FLAGS))
    validatehash = True
    # Depending on the operation (read or write), the order might be
    # reversed due to non-commutative transforms.
    orderedflags = REVIDX_FLAGS_ORDER
    if operation == 'write':

        # related operation transform and update result tuple.
        if flag & flags:
            vhash = True

            if flag not in revlog._flagprocessors:
                message = _("missing processor for flag '%#x'") % (flag)
                raise revlog._flagserrorclass(message)

            processor = revlog._flagprocessors[flag]
            if processor is not None:
                readtransform, writetransform, rawtransform = processor

                if operation == 'raw':
                    vhash = rawtransform(revlog, text)
                elif operation == 'read':
                    text, vhash, s = readtransform(revlog, text)
                    outsidedata.update(s)
                else: # write operation
                    text, vhash = writetransform(revlog, text, sidedata)
            validatehash = validatehash and vhash

    return text, validatehash, outsidedata

43076:2372284d9457

    REVIDX_ISCENSORED,
    REVIDX_RAWTEXT_CHANGING_FLAGS,
    REVIDX_SIDEDATA,
)

from .. import error, util

# blanked usage of all the name to prevent pyflakes constraints
# We need these name available in the module for extensions.
REVIDX_ISCENSORED
REVIDX_ELLIPSIS

# Store flag processors (cf. 'addflagprocessor()' to register)
flagprocessors = {
    REVIDX_ISCENSORED: None,
}


def addflagprocessor(flag, processor):
    """Register a flag processor on a revision data flag.

    Invariant:

      debug commands. In this case the transform only indicates whether the
      contents can be used for hash integrity checks.
    """
    insertflagprocessor(flag, processor, flagprocessors)


def insertflagprocessor(flag, processor, flagprocessors):
    if not flag & REVIDX_KNOWN_FLAGS:
        msg = _("cannot register processor on unknown flag '%#x'.") % flag
        raise error.ProgrammingError(msg)
    if flag not in REVIDX_FLAGS_ORDER:
        msg = _("flag '%#x' undefined in REVIDX_FLAGS_ORDER.") % flag
        raise error.ProgrammingError(msg)
    if flag in flagprocessors:
        msg = _("cannot register multiple processors on flag '%#x'.") % flag
        raise error.Abort(msg)
    flagprocessors[flag] = processor


def processflagswrite(revlog, text, flags, sidedata):
    """Inspect revision data flags and applies write transformations defined
    by registered flag processors.

    Returns a 2-tuple of ``(text, validatehash)`` where ``text`` is the
    processed text and ``validatehash`` is a bool indicating whether the
    returned text should be checked for hash integrity.
    """
    return _processflagsfunc(revlog, text, flags, 'write', sidedata=sidedata)[
        :2
    ]


def processflagsread(revlog, text, flags):
    """Inspect revision data flags and applies read transformations defined
    by registered flag processors.

    processed text and ``validatehash`` is a bool indicating whether the
    returned text should be checked for hash integrity.
    """
    return _processflagsfunc(revlog, text, flags, 'read')


def processflagsraw(revlog, text, flags):
    """Inspect revision data flags to check is the content hash should be
    validated.

    ``text`` - the revision data to process

    processed text and ``validatehash`` is a bool indicating whether the
    returned text should be checked for hash integrity.
    """
    return _processflagsfunc(revlog, text, flags, 'raw')[1]


def _processflagsfunc(revlog, text, flags, operation, sidedata=None):
    """internal function to process flag on a revlog

    This function is private to this module, code should never needs to call it
    directly."""
    # fast path: no flag processors will run
    if flags == 0:
        return text, True, {}
    if operation not in ('read', 'write', 'raw'):
        raise error.ProgrammingError(_("invalid '%s' operation") % operation)
    # Check all flags are known.
    if flags & ~REVIDX_KNOWN_FLAGS:
        raise revlog._flagserrorclass(
            _("incompatible revision flag '%#x'")
            % (flags & ~REVIDX_KNOWN_FLAGS)
        )
    validatehash = True
    # Depending on the operation (read or write), the order might be
    # reversed due to non-commutative transforms.
    orderedflags = REVIDX_FLAGS_ORDER
    if operation == 'write':

        # related operation transform and update result tuple.
        if flag & flags:
            vhash = True

            if flag not in revlog._flagprocessors:
                message = _("missing processor for flag '%#x'") % flag
                raise revlog._flagserrorclass(message)

            processor = revlog._flagprocessors[flag]
            if processor is not None:
                readtransform, writetransform, rawtransform = processor

                if operation == 'raw':
                    vhash = rawtransform(revlog, text)
                elif operation == 'read':
                    text, vhash, s = readtransform(revlog, text)
                    outsidedata.update(s)
                else:  # write operation
                    text, vhash = writetransform(revlog, text, sidedata)
            validatehash = validatehash and vhash

    return text, validatehash, outsidedata