yield z.flush()
-def _encode_looseobj(type, content, compression_level=1):
- z = zlib.compressobj(compression_level)
- yield z.compress(b'%s %d\0' % (type, len(content)))
- yield z.compress(content)
- yield z.flush()
-
-
-def _decode_looseobj(buf):
- assert(buf);
- s = zlib.decompress(buf)
- i = s.find(b'\0')
- assert(i > 0)
- l = s[:i].split(b' ')
- type = l[0]
- sz = int(l[1])
- content = s[i+1:]
- assert(type in _typemap)
- assert(sz == len(content))
- return (type, content)
-
-
def _decode_packobj(buf):
assert(buf)
c = byte_int(buf[0])
def test_encode():
s = b'hello world'
- looseb = b''.join(git._encode_looseobj(b'blob', s))
- looset = b''.join(git._encode_looseobj(b'tree', s))
- loosec = b''.join(git._encode_looseobj(b'commit', s))
packb = b''.join(git._encode_packobj(b'blob', s))
packt = b''.join(git._encode_packobj(b'tree', s))
packc = b''.join(git._encode_packobj(b'commit', s))
packlb = b''.join(git._encode_packobj(b'blob', s * 200))
- WVPASSEQ(git._decode_looseobj(looseb), (b'blob', s))
- WVPASSEQ(git._decode_looseobj(looset), (b'tree', s))
- WVPASSEQ(git._decode_looseobj(loosec), (b'commit', s))
WVPASSEQ(git._decode_packobj(packb), (b'blob', s))
WVPASSEQ(git._decode_packobj(packt), (b'tree', s))
WVPASSEQ(git._decode_packobj(packc), (b'commit', s))
WVPASSEQ(git._decode_packobj(packlb), (b'blob', s * 200))
- for i in range(10):
- WVPASS(git._encode_looseobj(b'blob', s, compression_level=i))
def encode_pobj(n):
return b''.join(git._encode_packobj(b'blob', s, compression_level=n))
WVEXCEPT(ValueError, encode_pobj, -1)