import os import shutil import tempfile from darc.helpers import Location from darc.remote import RemoteRepository from darc.repository import Repository from darc.testsuite import DarcTestCase class RepositoryTestCase(DarcTestCase): def open(self, create=False): return Repository(os.path.join(self.tmppath, 'repository'), create=create) def setUp(self): self.tmppath = tempfile.mkdtemp() self.repository = self.open(create=True) def tearDown(self): self.repository.close() shutil.rmtree(self.tmppath) def test1(self): for x in range(100): self.repository.put(('%-32d' % x).encode('ascii'), b'SOMEDATA') key50 = ('%-32d' % 50).encode('ascii') self.assert_equal(self.repository.get(key50), b'SOMEDATA') self.repository.delete(key50) self.assert_raises(Repository.DoesNotExist, lambda: self.repository.get(key50)) self.repository.commit() self.repository.close() repository2 = self.open() self.assert_raises(Repository.DoesNotExist, lambda: repository2.get(key50)) for x in range(100): if x == 50: continue self.assert_equal(repository2.get(('%-32d' % x).encode('ascii')), b'SOMEDATA') repository2.close() def test2(self): """Test multiple sequential transactions """ self.repository.put(b'00000000000000000000000000000000', b'foo') self.repository.put(b'00000000000000000000000000000001', b'foo') self.repository.commit() self.repository.delete(b'00000000000000000000000000000000') self.repository.put(b'00000000000000000000000000000001', b'bar') self.repository.commit() self.assert_equal(self.repository.get(b'00000000000000000000000000000001'), b'bar') def test_consistency(self): """Test cache consistency """ self.repository.put(b'00000000000000000000000000000000', b'foo') self.assert_equal(self.repository.get(b'00000000000000000000000000000000'), b'foo') self.repository.put(b'00000000000000000000000000000000', b'foo2') self.assert_equal(self.repository.get(b'00000000000000000000000000000000'), b'foo2') self.repository.put(b'00000000000000000000000000000000', b'bar') self.assert_equal(self.repository.get(b'00000000000000000000000000000000'), b'bar') self.repository.delete(b'00000000000000000000000000000000') self.assert_raises(Repository.DoesNotExist, lambda: self.repository.get(b'00000000000000000000000000000000')) def test_consistency2(self): """Test cache consistency2 """ self.repository.put(b'00000000000000000000000000000000', b'foo') self.assert_equal(self.repository.get(b'00000000000000000000000000000000'), b'foo') self.repository.commit() self.repository.put(b'00000000000000000000000000000000', b'foo2') self.assert_equal(self.repository.get(b'00000000000000000000000000000000'), b'foo2') self.repository.rollback() self.assert_equal(self.repository.get(b'00000000000000000000000000000000'), b'foo') def test_single_kind_transactions(self): # put self.repository.put(b'00000000000000000000000000000000', b'foo') self.repository.commit() self.repository.close() # replace self.repository = self.open() self.repository.put(b'00000000000000000000000000000000', b'bar') self.repository.commit() self.repository.close() # delete self.repository = self.open() self.repository.delete(b'00000000000000000000000000000000') self.repository.commit() class RemoteRepositoryTestCase(RepositoryTestCase): def open(self, create=False): return RemoteRepository(Location('localhost:' + os.path.join(self.tmppath, 'repository')), create=create)