Differential D10359 Diff 27454 tests/testlib/ext-sidedata.py

Changeset View

Standalone View

tests/testlib/ext-sidedata.py

# ext-sidedata.py - small extension to test the sidedata logic		# ext-sidedata.py - small extension to test the sidedata logic
#		#
# Copyright 2019 Pierre-Yves David <pierre-yves.david@octobus.net>		# Copyright 2019 Pierre-Yves David <pierre-yves.david@octobus.net>
#		#
# This software may be used and distributed according to the terms of the		# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.		# GNU General Public License version 2 or any later version.

from __future__ import absolute_import		from __future__ import absolute_import

import hashlib		import hashlib
import struct		import struct

from mercurial.node import nullrev		from mercurial.node import nullrev
from mercurial import (		from mercurial import (
		changegroup,
extensions,		extensions,
requirements,		requirements,
revlog,		revlog,
)		)

from mercurial.upgrade_utils import engine as upgrade_engine		from mercurial.upgrade_utils import engine as upgrade_engine

		from mercurial.revlogutils import constants
from mercurial.revlogutils import sidedata		from mercurial.revlogutils import sidedata


def wrapaddrevision(		def wrapaddrevision(
orig, self, text, transaction, link, p1, p2, args, *kwargs		orig, self, text, transaction, link, p1, p2, args, *kwargs
):		):
if kwargs.get('sidedata') is None:		if kwargs.get('sidedata') is None:
kwargs['sidedata'] = {}		kwargs['sidedata'] = {}
Show All 19 Lines	if nodeorrev != nullrev and nodeorrev != self.nullid:
raise RuntimeError('text size mismatch')		raise RuntimeError('text size mismatch')
expected = sd.get(sidedata.SD_TEST2)		expected = sd.get(sidedata.SD_TEST2)
got = hashlib.sha256(text).digest()		got = hashlib.sha256(text).digest()
if expected is not None and got != expected:		if expected is not None and got != expected:
raise RuntimeError('sha256 mismatch')		raise RuntimeError('sha256 mismatch')
return text, sd		return text, sd


def wrapgetsidedatacompanion(orig, srcrepo, dstrepo):		def wrapget_sidedata_helpers(orig, srcrepo, dstrepo):
sidedatacompanion = orig(srcrepo, dstrepo)		repo, computers, removers = orig(srcrepo, dstrepo)
		assert not computers and not removers # deal with composition later
addedreqs = dstrepo.requirements - srcrepo.requirements		addedreqs = dstrepo.requirements - srcrepo.requirements

if requirements.SIDEDATA_REQUIREMENT in addedreqs:		if requirements.SIDEDATA_REQUIREMENT in addedreqs:
assert sidedatacompanion is None # deal with composition later

def sidedatacompanion(revlog, rev):		def computer(repo, revlog, rev, old_sidedata):
		assert not old_sidedata # not supported yet
update = {}		update = {}
revlog.sidedatanocheck = True		revlog.sidedatanocheck = True
try:		try:
text = revlog.revision(rev)		text = revlog.revision(rev)
finally:		finally:
del revlog.sidedatanocheck		del revlog.sidedatanocheck
## let's store some arbitrary data just for testing		## let's store some arbitrary data just for testing
# text length		# text length
update[sidedata.SD_TEST1] = struct.pack('>I', len(text))		update[sidedata.SD_TEST1] = struct.pack('>I', len(text))
# and sha2 hashes		# and sha2 hashes
sha256 = hashlib.sha256(text).digest()		sha256 = hashlib.sha256(text).digest()
update[sidedata.SD_TEST2] = struct.pack('>32s', sha256)		update[sidedata.SD_TEST2] = struct.pack('>32s', sha256)
return False, (), update, 0, 0		return update, (0, 0)

		srcrepo.register_sidedata_computer(
		constants.KIND_CHANGELOG,
		b"whatever",
		(sidedata.SD_TEST1, sidedata.SD_TEST2),
		computer,
		0,
		)
		dstrepo.register_wanted_sidedata(b"whatever")

return sidedatacompanion		return changegroup.get_sidedata_helpers(srcrepo, dstrepo._wanted_sidedata)


def extsetup(ui):		def extsetup(ui):
extensions.wrapfunction(revlog.revlog, 'addrevision', wrapaddrevision)		extensions.wrapfunction(revlog.revlog, 'addrevision', wrapaddrevision)
extensions.wrapfunction(revlog.revlog, '_revisiondata', wrap_revisiondata)		extensions.wrapfunction(revlog.revlog, '_revisiondata', wrap_revisiondata)
extensions.wrapfunction(		extensions.wrapfunction(
upgrade_engine, 'getsidedatacompanion', wrapgetsidedatacompanion		upgrade_engine, 'get_sidedata_helpers', wrapget_sidedata_helpers
)		)


def reposetup(ui, repo):		def reposetup(ui, repo):
# We don't register sidedata computers because we don't care within these		# We don't register sidedata computers because we don't care within these
# tests		# tests
repo.register_wanted_sidedata(sidedata.SD_TEST1)		repo.register_wanted_sidedata(sidedata.SD_TEST1)
repo.register_wanted_sidedata(sidedata.SD_TEST2)		repo.register_wanted_sidedata(sidedata.SD_TEST2)