[Zope-CVS] CVS: Packages/pypes/pypes/tests - common.py:1.2
runtests.py:1.2 test_identity.py:1.2 test_services.py:1.2
Casey Duncan
casey at zope.com
Tue Aug 5 01:27:59 EDT 2003
Update of /cvs-repository/Packages/pypes/pypes/tests
In directory cvs.zope.org:/tmp/cvs-serv18617/tests
Modified Files:
common.py runtests.py test_identity.py test_services.py
Log Message:
Add real persistence and concurrency tests
=== Packages/pypes/pypes/tests/common.py 1.1.1.1 => 1.2 ===
--- Packages/pypes/pypes/tests/common.py:1.1.1.1 Mon Aug 4 00:46:04 2003
+++ Packages/pypes/pypes/tests/common.py Tue Aug 5 00:27:51 2003
@@ -15,6 +15,8 @@
$Id$"""
+import os
+import ZODB
from Persistence import Persistent
from unittest import TestCase
@@ -59,7 +61,32 @@
self.identity.register(obj)
return obj
-
+class PersistenceTest(TestCase):
+ """Basic test case for testing persistence"""
+
+ db = None
+
+ def setUp(self):
+ from ZODB.FileStorage import FileStorage
+ from ZODB.DB import DB
+ n = 'fs_tmp__%s' % os.getpid()
+ self.storage = FileStorage(n)
+ self.db = DB(self.storage)
+
+ def tearDown(self):
+ if self.db is not None:
+ self.db.close()
+ self.storage.cleanup()
+class PypesPersistentTest(PypesTestCase, PersistenceTest):
+ """Persistence test with standard pypes services available"""
+ def setUp(self):
+ import pypes
+ PersistenceTest.setUp(self)
+ self.conn = self.db.open()
+ self.root = self.conn.root()
+ pypes.initialize(self.conn)
+ self.identity = pypes.services.identity(self.conn)
+ get_transaction().commit()
=== Packages/pypes/pypes/tests/runtests.py 1.1.1.1 => 1.2 ===
--- Packages/pypes/pypes/tests/runtests.py:1.1.1.1 Mon Aug 4 00:46:06 2003
+++ Packages/pypes/pypes/tests/runtests.py Tue Aug 5 00:27:51 2003
@@ -5,11 +5,10 @@
$Id$"""
import unittest
-from test_services \
- import TestInitServices, TestServiceConfig, TestServiceAccessors
-from test_identity import TestIdentityService, TestIdentitySet
-from test_event import TestEventService
-from test_graph import TestDirectedGraph, TestDirectedIdGraph
+from test_services import *
+from test_identity import *
+from test_event import *
+from test_graph import *
if __name__ == '__main__':
unittest.main()
=== Packages/pypes/pypes/tests/test_identity.py 1.1.1.1 => 1.2 ===
--- Packages/pypes/pypes/tests/test_identity.py:1.1.1.1 Mon Aug 4 00:46:05 2003
+++ Packages/pypes/pypes/tests/test_identity.py Tue Aug 5 00:27:51 2003
@@ -15,13 +15,16 @@
$Id$"""
+import ZODB
+from ZODB.POSException import ConflictError
import unittest
from random import randint
from zope.interface.verify import verifyObject
from Persistence import Persistent
from BTrees.OOBTree import OOTreeSet, difference
from types import TypeType
-from common import PypesTestCase, TestClass
+from common \
+ import PypesTestCase, TestClass, PersistenceTest, PypesPersistentTest
class TestIdentityService(unittest.TestCase):
@@ -198,7 +201,42 @@
for i in xrange(100):
ids.insert(register(TestClass()))
self.failIf(difference(OOTreeSet(self.identity.idSet()), ids))
-
+
+class TestIdentityServicePersistence(PersistenceTest):
+
+ def testBasicPersistence(self):
+ from pypes.identity import IdentityService, pypesid
+ r1 = self.db.open().root()
+ ids = r1['t'] = IdentityService()
+ obj = TestClass()
+ obj.name = 'test'
+ objid = ids.register(obj)
+ get_transaction().commit()
+
+ r2 = self.db.open().root()
+ ids2 = r2['t']
+ obj = ids2.getObject(objid)
+ self.assertEqual(ids._p_serial, ids._p_serial)
+ self.failUnless(isinstance(obj, TestClass))
+ self.assertEqual(obj.name, 'test')
+ self.assertEqual(pypesid(obj), objid)
+
+ def testConcurrentRegistration(self):
+ from pypes.identity import IdentityService
+ r1 = self.db.open().root()
+ ids1 = r1['t'] = IdentityService()
+ get_transaction().commit()
+
+ r2 = self.db.open().root()
+ ids2 = r2['t']
+ list(ids2); list(ids2.iterIds()) # Make sure its loaded
+
+ ids1.register(TestClass())
+ get_transaction().commit()
+
+ ids2.register(TestClass())
+ get_transaction().commit()
+
class TestIdentitySet(PypesTestCase):
def setUp(self):
@@ -449,6 +487,75 @@
self.failUnless(set1.issuperset(set2))
self.failUnless(set2.issuperset(set1))
self.assertEqual(set1, set2)
+
+class TestIdentitySetPersistence(PypesPersistentTest):
+
+ def testSetPersistence(self):
+ from pypes.identity import IdentitySet
+ self.root['objs'] = [self._newObj() for i in range(10)]
+ set = IdentitySet(self.root['objs'])
+ self.root['set'] = set
+ get_transaction().commit()
+
+ r2 = self.db.open().root()
+ for obj in r2['objs']:
+ self.failUnless(obj in r2['set'])
+
+ def testConcurrentAdd(self):
+ from pypes.identity import IdentitySet
+
+ set1 = self.root['set'] = IdentitySet()
+ get_transaction().commit()
+
+ r2 = self.db.open().root()
+ set2 = r2['set']
+ list(set2) # Make sure its loaded
+
+ set1.add(self._newObj())
+ get_transaction().commit()
+
+ set2.add(self._newObj())
+ get_transaction().commit()
+
+ set3 = self.db.open().root()['set']
+ self.assertEqual(len(set3), 2)
+
+ def testDuplicateAddFails(self):
+ from pypes.identity import IdentitySet
+ set1 = self.root['set'] = IdentitySet()
+ get_transaction().commit()
+
+ r2 = self.db.open().root()
+ set2 = r2['set']
+ list(set2) # Make sure its loaded
+
+ obj = self._newObj()
+ set1.add(obj)
+ get_transaction().commit()
+
+ set2.add(obj)
+ self.assertRaises(ConflictError, get_transaction().commit)
+
+ set3 = self.db.open().root()['set']
+ self.assertEqual(len(set3), 1)
+
+ def testConcurrentUpdate(self):
+ from pypes.identity import IdentitySet
+ set1 = self.root['set'] = IdentitySet()
+ get_transaction().commit()
+
+ r2 = self.db.open().root()
+ set2 = r2['set']
+ list(set2) # Make sure its loaded
+
+ set1.update([self._newObj() for i in range(10)])
+ get_transaction().commit()
+
+ set2.update([self._newObj() for i in range(10)])
+ get_transaction().commit()
+
+ set3 = self.db.open().root()['set']
+ self.assertEqual(len(set3), 20)
if __name__ == '__main__':
unittest.main()
=== Packages/pypes/pypes/tests/test_services.py 1.1.1.1 => 1.2 ===
--- Packages/pypes/pypes/tests/test_services.py:1.1.1.1 Mon Aug 4 00:46:06 2003
+++ Packages/pypes/pypes/tests/test_services.py Tue Aug 5 00:27:51 2003
@@ -14,7 +14,8 @@
"""Service registration and accessor tests"""
import unittest
-from common import TestConnection, PypesTestCase
+import ZODB
+from common import TestConnection, PypesTestCase, PersistenceTest
from zope.interface.verify import verifyObject
from Persistence import Persistent
@@ -173,6 +174,21 @@
def testIdentity(self):
from pypes import services
self.failUnless(services.identity(self.conn) is self.identity)
+
+class TestInitializeServices(PersistenceTest):
+ def testInitDefaultServices(self):
+ import pypes
+ from pypes import services, identity, event
+ c1 = self.db.open()
+ pypes.initialize(c1)
+ get_transaction().commit()
+
+ c2 = self.db.open()
+ svcs = services.all(c2)
+ self.assertEqual(len(svcs), 2)
+ self.failUnless(isinstance(svcs['identity'], identity.IdentityService))
+ self.failUnless(isinstance(svcs['event'], event.EventService))
+
if __name__ == '__main__':
unittest.main()
More information about the Zope-CVS
mailing list