[Checkins] SVN: persistent/trunk/ Coverage for p.wref.WeakRef.
Tres Seaver
cvs-admin at zope.org
Thu Jun 28 22:50:22 UTC 2012
Log message for revision 127164:
Coverage for p.wref.WeakRef.
Changed:
_U persistent/trunk/
U persistent/trunk/persistent/tests/test_wref.py
U persistent/trunk/persistent/wref.py
U persistent/trunk/setup.cfg
-=-
Modified: persistent/trunk/persistent/tests/test_wref.py
===================================================================
--- persistent/trunk/persistent/tests/test_wref.py 2012-06-28 22:50:14 UTC (rev 127163)
+++ persistent/trunk/persistent/tests/test_wref.py 2012-06-28 22:50:18 UTC (rev 127164)
@@ -13,12 +13,137 @@
##############################################################################
import unittest
+class WeakRefTests(unittest.TestCase):
+
+ def _getTargetClass(self):
+ from persistent.wref import WeakRef
+ return WeakRef
+
+ def _makeTarget(self, **kw):
+ from persistent import Persistent
+ class Derived(Persistent):
+ def __eq__(self, other):
+ return self._p_oid == other._p_oid
+ derived = Derived()
+ for k, v in kw.items():
+ setattr(derived, k, v)
+ derived._p_oid = 'OID'
+ return derived
+
+ def _makeJar(self):
+ class _DB(object):
+ database_name = 'testing'
+ class _Jar(dict):
+ db = lambda self: _DB()
+ return _Jar()
+
+ def _makeOne(self, ob):
+ return self._getTargetClass()(ob)
+
+ def test_ctor_target_wo_jar(self):
+ target = self._makeTarget()
+ wref = self._makeOne(target)
+ self.assertTrue(wref._v_ob is target)
+ self.assertEqual(wref.oid, 'OID')
+ self.assertTrue(wref.dm is None)
+ self.assertFalse('database_name' in wref.__dict__)
+
+ def test_ctor_target_w_jar(self):
+ target = self._makeTarget()
+ target._p_jar = jar = self._makeJar()
+ wref = self._makeOne(target)
+ self.assertTrue(wref._v_ob is target)
+ self.assertEqual(wref.oid, 'OID')
+ self.assertTrue(wref.dm is jar)
+ self.assertEqual(wref.database_name, 'testing')
+
+ def test___call___target_in_volatile(self):
+ target = self._makeTarget()
+ target._p_jar = jar = self._makeJar()
+ wref = self._makeOne(target)
+ self.assertTrue(wref() is target)
+
+ def test___call___target_in_jar(self):
+ target = self._makeTarget()
+ target._p_jar = jar = self._makeJar()
+ jar[target._p_oid] = target
+ wref = self._makeOne(target)
+ del wref._v_ob
+ self.assertTrue(wref() is target)
+
+ def test___call___target_not_in_jar(self):
+ target = self._makeTarget()
+ target._p_jar = jar = self._makeJar()
+ wref = self._makeOne(target)
+ del wref._v_ob
+ self.assertTrue(wref() is None)
+
+ def test___hash___w_target(self):
+ target = self._makeTarget()
+ target._p_jar = jar = self._makeJar()
+ wref = self._makeOne(target)
+ self.assertEqual(hash(wref), hash(target))
+
+ def test___hash___wo_target(self):
+ target = self._makeTarget()
+ target._p_jar = jar = self._makeJar()
+ wref = self._makeOne(target)
+ del wref._v_ob
+ self.assertRaises(TypeError, hash, wref)
+
+ def test___eq___w_both_same_target(self):
+ target = self._makeTarget()
+ lhs = self._makeOne(target)
+ rhs_target = self._makeTarget()
+ rhs = self._makeOne(target)
+ self.assertEqual(lhs, rhs)
+
+ def test___eq___w_both_different_targets(self):
+ lhs_target = self._makeTarget()
+ lhs_target._p_oid = 'LHS'
+ lhs = self._makeOne(lhs_target)
+ rhs_target = self._makeTarget()
+ rhs_target._p_oid = 'RHS'
+ rhs = self._makeOne(rhs_target)
+ self.assertNotEqual(lhs, rhs)
+
+ def test___eq___w_lhs_gone_target_not_in_jar(self):
+ target = self._makeTarget()
+ target._p_jar = jar = self._makeJar()
+ lhs = self._makeOne(target)
+ del lhs._v_ob
+ rhs = self._makeOne(target)
+ self.assertRaises(TypeError, lambda: lhs == rhs)
+
+ def test___eq___w_lhs_gone_target_in_jar(self):
+ target = self._makeTarget()
+ target._p_jar = jar = self._makeJar()
+ jar[target._p_oid] = target
+ lhs = self._makeOne(target)
+ del lhs._v_ob
+ rhs_target = self._makeTarget()
+ rhs = self._makeOne(target)
+ self.assertEqual(lhs, rhs)
+
+ def test___eq___w_rhs_gone_target_not_in_jar(self):
+ target = self._makeTarget()
+ target._p_jar = jar = self._makeJar()
+ lhs = self._makeOne(target)
+ rhs = self._makeOne(target)
+ del rhs._v_ob
+ self.assertRaises(TypeError, lambda: lhs == rhs)
+
+ def test___eq___w_rhs_gone_target_in_jar(self):
+ target = self._makeTarget()
+ target._p_jar = jar = self._makeJar()
+ jar[target._p_oid] = target
+ lhs = self._makeOne(target)
+ rhs = self._makeOne(target)
+ del rhs._v_ob
+ self.assertEqual(lhs, rhs)
+
+
def test_suite():
- from doctest import DocTestSuite
- try:
- import transaction
- import ZODB
- except ImportError:
- return unittest.TestSuite()
- else:
- return DocTestSuite('persistent.wref')
+ return unittest.TestSuite((
+ unittest.makeSuite(WeakRefTests),
+ ))
Modified: persistent/trunk/persistent/wref.py
===================================================================
--- persistent/trunk/persistent/wref.py 2012-06-28 22:50:14 UTC (rev 127163)
+++ persistent/trunk/persistent/wref.py 2012-06-28 22:50:18 UTC (rev 127164)
@@ -14,8 +14,6 @@
"""ZODB-based persistent weakrefs
"""
-__docformat__ = "reStructuredText"
-
from persistent import Persistent
WeakRefMarker = object()
@@ -26,135 +24,7 @@
Persistent weak references are used much like Python weak
references. The major difference is that you can't specify an
object to be called when the object is removed from the database.
-
- Here's an example. We'll start by creating a persistent object and
- a reference to it:
-
- >>> import transaction
- >>> import persistent, ZODB.tests.MinPO
- >>> import ZODB.tests.util
- >>> ob = ZODB.tests.MinPO.MinPO()
- >>> ref = WeakRef(ob)
- >>> ref() is ob
- True
-
- The hash of the ref if the same as the hash of the referenced object:
-
- >>> hash(ref) == hash(ob)
- True
-
- Two refs to the same object are equal:
-
- >>> WeakRef(ob) == ref
- True
-
- >>> ob2 = ZODB.tests.MinPO.MinPO(1)
- >>> WeakRef(ob2) == ref
- False
-
- Lets save the reference and the referenced object in a database:
-
- >>> db = ZODB.tests.util.DB()
-
- >>> conn1 = db.open()
- >>> conn1.root()['ob'] = ob
- >>> conn1.root()['ref'] = ref
- >>> transaction.commit()
-
- If we open a new connection, we can use the reference:
-
- >>> conn2 = db.open()
- >>> conn2.root()['ref']() is conn2.root()['ob']
- True
- >>> hash(conn2.root()['ref']) == hash(conn2.root()['ob'])
- True
-
- But if we delete the referenced object and pack:
-
- >>> del conn2.root()['ob']
- >>> transaction.commit()
- >>> ZODB.tests.util.pack(db)
-
- And then look in a new connection:
-
- >>> conn3 = db.open()
- >>> conn3.root()['ob']
- Traceback (most recent call last):
- ...
- KeyError: 'ob'
-
- Trying to dereference the reference returns None:
-
- >>> conn3.root()['ref']()
-
- Trying to get a hash, raises a type error:
-
- >>> hash(conn3.root()['ref'])
- Traceback (most recent call last):
- ...
- TypeError: Weakly-referenced object has gone away
-
- Always explicitly close databases: :)
-
- >>> db.close()
- >>> del ob, ref, db, conn1, conn2, conn3
-
- When multiple databases are in use, a weakref in one database may
- point to an object in a different database. Let's create two new
- databases to demonstrate this.
-
- >>> dbA = ZODB.tests.util.DB(
- ... database_name = 'dbA',
- ... )
- >>> dbB = ZODB.tests.util.DB(
- ... database_name = 'dbB',
- ... databases = dbA.databases,
- ... )
- >>> connA1 = dbA.open()
- >>> connB1 = connA1.get_connection('dbB')
-
- Now create and add a new object and a weak reference, and add them
- to different databases.
-
- >>> ob = ZODB.tests.MinPO.MinPO()
- >>> ref = WeakRef(ob)
- >>> connA1.root()['ob'] = ob
- >>> connA1.add(ob)
- >>> connB1.root()['ref'] = ref
- >>> transaction.commit()
-
- After a succesful commit, the reference should know the oid,
- database name and connection of the object.
-
- >>> ref.oid == ob._p_oid
- True
- >>> ref.database_name == 'dbA'
- True
- >>> ref.dm is ob._p_jar is connA1
- True
-
- If we open new connections, we should be able to use the reference.
-
- >>> connA2 = dbA.open()
- >>> connB2 = connA2.get_connection('dbB')
- >>> ref2 = connB2.root()['ref']
- >>> ob2 = connA2.root()['ob']
- >>> ref2() is ob2
- True
- >>> ref2.oid == ob2._p_oid
- True
- >>> ref2.database_name == 'dbA'
- True
- >>> ref2.dm is ob2._p_jar is connA2
- True
-
- Always explicitly close databases: :)
-
- >>> dbA.close()
- >>> dbB.close()
-
"""
-
# We set _p_oid to a marker so that the serialization system can
# provide special handling of weakrefs.
_p_oid = WeakRefMarker
@@ -197,101 +67,7 @@
"""Persistent weak key dictionary
This is akin to WeakKeyDictionaries. Note, however, that removal
- of items is extremely lazy. See below.
-
- We'll start by creating a PersistentWeakKeyDictionary and adding
- some persistent objects to it.
-
- >>> import transaction
- >>> d = PersistentWeakKeyDictionary()
- >>> import ZODB.tests.util
- >>> p1 = ZODB.tests.util.P('p1')
- >>> p2 = ZODB.tests.util.P('p2')
- >>> p3 = ZODB.tests.util.P('p3')
- >>> d[p1] = 1
- >>> d[p2] = 2
- >>> d[p3] = 3
-
- We'll create an extra persistent object that's not in the dict:
-
- >>> p4 = ZODB.tests.util.P('p4')
-
- Now we'll excercise iteration and item access:
-
- >>> l = [(str(k), d[k], d.get(k)) for k in d]
- >>> l.sort()
- >>> l
- [('P(p1)', 1, 1), ('P(p2)', 2, 2), ('P(p3)', 3, 3)]
-
- And the containment operator:
-
- >>> [p in d for p in [p1, p2, p3, p4]]
- [True, True, True, False]
-
- We can add the dict and the referenced objects to a database:
-
- >>> db = ZODB.tests.util.DB()
-
- >>> conn1 = db.open()
- >>> conn1.root()['p1'] = p1
- >>> conn1.root()['d'] = d
- >>> conn1.root()['p2'] = p2
- >>> conn1.root()['p3'] = p3
- >>> transaction.commit()
-
- And things still work, as before:
-
- >>> l = [(str(k), d[k], d.get(k)) for k in d]
- >>> l.sort()
- >>> l
- [('P(p1)', 1, 1), ('P(p2)', 2, 2), ('P(p3)', 3, 3)]
- >>> [p in d for p in [p1, p2, p3, p4]]
- [True, True, True, False]
-
- Likewise, we can read the objects from another connection and
- things still work.
-
- >>> conn2 = db.open()
- >>> d = conn2.root()['d']
- >>> p1 = conn2.root()['p1']
- >>> p2 = conn2.root()['p2']
- >>> p3 = conn2.root()['p3']
- >>> l = [(str(k), d[k], d.get(k)) for k in d]
- >>> l.sort()
- >>> l
- [('P(p1)', 1, 1), ('P(p2)', 2, 2), ('P(p3)', 3, 3)]
- >>> [p in d for p in [p1, p2, p3, p4]]
- [True, True, True, False]
-
- Now, we'll delete one of the objects from the database, but *not*
- from the dictionary:
-
- >>> del conn2.root()['p2']
- >>> transaction.commit()
-
- And pack the database, so that the no-longer referenced p2 is
- actually removed from the database.
-
- >>> ZODB.tests.util.pack(db)
-
- Now if we access the dictionary in a new connection, it no longer
- has p2:
-
- >>> conn3 = db.open()
- >>> d = conn3.root()['d']
- >>> l = [(str(k), d[k], d.get(k)) for k in d]
- >>> l.sort()
- >>> l
- [('P(p1)', 1, 1), ('P(p3)', 3, 3)]
-
- It's worth nothing that that the versions of the dictionary in
- conn1 and conn2 still have p2, because p2 is still in the caches
- for those connections.
-
- Always explicitly close databases: :)
-
- >>> db.close()
-
+ of items is extremely lazy.
"""
# TODO: It's expensive trying to load dead objects from the database.
# It would be helpful if the data manager/connection cached these.
@@ -329,16 +105,6 @@
def get(self, key, default=None):
"""D.get(k[, d]) -> D[k] if k in D, else d.
-
- >>> import ZODB.tests.util
- >>> key = ZODB.tests.util.P("key")
- >>> missing = ZODB.tests.util.P("missing")
- >>> d = PersistentWeakKeyDictionary([(key, 1)])
- >>> d.get(key)
- 1
- >>> d.get(missing)
- >>> d.get(missing, 12)
- 12
"""
return self.data.get(WeakRef(key), default)
Modified: persistent/trunk/setup.cfg
===================================================================
--- persistent/trunk/setup.cfg 2012-06-28 22:50:14 UTC (rev 127163)
+++ persistent/trunk/setup.cfg 2012-06-28 22:50:18 UTC (rev 127164)
@@ -4,7 +4,6 @@
cover-erase=1
with-doctest=0
where=persistent
-ignore-files=wref.py
[aliases]
dev = develop easy_install persistent[testing]
More information about the checkins
mailing list