[Zope-CVS] CVS: Products/Ape/lib/apelib/tests - testscanner.py:1.5
teststorage.py:1.7 testzope2fs.py:1.7 zope2testbase.py:1.8
Shane Hathaway
shane at zope.com
Sat Feb 28 15:06:59 EST 2004
Update of /cvs-repository/Products/Ape/lib/apelib/tests
In directory cvs.zope.org:/tmp/cvs-serv30293/lib/apelib/tests
Modified Files:
testscanner.py teststorage.py testzope2fs.py zope2testbase.py
Log Message:
Merged ape-fs-oid-branch.
Ape now uses arbitrary OIDs on the filesystem, rather than using paths
as OIDs. This solves problems with moving and replacing objects and
further unifies SQL and filesystem databases.
=== Products/Ape/lib/apelib/tests/testscanner.py 1.4 => 1.5 ===
--- Products/Ape/lib/apelib/tests/testscanner.py:1.4 Tue Feb 17 00:25:12 2004
+++ Products/Ape/lib/apelib/tests/testscanner.py Sat Feb 28 15:06:28 2004
@@ -156,34 +156,18 @@
self.assertEqual(self.scanner.current[5], sources)
def testUseCommittedSources(self):
- # Verify the scanner sees sources according to transactions.
+ # Verify the scanner updates sources according to transactions.
repo = FakeRepository()
sources = {(repo, '999'): -1}
self.scanner.afterLoad(5, sources)
self.conn1.setOIDs([5])
sources_2 = {(repo, '999'): -2}
- self.scanner.afterStore(5, 123, sources_2)
- # Uncommitted data should have no effect on sources
- self.assertEqual(self.scanner.current[5], sources)
- self.scanner.afterCommit(123)
- # The scanner should have used data from the repo rather
- # than "-2".
+ self.scanner.afterCommit(5, sources_2)
final_sources = self.scanner.current[5]
self.assertEqual(len(final_sources), 1)
self.assertEqual(final_sources.keys()[0], (repo, '999'))
- self.assertEqual(final_sources.values()[0], 1001)
+ self.assertEqual(final_sources.values()[0], -2)
- def testAbort(self):
- # Verify the scanner ignores sources on transaction abort.
- repo = FakeRepository()
- sources = {(repo, '999'): -2}
- self.scanner.afterStore(5, 123, sources)
- self.assertEqual(self.scanner.uncommitted[123][5], sources)
- self.scanner.afterAbort(123)
- self.assert_(not self.scanner.uncommitted.has_key(123))
- self.assert_(not self.scanner.current.has_key(123))
- self.assert_(not self.scanner.future.has_key(123))
-
if __name__ == '__main__':
unittest.main()
=== Products/Ape/lib/apelib/tests/teststorage.py 1.6 => 1.7 ===
--- Products/Ape/lib/apelib/tests/teststorage.py:1.6 Mon Feb 2 10:07:22 2004
+++ Products/Ape/lib/apelib/tests/teststorage.py Sat Feb 28 15:06:28 2004
@@ -369,6 +369,19 @@
self.assertRaises(KeyError, self.storage.getPollSources, oid)
+ def testCleanChanged(self):
+ # Verify the storage discards the list of changed objects on
+ # commit or abort.
+ conn1 = self.db.open()
+ try:
+ ob1 = self._writeBasicObject(conn1)
+ self.assertEqual(len(self.storage.changed), 0)
+ ob1.strdata = 'def'
+ get_transaction().abort()
+ self.assertEqual(len(self.storage.changed), 0)
+ finally:
+ conn1.close()
+
+
if __name__ == '__main__':
unittest.main()
-
=== Products/Ape/lib/apelib/tests/testzope2fs.py 1.6 => 1.7 ===
--- Products/Ape/lib/apelib/tests/testzope2fs.py:1.6 Thu Feb 19 01:44:05 2004
+++ Products/Ape/lib/apelib/tests/testzope2fs.py Sat Feb 28 15:06:28 2004
@@ -50,33 +50,46 @@
class Zope2FSTests (unittest.TestCase, Zope2TestBase):
- def _createConnections(self, path):
- conn = FSConnection(path)
- return {'fs': conn}
+ annotation_prefix = '.'
def setUp(self):
+ self.db, self.conn = self.openDatabase()
+ self.conf = conf
+ self.path = tmpdir
+ c = self.db.open()
+ try:
+ if not c.root().has_key('Application'):
+ from OFS.Application import Application
+ c.root()['Application'] = Application()
+ get_transaction().commit()
+ finally:
+ c.close()
+ get_transaction().begin()
+ self.clearCaches()
+
+ def tearDown(self):
+ get_transaction().abort()
+ if self.db is not None:
+ self.db.close()
+ rmtree(self.path)
+
+ def openDatabase(self):
global conf
if conf is None:
conf = loadConf('filesystem')
if not os.path.exists(tmpdir):
os.mkdir(tmpdir)
- self.path = tmpdir
- self.conf = conf
- conns = self._createConnections(tmpdir)
- assert len(conns) == 1
- self.conn = conns['fs']
- resource = StaticResource(self.conf)
+ conn = FSConnection(tmpdir, annotation_prefix=self.annotation_prefix)
+ conns = {'fs': conn}
+ resource = StaticResource(conf)
storage = ApeStorage(resource, conns)
- self.storage = storage
- db = ApeDB(storage, resource)
- self.db = db
- get_transaction().begin()
- self.conn.afs.clearCache()
+ db = ApeDB(storage, resource, cache_size=0)
+ return db, conn
- def tearDown(self):
- get_transaction().abort()
- self.db.close()
- rmtree(self.path)
+ def clearCaches(self):
+ """Clears caches after a filesystem write.
+ """
+ self.conn.afs.clearCache()
def testClassificationPreservation(self):
# Ensure that classification doesn't get forgotten.
@@ -105,10 +118,10 @@
conn.close()
- def testMismatchedIdDetection(self):
- # FSAutoID should detect when the OID and ID don't match.
- # Normally, the only time they don't match is when an object
- # has been moved.
+ def testIgnoreMismatchedId(self):
+ # Verify that FSAutoID doesn't care if the ID of an item
+ # doesn't match what the folder thinks the item's ID should
+ # be.
conn = self.db.open()
try:
app = conn.root()['Application']
@@ -119,13 +132,13 @@
ob = app.Holidays
ob._setId('HolidayCalendar')
- self.assertRaises(ValueError, get_transaction().commit)
+ get_transaction().commit()
finally:
conn.close()
- def testReuseId(self):
- # Verifies that ApeConnection doesn't trip over reusing an OID that's
+ def testReusePath(self):
+ # Verifies that ApeConnection doesn't trip over reuse of a path that's
# no longer in use.
conn = self.db.open()
try:
@@ -156,7 +169,7 @@
app._setObject(template.id, template, set_owner=0)
get_transaction().commit()
- dir = self.conn.getPath('/')
+ dir = self.conn.basepath
names = os.listdir(dir)
self.assert_('template.html' in names, names)
self.assert_('template' not in names, names)
@@ -188,7 +201,7 @@
self.assert_(not hasattr(f, 'script%d.py' % n))
# white box test: verify the scripts were actually stored
# with .py extensions.
- dir = self.conn.getPath('/folder')
+ dir = os.path.join(self.conn.basepath, 'folder')
names = os.listdir(dir)
for n in range(3):
self.assert_(('script%d.py' % n) in names, names)
@@ -222,7 +235,7 @@
self.assert_(not hasattr(f, 'script%d' % n))
# white box test: verify the scripts were actually stored
# with .py extensions.
- dir = self.conn.getPath('/folder')
+ dir = os.path.join(self.conn.basepath, 'folder')
names = os.listdir(dir)
for n in range(3):
self.assert_(('script%d.py' % n) in names, names)
@@ -232,9 +245,11 @@
conn.close()
- def testNameExtensionConflictDetection(self):
- # Verifies that conflicting names resulting from automatic extensions
- # don't go unnoticed.
+ def testAutoRenameOnExtensionConflict(self):
+ # When you create a Python Script called "script0", Ape adds a
+ # .py extension. If, in a second transaction, you add
+ # "script0.py", Ape must rename the current "script0.py" to
+ # "script0" to make room for the new "script0.py".
conn = self.db.open()
try:
app = conn.root()['Application']
@@ -248,16 +263,24 @@
f._setObject(script.id, script, set_owner=0)
get_transaction().commit()
- dir = self.conn.getPath('/folder')
+ dir = os.path.join(self.conn.basepath, 'folder')
names = os.listdir(dir)
self.assert_(('script0.py') in names, names)
self.assert_(('script0') not in names, names)
- # script0.py already exists, so the transaction will fail.
+ # script0.py already exists, so Ape should automatically rename.
script = PythonScript('script0.py')
script.write('##title=test script\nreturn "Hello, world!"')
f._setObject(script.id, script, set_owner=0)
- self.assertRaises(OIDConflictError, get_transaction().commit)
+ get_transaction().commit()
+
+ # Did it write them correctly?
+ text = open(os.path.join(dir, 'script0')).read()
+ self.assert_(text.find('OK') > 0, text)
+ self.assert_(text.find('Hello, world!') < 0, text)
+ text = open(os.path.join(dir, 'script0.py')).read()
+ self.assert_(text.find('OK') < 0, text)
+ self.assert_(text.find('Hello, world!') > 0, text)
finally:
conn.close()
@@ -282,7 +305,7 @@
f._setObject(script.id, script, set_owner=0)
get_transaction().commit()
- dir = self.conn.getPath('/folder')
+ dir = os.path.join(self.conn.basepath, 'folder')
names = os.listdir(dir)
self.assert_(('script0.py') in names, names)
self.assert_(('script0') in names, names)
@@ -356,7 +379,7 @@
f._setObject(script.id, script, set_owner=0)
get_transaction().commit()
- dir = self.conn.getPath('/folder')
+ dir = os.path.join(self.conn.basepath, 'folder')
names = os.listdir(dir)
self.assert_(('script0.py') in names, names)
self.assert_(('script0.dtml') in names, names)
@@ -427,7 +450,7 @@
finally:
conn2.close()
- dir = self.conn.getPath('/')
+ dir = self.conn.basepath
names = os.listdir(dir)
self.assert_(('image.png') in names, names)
self.assert_(('image') not in names, names)
@@ -450,7 +473,7 @@
content_type='application/octet-stream')
get_transaction().commit()
- dir = self.conn.getPath('/')
+ dir = self.conn.basepath
names = os.listdir(dir)
self.assert_(('hello.txt') in names, names)
self.assert_(('world.dat') in names, names)
@@ -464,10 +487,11 @@
# Verify Zope chooses the right object type for
# a new object.
# White box test.
- dir = self.conn.getPath('/')
+ dir = self.conn.basepath
f = open(os.path.join(dir, 'test.py'), 'wt')
f.write('return "Ok!"')
f.close()
+ self.clearCaches()
conn = self.db.open()
try:
app = conn.root()['Application']
@@ -482,7 +506,7 @@
# Verify that even though the extension gets stripped off
# in Zope, Zope still sees the object as it should.
# White box test.
- dir = self.conn.getPath('/')
+ dir = self.conn.basepath
f = open(os.path.join(dir, 'test.py'), 'wt')
f.write('return "Ok!"')
f.close()
@@ -491,6 +515,7 @@
+ 'properties'), 'wt')
f.write('[object_names]\ntest\n')
f.close()
+ self.clearCaches()
conn = self.db.open()
try:
app = conn.root()['Application']
@@ -505,10 +530,11 @@
# Verify Zope uses a File object for unrecognized files on
# the filesystem. White box test.
data = 'data goes here'
- dir = self.conn.getPath('/')
+ dir = self.conn.basepath
f = open(os.path.join(dir, 'test'), 'wt')
f.write(data)
f.close()
+ self.clearCaches()
conn = self.db.open()
try:
app = conn.root()['Application']
@@ -522,8 +548,9 @@
def testDefaultPropertySchema(self):
# Verify Zope uses the default property schema when no properties
# are set.
- dir = self.conn.getPath('/')
+ dir = self.conn.basepath
os.mkdir(os.path.join(dir, 'test'))
+ self.clearCaches()
conn = self.db.open()
try:
app = conn.root()['Application']
@@ -558,7 +585,7 @@
conn2.close()
# Verify the stowaway is in the properties file.
- dir = self.conn.getPath('/')
+ dir = self.conn.basepath
p = os.path.join(
dir, self.conn.afs.annotation_prefix + 'properties')
f = open(p, 'rt')
@@ -589,7 +616,7 @@
def testGuessFileContentType(self):
# Verify that file content type guessing happens.
data = '<html><body>Cool stuff</body></html>'
- dir = self.conn.getPath('/')
+ dir = self.conn.basepath
f = open(os.path.join(dir, 'testobj'), 'wt')
f.write(data)
f.close()
@@ -606,8 +633,7 @@
def testWriteToRoot(self):
# Verify it's possible to write to the _root object as well as
# the Application object without either one stomping on each
- # other's data. This is a functional test of
- # apelib.fs.structure.RootDirectoryItems.
+ # other's data.
conn = self.db.open()
conn2 = None
try:
@@ -630,11 +656,69 @@
conn2.close()
+ def testOpenExisting(self):
+ # Verifies that opening an existing database finds the same
+ # data.
+ conn = self.db.open()
+ try:
+ app = conn.root()['Application']
+ app.test_attribute = '123'
+ get_transaction().commit()
+ finally:
+ conn.close()
+
+ # Close the database and open a new one pointing at the same
+ # directory.
+ self.db.close()
+ self.db = None
+ self.db, self.conn = self.openDatabase()
+ conn = self.db.open()
+ try:
+ root = conn.root()
+ app = root['Application']
+ self.assertEqual(app.test_attribute, '123')
+ finally:
+ conn.close()
+
+
+ def testNoClobberOnOpen(self):
+ # Opening a database with no "_root" shouldn't clobber the
+ # existing contents.
+ conn = self.db.open()
+ try:
+ app = conn.root()['Application']
+ f = Folder()
+ f.id = 'bar'
+ app._setObject(f.id, f)
+ get_transaction().commit()
+ finally:
+ conn.close()
+ self.db.close()
+ self.db = None
+
+ # Destroy the _root and the annotations at the app root.
+ basepath = self.conn.basepath
+ root_p = os.path.join(basepath, '_root')
+ if os.path.exists(root_p):
+ rmtree(root_p)
+ paths = self.conn.afs.getAnnotationPaths(basepath)
+ for path in paths:
+ if os.path.exists(path):
+ os.remove(path)
+
+ # Now look for the 'bar' folder.
+ self.db, self.conn = self.openDatabase()
+ conn = self.db.open()
+ try:
+ root = conn.root()
+ app = root['Application']
+ self.assertEqual(app.bar.id, 'bar')
+ finally:
+ conn.close()
+
+
class Zope2FSUnderscoreTests (Zope2FSTests):
-
- def _createConnections(self, path):
- conn = FSConnection(path, annotation_prefix='_')
- return {'fs': conn}
+ annotation_prefix = '_'
if __name__ == '__main__':
=== Products/Ape/lib/apelib/tests/zope2testbase.py 1.7 => 1.8 ===
--- Products/Ape/lib/apelib/tests/zope2testbase.py:1.7 Tue Feb 17 00:25:12 2004
+++ Products/Ape/lib/apelib/tests/zope2testbase.py Sat Feb 28 15:06:28 2004
@@ -33,7 +33,6 @@
from Products.ZSQLMethods.SQL import manage_addZSQLMethod
from Products.PageTemplates.ZopePageTemplate import ZopePageTemplate
-from apelib.zope2.setup.patches import applySetObPatch
from apelib.core.interfaces import OIDConflictError
@@ -377,8 +376,6 @@
def testRename(self):
- applySetObPatch() # Required for this test to work with Zope2FS
-
conn = self.db.open()
try:
app = conn.root()['Application']
@@ -636,5 +633,28 @@
get_transaction().commit()
self.assert_(app._p_mtime > now - 10)
self.assert_(app._p_mtime < now + 10)
+ finally:
+ conn.close()
+
+
+ def testWriteWithGhosts(self):
+ # It should be possible to write a container even if one
+ # or more of its subobjects are ghosts.
+ conn = self.db.open()
+ try:
+ root = conn.root()
+ root['foo'] = 1
+ f = Folder()
+ f.id = 'bar'
+ root['bar'] = f
+ get_transaction().commit()
+ conn2 = self.db.open()
+ try:
+ root2 = conn2.root()
+ root2['foo'] = 2
+ self.assertEqual(root2['bar']._p_changed, None)
+ get_transaction().commit()
+ finally:
+ conn2.close()
finally:
conn.close()
More information about the Zope-CVS
mailing list