Source Code for Module grassyknoll.tests.test_concurrent.test_Fault

 1  from grassyknoll.concurrent.Fault import current_tb_message, Fault 
 2  from grassyknoll.concurrent import Message 
 3  from grassyknoll.concurrent import MailBox 
 4   
 5 -class TestCurrentTBMessage(object): 
 6   
 7 -    def test_no_tb(self): 
 8          try: 
 9              raise RuntimeError, "OMFG" 
10          except RuntimeError: 
11              m=current_tb_message(include_tb=False) 
12          assert isinstance(m.payload, Fault) 
13          assert m.payload.type is RuntimeError 
14          assert m.payload.value.args == ("OMFG",) 
15          assert m.payload.traceback is None 
16          assert m.reply_box is None 
17          assert m.re_id is None 
18          del m 
19       
20 -    def test_tb(self): 
21          try: 
22              raise RuntimeError, "OMFG" 
23          except RuntimeError: 
24              m=current_tb_message(include_tb=True) 
25          assert isinstance(m.payload, Fault) 
26          assert isinstance(m.payload.traceback , str) 
27          assert m.payload.traceback.endswith('''raise RuntimeError, "OMFG"\nRuntimeError: OMFG\n''') 
28          del m 
29       
30 -    def test_reply(self): 
31          x=Message.Message("pants") 
32          try: 
33              raise RuntimeError, "OMFG" 
34          except RuntimeError: 
35              m=current_tb_message(re_id=x.id) 
36          assert isinstance(m.payload, Fault) 
37          assert m.re_id == x.id 
38          del m 
39       
40 -    def test_reply_box(self): 
41          box=MailBox.MailBox() 
42          try: 
43              raise RuntimeError, "OMFG" 
44          except RuntimeError: 
45              m=current_tb_message(reply_box=box) 
46          assert isinstance(m.payload, Fault) 
47          assert m.reply_box is box 
48          del m 
49