#!/usr/bin/env python
from __future__ import absolute_import
import hashlib
import os
import random
import shutil
import stat
import struct
import sys
import tempfile
import unittest
import silenttestrunner
from mercurial.node import sha1nodeconstants
from mercurial import (
pycompat,
ui as uimod,
)
# Load the local remotefilelog, not the system one
sys.path[0:0] = [os.path.join(os.path.dirname(__file__), '..')]
from hgext.remotefilelog import (
basepack,
historypack,
)
class histpacktests(unittest.TestCase):
def setUp(self):
self.tempdirs = []
def tearDown(self):
for d in self.tempdirs:
shutil.rmtree(d)
def makeTempDir(self):
tempdir = tempfile.mkdtemp()
self.tempdirs.append(tempdir)
return pycompat.fsencode(tempdir)
def getHash(self, content):
return hashlib.sha1(content).digest()
def getFakeHash(self):
return b''.join(
pycompat.bytechr(random.randint(0, 255)) for _ in range(20)
)
def createPack(self, revisions=None):
"""Creates and returns a historypack containing the specified revisions.
`revisions` is a list of tuples, where each tuple contains a filanem,
node, p1node, p2node, and linknode.
"""
if revisions is None:
revisions = [
(
b"filename",
self.getFakeHash(),
sha1nodeconstants.nullid,
sha1nodeconstants.nullid,
self.getFakeHash(),
None,
)
]
packdir = pycompat.fsencode(self.makeTempDir())
packer = historypack.mutablehistorypack(uimod.ui(), packdir, version=2)
for filename, node, p1, p2, linknode, copyfrom in revisions:
packer.add(filename, node, p1, p2, linknode, copyfrom)
path = packer.close()
return historypack.historypack(path)
def testAddSingle(self):
"""Test putting a single entry into a pack and reading it out."""
filename = b"foo"
node = self.getFakeHash()
p1 = self.getFakeHash()
p2 = self.getFakeHash()
linknode = self.getFakeHash()
revisions = [(filename, node, p1, p2, linknode, None)]
pack = self.createPack(revisions)
actual = pack.getancestors(filename, node)[node]
self.assertEqual(p1, actual[0])
self.assertEqual(p2, actual[1])
self.assertEqual(linknode, actual[2])
def testAddMultiple(self):
"""Test putting multiple unrelated revisions into a pack and reading
them out.
"""
revisions = []
for i in range(10):
filename = b"foo-%d" % i
node = self.getFakeHash()
p1 = self.getFakeHash()
p2 = self.getFakeHash()
linknode = self.getFakeHash()
revisions.append((filename, node, p1, p2, linknode, None))
pack = self.createPack(revisions)
for filename, node, p1, p2, linknode, copyfrom in revisions:
actual = pack.getancestors(filename, node)[node]
self.assertEqual(p1, actual[0])
self.assertEqual(p2, actual[1])
self.assertEqual(linknode, actual[2])
self.assertEqual(copyfrom, actual[3])
def testAddAncestorChain(self):
"""Test putting multiple revisions in into a pack and read the ancestor
chain.
"""
revisions = []
filename = b"foo"
lastnode = sha1nodeconstants.nullid
for i in range(10):
node = self.getFakeHash()
revisions.append(
(
filename,
node,
lastnode,
sha1nodeconstants.nullid,
sha1nodeconstants.nullid,
None,
)
)
lastnode = node
# revisions must be added in topological order, newest first
revisions = list(reversed(revisions))
pack = self.createPack(revisions)
# Test that the chain has all the entries
ancestors = pack.getancestors(revisions[0][0], revisions[0][1])
for filename, node, p1, p2, linknode, copyfrom in revisions:
ap1, ap2, alinknode, acopyfrom = ancestors[node]
self.assertEqual(ap1, p1)
self.assertEqual(ap2, p2)
self.assertEqual(alinknode, linknode)
self.assertEqual(acopyfrom, copyfrom)
def testPackMany(self):
"""Pack many related and unrelated ancestors."""
# Build a random pack file
allentries = {}
ancestorcounts = {}
revisions = []
random.seed(0)
for i in range(100):
filename = b"filename-%d" % i
entries = []
p2 = sha1nodeconstants.nullid
linknode = sha1nodeconstants.nullid
for j in range(random.randint(1, 100)):
node = self.getFakeHash()
p1 = sha1nodeconstants.nullid
if len(entries) > 0:
p1 = entries[random.randint(0, len(entries) - 1)]
entries.append(node)
revisions.append((filename, node, p1, p2, linknode, None))
allentries[(filename, node)] = (p1, p2, linknode)
if p1 == sha1nodeconstants.nullid:
ancestorcounts[(filename, node)] = 1
else:
newcount = ancestorcounts[(filename, p1)] + 1
ancestorcounts[(filename, node)] = newcount
# Must add file entries in reverse topological order
revisions = list(reversed(revisions))
pack = self.createPack(revisions)
# Verify the pack contents
for (filename, node) in allentries:
ancestors = pack.getancestors(filename, node)
self.assertEqual(ancestorcounts[(filename, node)], len(ancestors))
for anode, (ap1, ap2, alinknode, copyfrom) in ancestors.items():
ep1, ep2, elinknode = allentries[(filename, anode)]
self.assertEqual(ap1, ep1)
self.assertEqual(ap2, ep2)
self.assertEqual(alinknode, elinknode)
self.assertEqual(copyfrom, None)
def testGetNodeInfo(self):
revisions = []
filename = b"foo"
lastnode = sha1nodeconstants.nullid
for i in range(10):
node = self.getFakeHash()
revisions.append(
(
filename,
node,
lastnode,
sha1nodeconstants.nullid,
sha1nodeconstants.nullid,
None,
)
)
lastnode = node
pack = self.createPack(revisions)
# Test that getnodeinfo returns the expected results
for filename, node, p1, p2, linknode, copyfrom in revisions:
ap1, ap2, alinknode, acopyfrom = pack.getnodeinfo(filename, node)
self.assertEqual(ap1, p1)
self.assertEqual(ap2, p2)
self.assertEqual(alinknode, linknode)
self.assertEqual(acopyfrom, copyfrom)
def testGetMissing(self):
"""Test the getmissing() api."""
revisions = []
filename = b"foo"
for i in range(10):
node = self.getFakeHash()
p1 = self.getFakeHash()
p2 = self.getFakeHash()
linknode = self.getFakeHash()
revisions.append((filename, node, p1, p2, linknode, None))
pack = self.createPack(revisions)
missing = pack.getmissing([(filename, revisions[0][1])])
self.assertFalse(missing)
missing = pack.getmissing(
[(filename, revisions[0][1]), (filename, revisions[1][1])]
)
self.assertFalse(missing)
fakenode = self.getFakeHash()
missing = pack.getmissing(
[(filename, revisions[0][1]), (filename, fakenode)]
)
self.assertEqual(missing, [(filename, fakenode)])
# Test getmissing on a non-existant filename
missing = pack.getmissing([(b"bar", fakenode)])
self.assertEqual(missing, [(b"bar", fakenode)])
def testAddThrows(self):
pack = self.createPack()
try:
pack.add(
b'filename',
sha1nodeconstants.nullid,
sha1nodeconstants.nullid,
sha1nodeconstants.nullid,
sha1nodeconstants.nullid,
None,
)
self.assertTrue(False, "historypack.add should throw")
except RuntimeError:
pass
def testBadVersionThrows(self):
pack = self.createPack()
path = pack.path + b'.histpack'
with open(path, 'rb') as f:
raw = f.read()
raw = struct.pack('!B', 255) + raw[1:]
os.chmod(path, os.stat(path).st_mode | stat.S_IWRITE)
with open(path, 'wb+') as f:
f.write(raw)
try:
historypack.historypack(pack.path)
self.assertTrue(False, "bad version number should have thrown")
except RuntimeError:
pass
def testLargePack(self):
"""Test creating and reading from a large pack with over X entries.
This causes it to use a 2^16 fanout table instead."""
total = basepack.SMALLFANOUTCUTOFF + 1
revisions = []
for i in pycompat.xrange(total):
filename = b"foo-%d" % i
node = self.getFakeHash()
p1 = self.getFakeHash()
p2 = self.getFakeHash()
linknode = self.getFakeHash()
revisions.append((filename, node, p1, p2, linknode, None))
pack = self.createPack(revisions)
self.assertEqual(pack.params.fanoutprefix, basepack.LARGEFANOUTPREFIX)
for filename, node, p1, p2, linknode, copyfrom in revisions:
actual = pack.getancestors(filename, node)[node]
self.assertEqual(p1, actual[0])
self.assertEqual(p2, actual[1])
self.assertEqual(linknode, actual[2])
self.assertEqual(copyfrom, actual[3])
# TODO:
# histpack store:
# - repack two packs into one
if __name__ == '__main__':
if pycompat.iswindows:
sys.exit(80) # Skip on Windows
silenttestrunner.main(__name__)