mercurial: comparison mercurial/testing/storage.py

equal deleted inserted replaced

-:57875cf423c9
+:2372284d9457
 )
 from .. import (
 error,
 mdiff,
 )
-from ..interfaces import (
+from ..interfaces import repository
-repository,
+from ..utils import storageutil
-)
-from ..utils import (
-storageutil,
-)
 class basetestcase(unittest.TestCase):
 if not getattr(unittest.TestCase, r'assertRaisesRegex', False):
-assertRaisesRegex = (# camelcase-required
+assertRaisesRegex = (  # camelcase-required
-unittest.TestCase.assertRaisesRegexp)
+unittest.TestCase.assertRaisesRegexp
+)
 class ifileindextests(basetestcase):
 """Generic tests for the ifileindex interface.
 All file storage backends for index data should conform to the tests in this
 class.
 Use ``makeifileindextests()`` to create an instance of this type.
 """
 def testempty(self):
 f = self._makefilefn()
 self.assertEqual(len(f), 0, 'new file store has 0 length by default')
 self.assertEqual(list(f), [], 'iter yields nothing by default')
 self.assertEqual(f.children(node2), [])
 self.assertEqual(f.children(node3), [node4])
 self.assertEqual(f.children(node4), [])
 self.assertEqual(f.children(node5), [])
 class ifiledatatests(basetestcase):
 """Generic tests for the ifiledata interface.
 All file storage backends for data should conform to the tests in this
 class.
 Use ``makeifiledatatests()`` to create an instance of this type.
 """
 def testempty(self):
 f = self._makefilefn()
 self.assertEqual(f.storageinfo(), {})
-self.assertEqual(f.storageinfo(revisionscount=True, trackedsize=True),
+self.assertEqual(
-{'revisionscount': 0, 'trackedsize': 0})
+f.storageinfo(revisionscount=True, trackedsize=True),
+{'revisionscount': 0, 'trackedsize': 0},
+)
 self.assertEqual(f.size(nullrev), 0)
 for i in range(-5, 5):
 if i == nullrev:
 f = self._makefilefn()
 with self._maketransactionfn() as tr:
 node = f.add(fulltext, None, tr, 0, nullid, nullid)
 self.assertEqual(f.storageinfo(), {})
-self.assertEqual(f.storageinfo(revisionscount=True, trackedsize=True),
+self.assertEqual(
-{'revisionscount': 1, 'trackedsize': len(fulltext)})
+f.storageinfo(revisionscount=True, trackedsize=True),
+{'revisionscount': 1, 'trackedsize': len(fulltext)},
+)
 self.assertEqual(f.size(0), len(fulltext))
 with self.assertRaises(IndexError):
 f.size(1)
 self.assertEqual(
 f.storageinfo(revisionscount=True, trackedsize=True),
 {
 'revisionscount': 3,
 'trackedsize': len(fulltext0) + len(fulltext1) + len(fulltext2),
-})
+},
+)
 self.assertEqual(f.size(0), len(fulltext0))
 self.assertEqual(f.size(1), len(fulltext1))
 self.assertEqual(f.size(2), len(fulltext2))
 self.assertEqual(rev.p2node, nullid)
 self.assertIsNone(rev.linknode)
 self.assertEqual(rev.basenode, node0)
 self.assertIsNone(rev.baserevisionsize)
 self.assertIsNone(rev.revision)
-self.assertEqual(rev.delta,
+self.assertEqual(
-b'\x00\x00\x00\x00\x00\x00\x04\x00\x00\x00\x04\x01' +
+rev.delta,
-fulltext1)
+b'\x00\x00\x00\x00\x00\x00\x04\x00\x00\x00\x04\x01' + fulltext1,
+)
 rev = next(gen)
 self.assertEqual(rev.node, node2)
 self.assertEqual(rev.p1node, node1)
 self.assertEqual(rev.p2node, nullid)
 self.assertIsNone(rev.linknode)
 self.assertEqual(rev.basenode, node1)
 self.assertIsNone(rev.baserevisionsize)
 self.assertIsNone(rev.revision)
-self.assertEqual(rev.delta,
+self.assertEqual(
-b'\x00\x00\x00\x00\x00\x00\x04\x01\x00\x00\x04\x02' +
+rev.delta,
-fulltext2)
+b'\x00\x00\x00\x00\x00\x00\x04\x01\x00\x00\x04\x02' + fulltext2,
+)
 with self.assertRaises(StopIteration):
 next(gen)
 # Request not in DAG order is reordered to be in DAG order.
 self.assertEqual(rev.p2node, nullid)
 self.assertIsNone(rev.linknode)
 self.assertEqual(rev.basenode, node0)
 self.assertIsNone(rev.baserevisionsize)
 self.assertIsNone(rev.revision)
-self.assertEqual(rev.delta,
+self.assertEqual(
-b'\x00\x00\x00\x00\x00\x00\x04\x00\x00\x00\x04\x01' +
+rev.delta,
-fulltext1)
+b'\x00\x00\x00\x00\x00\x00\x04\x00\x00\x00\x04\x01' + fulltext1,
+)
 rev = next(gen)
 self.assertEqual(rev.node, node2)
 self.assertEqual(rev.p1node, node1)
 self.assertEqual(rev.p2node, nullid)
 self.assertIsNone(rev.linknode)
 self.assertEqual(rev.basenode, node1)
 self.assertIsNone(rev.baserevisionsize)
 self.assertIsNone(rev.revision)
-self.assertEqual(rev.delta,
+self.assertEqual(
-b'\x00\x00\x00\x00\x00\x00\x04\x01\x00\x00\x04\x02' +
+rev.delta,
-fulltext2)
+b'\x00\x00\x00\x00\x00\x00\x04\x01\x00\x00\x04\x02' + fulltext2,
+)
 with self.assertRaises(StopIteration):
 next(gen)
 # Unrecognized nodesorder value raises ProgrammingError.
 with self.assertRaises(error.ProgrammingError):
 list(f.emitrevisions([], nodesorder='bad'))
 # nodesorder=storage is recognized. But we can't test it thoroughly
 # because behavior is storage-dependent.
-res = list(f.emitrevisions([node2, node1, node0],
+res = list(f.emitrevisions([node2, node1, node0], nodesorder='storage'))
-nodesorder='storage'))
 self.assertEqual(len(res), 3)
 self.assertEqual({o.node for o in res}, {node0, node1, node2})
 # nodesorder=nodes forces the order.
-gen = f.emitrevisions([node2, node0], nodesorder='nodes',
+gen = f.emitrevisions(
-revisiondata=True)
+[node2, node0], nodesorder='nodes', revisiondata=True
+)
 rev = next(gen)
 self.assertEqual(rev.node, node2)
 self.assertEqual(rev.p1node, node1)
 self.assertEqual(rev.p2node, nullid)
 self.assertEqual(rev.p1node, node1)
 self.assertEqual(rev.p2node, nullid)
 self.assertEqual(rev.basenode, node1)
 self.assertIsNone(rev.baserevisionsize)
 self.assertIsNone(rev.revision)
-self.assertEqual(rev.delta,
+self.assertEqual(
-b'\x00\x00\x00\x00\x00\x00\x04\x01\x00\x00\x04\x02' +
+rev.delta,
-fulltext2)
+b'\x00\x00\x00\x00\x00\x00\x04\x01\x00\x00\x04\x02' + fulltext2,
+)
 with self.assertRaises(StopIteration):
 next(gen)
 # assumehaveparentrevisions=True allows delta against initial revision.
-gen = f.emitrevisions([node2, node1],
+gen = f.emitrevisions(
-revisiondata=True, assumehaveparentrevisions=True)
+[node2, node1], revisiondata=True, assumehaveparentrevisions=True
+)
 rev = next(gen)
 self.assertEqual(rev.node, node1)
 self.assertEqual(rev.p1node, node0)
 self.assertEqual(rev.p2node, nullid)
 self.assertEqual(rev.basenode, node0)
 self.assertIsNone(rev.baserevisionsize)
 self.assertIsNone(rev.revision)
-self.assertEqual(rev.delta,
+self.assertEqual(
-b'\x00\x00\x00\x00\x00\x00\x04\x00\x00\x00\x04\x01' +
+rev.delta,
-fulltext1)
+b'\x00\x00\x00\x00\x00\x00\x04\x00\x00\x00\x04\x01' + fulltext1,
+)
 # forceprevious=True forces a delta against the previous revision.
 # Special case for initial revision.
-gen = f.emitrevisions([node0], revisiondata=True,
+gen = f.emitrevisions(
-deltamode=repository.CG_DELTAMODE_PREV)
+[node0], revisiondata=True, deltamode=repository.CG_DELTAMODE_PREV
+)
 rev = next(gen)
 self.assertEqual(rev.node, node0)
 self.assertEqual(rev.p1node, nullid)
 self.assertEqual(rev.p2node, nullid)
 self.assertEqual(rev.basenode, nullid)
 self.assertIsNone(rev.baserevisionsize)
 self.assertIsNone(rev.revision)
-self.assertEqual(rev.delta,
+self.assertEqual(
-b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\x00' +
+rev.delta,
-fulltext0)
+b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\x00' + fulltext0,
+)
 with self.assertRaises(StopIteration):
 next(gen)
-gen = f.emitrevisions([node0, node2], revisiondata=True,
+gen = f.emitrevisions(
-deltamode=repository.CG_DELTAMODE_PREV)
+[node0, node2],
+revisiondata=True,
+deltamode=repository.CG_DELTAMODE_PREV,
+)
 rev = next(gen)
 self.assertEqual(rev.node, node0)
 self.assertEqual(rev.p1node, nullid)
 self.assertEqual(rev.p2node, nullid)
 self.assertEqual(rev.basenode, nullid)
 self.assertIsNone(rev.baserevisionsize)
 self.assertIsNone(rev.revision)
-self.assertEqual(rev.delta,
+self.assertEqual(
-b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\x00' +
+rev.delta,
-fulltext0)
+b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\x00' + fulltext0,
+)
 rev = next(gen)
 self.assertEqual(rev.node, node2)
 self.assertEqual(rev.p1node, node1)
 self.assertEqual(rev.p2node, nullid)
 meta2 = {
 b'copy': b'source1',
 b'copyrev': b'b' * 40,
 }
-stored1 = b''.join([
+stored1 = b''.join(
-b'\x01\ncopy: source0\n',
+[
-b'copyrev: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n\x01\n',
+b'\x01\ncopy: source0\n',
-fulltext1,
+b'copyrev: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n\x01\n',
-])
+fulltext1,
+]
-stored2 = b''.join([
+)
-b'\x01\ncopy: source1\n',
-b'copyrev: bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb\n\x01\n',
+stored2 = b''.join(
-fulltext2,
+[
-])
+b'\x01\ncopy: source1\n',
+b'copyrev: bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb\n\x01\n',
+fulltext2,
+]
+)
 f = self._makefilefn()
 with self._maketransactionfn() as tr:
 node0 = f.add(fulltext0, None, tr, 0, nullid, nullid)
 node1 = f.add(fulltext1, meta1, tr, 1, node0, nullid)
 fulltext1 = b'\x01\nbar'
 meta1 = {
 b'copy': b'source0',
 b'copyrev': b'b' * 40,
 }
-stored1 = b''.join([
+stored1 = b''.join(
-b'\x01\ncopy: source0\n',
+[
-b'copyrev: %s\n' % (b'b' * 40),
+b'\x01\ncopy: source0\n',
-b'\x01\n\x01\nbar',
+b'copyrev: %s\n' % (b'b' * 40),
-])
+b'\x01\n\x01\nbar',
+]
+)
 f = self._makefilefn()
 with self._maketransactionfn() as tr:
 node0 = f.add(fulltext0, {}, tr, 0, nullid, nullid)
 node1 = f.add(fulltext1, meta1, tr, 1, nullid, nullid)
 with self._maketransactionfn() as tr:
 node0 = f.add(fulltext0, None, tr, 0, nullid, nullid)
 node1 = b'\xaa' * 20
-self._addrawrevisionfn(f, tr, node1, node0, nullid, 1,
+self._addrawrevisionfn(
-rawtext=fulltext1)
+f, tr, node1, node0, nullid, 1, rawtext=fulltext1
+)
 self.assertEqual(len(f), 2)
 self.assertEqual(f.parents(node1), (node0, nullid))
 # revision() raises since it performs hash verification.
 with self._maketransactionfn() as tr:
 node0 = f.add(fulltext0, None, tr, 0, nullid, nullid)
 node1 = b'\xaa' * 20
-self._addrawrevisionfn(f, tr, node1, node0, nullid, 1,
+self._addrawrevisionfn(
-rawtext=fulltext1)
+f, tr, node1, node0, nullid, 1, rawtext=fulltext1
+)
 with self.assertRaises(error.StorageError):
 f.rawdata(node1)
 with self.assertRaises(error.StorageError):
 with self._maketransactionfn() as tr:
 node0 = f.add(fulltext0, None, tr, 0, nullid, nullid)
 node1 = b'\xaa' * 20
-self._addrawrevisionfn(f, tr, node1, node0, nullid, 1,
+self._addrawrevisionfn(
-rawtext=fulltext1)
+f, tr, node1, node0, nullid, 1, rawtext=fulltext1
+)
 with self.assertRaises(error.StorageError):
 f.read(node1)
 with self.assertRaises(error.StorageError):
 with self._maketransactionfn() as tr:
 node0 = f.add(fulltext0, None, tr, 0, nullid, nullid)
 node1 = b'\xaa' * 20
-self._addrawrevisionfn(f, tr, node1, node0, nullid, 1,
+self._addrawrevisionfn(
-rawtext=fulltext1)
+f, tr, node1, node0, nullid, 1, rawtext=fulltext1
+)
 with self.assertRaises(error.StorageError):
 f.read(node1)
 node2 = storageutil.hashrevisionsha1(fulltext2, node1, nullid)
 with self._maketransactionfn() as tr:
 delta = mdiff.textdiff(fulltext1, fulltext2)
-self._addrawrevisionfn(f, tr, node2, node1, nullid,
+self._addrawrevisionfn(
-2, delta=(1, delta))
+f, tr, node2, node1, nullid, 2, delta=(1, delta)
+)
 self.assertEqual(len(f), 3)
 # Assuming a delta is stored, we shouldn't need to validate node1 in
 # order to retrieve node2.
 self.assertEqual(f.read(node2), fulltext2)
 def testcensored(self):
 f = self._makefilefn()
-stored1 = storageutil.packmeta({
+stored1 = storageutil.packmeta({b'censored': b'tombstone',}, b'')
-b'censored': b'tombstone',
-}, b'')
 with self._maketransactionfn() as tr:
 node0 = f.add(b'foo', None, tr, 0, nullid, nullid)
 # The node value doesn't matter since we can't verify it.
 node1 = b'\xbb' * 20
-self._addrawrevisionfn(f, tr, node1, node0, nullid, 1, stored1,
+self._addrawrevisionfn(
-censored=True)
+f, tr, node1, node0, nullid, 1, stored1, censored=True
+)
 self.assertTrue(f.iscensored(1))
 with self.assertRaises(error.CensoredNodeError):
 f.revision(1)
 # Like above, except we do the rawdata() request first to
 # isolate revision caching behavior.
 f = self._makefilefn()
-stored1 = storageutil.packmeta({
+stored1 = storageutil.packmeta({b'censored': b'tombstone',}, b'')
-b'censored': b'tombstone',
-}, b'')
 with self._maketransactionfn() as tr:
 node0 = f.add(b'foo', None, tr, 0, nullid, nullid)
 # The node value doesn't matter since we can't verify it.
 node1 = b'\xbb' * 20
-self._addrawrevisionfn(f, tr, node1, node0, nullid, 1, stored1,
+self._addrawrevisionfn(
-censored=True)
+f, tr, node1, node0, nullid, 1, stored1, censored=True
+)
 with self.assertRaises(error.CensoredNodeError):
 f.rawdata(1)
 class ifilemutationtests(basetestcase):
 """Generic tests for the ifilemutation interface.
 All file storage backends that support writing should conform to this
 interface.
 Use ``makeifilemutationtests()`` to create an instance of this type.
 """
 def testaddnoop(self):
 f = self._makefilefn()
 with self._maketransactionfn() as tr:
 node0 = f.add(b'foo', None, tr, 0, nullid, nullid)
 node1 = f.add(b'foo', None, tr, 0, nullid, nullid)
 def testaddgroupsimple(self):
 f = self._makefilefn()
 callbackargs = []
 def cb(*args, **kwargs):
 callbackargs.append((args, kwargs))
 def linkmapper(node):
 return 0
 ]
 with self._maketransactionfn() as tr:
 nodes = f.addgroup(deltas, linkmapper, tr, addrevisioncb=cb)
-self.assertEqual(nodes, [
+self.assertEqual(
-b'\x49\xd8\xcb\xb1\x5c\xe2\x57\x92\x04\x47'
+nodes,
-b'\x00\x6b\x46\x97\x8b\x7a\xf9\x80\xa9\x79'])
+[
+b'\x49\xd8\xcb\xb1\x5c\xe2\x57\x92\x04\x47'
+b'\x00\x6b\x46\x97\x8b\x7a\xf9\x80\xa9\x79'
+],
+)
 self.assertEqual(len(callbackargs), 1)
 self.assertEqual(callbackargs[0][0][1], nodes[0])
 self.assertEqual(list(f.revs()), [0])
 def testdeltaagainstcensored(self):
 # Attempt to apply a delta made against a censored revision.
 f = self._makefilefn()
-stored1 = storageutil.packmeta({
+stored1 = storageutil.packmeta({b'censored': b'tombstone',}, b'')
-b'censored': b'tombstone',
-}, b'')
 with self._maketransactionfn() as tr:
 node0 = f.add(b'foo\n' * 30, None, tr, 0, nullid, nullid)
 # The node value doesn't matter since we can't verify it.
 node1 = b'\xbb' * 20
-self._addrawrevisionfn(f, tr, node1, node0, nullid, 1, stored1,
+self._addrawrevisionfn(
-censored=True)
+f, tr, node1, node0, nullid, 1, stored1, censored=True
+)
 delta = mdiff.textdiff(b'bar\n' * 30, (b'bar\n' * 30) + b'baz\n')
 deltas = [(b'\xcc' * 20, node1, nullid, b'\x01' * 20, node1, delta, 0)]
 with self._maketransactionfn() as tr:
 self.assertEqual(len(f), 1)
 with self.assertRaises(error.LookupError):
 f.rev(node1)
 def makeifileindextests(makefilefn, maketransactionfn, addrawrevisionfn):
 """Create a unittest.TestCase class suitable for testing file storage.
 ``makefilefn`` is a callable which receives the test case as an
 argument and returns an object implementing the ``ifilestorage`` interface.
 r'_maketransactionfn': maketransactionfn,
 r'_addrawrevisionfn': addrawrevisionfn,
 }
 return type(r'ifileindextests', (ifileindextests,), d)
 def makeifiledatatests(makefilefn, maketransactionfn, addrawrevisionfn):
 d = {
 r'_makefilefn': makefilefn,
 r'_maketransactionfn': maketransactionfn,
 r'_addrawrevisionfn': addrawrevisionfn,
 }
 return type(r'ifiledatatests', (ifiledatatests,), d)
 def makeifilemutationtests(makefilefn, maketransactionfn, addrawrevisionfn):
 d = {
 r'_makefilefn': makefilefn,
 r'_maketransactionfn': maketransactionfn,
 r'_addrawrevisionfn': addrawrevisionfn,

changeset 43076	2372284d9457
parent 42813	268662aac075
child 43077	687b865b95ad