#! /usr/bin/env python
# -*- coding: iso-8859-1 -*-
#-----------------------------------------------------------------------------
# Modeling Framework: an Object-Relational Bridge for python
#
# Copyright (c) 2001-2004 Sbastien Bigaret <sbigaret@users.sourceforge.net>
# All rights reserved.
#
# This file is part of the Modeling Framework.
#
# This code is distributed under a "3-clause BSD"-style license;
# see the LICENSE file for details.
#-----------------------------------------------------------------------------
"""
Tests ObserverCenter
...
CVS information
$Id: test_ObserverCenter.py 932 2004-07-20 06:21:57Z sbigaret $
"""
import unittest
if __name__ == "__main__":
import utils, sys
utils.fixpath()
from Modeling import ObserverCenter
from Modeling.interfaces.ObservingInterface import ObservingInterface
OC=ObserverCenter
class Observed:
def __init__(self):
pass
def willChange(self):
ObserverCenter.notifyObserversObjectWillChange(self)
class Observer:
__implements__=(ObservingInterface,)
def __init__(self):
self.object=None
def objectWillChange(self, anObject):
self.object=anObject
def reset(self):
self.object=None
class TestObserverCenter(unittest.TestCase):
"Empty for now"
def test_01_add_notify_remove_and_observersForObject(self):
"[ObserverCenter] add/notify x2/remove + observersForObject()"
observer=Observer()
o1=Observed()
OC.addObserver(observer, o1)
o1.willChange()
self.failUnless(observer.object==o1)
observer.reset()
o1.willChange()
self.failUnless(observer.object is None)
self.failUnless(OC.observersForObject(o1)==(observer,))
OC.removeObserver(observer, o1)
self.failUnless(OC.observersForObject(o1)==())
def test_02_add_notify_add_notify(self):
"[ObserverCenter] ..."
observer1=Observer()
observer2=Observer()
o1=Observed()
o2=Observed()
self.failIf(o2.willChange(), "Should not raise")
OC.addObserver(observer1, o1)
OC.addObserver(observer1, o2)
o2.willChange()
self.failUnless(observer1.object==o2)
observer1.reset()
OC.addObserver(observer2, o2)
o2.willChange()
self.failUnless(observer1.object==o2)
self.failUnless(observer2.object==o2)
def test_03_ensureSubsequentChangeWillBeNotifiedForObject(self):
"[ObserverCenter] ensureSubsequentChangeWillBeNotifiedForObject"
observer1=Observer()
o1=Observed()
OC.addObserver(observer1, o1)
o1.willChange()
observer1.reset()
OC.ensureSubsequentChangeWillBeNotifiedForObject(o1)
o1.willChange()
self.failUnless(observer1.object==o1)
def test_04_notifyObserversObjectWillChange_None(self):
"[ObserverCenter] notifyObserversObjectWillChange(None)"
observer1=Observer()
o1=Observed()
o2=Observed()
OC.addObserver(observer1, o1)
OC.addObserver(observer1, o2)
o1.willChange(); o2.willChange()
observer1.reset()
self.failIf(OC.notifyObserversObjectWillChange(None), 'Should not raise')
o1.willChange()
self.failUnless(observer1.object==o1)
o2.willChange()
self.failUnless(observer1.object==o2)
def test_05_omniscientObservers(self):
"[ObserverCenter] omniscientObservers"
omni=Observer()
OC.addOmniscientObserver(omni)
observer1=Observer() ; o1=Observed() ; o2=Observed()
OC.addObserver(observer1, o1)
OC.addObserver(observer1, o2)
o1.willChange()
self.failUnless(omni.object==o1)
o2.willChange()
self.failUnless(omni.object==o2)
self.failIf(OC.notifyObserversObjectWillChange(None), 'Should not raise')
self.failUnless(omni.object==None)
#del omni ## ???
o2.willChange()
def test_06_observerGarbageCollected(self):
"[ObserverCenter] Tests behaviour when an observer is garbage-collected"
observer=Observer()
o=Observed()
OC.addObserver(observer, o)
del observer
o.willChange()
# test suite
def test_suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(TestObserverCenter, "test_"))
return suite
if __name__ == "__main__":
errs = utils.run_suite(test_suite())
sys.exit(errs and 1 or 0)
|