##// END OF EJS Templates
exchange: move disabling of rev-branch-cache bundle part out of narrow...
exchange: move disabling of rev-branch-cache bundle part out of narrow I'm attempting to refactor changegroup code in order to better support alternate storage backends. The narrow extension is performing a lot of monkeypatching to this code and it is making it difficult to reason about how everything works. I'm reasonably certain I would be unable to abstract storage without requiring extensive rework of narrow. I believe it is less effort to move narrow code into core so it can be accounted for when changegroup code is refactored. So I'll be doing that. The first part of this is integrating the disabling of the cache:rev-branch-cache bundle2 part into core. This doesn't seem like it is related to changegroup, but narrow's modifications to changegroup are invasive and also require taking its code for bundle generation and exchange into core in order for the changegroup code to work. Differential Revision: https://phab.mercurial-scm.org/D4007

File last commit:

r37513:b1fb341d default
r38813:ab765bc4 default
Show More
test_buffer_util.py
108 lines | 4.0 KiB | text/x-python | PythonLexer
import struct
import unittest
import zstandard as zstd
ss = struct.Struct('=QQ')
class TestBufferWithSegments(unittest.TestCase):
def test_arguments(self):
with self.assertRaises(TypeError):
zstd.BufferWithSegments()
with self.assertRaises(TypeError):
zstd.BufferWithSegments(b'foo')
# Segments data should be a multiple of 16.
with self.assertRaisesRegexp(ValueError, 'segments array size is not a multiple of 16'):
zstd.BufferWithSegments(b'foo', b'\x00\x00')
def test_invalid_offset(self):
with self.assertRaisesRegexp(ValueError, 'offset within segments array references memory'):
zstd.BufferWithSegments(b'foo', ss.pack(0, 4))
def test_invalid_getitem(self):
b = zstd.BufferWithSegments(b'foo', ss.pack(0, 3))
with self.assertRaisesRegexp(IndexError, 'offset must be non-negative'):
test = b[-10]
with self.assertRaisesRegexp(IndexError, 'offset must be less than 1'):
test = b[1]
with self.assertRaisesRegexp(IndexError, 'offset must be less than 1'):
test = b[2]
def test_single(self):
b = zstd.BufferWithSegments(b'foo', ss.pack(0, 3))
self.assertEqual(len(b), 1)
self.assertEqual(b.size, 3)
self.assertEqual(b.tobytes(), b'foo')
self.assertEqual(len(b[0]), 3)
self.assertEqual(b[0].offset, 0)
self.assertEqual(b[0].tobytes(), b'foo')
def test_multiple(self):
b = zstd.BufferWithSegments(b'foofooxfooxy', b''.join([ss.pack(0, 3),
ss.pack(3, 4),
ss.pack(7, 5)]))
self.assertEqual(len(b), 3)
self.assertEqual(b.size, 12)
self.assertEqual(b.tobytes(), b'foofooxfooxy')
self.assertEqual(b[0].tobytes(), b'foo')
self.assertEqual(b[1].tobytes(), b'foox')
self.assertEqual(b[2].tobytes(), b'fooxy')
class TestBufferWithSegmentsCollection(unittest.TestCase):
def test_empty_constructor(self):
with self.assertRaisesRegexp(ValueError, 'must pass at least 1 argument'):
zstd.BufferWithSegmentsCollection()
def test_argument_validation(self):
with self.assertRaisesRegexp(TypeError, 'arguments must be BufferWithSegments'):
zstd.BufferWithSegmentsCollection(None)
with self.assertRaisesRegexp(TypeError, 'arguments must be BufferWithSegments'):
zstd.BufferWithSegmentsCollection(zstd.BufferWithSegments(b'foo', ss.pack(0, 3)),
None)
with self.assertRaisesRegexp(ValueError, 'ZstdBufferWithSegments cannot be empty'):
zstd.BufferWithSegmentsCollection(zstd.BufferWithSegments(b'', b''))
def test_length(self):
b1 = zstd.BufferWithSegments(b'foo', ss.pack(0, 3))
b2 = zstd.BufferWithSegments(b'barbaz', b''.join([ss.pack(0, 3),
ss.pack(3, 3)]))
c = zstd.BufferWithSegmentsCollection(b1)
self.assertEqual(len(c), 1)
self.assertEqual(c.size(), 3)
c = zstd.BufferWithSegmentsCollection(b2)
self.assertEqual(len(c), 2)
self.assertEqual(c.size(), 6)
c = zstd.BufferWithSegmentsCollection(b1, b2)
self.assertEqual(len(c), 3)
self.assertEqual(c.size(), 9)
def test_getitem(self):
b1 = zstd.BufferWithSegments(b'foo', ss.pack(0, 3))
b2 = zstd.BufferWithSegments(b'barbaz', b''.join([ss.pack(0, 3),
ss.pack(3, 3)]))
c = zstd.BufferWithSegmentsCollection(b1, b2)
with self.assertRaisesRegexp(IndexError, 'offset must be less than 3'):
c[3]
with self.assertRaisesRegexp(IndexError, 'offset must be less than 3'):
c[4]
self.assertEqual(c[0].tobytes(), b'foo')
self.assertEqual(c[1].tobytes(), b'bar')
self.assertEqual(c[2].tobytes(), b'baz')