Added basic channel implementation including test

restructured odb tests, they are now in an own module to keep the modules small

Author: Byron

lib/git/odb/channel.py

@@ -0,0 +1,106 @@
+"""Contains a queue based channel implementation"""
+from Queue import (
+	Queue, 
+	Empty, 
+	Full
+	)
+
+#{ Classes 
+class Channel(object):
+	"""A channel is similar to a system pipe. It has a write end as well as one or
+	more read ends. If Data is in the channel, it can be read, if not the read operation
+	will block until data becomes available.
+	If the channel is closed, any read operation will result in an exception
+	
+	This base class is not instantiated directly, but instead serves as constructor
+	for RWChannel pairs.
+	
+	Create a new channel """
+	__slots__ = tuple()
+	def __new__(cls, *args):
+		if cls is Channel:
+			max_items = 0
+			if len(args) == 1:
+				max_items = args[0]
+			if len(args) > 1:
+				raise ValueError("Specify not more than the number of items the channel should take")
+			wc = WChannel(max_items)
+			rc = RChannel(wc)
+			return wc, rc
+		# END constructor mode
+		return object.__new__(cls)
+
+class WChannel(Channel):
+	"""The write end of a channel"""
+	__slots__ = ('_closed', '_queue')
+	
+	def __init__(self, max_items=0):
+		"""initialize this instance, able to hold max_items at once
+		Write calls will block if the channel is full, until someone reads from it"""
+		self._closed = False
+		self._queue = Queue(max_items)
+		
+	
+	#{ Interface 
+	def write(self, item, block=True, timeout=None):
+		"""Send an item into the channel, it can be read from the read end of the 
+		channel accordingly
+		:param item: Item to send
+		:param block: If True, the call will block until there is free space in the 
+			channel
+		:param timeout: timeout in seconds for blocking calls.
+		:raise IOError: when writing into closed file or when writing into a non-blocking
+			full channel
+		:note: may block if the channel has a limited capacity"""
+		if self._closed:
+			raise IOError("Cannot write to a closed channel")
+			
+		try:
+			self._queue.put(item, block, timeout)
+		except Full:
+			raise IOError("Capacity of the channel was exeeded")
+		# END exception handling
+		
+	def close(self):
+		"""Close the channel. Multiple close calls on a closed channel are no 
+		an error"""
+		self._closed = True
+		
+	@property
+	def closed(self):
+		""":return: True if the channel was closed"""
+		return self._closed
+	#} END interface 
+	
+
+class RChannel(Channel):
+	"""The read-end of a corresponding write channel"""
+	__slots__ = '_wc'
+	
+	def __init__(self, wchannel):
+		"""Initialize this instance from its parent write channel"""
+		self._wc = wchannel
+		
+		
+	#{ Interface
+	
+	def read(self, block=True, timeout=None):
+		""":return: an item read from the channel
+		:param block: if True, the call will block until an item is available
+		:param timeout: if positive and block is True, it will block only for the 
+			given amount of seconds.
+		:raise IOError: When reading from an empty channel ( if non-blocking, or 
+			if the channel is still empty after the timeout"""
+		# if the channel is closed for writing, we never block
+		if self._wc.closed:
+			block = False
+			
+		try:
+			return self._wc._queue.get(block, timeout)
+		except Empty:
+			raise IOError("Error reading from an empty channel")
+		# END handle reading
+		
+	#} END interface 
+	
+#} END classes

lib/git/odb/db.py

@@ -142,11 +142,10 @@ def __init__(self, root_path):
 		"""Initialize this instance to look for its files at the given root path
 		All subsequent operations will be relative to this path
 		:raise InvalidDBRoot: 
-		:note: The base will perform basic checking for accessability, but the subclass
-			is required to verify that the root_path contains the database structure it needs"""
+		:note: The base will not perform any accessablity checking as the base
+			might not yet be accessible, but become accessible before the first 
+			access."""
 		super(FileDBBase, self).__init__()
-		if not os.path.isdir(root_path):
-			raise InvalidDBRoot(root_path)
 		self._root_path = root_path
 
 
@@ -333,10 +332,10 @@ def __init__(self, root_path, git):
 
 	def info(self, sha):
 		t = self._git.get_object_header(sha)
-		return OInfo(t[0], t[1], t[2])
+		return OInfo(*t)
 
 	def stream(self, sha):
 		"""For now, all lookup is done by git itself"""
 		t = self._git.stream_object_data(sha)
-		return OStream(t[0], t[1], t[2], t[3])
+		return OStream(*t)
 

test/git/odb/__init__.py

@@ -0,0 +1 @@
+

test/git/odb/lib.py

@@ -0,0 +1,60 @@
+"""Utilities used in ODB testing"""
+from git.odb import (
+	OStream, 
+	)
+from git.odb.stream import Sha1Writer
+
+import zlib
+from cStringIO import StringIO
+
+#{ 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
+
+
+class ZippedStoreShaWriter(Sha1Writer):
+	"""Remembers everything someone writes to it"""
+	__slots__ = ('buf', 'zip')
+	def __init__(self):
+		Sha1Writer.__init__(self)
+		self.buf = StringIO()
+		self.zip = zlib.compressobj(1)	# fastest
+	
+	def __getattr__(self, attr):
+		return getattr(self.buf, attr)
+	
+	def write(self, data):
+		alen = Sha1Writer.write(self, data)
+		self.buf.write(self.zip.compress(data))
+		return alen
+		
+	def close(self):
+		self.buf.write(self.zip.flush())
+
+
+#} END stream utilitiess
+

test/git/odb/test_channel.py

@@ -0,0 +1,61 @@
+"""Channel testing"""
+from test.testlib import *
+from git.odb.channel import *
+
+import time
+
+class TestDB(TestBase):
+	
+	def test_base(self):
+		# creating channel yields a write and a read channal
+		wc, rc = Channel()
+		assert isinstance(wc, WChannel)
+		assert isinstance(rc, RChannel)
+		
+		# everything else fails
+		self.failUnlessRaises(ValueError, Channel, 1, "too many args")
+		
+		# TEST UNLIMITED SIZE CHANNEL - writing+reading is FIFO
+		item = 1
+		item2 = 2
+		wc.write(item)
+		wc.write(item2)
+		assert rc.read() == item
+		assert rc.read() == item2
+		
+		# next read blocks, then raises - it waits a second
+		st = time.time()
+		self.failUnlessRaises(IOError, rc.read, True, 1)
+		assert time.time() - st >= 1.0
+		
+		# writing to a closed channel raises
+		assert not wc.closed
+		wc.close()
+		assert wc.closed
+		wc.close()	# fine
+		assert wc.closed
+		
+		self.failUnlessRaises(IOError, wc.write, 1)
+		
+		# reading from a closed channel never blocks
+		self.failUnlessRaises(IOError, rc.read)
+		
+		
+		
+		# TEST LIMITED SIZE CHANNEL
+		# channel with max-items set
+		wc, rc = Channel(1)
+		wc.write(item)			# fine
+		
+		# blocks for a second, its full
+		st = time.time()
+		self.failUnlessRaises(IOError, wc.write, item, True, 1)
+		assert time.time() - st >= 1.0
+		
+		# get one
+		assert rc.read() == item
+		
+		# its empty,can put one again
+		wc.write(item2)
+		assert rc.read() == item2
+		wc.close()

test/git/odb/test_db.py

@@ -0,0 +1,90 @@
+"""Test for object db"""
+from test.testlib import *
+from lib import ZippedStoreShaWriter
+
+from git.odb import *
+from git.odb.stream import Sha1Writer
+from git import Blob
+from git.errors import BadObject
+
+
+from cStringIO import StringIO
+import os
+		
+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 = (ZippedStoreShaWriter, 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()
+					assert isinstance(ostream, Sha1Writer)
+				# 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)
+					
+					# DIRECT STREAM COPY
+					# our data hase been written in object format to the StringIO
+					# we pasesd as output stream. No physical database representation
+					# was created.
+					# Test direct stream copy of object streams, the result must be 
+					# identical to what we fed in
+					ostream.seek(0)
+					istream.stream = ostream
+					assert istream.sha is not None
+					prev_sha = istream.sha
+					
+					db.set_ostream(ZippedStoreShaWriter())
+					db.store(istream)
+					assert istream.sha == prev_sha
+					new_ostream = db.ostream()
+					
+					# note: only works as long our store write uses the same compression
+					# level, which is zip
+					assert ostream.getvalue() == new_ostream.getvalue()
+			# 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)
+