Package grassyknoll :: Package tests :: Package test_concurrent :: Module test_MailBox
[hide private]

Source Code for Module grassyknoll.tests.test_concurrent.test_MailBox

 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 
 7      
8 -class TestMailBox(object):
9
10 - def setup(self):
11 self.box=ThreadMailBox(qsize=3)
12
13 - def test_empty(self):
14 assert hasattr(self.box, 'name') 15 assert len(self.box) == 0 16 assert_raises(BoxEmptyError, self.box.recvNow)
17
18 - def test_send_recv(self):
19 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) == 0
25
26 - def test_send_full(self):
27 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))
32
33 - def test_recv_order(self):
34 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)
47
48 - def test_close(self):
49 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)
55
56 - def test_bad_send(self):
57 assert_raises(TypeError, self.box.send, "pants")
58