| Home | Trees | Indices | Help |
|
|---|
|
|
1 from nose.tools import * 2 from grassyknoll.concurrent.MailBox import * 3 from grassyknoll.concurrent.Message import * 4 from grassyknoll.concurrent.errors import * 5 6 # for tests of non-zero timeouts, see test_Worker.py 795811 self.box=ThreadMailBox(qsize=3)1214 assert hasattr(self.box, 'name') 15 assert len(self.box) == 0 16 assert_raises(BoxEmptyError, self.box.recvNow)1719 m1 = Message(111) 20 self.box.sendNow(m1) 21 assert len(self.box) == 1 22 m1_ = self.box.recvNow() 23 assert m1_ is m1 24 assert len(self.box) == 02527 for i in xrange(1,4): 28 self.box.sendNow(Message(i*100)) 29 30 assert len(self.box) == 3 31 assert_raises(BoxFullError, self.box.sendNow, Message(400))3234 for i in xrange(1,4): 35 self.box.sendNow(Message(i*100)) 36 37 mesg = self.box.recvNow() 38 assert mesg.payload == 100 39 40 self.box.sendNow(Message(500)) 41 42 for i in (200, 300, 500): 43 mesg = self.box.recvNow() 44 assert mesg.payload == i 45 46 assert_raises(BoxEmptyError, self.box.recvNow)4749 self.box.sendNow(Message(600)) 50 self.box.close() 51 assert_raises(BoxClosedError, self.box.sendNow, Message(700)) 52 mesg = self.box.recvNow() 53 assert mesg.payload == 600 54 assert_raises(BoxEmptyError, self.box.recvNow)5557 assert_raises(TypeError, self.box.send, "pants")
| Home | Trees | Indices | Help |
|
|---|
| Generated by Epydoc 3.0beta1 on Mon Mar 10 05:37:18 2008 | http://epydoc.sourceforge.net |