From c69b6b979e3d6bd01ec40e75b92b21f7a391f0ca Mon Sep 17 00:00:00 2001 From: Sebastian Thiel Date: Sat, 5 Jun 2010 15:56:14 +0200 Subject: Added basic channel implementation including test restructured odb tests, they are now in an own module to keep the modules small --- lib/git/odb/channel.py | 106 +++++++++++++++ lib/git/odb/db.py | 11 +- test/git/odb/__init__.py | 1 + test/git/odb/lib.py | 60 +++++++++ test/git/odb/test_channel.py | 61 +++++++++ test/git/odb/test_db.py | 90 +++++++++++++ test/git/odb/test_stream.py | 172 ++++++++++++++++++++++++ test/git/odb/test_utils.py | 15 +++ test/git/test_odb.py | 307 ------------------------------------------- 9 files changed, 510 insertions(+), 313 deletions(-) create mode 100644 lib/git/odb/channel.py create mode 100644 test/git/odb/__init__.py create mode 100644 test/git/odb/lib.py create mode 100644 test/git/odb/test_channel.py create mode 100644 test/git/odb/test_db.py create mode 100644 test/git/odb/test_stream.py create mode 100644 test/git/odb/test_utils.py delete mode 100644 test/git/test_odb.py diff --git a/lib/git/odb/channel.py b/lib/git/odb/channel.py new file mode 100644 index 00000000..f6469d42 --- /dev/null +++ b/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 diff --git a/lib/git/odb/db.py b/lib/git/odb/db.py index d970b0bf..5d3cc6a3 100644 --- a/lib/git/odb/db.py +++ b/lib/git/odb/db.py @@ -142,11 +142,10 @@ class FileDBBase(object): """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 @@ class GitObjectDB(LooseObjectDB): 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) diff --git a/test/git/odb/__init__.py b/test/git/odb/__init__.py new file mode 100644 index 00000000..8b137891 --- /dev/null +++ b/test/git/odb/__init__.py @@ -0,0 +1 @@ + diff --git a/test/git/odb/lib.py b/test/git/odb/lib.py new file mode 100644 index 00000000..d5199748 --- /dev/null +++ b/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 + diff --git a/test/git/odb/test_channel.py b/test/git/odb/test_channel.py new file mode 100644 index 00000000..89b26582 --- /dev/null +++ b/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() diff --git a/test/git/odb/test_db.py b/test/git/odb/test_db.py new file mode 100644 index 00000000..35ba8680 --- /dev/null +++ b/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) + diff --git a/test/git/odb/test_stream.py b/test/git/odb/test_stream.py new file mode 100644 index 00000000..020fe6bd --- /dev/null +++ b/test/git/odb/test_stream.py @@ -0,0 +1,172 @@ +"""Test for object db""" +from test.testlib import * +from lib import ( + DummyStream, + DeriveTest, + Sha1Writer + ) + +from git.odb import * +from git import Blob +from cStringIO import StringIO +import tempfile +import os +import zlib + + + + +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 + + # derive with own args + DeriveTest(sha, Blob.type, s, stream, 'mine',myarg = 3)._assert() + + # test istream + istream = IStream(Blob.type, s, stream) + 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 + + assert istream.error is None + istream.error = Exception() + assert isinstance(istream.error, Exception) + + 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 + + diff --git a/test/git/odb/test_utils.py b/test/git/odb/test_utils.py new file mode 100644 index 00000000..34572b37 --- /dev/null +++ b/test/git/odb/test_utils.py @@ -0,0 +1,15 @@ +"""Test for object db""" +from test.testlib import * +from git import Blob +from git.odb.utils import ( + to_hex_sha, + to_bin_sha + ) + + +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 + diff --git a/test/git/test_odb.py b/test/git/test_odb.py deleted file mode 100644 index 5c8268cd..00000000 --- a/test/git/test_odb.py +++ /dev/null @@ -1,307 +0,0 @@ -"""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 - - -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 - - - -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 - - # derive with own args - DeriveTest(sha, Blob.type, s, stream, 'mine',myarg = 3)._assert() - - # test istream - istream = IStream(Blob.type, s, stream) - 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 - - assert istream.error is None - istream.error = Exception() - assert isinstance(istream.error, Exception) - - 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 = (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) - -- cgit v1.2.3