path: root/test/git/test_odb.py

"""Test for object db"""
from test.testlib import *
from git.odb import *
from git.odb.utils import (
	to_hex_sha, 
	to_bin_sha
	)
from git.odb.stream import Sha1Writer
from git import Blob
from git.errors import BadObject
from cStringIO import StringIO
import tempfile
import os
import zlib


#{ Stream Utilities

class DummyStream(object):
		def __init__(self):
			self.was_read = False
			self.bytes = 0
			self.closed = False
			
		def read(self, size):
			self.was_read = True
			self.bytes = size
			
		def close(self):
			self.closed = True
			
		def _assert(self):
			assert self.was_read
	
class DeriveTest(OStream):
	def __init__(self, sha, type, size, stream, *args, **kwargs):
		self.myarg = kwargs.pop('myarg')
		self.args = args
		
	def _assert(self):
		assert self.args
		assert self.myarg

#} END stream utilitiess
	
	

class TestStream(TestBase):
	"""Test stream classes"""
	
	data_sizes = (15, 10000, 1000*1024+512)
	
	def test_streams(self):
		# test info
		sha = Blob.NULL_HEX_SHA
		s = 20
		info = OInfo(sha, Blob.type, s)
		assert info.sha == sha
		assert info.type == Blob.type
		assert info.size == s
		
		# test ostream
		stream = DummyStream()
		ostream = OStream(*(info + (stream, )))
		ostream.read(15)
		stream._assert()
		assert stream.bytes == 15
		ostream.read(20)
		assert stream.bytes == 20
		
		# defaults false
		assert not ostream.is_compressed()
		
		# derive with own args
		DeriveTest(sha, Blob.type, s, stream, 'mine',myarg = 3)._assert()
		
		# test istream
		istream = IStream(Blob.type, s, stream)
		assert not istream.is_compressed()
		assert istream.sha == None
		istream.sha = sha
		assert istream.sha == sha
		
		assert len(istream.binsha) == 20
		assert len(istream.hexsha) == 40
		
		assert istream.size == s
		istream.size = s * 2
		istream.size == s * 2
		assert istream.type == Blob.type
		istream.type = "something"
		assert istream.type == "something"
		assert istream.stream is stream
		istream.stream = None
		assert istream.stream is None
		
	def _assert_stream_reader(self, stream, cdata, rewind_stream=lambda s: None):
		"""Make stream tests - the orig_stream is seekable, allowing it to be 
		rewound and reused
		:param cdata: the data we expect to read from stream, the contents
		:param rewind_stream: function called to rewind the stream to make it ready
			for reuse"""
		ns = 10
		assert len(cdata) > ns-1, "Data must be larger than %i, was %i" % (ns, len(cdata))
		
		# read in small steps
		ss = len(cdata) / ns
		for i in range(ns):
			data = stream.read(ss)
			chunk = cdata[i*ss:(i+1)*ss]
			assert data == chunk
		# END for each step
		rest = stream.read()
		if rest:
			assert rest == cdata[-len(rest):]
		# END handle rest
		
		rewind_stream(stream)
		
		# read everything
		rdata = stream.read()
		assert rdata == cdata
		
	def test_decompress_reader(self):
		for close_on_deletion in range(2):
			for with_size in range(2):
				for ds in self.data_sizes:
					cdata = make_bytes(ds, randomize=False)
					
					# zdata = zipped actual data
					# cdata = original content data
					
					# create reader
					if with_size:
						# need object data
						zdata = zlib.compress(make_object(Blob.type, cdata))
						type, size, reader = DecompressMemMapReader.new(zdata, close_on_deletion)
						assert size == len(cdata)
						assert type == Blob.type
					else:
						# here we need content data
						zdata = zlib.compress(cdata)
						reader = DecompressMemMapReader(zdata, close_on_deletion, len(cdata))
						assert reader._s == len(cdata)
					# END get reader 
					
					def rewind(r):
						r._zip = zlib.decompressobj()
						r._br = r._cws = r._cwe = 0
						if with_size:
							r._parse_header_info()
						# END skip header
					# END make rewind func
					
					self._assert_stream_reader(reader, cdata, rewind)
					
					# put in a dummy stream for closing
					dummy = DummyStream()
					reader._m = dummy
					
					assert not dummy.closed
					del(reader)
					assert dummy.closed == close_on_deletion
					#zdi#
				# END for each datasize
			# END whether size should be used
		# END whether stream should be closed when deleted
		
	def test_sha_writer(self):
		writer = Sha1Writer()
		assert 2 == writer.write("hi")
		assert len(writer.sha(as_hex=1)) == 40
		assert len(writer.sha(as_hex=0)) == 20
		
		# make sure it does something ;)
		prev_sha = writer.sha()
		writer.write("hi again")
		assert writer.sha() != prev_sha
		
	def test_compressed_writer(self):
		for ds in self.data_sizes:
			fd, path = tempfile.mkstemp()
			ostream = FDCompressedSha1Writer(fd)
			data = make_bytes(ds, randomize=False)
			
			# for now, just a single write, code doesn't care about chunking
			assert len(data) == ostream.write(data)
			ostream.close()
			# its closed already
			self.failUnlessRaises(OSError, os.close, fd)
			
			# read everything back, compare to data we zip
			fd = os.open(path, os.O_RDONLY)
			written_data = os.read(fd, os.path.getsize(path))
			os.close(fd)
			assert written_data == zlib.compress(data, 1)	# best speed
			
			os.remove(path)
		# END for each os
	
	
class TestUtils(TestBase):
	def test_basics(self):
		assert to_hex_sha(Blob.NULL_HEX_SHA) == Blob.NULL_HEX_SHA
		assert len(to_bin_sha(Blob.NULL_HEX_SHA)) == 20
		assert to_hex_sha(to_bin_sha(Blob.NULL_HEX_SHA)) == Blob.NULL_HEX_SHA
		

class TestDB(TestBase):
	"""Test the different db class implementations"""
	
	# data
	two_lines = "1234\nhello world"
	
	all_data = (two_lines, )
	
	def _assert_object_writing(self, db):
		"""General tests to verify object writing, compatible to ObjectDBW
		:note: requires write access to the database"""
		# start in 'dry-run' mode, using a simple sha1 writer
		ostreams = (Sha1Writer, None)
		for ostreamcls in ostreams:
			for data in self.all_data:
				dry_run = ostreamcls is not None
				ostream = None
				if ostreamcls is not None:
					ostream = ostreamcls()
				# END create ostream
				
				prev_ostream = db.set_ostream(ostream)
				assert type(prev_ostream) in ostreams or prev_ostream in ostreams 
					
				istream = IStream(Blob.type, len(data), StringIO(data))
				
				# store returns same istream instance, with new sha set
				my_istream = db.store(istream)
				sha = istream.sha
				assert my_istream is istream
				assert db.has_object(sha) != dry_run
				assert len(sha) == 40		# for now we require 40 byte shas as default
				
				# verify data - the slow way, we want to run code
				if not dry_run:
					info = db.info(sha)
					assert Blob.type == info.type
					assert info.size == len(data)
					
					ostream = db.stream(sha)
					assert ostream.read() == data
					assert ostream.type == Blob.type
					assert ostream.size == len(data)
				else:
					self.failUnlessRaises(BadObject, db.info, sha)
					self.failUnlessRaises(BadObject, db.stream, sha)
			# END for each data set
		# END for each dry_run mode
				
	@with_bare_rw_repo
	def test_writing(self, rwrepo):
		ldb = LooseObjectDB(os.path.join(rwrepo.git_dir, 'objects'))
		
		# write data
		self._assert_object_writing(ldb)