##// END OF EJS Templates
Replace all use of the ast module with the codeop module, and all use of...
Gael Varoquaux -
Show More
@@ -0,0 +1,155 b''
1 # encoding: utf-8
2
3 """This file contains unittests for the frontendbase module."""
4
5 __docformat__ = "restructuredtext en"
6
7 #---------------------------------------------------------------------------
8 # Copyright (C) 2008 The IPython Development Team
9 #
10 # Distributed under the terms of the BSD License. The full license is in
11 # the file COPYING, distributed as part of this software.
12 #---------------------------------------------------------------------------
13
14 #---------------------------------------------------------------------------
15 # Imports
16 #---------------------------------------------------------------------------
17
18 import unittest
19
20 try:
21 from IPython.frontend.asyncfrontendbase import AsyncFrontEndBase
22 from IPython.frontend import frontendbase
23 from IPython.kernel.engineservice import EngineService
24 except ImportError:
25 import nose
26 raise nose.SkipTest("This test requires zope.interface, Twisted and Foolscap")
27
28 from IPython.testing.decorators import skip
29
30 class FrontEndCallbackChecker(AsyncFrontEndBase):
31 """FrontEndBase subclass for checking callbacks"""
32 def __init__(self, engine=None, history=None):
33 super(FrontEndCallbackChecker, self).__init__(engine=engine,
34 history=history)
35 self.updateCalled = False
36 self.renderResultCalled = False
37 self.renderErrorCalled = False
38
39 def update_cell_prompt(self, result, blockID=None):
40 self.updateCalled = True
41 return result
42
43 def render_result(self, result):
44 self.renderResultCalled = True
45 return result
46
47
48 def render_error(self, failure):
49 self.renderErrorCalled = True
50 return failure
51
52
53
54
55 class TestAsyncFrontendBase(unittest.TestCase):
56 def setUp(self):
57 """Setup the EngineService and FrontEndBase"""
58
59 self.fb = FrontEndCallbackChecker(engine=EngineService())
60
61 def test_implements_IFrontEnd(self):
62 assert(frontendbase.IFrontEnd.implementedBy(
63 AsyncFrontEndBase))
64
65 def test_is_complete_returns_False_for_incomplete_block(self):
66 """"""
67
68 block = """def test(a):"""
69
70 assert(self.fb.is_complete(block) == False)
71
72 def test_is_complete_returns_True_for_complete_block(self):
73 """"""
74
75 block = """def test(a): pass"""
76
77 assert(self.fb.is_complete(block))
78
79 block = """a=3"""
80
81 assert(self.fb.is_complete(block))
82
83 def test_blockID_added_to_result(self):
84 block = """3+3"""
85
86 d = self.fb.execute(block, blockID='TEST_ID')
87
88 d.addCallback(self.checkBlockID, expected='TEST_ID')
89
90 def test_blockID_added_to_failure(self):
91 block = "raise Exception()"
92
93 d = self.fb.execute(block,blockID='TEST_ID')
94 d.addErrback(self.checkFailureID, expected='TEST_ID')
95
96 def checkBlockID(self, result, expected=""):
97 assert(result['blockID'] == expected)
98
99
100 def checkFailureID(self, failure, expected=""):
101 assert(failure.blockID == expected)
102
103
104 def test_callbacks_added_to_execute(self):
105 """test that
106 update_cell_prompt
107 render_result
108
109 are added to execute request
110 """
111
112 d = self.fb.execute("10+10")
113 d.addCallback(self.checkCallbacks)
114
115 def checkCallbacks(self, result):
116 assert(self.fb.updateCalled)
117 assert(self.fb.renderResultCalled)
118
119 @skip("This test fails and lead to an unhandled error in a Deferred.")
120 def test_error_callback_added_to_execute(self):
121 """test that render_error called on execution error"""
122
123 d = self.fb.execute("raise Exception()")
124 d.addCallback(self.checkRenderError)
125
126 def checkRenderError(self, result):
127 assert(self.fb.renderErrorCalled)
128
129 def test_history_returns_expected_block(self):
130 """Make sure history browsing doesn't fail"""
131
132 blocks = ["a=1","a=2","a=3"]
133 for b in blocks:
134 d = self.fb.execute(b)
135
136 # d is now the deferred for the last executed block
137 d.addCallback(self.historyTests, blocks)
138
139
140 def historyTests(self, result, blocks):
141 """historyTests"""
142
143 assert(len(blocks) >= 3)
144 assert(self.fb.get_history_previous("") == blocks[-2])
145 assert(self.fb.get_history_previous("") == blocks[-3])
146 assert(self.fb.get_history_next() == blocks[-2])
147
148
149 def test_history_returns_none_at_startup(self):
150 """test_history_returns_none_at_startup"""
151
152 assert(self.fb.get_history_previous("")==None)
153 assert(self.fb.get_history_next()==None)
154
155
@@ -14,7 +14,7 b' __docformat__ = "restructuredtext en"'
14 #-------------------------------------------------------------------------------
14 #-------------------------------------------------------------------------------
15 # Imports
15 # Imports
16 #-------------------------------------------------------------------------------
16 #-------------------------------------------------------------------------------
17 import uuid
17 from IPython.external import guid
18
18
19
19
20 from zope.interface import Interface, Attribute, implements, classProvides
20 from zope.interface import Interface, Attribute, implements, classProvides
@@ -59,7 +59,7 b' class AsyncFrontEndBase(FrontEndBase):'
59 return Failure(Exception("Block is not compilable"))
59 return Failure(Exception("Block is not compilable"))
60
60
61 if(blockID == None):
61 if(blockID == None):
62 blockID = uuid.uuid4() #random UUID
62 blockID = guid.generate()
63
63
64 d = self.engine.execute(block)
64 d = self.engine.execute(block)
65 d.addCallback(self._add_history, block=block)
65 d.addCallback(self._add_history, block=block)
@@ -26,7 +26,7 b' __docformat__ = "restructuredtext en"'
26
26
27 import sys
27 import sys
28 import objc
28 import objc
29 import uuid
29 from IPython.external import guid
30
30
31 from Foundation import NSObject, NSMutableArray, NSMutableDictionary,\
31 from Foundation import NSObject, NSMutableArray, NSMutableDictionary,\
32 NSLog, NSNotificationCenter, NSMakeRange,\
32 NSLog, NSNotificationCenter, NSMakeRange,\
@@ -361,7 +361,7 b' class IPythonCocoaController(NSObject, AsyncFrontEndBase):'
361
361
362 def next_block_ID(self):
362 def next_block_ID(self):
363
363
364 return uuid.uuid4()
364 return guid.generate()
365
365
366 def new_cell_block(self):
366 def new_cell_block(self):
367 """A new CellBlock at the end of self.textView.textStorage()"""
367 """A new CellBlock at the end of self.textView.textStorage()"""
@@ -21,76 +21,8 b' __docformat__ = "restructuredtext en"'
21 # Imports
21 # Imports
22 #-------------------------------------------------------------------------------
22 #-------------------------------------------------------------------------------
23 import string
23 import string
24
24 import codeop
25 try:
25 from IPython.external import guid
26 import _ast
27 except ImportError:
28 # Python 2.4 hackish workaround.
29 class bunch: pass
30 _ast = bunch()
31 _ast.PyCF_ONLY_AST = 1024
32
33
34
35 try:
36 import uuid
37 except ImportError:
38 # Python 2.4 hackish workaround.
39 class UUID:
40 def __init__(self,bytes):
41 version = 4
42 int = long(('%02x'*16) % tuple(map(ord, bytes)), 16)
43 # Set the variant to RFC 4122.
44 int &= ~(0xc000 << 48L)
45 int |= 0x8000 << 48L
46 # Set the version number.
47 int &= ~(0xf000 << 64L)
48 int |= version << 76L
49 self.__dict__['int'] = int
50
51 def __cmp__(self, other):
52 if isinstance(other, UUID):
53 return cmp(self.int, other.int)
54 return NotImplemented
55
56 def __hash__(self):
57 return hash(self.int)
58
59 def __int__(self):
60 return self.int
61
62 def __repr__(self):
63 return 'UUID(%r)' % str(self)
64
65 def __setattr__(self, name, value):
66 raise TypeError('UUID objects are immutable')
67
68 def __str__(self):
69 hex = '%032x' % self.int
70 return '%s-%s-%s-%s-%s' % (
71 hex[:8], hex[8:12], hex[12:16], hex[16:20], hex[20:])
72
73 def get_bytes(self):
74 bytes = ''
75 for shift in range(0, 128, 8):
76 bytes = chr((self.int >> shift) & 0xff) + bytes
77 return bytes
78
79 bytes = property(get_bytes)
80
81
82 def _u4():
83 "Fake random uuid"
84
85 import random
86 bytes = [chr(random.randrange(256)) for i in range(16)]
87 return UUID(bytes)
88
89 class bunch: pass
90 uuid = bunch()
91 uuid.uuid4 = _u4
92 del _u4
93
94
26
95
27
96 from IPython.frontend.zopeinterface import (
28 from IPython.frontend.zopeinterface import (
@@ -201,10 +133,6 b' class IFrontEnd(Interface):'
201
133
202 pass
134 pass
203
135
204 def compile_ast(block):
205 """Compiles block to an _ast.AST"""
206
207 pass
208
136
209 def get_history_previous(current_block):
137 def get_history_previous(current_block):
210 """Returns the block previous in the history. Saves currentBlock if
138 """Returns the block previous in the history. Saves currentBlock if
@@ -287,28 +215,14 b' class FrontEndBase(object):'
287 """
215 """
288
216
289 try:
217 try:
290 ast = self.compile_ast(block)
218 is_complete = codeop.compile_command(block.rstrip() + '\n\n',
219 "<string>", "exec")
291 except:
220 except:
292 return False
221 return False
293
222
294 lines = block.split('\n')
223 lines = block.split('\n')
295 return (len(lines)==1 or str(lines[-1])=='')
224 return ((is_complete is not None)
296
225 and (len(lines)==1 or str(lines[-1])==''))
297
298 def compile_ast(self, block):
299 """Compile block to an AST
300
301 Parameters:
302 block : str
303
304 Result:
305 AST
306
307 Throws:
308 Exception if block cannot be compiled
309 """
310
311 return compile(block, "<string>", "exec", _ast.PyCF_ONLY_AST)
312
226
313
227
314 def execute(self, block, blockID=None):
228 def execute(self, block, blockID=None):
@@ -328,7 +242,7 b' class FrontEndBase(object):'
328 raise Exception("Block is not compilable")
242 raise Exception("Block is not compilable")
329
243
330 if(blockID == None):
244 if(blockID == None):
331 blockID = uuid.uuid4() #random UUID
245 blockID = guid.generate()
332
246
333 try:
247 try:
334 result = self.shell.execute(block)
248 result = self.shell.execute(block)
@@ -1,155 +1,32 b''
1 # encoding: utf-8
1 # encoding: utf-8
2
2 """
3 """This file contains unittests for the frontendbase module."""
3 Test the basic functionality of frontendbase.
4 """
4
5
5 __docformat__ = "restructuredtext en"
6 __docformat__ = "restructuredtext en"
6
7
7 #---------------------------------------------------------------------------
8 #-------------------------------------------------------------------------------
8 # Copyright (C) 2008 The IPython Development Team
9 # Copyright (C) 2008 The IPython Development Team
9 #
10 #
10 # Distributed under the terms of the BSD License. The full license is in
11 # Distributed under the terms of the BSD License. The full license is
11 # the file COPYING, distributed as part of this software.
12 # in the file COPYING, distributed as part of this software.
12 #---------------------------------------------------------------------------
13 #-------------------------------------------------------------------------------
13
14 #---------------------------------------------------------------------------
15 # Imports
16 #---------------------------------------------------------------------------
17
18 import unittest
19
20 try:
21 from IPython.frontend.asyncfrontendbase import AsyncFrontEndBase
22 from IPython.frontend import frontendbase
23 from IPython.kernel.engineservice import EngineService
24 except ImportError:
25 import nose
26 raise nose.SkipTest("This test requires zope.interface, Twisted and Foolscap")
27
28 from IPython.testing.decorators import skip
29
30 class FrontEndCallbackChecker(AsyncFrontEndBase):
31 """FrontEndBase subclass for checking callbacks"""
32 def __init__(self, engine=None, history=None):
33 super(FrontEndCallbackChecker, self).__init__(engine=engine,
34 history=history)
35 self.updateCalled = False
36 self.renderResultCalled = False
37 self.renderErrorCalled = False
38
39 def update_cell_prompt(self, result, blockID=None):
40 self.updateCalled = True
41 return result
42
43 def render_result(self, result):
44 self.renderResultCalled = True
45 return result
46
47
48 def render_error(self, failure):
49 self.renderErrorCalled = True
50 return failure
51
52
53
54
55 class TestAsyncFrontendBase(unittest.TestCase):
56 def setUp(self):
57 """Setup the EngineService and FrontEndBase"""
58
59 self.fb = FrontEndCallbackChecker(engine=EngineService())
60
61 def test_implements_IFrontEnd(self):
62 assert(frontendbase.IFrontEnd.implementedBy(
63 AsyncFrontEndBase))
64
65 def test_is_complete_returns_False_for_incomplete_block(self):
66 """"""
67
68 block = """def test(a):"""
69
70 assert(self.fb.is_complete(block) == False)
71
72 def test_is_complete_returns_True_for_complete_block(self):
73 """"""
74
75 block = """def test(a): pass"""
76
77 assert(self.fb.is_complete(block))
78
79 block = """a=3"""
80
14
81 assert(self.fb.is_complete(block))
15 from IPython.frontend.frontendbase import FrontEndBase
82
16
83 def test_blockID_added_to_result(self):
17 def test_iscomplete():
84 block = """3+3"""
18 """ Check that is_complete works.
85
86 d = self.fb.execute(block, blockID='TEST_ID')
87
88 d.addCallback(self.checkBlockID, expected='TEST_ID')
89
90 def test_blockID_added_to_failure(self):
91 block = "raise Exception()"
92
93 d = self.fb.execute(block,blockID='TEST_ID')
94 d.addErrback(self.checkFailureID, expected='TEST_ID')
95
96 def checkBlockID(self, result, expected=""):
97 assert(result['blockID'] == expected)
98
99
100 def checkFailureID(self, failure, expected=""):
101 assert(failure.blockID == expected)
102
103
104 def test_callbacks_added_to_execute(self):
105 """test that
106 update_cell_prompt
107 render_result
108
109 are added to execute request
110 """
19 """
111
20 f = FrontEndBase()
112 d = self.fb.execute("10+10")
21 assert f.is_complete('(a + a)')
113 d.addCallback(self.checkCallbacks)
22 assert not f.is_complete('(a + a')
114
23 assert f.is_complete('1')
115 def checkCallbacks(self, result):
24 assert not f.is_complete('1 + ')
116 assert(self.fb.updateCalled)
25 assert not f.is_complete('1 + \n\n')
117 assert(self.fb.renderResultCalled)
26 assert f.is_complete('if True:\n print 1\n')
118
27 assert not f.is_complete('if True:\n print 1')
119 @skip("This test fails and lead to an unhandled error in a Deferred.")
28 assert f.is_complete('def f():\n print 1\n')
120 def test_error_callback_added_to_execute(self):
29
121 """test that render_error called on execution error"""
30 if __name__ == '__main__':
122
31 test_iscomplete()
123 d = self.fb.execute("raise Exception()")
124 d.addCallback(self.checkRenderError)
125
126 def checkRenderError(self, result):
127 assert(self.fb.renderErrorCalled)
128
129 def test_history_returns_expected_block(self):
130 """Make sure history browsing doesn't fail"""
131
132 blocks = ["a=1","a=2","a=3"]
133 for b in blocks:
134 d = self.fb.execute(b)
135
136 # d is now the deferred for the last executed block
137 d.addCallback(self.historyTests, blocks)
138
139
140 def historyTests(self, result, blocks):
141 """historyTests"""
142
143 assert(len(blocks) >= 3)
144 assert(self.fb.get_history_previous("") == blocks[-2])
145 assert(self.fb.get_history_previous("") == blocks[-3])
146 assert(self.fb.get_history_next() == blocks[-2])
147
148
149 def test_history_returns_none_at_startup(self):
150 """test_history_returns_none_at_startup"""
151
152 assert(self.fb.get_history_previous("")==None)
153 assert(self.fb.get_history_next()==None)
154
155
32
General Comments 0
You need to be logged in to leave comments. Login now