1# Copyright (C) 2010, 2011 Sebastian Thiel (byronimo@gmail.com) and contributors 2# 3# This module is part of GitDB and is released under 4# the New BSD License: http://www.opensource.org/licenses/bsd-license.php 5"""Base classes for object db testing""" 6from gitdb.test.lib import ( 7 with_rw_directory, 8 with_packs_rw, 9 fixture_path, 10 TestBase 11) 12 13from gitdb.stream import ( 14 Sha1Writer, 15 ZippedStoreShaWriter 16) 17 18from gitdb.base import ( 19 IStream, 20 OStream, 21 OInfo 22) 23 24from gitdb.exc import BadObject 25from gitdb.typ import str_blob_type 26 27from io import BytesIO 28 29from struct import pack 30 31 32__all__ = ('TestDBBase', 'with_rw_directory', 'with_packs_rw', 'fixture_path') 33 34 35class TestDBBase(TestBase): 36 37 """Base class providing testing routines on databases""" 38 39 # data 40 two_lines = b'1234\nhello world' 41 all_data = (two_lines, ) 42 43 def _assert_object_writing_simple(self, db): 44 # write a bunch of objects and query their streams and info 45 null_objs = db.size() 46 ni = 250 47 for i in range(ni): 48 data = pack(">L", i) 49 istream = IStream(str_blob_type, len(data), BytesIO(data)) 50 new_istream = db.store(istream) 51 assert new_istream is istream 52 assert db.has_object(istream.binsha) 53 54 info = db.info(istream.binsha) 55 assert isinstance(info, OInfo) 56 assert info.type == istream.type and info.size == istream.size 57 58 stream = db.stream(istream.binsha) 59 assert isinstance(stream, OStream) 60 assert stream.binsha == info.binsha and stream.type == info.type 61 assert stream.read() == data 62 # END for each item 63 64 assert db.size() == null_objs + ni 65 shas = list(db.sha_iter()) 66 assert len(shas) == db.size() 67 assert len(shas[0]) == 20 68 69 def _assert_object_writing(self, db): 70 """General tests to verify object writing, compatible to ObjectDBW 71 **Note:** requires write access to the database""" 72 # start in 'dry-run' mode, using a simple sha1 writer 73 ostreams = (ZippedStoreShaWriter, None) 74 for ostreamcls in ostreams: 75 for data in self.all_data: 76 dry_run = ostreamcls is not None 77 ostream = None 78 if ostreamcls is not None: 79 ostream = ostreamcls() 80 assert isinstance(ostream, Sha1Writer) 81 # END create ostream 82 83 prev_ostream = db.set_ostream(ostream) 84 assert type(prev_ostream) in ostreams or prev_ostream in ostreams 85 istream = IStream(str_blob_type, len(data), BytesIO(data)) 86 87 # store returns same istream instance, with new sha set 88 my_istream = db.store(istream) 89 sha = istream.binsha 90 assert my_istream is istream 91 assert db.has_object(sha) != dry_run 92 assert len(sha) == 20 93 94 # verify data - the slow way, we want to run code 95 if not dry_run: 96 info = db.info(sha) 97 assert str_blob_type == info.type 98 assert info.size == len(data) 99 100 ostream = db.stream(sha) 101 assert ostream.read() == data 102 assert ostream.type == str_blob_type 103 assert ostream.size == len(data) 104 else: 105 self.assertRaises(BadObject, db.info, sha) 106 self.assertRaises(BadObject, db.stream, sha) 107 108 # DIRECT STREAM COPY 109 # our data hase been written in object format to the StringIO 110 # we pasesd as output stream. No physical database representation 111 # was created. 112 # Test direct stream copy of object streams, the result must be 113 # identical to what we fed in 114 ostream.seek(0) 115 istream.stream = ostream 116 assert istream.binsha is not None 117 prev_sha = istream.binsha 118 119 db.set_ostream(ZippedStoreShaWriter()) 120 db.store(istream) 121 assert istream.binsha == prev_sha 122 new_ostream = db.ostream() 123 124 # note: only works as long our store write uses the same compression 125 # level, which is zip_best 126 assert ostream.getvalue() == new_ostream.getvalue() 127 # END for each data set 128 # END for each dry_run mode 129