Viewing file: test_reference.py (6.72 KB) -rw-r--r-- Select action/file-type: (+) | (+) | (+) | Code (+) | Session (+) | (+) | SDB (+) | (+) | (+) | (+) | (+) | (+) |
import gc
from twisted.trial.unittest import TestCase
from axiom.store import Store from axiom.upgrade import registerUpgrader from axiom.item import Item, declareLegacyItem from axiom.attributes import integer, reference from axiom.errors import BrokenReference, DeletionDisallowed
class Referee(Item): schemaVersion = 1 typeName = "test_reference_referee"
topSecret = integer()
class SimpleReferent(Item): schemaVersion = 1 typeName = "test_reference_referent"
ref = reference()
class BreakingReferent(Item): schemaVersion = 1 typeName = "test_reference_breaking_referent"
ref = reference(whenDeleted=reference.NULLIFY)
class DependentReferent(Item): ref = reference(whenDeleted=reference.CASCADE, reftype=Referee)
class DisallowReferent(Item): ref = reference(whenDeleted=reference.DISALLOW, reftype=Referee)
class BadReferenceTestCase(TestCase): ntimes = 10
def testSanity(self): store = Store() for i in xrange(self.ntimes): SimpleReferent(store=store, ref=Referee(store=store, topSecret=i)) (referee,) = list(store.query(Referee)) (referent,) = list(store.query(SimpleReferent)) self.assertEqual(referent.ref.topSecret, referee.topSecret) referee.deleteFromStore() referent.deleteFromStore()
def testBadReferenceNone(self): """ Test that accessing a broken reference on an Item that has already been loaded into memory correctly nullifies the attribute. """ store = Store() referee = Referee(store=store, topSecret=0) referent = SimpleReferent(store=store, ref=referee) referee.deleteFromStore()
referee = None gc.collect()
(referent,) = list(store.query(SimpleReferent)) self.assertEqual(referent.ref, None)
def testBadReferenceNoneLoading(self): """ Test that accessing a broken reference on an Item that has not yet been loaded correctly nullifies the attribute. """ store = Store() referee = Referee(store=store, topSecret=0) referent = SimpleReferent(store=store, ref=referee) referee.deleteFromStore()
referee = None referent = None gc.collect()
(referent,) = list(store.query(SimpleReferent)) self.assertEqual(referent.ref, None)
def test_brokenReferenceException(self): """ Test that an exception is raised when a broken reference is detected when this should be impossible (ie. CASCADE or NULLIFY). """ store = Store()
referee = Referee(store=store, topSecret=0) referent = BreakingReferent(store=store, ref=referee)
referee.deleteFromStore() referent = None gc.collect()
referent = store.findFirst(BreakingReferent) self.patch(BreakingReferent.ref, 'whenDeleted', reference.CASCADE) self.assertRaises(BrokenReference, lambda: referent.ref)
def testBadReferenceNoneRevert(self): store = Store() referee = Referee(store=store, topSecret=0) referent = SimpleReferent(store=store, ref=referee) def txn(): referee.deleteFromStore() self.assertEqual(referent.ref, None) 1 / 0 self.assertRaises(ZeroDivisionError, store.transact, txn) self.assertEqual(referent.ref, referee)
referent = None referee = None gc.collect()
referent = store.findUnique(SimpleReferent) referee = store.findUnique(Referee) self.assertEqual(referent.ref, referee)
def testBrokenReferenceDisallow(self): """ Test that deleting an item referred to by a whenDeleted == DISALLOW reference raises an exception. """ store = Store() referee = Referee(store=store, topSecret=0) referent = DisallowReferent(store=store, ref=referee)
self.assertRaises(DeletionDisallowed, referee.deleteFromStore) self.assertRaises(DeletionDisallowed, store.query(Referee).deleteFromStore)
def testReferenceQuery(self): store = Store() referee = Referee(store=store, topSecret=0) self.assertEqual( list(store.query(SimpleReferent, SimpleReferent.ref == Referee.storeID)), [])
def testReferenceDeletion(self): store = Store() referee = Referee(store=store, topSecret=0) dep = DependentReferent(store=store, ref=referee) sid = dep.storeID self.assertIdentical(store.getItemByID(sid), dep) # sanity referee.deleteFromStore() self.assertRaises(KeyError, store.getItemByID, sid)
def testBatchReferenceDeletion(self): """ Test that batch deletion removes dependent items correctly. """ store = Store() referee = Referee(store=store, topSecret=0) dep = DependentReferent(store=store, ref=referee) sid = dep.storeID store.query(Referee).deleteFromStore() self.assertRaises(KeyError, store.getItemByID, sid)
def test_dummyItemReference(self): """ Getting the value of a reference attribute which has previously been set to a legacy item results in an instance of the most recent type for that item. """ store = Store() referent = SimpleReferent(store=store) oldReferee = nonUpgradedItem(store=store) referent.ref = oldReferee newReferee = referent.ref self.assertTrue( isinstance(newReferee, UpgradedItem), "%r was instance of %r, expected %r" % (newReferee, type(newReferee), UpgradedItem))
def test_dummyItemGetItemByID(self): """ Instantiating a dummy item and then getting it by its storeID should upgrade it. """ store = Store() t = nonUpgradedItem(store=store) self.assertEquals(t.__legacy__, True) self.assertFalse(store.objectCache.has(t.storeID)) t2 = store.getItemByID(t.storeID) self.assertNotIdentical(t, t2) self.assertTrue(isinstance(t2, UpgradedItem))
class UpgradedItem(Item): """ A simple item which is the current version of L{nonUpgradedItem}. """ schemaVersion = 2 dummy = integer()
nonUpgradedItem = declareLegacyItem( UpgradedItem.typeName, 1, dict(dummy=integer()))
def item1to2(old): """ Upgrade an nonUpgradedItem to UpgradedItem """ return old.upgradeVersion(UpgradedItem.typeName, 1, 2, dummy=old.dummy)
registerUpgrader(item1to2, UpgradedItem.typeName, 1, 2)
|