C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// # Copyright (C) 2010, 2011 Sebastian Thiel (byronimo@gmail.com) and contributors # # This module is part of GitDB and is released under # the New BSD License: http://www.opensource.org/licenses/bsd-license.php """Test everything about packs reading and writing""" from gitdb.test.lib import ( TestBase, with_rw_directory, fixture_path ) from gitdb.stream import DeltaApplyReader from gitdb.pack import ( PackEntity, PackIndexFile, PackFile ) from gitdb.base import ( OInfo, OStream, ) from gitdb.fun import delta_types from gitdb.exc import UnsupportedOperation from gitdb.util import to_bin_sha import pytest import os import tempfile #{ Utilities def bin_sha_from_filename(filename): return to_bin_sha(os.path.splitext(os.path.basename(filename))[0][5:]) #} END utilities class TestPack(TestBase): packindexfile_v1 = (fixture_path('packs/pack-c0438c19fb16422b6bbcce24387b3264416d485b.idx'), 1, 67) packindexfile_v2 = (fixture_path('packs/pack-11fdfa9e156ab73caae3b6da867192221f2089c2.idx'), 2, 30) packindexfile_v2_3_ascii = (fixture_path('packs/pack-a2bf8e71d8c18879e499335762dd95119d93d9f1.idx'), 2, 42) packfile_v2_1 = (fixture_path('packs/pack-c0438c19fb16422b6bbcce24387b3264416d485b.pack'), 2, packindexfile_v1[2]) packfile_v2_2 = (fixture_path('packs/pack-11fdfa9e156ab73caae3b6da867192221f2089c2.pack'), 2, packindexfile_v2[2]) packfile_v2_3_ascii = ( fixture_path('packs/pack-a2bf8e71d8c18879e499335762dd95119d93d9f1.pack'), 2, packindexfile_v2_3_ascii[2]) def _assert_index_file(self, index, version, size): assert index.packfile_checksum() != index.indexfile_checksum() assert len(index.packfile_checksum()) == 20 assert len(index.indexfile_checksum()) == 20 assert index.version() == version assert index.size() == size assert len(index.offsets()) == size # get all data of all objects for oidx in range(index.size()): sha = index.sha(oidx) assert oidx == index.sha_to_index(sha) entry = index.entry(oidx) assert len(entry) == 3 assert entry[0] == index.offset(oidx) assert entry[1] == sha assert entry[2] == index.crc(oidx) # verify partial sha for l in (4, 8, 11, 17, 20): assert index.partial_sha_to_index(sha[:l], l * 2) == oidx # END for each object index in indexfile self.assertRaises(ValueError, index.partial_sha_to_index, "\0", 2) def _assert_pack_file(self, pack, version, size): assert pack.version() == 2 assert pack.size() == size assert len(pack.checksum()) == 20 num_obj = 0 for obj in pack.stream_iter(): num_obj += 1 info = pack.info(obj.pack_offset) stream = pack.stream(obj.pack_offset) assert info.pack_offset == stream.pack_offset assert info.type_id == stream.type_id assert hasattr(stream, 'read') # it should be possible to read from both streams assert obj.read() == stream.read() streams = pack.collect_streams(obj.pack_offset) assert streams # read the stream try: dstream = DeltaApplyReader.new(streams) except ValueError: # ignore these, old git versions use only ref deltas, # which we haven't resolved ( as we are without an index ) # Also ignore non-delta streams continue # END get deltastream # read all data = dstream.read() assert len(data) == dstream.size # test seek dstream.seek(0) assert dstream.read() == data # read chunks # NOTE: the current implementation is safe, it basically transfers # all calls to the underlying memory map # END for each object assert num_obj == size def test_pack_index(self): # check version 1 and 2 for indexfile, version, size in (self.packindexfile_v1, self.packindexfile_v2): index = PackIndexFile(indexfile) self._assert_index_file(index, version, size) # END run tests def test_pack(self): # there is this special version 3, but apparently its like 2 ... for packfile, version, size in (self.packfile_v2_3_ascii, self.packfile_v2_1, self.packfile_v2_2): pack = PackFile(packfile) self._assert_pack_file(pack, version, size) # END for each pack to test @with_rw_directory def test_pack_entity(self, rw_dir): pack_objs = list() for packinfo, indexinfo in ((self.packfile_v2_1, self.packindexfile_v1), (self.packfile_v2_2, self.packindexfile_v2), (self.packfile_v2_3_ascii, self.packindexfile_v2_3_ascii)): packfile, version, size = packinfo indexfile, version, size = indexinfo entity = PackEntity(packfile) assert entity.pack().path() == packfile assert entity.index().path() == indexfile pack_objs.extend(entity.stream_iter()) count = 0 for info, stream in zip(entity.info_iter(), entity.stream_iter()): count += 1 assert info.binsha == stream.binsha assert len(info.binsha) == 20 assert info.type_id == stream.type_id assert info.size == stream.size # we return fully resolved items, which is implied by the sha centric access assert not info.type_id in delta_types # try all calls assert len(entity.collect_streams(info.binsha)) oinfo = entity.info(info.binsha) assert isinstance(oinfo, OInfo) assert oinfo.binsha is not None ostream = entity.stream(info.binsha) assert isinstance(ostream, OStream) assert ostream.binsha is not None # verify the stream try: assert entity.is_valid_stream(info.binsha, use_crc=True) except UnsupportedOperation: pass # END ignore version issues assert entity.is_valid_stream(info.binsha, use_crc=False) # END for each info, stream tuple assert count == size # END for each entity # pack writing - write all packs into one # index path can be None pack_path1 = tempfile.mktemp('', "pack1", rw_dir) pack_path2 = tempfile.mktemp('', "pack2", rw_dir) index_path = tempfile.mktemp('', 'index', rw_dir) iteration = 0 def rewind_streams(): for obj in pack_objs: obj.stream.seek(0) # END utility for ppath, ipath, num_obj in zip((pack_path1, pack_path2), (index_path, None), (len(pack_objs), None)): iwrite = None if ipath: ifile = open(ipath, 'wb') iwrite = ifile.write # END handle ip # make sure we rewind the streams ... we work on the same objects over and over again if