comparison tests/test-revlog-raw.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

# TESTTMP is optional. This makes it convenient to run without run-tests.py
tvfs = vfs.vfs(encoding.environ.get(b'TESTTMP', b'/tmp'))

# Enable generaldelta otherwise revlog won't use delta as expected by the test
tvfs.options = {b'generaldelta': True, b'revlogv1': True,
                b'sparse-revlog': True}

# The test wants to control whether to use delta explicitly, based on
# "storedeltachains".
revlog.revlog._isgooddeltainfo = lambda self, d, textlen: self._storedeltachains

def abort(msg):
    print('abort: %s' % msg)
    # Return 0 so run-tests.py could compare the output.
    sys.exit()

# Register a revlog processor for flag EXTSTORED.
#
# It simply prepends a fixed header, and replaces '1' to 'i'. So it has
# insertion and replacement, and may be interesting to test revlog's line-based
# deltas.
_extheader = b'E\n'

def readprocessor(self, rawtext):
    # True: the returned text could be used to verify hash
    text = rawtext[len(_extheader):].replace(b'i', b'1')
    return text, True, {}

def writeprocessor(self, text, sidedata):
    # False: the returned rawtext shouldn't be used to verify hash
    rawtext = _extheader + text.replace(b'1', b'i')
    return rawtext, False

def rawprocessor(self, rawtext):
    # False: do not verify hash. Only the content returned by "readprocessor"
    # can be used to verify hash.
    return False

flagutil.addflagprocessor(revlog.REVIDX_EXTSTORED,
                        (readprocessor, writeprocessor, rawprocessor))

# Utilities about reading and appending revlog

def newtransaction():
    # A transaction is required to write revlogs
    report = lambda msg: None
    return transaction.transaction(report, tvfs, {'plain': tvfs}, b'journal')

def newrevlog(name=b'_testrevlog.i', recreate=False):
    if recreate:
        tvfs.tryunlink(name)
    rlog = revlog.revlog(tvfs, name)
    return rlog

def appendrev(rlog, text, tr, isext=False, isdelta=True):
    '''Append a revision. If isext is True, set the EXTSTORED flag so flag
    processor will be used (and rawtext is different from text). If isdelta is
    True, force the revision to be a delta, otherwise it's full text.

        abort('rev %d: failed to append: %s' % (nextrev, ex))
    finally:
        # Restore storedeltachains. It is always True, see revlog.__init__
        rlog._storedeltachains = True

def addgroupcopy(rlog, tr, destname=b'_destrevlog.i', optimaldelta=True):
    '''Copy revlog to destname using revlog.addgroup. Return the copied revlog.

    This emulates push or pull. They use changegroup. Changegroup requires
    repo to work. We don't have a repo, so a dummy changegroup is used.

    the destination revlog needs more work to use it.

    This exercises some revlog.addgroup (and revlog._addrevision(text=None))
    code path, which is not covered by "appendrev" alone.
    '''
    class dummychangegroup(object):
        @staticmethod
        def deltachunk(pnode):
            pnode = pnode or node.nullid
            parentrev = rlog.rev(pnode)

            else:
                # suboptimal deltaparent
                deltaparent = min(0, parentrev)
            if not rlog.candelta(deltaparent, r):
                deltaparent = -1
            return {b'node': rlog.node(r), b'p1': pnode, b'p2': node.nullid,
                    b'cs': rlog.node(rlog.linkrev(r)), b'flags': rlog.flags(r),
                    b'deltabase': rlog.node(deltaparent),
                    b'delta': rlog.revdiff(deltaparent, r)}

        def deltaiter(self):
            chain = None
            for chunkdata in iter(lambda: self.deltachunk(chain), {}):
                node = chunkdata[b'node']

    dlog = newrevlog(destname, recreate=True)
    dummydeltas = dummychangegroup().deltaiter()
    dlog.addgroup(dummydeltas, linkmap, tr)
    return dlog

def lowlevelcopy(rlog, tr, destname=b'_destrevlog.i'):
    '''Like addgroupcopy, but use the low level revlog._addrevision directly.

    It exercises some code paths that are hard to reach easily otherwise.

        if r == 0 or (rlog.flags(r) & revlog.REVIDX_EXTSTORED):
            text = rlog.rawdata(r)
            cachedelta = None
        else:
            # deltaparent cannot have EXTSTORED flag.
            deltaparent = max([-1] +
                              [p for p in range(r)
                               if rlog.flags(p) & revlog.REVIDX_EXTSTORED == 0])
            text = None
            cachedelta = (deltaparent, rlog.revdiff(deltaparent, r))
        flags = rlog.flags(r)
        ifh = dfh = None
        try:
            ifh = dlog.opener(dlog.indexfile, b'a+')
            if not dlog._inline:
                dfh = dlog.opener(dlog.datafile, b'a+')
            dlog._addrevision(rlog.node(r), text, tr, r, p1, p2, flags,
                              cachedelta, ifh, dfh)
        finally:
            if dfh is not None:
                dfh.close()
            if ifh is not None:
                ifh.close()
    return dlog

# Utilities to generate revisions for testing

def genbits(n):
    '''Given a number n, generate (2 ** (n * 2) + 1) numbers in range(2 ** n).
    i.e. the generated numbers have a width of n bits.

        assert a[x] < m
        a[x] += 1
        x = y
        yield x

def gentext(rev):
    '''Given a revision number, generate dummy text'''
    return b''.join(b'%d\n' % j for j in range(-1, rev % 5))

def writecases(rlog, tr):
    '''Write some revisions interested to the test.

    The test is interested in 3 properties of a revision:

            abort('rev %d: isdelta is unexpected' % rev)
        if bool(rlog.flags(rev)) != isext:
            abort('rev %d: isext is ineffective' % rev)
    return result

# Main test and checking

def checkrevlog(rlog, expected):
    '''Check if revlog has expected contents. expected is [(text, rawtext)]'''
    # Test using different access orders. This could expose some issues
    # depending on revlog caching (see revlog._cache).

                        if raw:
                            t = nlog.rawdata(rev)
                        else:
                            t = nlog.revision(rev)
                        if t != expected[rev][int(raw)]:
                            abort('rev %d: corrupted %stext'
                                  % (rev, raw and 'raw' or ''))

slicingdata = [
    ([0, 1, 2, 3, 55, 56, 58, 59, 60],
     [[0, 1], [2], [58], [59, 60]],
     10),
    ([0, 1, 2, 3, 55, 56, 58, 59, 60],
     [[0, 1], [2], [58], [59, 60]],
     10),
    ([-1, 0, 1, 2, 3, 55, 56, 58, 59, 60],
     [[-1, 0, 1], [2], [58], [59, 60]],
     10),
]

def slicingtest(rlog):
    oldmin = rlog._srmingapsize
    try:
        # the test revlog is small, we remove the floor under which we

                print('  expected: %s' % expected)
                print('  result:   %s' % result)
    finally:
        rlog._srmingapsize = oldmin

def md5sum(s):
    return hashlib.md5(s).digest()

def _maketext(*coord):
    """create piece of text according to range of integers

    The test returned use a md5sum of the integer to make it less

    for start, size in coord:
        num = range(start, start + size)
        p = [md5sum(b'%d' % r) for r in num]
        pieces.append(b'\n'.join(p))
    return b'\n'.join(pieces) + b'\n'

data = [
    _maketext((0, 120), (456, 60)),
    _maketext((0, 120), (345, 60)),
    _maketext((0, 120), (734, 60)),

    _maketext((0, 120), (60, 60), (569, 30), (745, 40)),
    _maketext((0, 120), (60, 60), (777, 30), (700, 40)),
    _maketext((0, 120), (60, 60), (618, 30), (398, 40), (158, 10)),
]

def makesnapshot(tr):
    rl = newrevlog(name=b'_snaprevlog3.i', recreate=True)
    for i in data:
        appendrev(rl, i, tr)
    return rl

snapshots = [-1, 0, 6, 8, 11, 17, 19, 21, 25, 30]
def issnapshottest(rlog):
    result = []
    if rlog.issnapshot(-1):
        result.append(-1)
    for rev in rlog:

    if snapshots != result:
        print('snapshot differ:')
        print('  expected: %s' % snapshots)
        print('  got:      %s' % result)

snapshotmapall = {0: [6, 8, 11, 17, 19, 25], 8: [21], -1: [0, 30]}
snapshotmap15 = {0: [17, 19, 25], 8: [21], -1: [30]}
def findsnapshottest(rlog):
    resultall = collections.defaultdict(list)
    deltas._findsnapshots(rlog, resultall, 0)
    resultall = dict(resultall.items())
    if resultall != snapshotmapall:

    result15 = dict(result15.items())
    if result15 != snapshotmap15:
        print('snapshot map  differ:')
        print('  expected: %s' % snapshotmap15)
        print('  got:      %s' % result15)

def maintest():
    with newtransaction() as tr:
        rl = newrevlog(recreate=True)
        expected = writecases(rl, tr)

        issnapshottest(rl5)
        print('issnapshot test passed')
        findsnapshottest(rl5)
        print('findsnapshot test passed')

try:
    maintest()
except Exception as ex:
    abort('crashed: %s' % ex)

43076:2372284d9457

# TESTTMP is optional. This makes it convenient to run without run-tests.py
tvfs = vfs.vfs(encoding.environ.get(b'TESTTMP', b'/tmp'))

# Enable generaldelta otherwise revlog won't use delta as expected by the test
tvfs.options = {
    b'generaldelta': True,
    b'revlogv1': True,
    b'sparse-revlog': True,
}

# The test wants to control whether to use delta explicitly, based on
# "storedeltachains".
revlog.revlog._isgooddeltainfo = lambda self, d, textlen: self._storedeltachains


def abort(msg):
    print('abort: %s' % msg)
    # Return 0 so run-tests.py could compare the output.
    sys.exit()


# Register a revlog processor for flag EXTSTORED.
#
# It simply prepends a fixed header, and replaces '1' to 'i'. So it has
# insertion and replacement, and may be interesting to test revlog's line-based
# deltas.
_extheader = b'E\n'


def readprocessor(self, rawtext):
    # True: the returned text could be used to verify hash
    text = rawtext[len(_extheader) :].replace(b'i', b'1')
    return text, True, {}


def writeprocessor(self, text, sidedata):
    # False: the returned rawtext shouldn't be used to verify hash
    rawtext = _extheader + text.replace(b'1', b'i')
    return rawtext, False


def rawprocessor(self, rawtext):
    # False: do not verify hash. Only the content returned by "readprocessor"
    # can be used to verify hash.
    return False


flagutil.addflagprocessor(
    revlog.REVIDX_EXTSTORED, (readprocessor, writeprocessor, rawprocessor)
)

# Utilities about reading and appending revlog


def newtransaction():
    # A transaction is required to write revlogs
    report = lambda msg: None
    return transaction.transaction(report, tvfs, {'plain': tvfs}, b'journal')


def newrevlog(name=b'_testrevlog.i', recreate=False):
    if recreate:
        tvfs.tryunlink(name)
    rlog = revlog.revlog(tvfs, name)
    return rlog


def appendrev(rlog, text, tr, isext=False, isdelta=True):
    '''Append a revision. If isext is True, set the EXTSTORED flag so flag
    processor will be used (and rawtext is different from text). If isdelta is
    True, force the revision to be a delta, otherwise it's full text.

        abort('rev %d: failed to append: %s' % (nextrev, ex))
    finally:
        # Restore storedeltachains. It is always True, see revlog.__init__
        rlog._storedeltachains = True


def addgroupcopy(rlog, tr, destname=b'_destrevlog.i', optimaldelta=True):
    '''Copy revlog to destname using revlog.addgroup. Return the copied revlog.

    This emulates push or pull. They use changegroup. Changegroup requires
    repo to work. We don't have a repo, so a dummy changegroup is used.

    the destination revlog needs more work to use it.

    This exercises some revlog.addgroup (and revlog._addrevision(text=None))
    code path, which is not covered by "appendrev" alone.
    '''

    class dummychangegroup(object):
        @staticmethod
        def deltachunk(pnode):
            pnode = pnode or node.nullid
            parentrev = rlog.rev(pnode)

            else:
                # suboptimal deltaparent
                deltaparent = min(0, parentrev)
            if not rlog.candelta(deltaparent, r):
                deltaparent = -1
            return {
                b'node': rlog.node(r),
                b'p1': pnode,
                b'p2': node.nullid,
                b'cs': rlog.node(rlog.linkrev(r)),
                b'flags': rlog.flags(r),
                b'deltabase': rlog.node(deltaparent),
                b'delta': rlog.revdiff(deltaparent, r),
            }

        def deltaiter(self):
            chain = None
            for chunkdata in iter(lambda: self.deltachunk(chain), {}):
                node = chunkdata[b'node']

    dlog = newrevlog(destname, recreate=True)
    dummydeltas = dummychangegroup().deltaiter()
    dlog.addgroup(dummydeltas, linkmap, tr)
    return dlog


def lowlevelcopy(rlog, tr, destname=b'_destrevlog.i'):
    '''Like addgroupcopy, but use the low level revlog._addrevision directly.

    It exercises some code paths that are hard to reach easily otherwise.

        if r == 0 or (rlog.flags(r) & revlog.REVIDX_EXTSTORED):
            text = rlog.rawdata(r)
            cachedelta = None
        else:
            # deltaparent cannot have EXTSTORED flag.
            deltaparent = max(
                [-1]
                + [
                    p
                    for p in range(r)
                    if rlog.flags(p) & revlog.REVIDX_EXTSTORED == 0
                ]
            )
            text = None
            cachedelta = (deltaparent, rlog.revdiff(deltaparent, r))
        flags = rlog.flags(r)
        ifh = dfh = None
        try:
            ifh = dlog.opener(dlog.indexfile, b'a+')
            if not dlog._inline:
                dfh = dlog.opener(dlog.datafile, b'a+')
            dlog._addrevision(
                rlog.node(r), text, tr, r, p1, p2, flags, cachedelta, ifh, dfh
            )
        finally:
            if dfh is not None:
                dfh.close()
            if ifh is not None:
                ifh.close()
    return dlog


# Utilities to generate revisions for testing


def genbits(n):
    '''Given a number n, generate (2 ** (n * 2) + 1) numbers in range(2 ** n).
    i.e. the generated numbers have a width of n bits.

        assert a[x] < m
        a[x] += 1
        x = y
        yield x


def gentext(rev):
    '''Given a revision number, generate dummy text'''
    return b''.join(b'%d\n' % j for j in range(-1, rev % 5))


def writecases(rlog, tr):
    '''Write some revisions interested to the test.

    The test is interested in 3 properties of a revision:

            abort('rev %d: isdelta is unexpected' % rev)
        if bool(rlog.flags(rev)) != isext:
            abort('rev %d: isext is ineffective' % rev)
    return result


# Main test and checking


def checkrevlog(rlog, expected):
    '''Check if revlog has expected contents. expected is [(text, rawtext)]'''
    # Test using different access orders. This could expose some issues
    # depending on revlog caching (see revlog._cache).

                        if raw:
                            t = nlog.rawdata(rev)
                        else:
                            t = nlog.revision(rev)
                        if t != expected[rev][int(raw)]:
                            abort(
                                'rev %d: corrupted %stext'
                                % (rev, raw and 'raw' or '')
                            )


slicingdata = [
    ([0, 1, 2, 3, 55, 56, 58, 59, 60], [[0, 1], [2], [58], [59, 60]], 10),
    ([0, 1, 2, 3, 55, 56, 58, 59, 60], [[0, 1], [2], [58], [59, 60]], 10),
    (
        [-1, 0, 1, 2, 3, 55, 56, 58, 59, 60],
        [[-1, 0, 1], [2], [58], [59, 60]],
        10,
    ),
]


def slicingtest(rlog):
    oldmin = rlog._srmingapsize
    try:
        # the test revlog is small, we remove the floor under which we

                print('  expected: %s' % expected)
                print('  result:   %s' % result)
    finally:
        rlog._srmingapsize = oldmin


def md5sum(s):
    return hashlib.md5(s).digest()


def _maketext(*coord):
    """create piece of text according to range of integers

    The test returned use a md5sum of the integer to make it less

    for start, size in coord:
        num = range(start, start + size)
        p = [md5sum(b'%d' % r) for r in num]
        pieces.append(b'\n'.join(p))
    return b'\n'.join(pieces) + b'\n'


data = [
    _maketext((0, 120), (456, 60)),
    _maketext((0, 120), (345, 60)),
    _maketext((0, 120), (734, 60)),

    _maketext((0, 120), (60, 60), (569, 30), (745, 40)),
    _maketext((0, 120), (60, 60), (777, 30), (700, 40)),
    _maketext((0, 120), (60, 60), (618, 30), (398, 40), (158, 10)),
]


def makesnapshot(tr):
    rl = newrevlog(name=b'_snaprevlog3.i', recreate=True)
    for i in data:
        appendrev(rl, i, tr)
    return rl


snapshots = [-1, 0, 6, 8, 11, 17, 19, 21, 25, 30]


def issnapshottest(rlog):
    result = []
    if rlog.issnapshot(-1):
        result.append(-1)
    for rev in rlog:

    if snapshots != result:
        print('snapshot differ:')
        print('  expected: %s' % snapshots)
        print('  got:      %s' % result)


snapshotmapall = {0: [6, 8, 11, 17, 19, 25], 8: [21], -1: [0, 30]}
snapshotmap15 = {0: [17, 19, 25], 8: [21], -1: [30]}


def findsnapshottest(rlog):
    resultall = collections.defaultdict(list)
    deltas._findsnapshots(rlog, resultall, 0)
    resultall = dict(resultall.items())
    if resultall != snapshotmapall:

    result15 = dict(result15.items())
    if result15 != snapshotmap15:
        print('snapshot map  differ:')
        print('  expected: %s' % snapshotmap15)
        print('  got:      %s' % result15)


def maintest():
    with newtransaction() as tr:
        rl = newrevlog(recreate=True)
        expected = writecases(rl, tr)

        issnapshottest(rl5)
        print('issnapshot test passed')
        findsnapshottest(rl5)
        print('findsnapshot test passed')


try:
    maintest()
except Exception as ex:
    abort('crashed: %s' % ex)