Replace all use of the ast module with the codeop module, and all use of

the uuid module with the IPython.external.guid module.

Add tests for the is_complete method.

IPython/frontend/tests/test_asyncfrontendbase.py

@@ -0,0 +1,155 b''
+# encoding: utf-8
+
+"""This file contains unittests for the frontendbase module."""
+
+__docformat__ = "restructuredtext en"
+
+#---------------------------------------------------------------------------
+#  Copyright (C) 2008  The IPython Development Team                         
+#                                                                           
+#  Distributed under the terms of the BSD License.  The full license is in  
+#  the file COPYING, distributed as part of this software.                  
+#---------------------------------------------------------------------------
+                                                                            
+#---------------------------------------------------------------------------
+# Imports                                                                   
+#---------------------------------------------------------------------------
+
+import unittest
+
+try:
+    from IPython.frontend.asyncfrontendbase import AsyncFrontEndBase
+    from IPython.frontend import frontendbase 
+    from IPython.kernel.engineservice import EngineService
+except ImportError:
+    import nose
+    raise nose.SkipTest("This test requires zope.interface, Twisted and Foolscap")
+
+from IPython.testing.decorators import skip
+
+class FrontEndCallbackChecker(AsyncFrontEndBase):
+    """FrontEndBase subclass for checking callbacks"""
+    def __init__(self, engine=None, history=None):
+        super(FrontEndCallbackChecker, self).__init__(engine=engine, 
+                                                    history=history)
+        self.updateCalled = False
+        self.renderResultCalled = False
+        self.renderErrorCalled = False
+   
+    def update_cell_prompt(self, result, blockID=None):
+        self.updateCalled = True
+        return result
+   
+    def render_result(self, result):
+        self.renderResultCalled = True
+        return result
+    
+    
+    def render_error(self, failure):
+        self.renderErrorCalled = True
+        return failure
+    
+
+
+    
+class TestAsyncFrontendBase(unittest.TestCase):
+    def setUp(self):
+        """Setup the EngineService and FrontEndBase"""
+        
+        self.fb = FrontEndCallbackChecker(engine=EngineService())
+    
+    def test_implements_IFrontEnd(self):
+        assert(frontendbase.IFrontEnd.implementedBy(
+                                    AsyncFrontEndBase))
+    
+    def test_is_complete_returns_False_for_incomplete_block(self):
+        """"""
+        
+        block = """def test(a):"""
+        
+        assert(self.fb.is_complete(block) == False)
+    
+    def test_is_complete_returns_True_for_complete_block(self):
+        """"""
+        
+        block = """def test(a): pass"""
+            
+        assert(self.fb.is_complete(block))
+        
+        block = """a=3"""
+        
+        assert(self.fb.is_complete(block))
+    
+    def test_blockID_added_to_result(self):
+        block = """3+3"""
+        
+        d = self.fb.execute(block, blockID='TEST_ID')
+        
+        d.addCallback(self.checkBlockID, expected='TEST_ID')
+    
+    def test_blockID_added_to_failure(self):
+        block = "raise Exception()"
+        
+        d = self.fb.execute(block,blockID='TEST_ID')
+        d.addErrback(self.checkFailureID, expected='TEST_ID')
+    
+    def checkBlockID(self, result, expected=""):
+        assert(result['blockID'] == expected)
+    
+    
+    def checkFailureID(self, failure, expected=""):
+        assert(failure.blockID == expected)
+    
+    
+    def test_callbacks_added_to_execute(self):
+        """test that
+        update_cell_prompt
+        render_result
+        
+        are added to execute request
+        """
+        
+        d = self.fb.execute("10+10")
+        d.addCallback(self.checkCallbacks)
+    
+    def checkCallbacks(self, result):
+        assert(self.fb.updateCalled)
+        assert(self.fb.renderResultCalled)
+    
+    @skip("This test fails and lead to an unhandled error in a Deferred.")
+    def test_error_callback_added_to_execute(self):
+        """test that render_error called on execution error"""
+        
+        d = self.fb.execute("raise Exception()")
+        d.addCallback(self.checkRenderError)
+    
+    def checkRenderError(self, result):
+        assert(self.fb.renderErrorCalled)
+    
+    def test_history_returns_expected_block(self):
+        """Make sure history browsing doesn't fail"""
+        
+        blocks = ["a=1","a=2","a=3"]
+        for b in blocks:
+            d = self.fb.execute(b)
+        
+        # d is now the deferred for the last executed block
+        d.addCallback(self.historyTests, blocks)
+        
+    
+    def historyTests(self, result, blocks):
+        """historyTests"""
+        
+        assert(len(blocks) >= 3)
+        assert(self.fb.get_history_previous("") == blocks[-2])
+        assert(self.fb.get_history_previous("") == blocks[-3])
+        assert(self.fb.get_history_next() == blocks[-2])
+    
+    
+    def test_history_returns_none_at_startup(self):
+        """test_history_returns_none_at_startup"""
+        
+        assert(self.fb.get_history_previous("")==None)
+        assert(self.fb.get_history_next()==None)
+    
+    

IPython/frontend/asyncfrontendbase.py

@@ -1,76 +1,76 b''
 """
 Base front end class for all async frontends.
 """
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
-import uuid
+from IPython.external import guid
 
 
 from zope.interface import Interface, Attribute, implements, classProvides
 from twisted.python.failure import Failure
 from IPython.frontend.frontendbase import FrontEndBase, IFrontEnd, IFrontEndFactory
 from IPython.kernel.core.history import FrontEndHistory
 from IPython.kernel.engineservice import IEngineCore
 
 
 class AsyncFrontEndBase(FrontEndBase):
     """
     Overrides FrontEndBase to wrap execute in a deferred result.
     All callbacks are made as callbacks on the deferred result.
     """
     
     implements(IFrontEnd)
     classProvides(IFrontEndFactory)
     
     def __init__(self, engine=None, history=None):
         assert(engine==None or IEngineCore.providedBy(engine))
         self.engine = IEngineCore(engine)
         if history is None:
                 self.history = FrontEndHistory(input_cache=[''])
         else:
             self.history = history
     
     
     def execute(self, block, blockID=None):
         """Execute the block and return the deferred result.
         
         Parameters:
             block : {str, AST}
             blockID : any
                 Caller may provide an ID to identify this block. 
                 result['blockID'] := blockID
         
         Result:
             Deferred result of self.interpreter.execute
         """
         
         if(not self.is_complete(block)):
             return Failure(Exception("Block is not compilable"))
         
         if(blockID == None):
-            blockID = uuid.uuid4() #random UUID
+            blockID = guid.generate() 
         
         d = self.engine.execute(block)
         d.addCallback(self._add_history, block=block)
         d.addCallbacks(self._add_block_id_for_result,
                 errback=self._add_block_id_for_failure,
                 callbackArgs=(blockID,),
                 errbackArgs=(blockID,))
         d.addBoth(self.update_cell_prompt, blockID=blockID)
         d.addCallbacks(self.render_result, 
             errback=self.render_error)
         
         return d
     
 

IPython/frontend/cocoa/cocoa_frontend.py

@@ -1,560 +1,560 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.frontend.cocoa.tests.test_cocoa_frontend -*-
 
 """PyObjC classes to provide a Cocoa frontend to the 
 IPython.kernel.engineservice.IEngineBase.
 
 To add an IPython interpreter to a cocoa app, instantiate an 
 IPythonCocoaController in a XIB and connect its textView outlet to an
 NSTextView instance in your UI. That's it.
 
 Author: Barry Wark
 """
 
 __docformat__ = "restructuredtext en"
 
 #-----------------------------------------------------------------------------
 #       Copyright (C) 2008  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import sys
 import objc
-import uuid
+from IPython.external import guid
 
 from Foundation import NSObject, NSMutableArray, NSMutableDictionary,\
                         NSLog, NSNotificationCenter, NSMakeRange,\
                         NSLocalizedString, NSIntersectionRange,\
                         NSString, NSAutoreleasePool
                         
 from AppKit import NSApplicationWillTerminateNotification, NSBeep,\
                     NSTextView, NSRulerView, NSVerticalRuler
 
 from pprint import saferepr
 
 import IPython
 from IPython.kernel.engineservice import ThreadedEngineService
 from IPython.frontend.asyncfrontendbase import AsyncFrontEndBase
 
 from twisted.internet.threads import blockingCallFromThread
 from twisted.python.failure import Failure
 
 #-----------------------------------------------------------------------------
 # Classes to implement the Cocoa frontend
 #-----------------------------------------------------------------------------
 
 # TODO: 
 #   1. use MultiEngineClient and out-of-process engine rather than 
 #       ThreadedEngineService?
 #   2. integrate Xgrid launching of engines
         
 class AutoreleasePoolWrappedThreadedEngineService(ThreadedEngineService):
     """Wrap all blocks in an NSAutoreleasePool"""
     
     def wrapped_execute(self, msg, lines):
         """wrapped_execute"""
         try:
             p = NSAutoreleasePool.alloc().init()
             result = super(AutoreleasePoolWrappedThreadedEngineService,
                             self).wrapped_execute(msg, lines)
         finally:
             p.drain()
         
         return result
     
 
 
 class Cell(NSObject):
     """
     Representation of the prompts, input and output of a cell in the
     frontend
     """
     
     blockNumber = objc.ivar().unsigned_long()
     blockID = objc.ivar()
     inputBlock = objc.ivar()
     output = objc.ivar()
      
 
    
 class CellBlock(object):
     """
     Storage for information about text ranges relating to a single cell
     """
 
         
     def __init__(self, inputPromptRange, inputRange=None, outputPromptRange=None,
                 outputRange=None):
         super(CellBlock, self).__init__()
         self.inputPromptRange = inputPromptRange
         self.inputRange = inputRange
         self.outputPromptRange = outputPromptRange
         self.outputRange = outputRange
     
     def update_ranges_for_insertion(self, text, textRange):
         """Update ranges for text insertion at textRange"""
         
         for r in [self.inputPromptRange,self.inputRange,
                     self.outputPromptRange, self.outputRange]:
             if(r == None):
                 continue
             intersection = NSIntersectionRange(r,textRange)
             if(intersection.length == 0): #ranges don't intersect
                 if r.location >= textRange.location:
                     r.location += len(text)
             else: #ranges intersect
                 if(r.location > textRange.location):
                     offset = len(text) - intersection.length
                     r.length -= offset
                     r.location += offset
                 elif(r.location == textRange.location):
                     r.length += len(text) - intersection.length
                 else:
                     r.length -= intersection.length
     
     
     def update_ranges_for_deletion(self, textRange):
         """Update ranges for text deletion at textRange"""
         
         for r in [self.inputPromptRange,self.inputRange,
                     self.outputPromptRange, self.outputRange]:
             if(r==None):
                 continue
             intersection = NSIntersectionRange(r, textRange)
             if(intersection.length == 0): #ranges don't intersect
                 if r.location >= textRange.location:
                     r.location -= textRange.length
             else: #ranges intersect
                 if(r.location > textRange.location):
                     offset = intersection.length
                     r.length -= offset
                     r.location += offset
                 elif(r.location == textRange.location):
                     r.length += intersection.length
                 else:
                     r.length -= intersection.length
     
     def __repr__(self):
         return 'CellBlock('+ str((self.inputPromptRange,
                                 self.inputRange,
                                 self.outputPromptRange,
                                 self.outputRange)) + ')'
     
 
 
         
 class IPythonCocoaController(NSObject, AsyncFrontEndBase):
     userNS = objc.ivar() #mirror of engine.user_ns (key=>str(value))
     waitingForEngine = objc.ivar().bool()
     textView = objc.IBOutlet()
     
     def init(self):
         self = super(IPythonCocoaController, self).init()
         AsyncFrontEndBase.__init__(self,
                     engine=AutoreleasePoolWrappedThreadedEngineService())
         if(self != None):
             self._common_init()
         
         return self
     
     def _common_init(self):
         """_common_init"""
         
         self.userNS = NSMutableDictionary.dictionary()
         self.waitingForEngine = False
         
         self.lines = {}
         self.tabSpaces = 4
         self.tabUsesSpaces = True
         self.currentBlockID = self.next_block_ID()
         self.blockRanges = {} # blockID=>CellBlock
     
     
     def awakeFromNib(self):
         """awakeFromNib"""
         
         self._common_init()
         
         # Start the IPython engine
         self.engine.startService()
         NSLog('IPython engine started')
         
         # Register for app termination
         nc = NSNotificationCenter.defaultCenter()
         nc.addObserver_selector_name_object_(
                                     self,
                                     'appWillTerminate:',
                                     NSApplicationWillTerminateNotification,
                                     None)
         
         self.textView.setDelegate_(self)
         self.textView.enclosingScrollView().setHasVerticalRuler_(True)
         r = NSRulerView.alloc().initWithScrollView_orientation_(
                                         self.textView.enclosingScrollView(),
                                         NSVerticalRuler)
         self.verticalRulerView = r
         self.verticalRulerView.setClientView_(self.textView)
         self._start_cli_banner()
         self.start_new_block()
     
     
     def appWillTerminate_(self, notification):
         """appWillTerminate"""
         
         self.engine.stopService()
     
     
     def complete(self, token):
         """Complete token in engine's user_ns
         
         Parameters
         ----------
         token : string
         
         Result
         ------
         Deferred result of 
         IPython.kernel.engineservice.IEngineBase.complete
         """
         
         return self.engine.complete(token)
     
     
     def execute(self, block, blockID=None):
         self.waitingForEngine = True
         self.willChangeValueForKey_('commandHistory')
         d = super(IPythonCocoaController, self).execute(block,
                                                         blockID)
         d.addBoth(self._engine_done)
         d.addCallback(self._update_user_ns)
         
         return d
     
     
     def push_(self, namespace):
         """Push dictionary of key=>values to python namespace"""
         
         self.waitingForEngine = True
         self.willChangeValueForKey_('commandHistory')
         d = self.engine.push(namespace)
         d.addBoth(self._engine_done)
         d.addCallback(self._update_user_ns)
     
     
     def pull_(self, keys):
         """Pull keys from python namespace"""
         
         self.waitingForEngine = True
         result = blockingCallFromThread(self.engine.pull, keys)
         self.waitingForEngine = False
     
     @objc.signature('v@:@I')
     def executeFileAtPath_encoding_(self, path, encoding):
         """Execute file at path in an empty namespace. Update the engine
         user_ns with the resulting locals."""
         
         lines,err = NSString.stringWithContentsOfFile_encoding_error_(
             path,
             encoding,
             None)
         self.engine.execute(lines)
     
     
     def _engine_done(self, x):
         self.waitingForEngine = False
         self.didChangeValueForKey_('commandHistory')
         return x
     
     def _update_user_ns(self, result):
         """Update self.userNS from self.engine's namespace"""
         d = self.engine.keys()
         d.addCallback(self._get_engine_namespace_values_for_keys)
         
         return result
     
     
     def _get_engine_namespace_values_for_keys(self, keys):
         d = self.engine.pull(keys)
         d.addCallback(self._store_engine_namespace_values, keys=keys)
     
     
     def _store_engine_namespace_values(self, values, keys=[]):
         assert(len(values) == len(keys))
         self.willChangeValueForKey_('userNS')
         for (k,v) in zip(keys,values):
             self.userNS[k] = saferepr(v)
         self.didChangeValueForKey_('userNS')
     
     
     def update_cell_prompt(self, result, blockID=None):
         print self.blockRanges
         if(isinstance(result, Failure)):
             prompt = self.input_prompt()
             
         else:
             prompt = self.input_prompt(number=result['number'])
         
         r = self.blockRanges[blockID].inputPromptRange
         self.insert_text(prompt,
                 textRange=r,
                 scrollToVisible=False
                 )
         
         return result
     
     
     def render_result(self, result):
         blockID = result['blockID']
         inputRange = self.blockRanges[blockID].inputRange
         del self.blockRanges[blockID]
         
         #print inputRange,self.current_block_range()
         self.insert_text('\n' +
                 self.output_prompt(number=result['number']) +
                 result.get('display',{}).get('pprint','') +
                 '\n\n',
                 textRange=NSMakeRange(inputRange.location+inputRange.length,
                                     0))
         return result
     
         
     def render_error(self, failure):
         print failure
         blockID = failure.blockID
         inputRange = self.blockRanges[blockID].inputRange
         self.insert_text('\n' +
                         self.output_prompt() +
                         '\n' +
                         failure.getErrorMessage() +
                         '\n\n',
                         textRange=NSMakeRange(inputRange.location +
                                                 inputRange.length,
                                                 0))
         self.start_new_block()
         return failure
     
     
     def _start_cli_banner(self):
         """Print banner"""
         
         banner = """IPython1 %s -- An enhanced Interactive Python.""" % \
                     IPython.__version__
         
         self.insert_text(banner + '\n\n')
     
     
     def start_new_block(self):
         """"""
         
         self.currentBlockID = self.next_block_ID()
         self.blockRanges[self.currentBlockID] = self.new_cell_block()
         self.insert_text(self.input_prompt(), 
             textRange=self.current_block_range().inputPromptRange)
     
     
     
     def next_block_ID(self):
         
-        return uuid.uuid4()
+        return guid.generate() 
     
     def new_cell_block(self):
         """A new CellBlock at the end of self.textView.textStorage()"""
         
         return CellBlock(NSMakeRange(self.textView.textStorage().length(), 
                                     0), #len(self.input_prompt())),
                         NSMakeRange(self.textView.textStorage().length(),# + len(self.input_prompt()),
                                     0))
     
     
     def current_block_range(self):
         return self.blockRanges.get(self.currentBlockID, 
                         self.new_cell_block())
     
     def current_block(self):
         """The current block's text"""
         
         return self.text_for_range(self.current_block_range().inputRange)
     
     def text_for_range(self, textRange):
         """text_for_range"""
         
         ts = self.textView.textStorage()
         return ts.string().substringWithRange_(textRange)
     
     def current_line(self):
         block = self.text_for_range(self.current_block_range().inputRange)
         block = block.split('\n')
         return block[-1]
     
     
     def insert_text(self, string=None, textRange=None, scrollToVisible=True):
         """Insert text into textView at textRange, updating blockRanges 
         as necessary
         """
         if(textRange == None):
             #range for end of text
             textRange = NSMakeRange(self.textView.textStorage().length(), 0) 
         
         
         self.textView.replaceCharactersInRange_withString_(
             textRange, string)
             
         for r in self.blockRanges.itervalues():
             r.update_ranges_for_insertion(string, textRange)
         
         self.textView.setSelectedRange_(textRange)
         if(scrollToVisible):
             self.textView.scrollRangeToVisible_(textRange)
     
     
     
     def replace_current_block_with_string(self, textView, string):
         textView.replaceCharactersInRange_withString_(
                                     self.current_block_range().inputRange,
                                     string)
         self.current_block_range().inputRange.length = len(string)
         r = NSMakeRange(textView.textStorage().length(), 0)
         textView.scrollRangeToVisible_(r)
         textView.setSelectedRange_(r)
     
     
     def current_indent_string(self):
         """returns string for indent or None if no indent"""
         
         return self._indent_for_block(self.current_block())
     
     
     def _indent_for_block(self, block):
         lines = block.split('\n')
         if(len(lines) > 1):
             currentIndent = len(lines[-1]) - len(lines[-1].lstrip())
             if(currentIndent == 0):
                 currentIndent = self.tabSpaces
         
             if(self.tabUsesSpaces):
                 result = ' ' * currentIndent
             else:
                 result = '\t' * (currentIndent/self.tabSpaces)
         else:
             result = None
         
         return result
     
     
     # NSTextView delegate methods...
     def textView_doCommandBySelector_(self, textView, selector):
         assert(textView == self.textView)
         NSLog("textView_doCommandBySelector_: "+selector)
         
         
         if(selector == 'insertNewline:'):
             indent = self.current_indent_string()
             if(indent):
                 line = indent + self.current_line()
             else:
                 line = self.current_line()
             
             if(self.is_complete(self.current_block())):
                 self.execute(self.current_block(),
                                 blockID=self.currentBlockID)
                 self.start_new_block()
                 
                 return True
             
             return False
         
         elif(selector == 'moveUp:'):
             prevBlock = self.get_history_previous(self.current_block())
             if(prevBlock != None):
                 self.replace_current_block_with_string(textView, prevBlock)
             else:
                 NSBeep()
             return True
         
         elif(selector == 'moveDown:'):
             nextBlock = self.get_history_next()
             if(nextBlock != None):
                 self.replace_current_block_with_string(textView, nextBlock)
             else:
                 NSBeep()
             return True
         
         elif(selector == 'moveToBeginningOfParagraph:'):
             textView.setSelectedRange_(NSMakeRange(
                             self.current_block_range().inputRange.location, 
                             0))
             return True
         elif(selector == 'moveToEndOfParagraph:'):
             textView.setSelectedRange_(NSMakeRange(
                             self.current_block_range().inputRange.location + \
                             self.current_block_range().inputRange.length, 0))
             return True
         elif(selector == 'deleteToEndOfParagraph:'):
             if(textView.selectedRange().location <= \
                 self.current_block_range().location):
                 raise NotImplemented()
             
             return False # don't actually handle the delete
         
         elif(selector == 'insertTab:'):
             if(len(self.current_line().strip()) == 0): #only white space
                 return False
             else:
                 self.textView.complete_(self)
                 return True
         
         elif(selector == 'deleteBackward:'):
             #if we're at the beginning of the current block, ignore
             if(textView.selectedRange().location == \
                 self.current_block_range().inputRange.location):
                 return True
             else:
                 for r in self.blockRanges.itervalues():
                     deleteRange = textView.selectedRange
                     if(deleteRange.length == 0):
                         deleteRange.location -= 1
                         deleteRange.length = 1
                     r.update_ranges_for_deletion(deleteRange)
                 return False
         return False
     
     
     def textView_shouldChangeTextInRanges_replacementStrings_(self, 
         textView, ranges, replacementStrings):
         """
         Delegate method for NSTextView.
         
         Refuse change text in ranges not at end, but make those changes at 
         end.
         """
         
         assert(len(ranges) == len(replacementStrings))
         allow = True
         for r,s in zip(ranges, replacementStrings):
             r = r.rangeValue()
             if(textView.textStorage().length() > 0 and
                 r.location < self.current_block_range().inputRange.location):
                 self.insert_text(s)
                 allow = False
         
         return allow
     
     def textView_completions_forPartialWordRange_indexOfSelectedItem_(self, 
         textView, words, charRange, index):
         try:
             ts = textView.textStorage()
             token = ts.string().substringWithRange_(charRange)
             completions = blockingCallFromThread(self.complete, token)
         except:
             completions = objc.nil
             NSBeep()
         
         return (completions,0)
     
 

IPython/frontend/frontendbase.py

@@ -1,429 +1,343 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.frontend.tests.test_frontendbase -*-
 """
 frontendbase provides an interface and base class for GUI frontends for 
 IPython.kernel/IPython.kernel.core.
 
 Frontend implementations will likely want to subclass FrontEndBase. 
 
 Author: Barry Wark
 """
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 import string
-
-try:
-    import _ast
-except ImportError:
-    # Python 2.4 hackish workaround.
-    class bunch: pass
-    _ast = bunch()
-    _ast.PyCF_ONLY_AST = 1024
-    
-
-
-try:
-    import uuid
-except ImportError:
-    # Python 2.4 hackish workaround.
-    class UUID:
-        def __init__(self,bytes):
-            version = 4
-            int = long(('%02x'*16) % tuple(map(ord, bytes)), 16)
-            # Set the variant to RFC 4122.
-            int &= ~(0xc000 << 48L)
-            int |= 0x8000 << 48L
-            # Set the version number.
-            int &= ~(0xf000 << 64L)
-            int |= version << 76L
-            self.__dict__['int'] = int
-
-        def __cmp__(self, other):
-            if isinstance(other, UUID):
-                return cmp(self.int, other.int)
-            return NotImplemented
-
-        def __hash__(self):
-            return hash(self.int)
-
-        def __int__(self):
-            return self.int
-
-        def __repr__(self):
-            return 'UUID(%r)' % str(self)
-
-        def __setattr__(self, name, value):
-            raise TypeError('UUID objects are immutable')
-
-        def __str__(self):
-            hex = '%032x' % self.int
-            return '%s-%s-%s-%s-%s' % (
-                hex[:8], hex[8:12], hex[12:16], hex[16:20], hex[20:])
-
-        def get_bytes(self):
-            bytes = ''
-            for shift in range(0, 128, 8):
-                bytes = chr((self.int >> shift) & 0xff) + bytes
-            return bytes
-
-        bytes = property(get_bytes)
- 
-    
-    def _u4():
-        "Fake random uuid"
-        
-        import random
-        bytes = [chr(random.randrange(256)) for i in range(16)]
-        return UUID(bytes)
-
-    class bunch: pass
-    uuid = bunch()
-    uuid.uuid4 = _u4
-    del _u4
-
+import codeop
+from IPython.external import guid
 
 
 from IPython.frontend.zopeinterface import (
     Interface, 
     Attribute, 
 )
 from IPython.kernel.core.history import FrontEndHistory
 from IPython.kernel.core.util import Bunch
 
 ##############################################################################
 # TEMPORARY!!! fake configuration, while we decide whether to use tconfig or
 # not
 
 rc = Bunch()
 rc.prompt_in1 = r'In [$number]:  '
 rc.prompt_in2 = r'...'
 rc.prompt_out = r'Out [$number]:  '
 
 ##############################################################################
 # Interface definitions
 ##############################################################################
 
 class IFrontEndFactory(Interface):
     """Factory interface for frontends."""
     
     def __call__(engine=None, history=None):
         """
         Parameters:
         interpreter : IPython.kernel.engineservice.IEngineCore
         """
         
         pass
 
 
 class IFrontEnd(Interface):
     """Interface for frontends. All methods return t.i.d.Deferred"""
     
     Attribute("input_prompt_template", "string.Template instance\
                 substituteable with execute result.")
     Attribute("output_prompt_template", "string.Template instance\
                 substituteable with execute result.")
     Attribute("continuation_prompt_template", "string.Template instance\
                 substituteable with execute result.")
     
     def update_cell_prompt(result, blockID=None):
         """Subclass may override to update the input prompt for a block. 
 
         In asynchronous frontends, this method will be called as a
         twisted.internet.defer.Deferred's callback/errback.
         Implementations should thus return result when finished.
         
         Result is a result dict in case of success, and a
         twisted.python.util.failure.Failure in case of an error
         """
         
         pass
     
     def render_result(result):
         """Render the result of an execute call. Implementors may choose the
          method of rendering.
         For example, a notebook-style frontend might render a Chaco plot 
         inline.
         
         Parameters:
             result : dict (result of IEngineBase.execute )
                 blockID = result['blockID']
         
         Result:
             Output of frontend rendering
         """
         
         pass
     
     def render_error(failure):
         """Subclasses must override to render the failure. 
         
         In asynchronous frontend, since this method will be called as a
         twisted.internet.defer.Deferred's callback. Implementations
         should thus return result when finished.
         
         blockID = failure.blockID
         """
         
         pass
     
     def input_prompt(number=''):
         """Returns the input prompt by subsituting into 
         self.input_prompt_template
         """
         pass
     
     def output_prompt(number=''):
         """Returns the output prompt by subsituting into 
         self.output_prompt_template
         """
         
         pass
     
     def continuation_prompt():
         """Returns the continuation prompt by subsituting into 
         self.continuation_prompt_template
         """
         
         pass
     
     def is_complete(block):
         """Returns True if block is complete, False otherwise."""
         
         pass
     
-    def compile_ast(block):
-        """Compiles block to an _ast.AST"""
-        
-        pass
     
     def get_history_previous(current_block):
         """Returns the block previous in  the history. Saves currentBlock if
         the history_cursor is currently at the end of the input history"""
         pass
     
     def get_history_next():
         """Returns the next block in the history."""
         
         pass
     
     def complete(self, line):
         """Returns the list of possible completions, and the completed
             line.
 
         The input argument is the full line to be completed. This method
         returns both the line completed as much as possible, and the list
         of further possible completions (full words).
         """
         pass
 
 
 ##############################################################################
 # Base class for all the frontends. 
 ##############################################################################
 
 class FrontEndBase(object):
     """
     FrontEndBase manages the state tasks for a CLI frontend:
         - Input and output history management
         - Input/continuation and output prompt generation
         
     Some issues (due to possibly unavailable engine):
         - How do we get the current cell number for the engine?
         - How do we handle completions?
     """
     
     history_cursor = 0
     
     input_prompt_template = string.Template(rc.prompt_in1)
     output_prompt_template = string.Template(rc.prompt_out)
     continuation_prompt_template = string.Template(rc.prompt_in2)
     
     def __init__(self, shell=None, history=None):
         self.shell = shell
         if history is None:
                 self.history = FrontEndHistory(input_cache=[''])
         else:
             self.history = history
         
     
     def input_prompt(self, number=''):
         """Returns the current input prompt
         
         It would be great to use ipython1.core.prompts.Prompt1 here
         """
         return self.input_prompt_template.safe_substitute({'number':number})
     
     
     def continuation_prompt(self):
         """Returns the current continuation prompt"""
         
         return self.continuation_prompt_template.safe_substitute()
     
     def output_prompt(self, number=''):
         """Returns the output prompt for result"""
         
         return self.output_prompt_template.safe_substitute({'number':number})
     
     
     def is_complete(self, block):
         """Determine if block is complete.
         
         Parameters
         block : string
         
         Result 
         True if block can be sent to the engine without compile errors.
         False otherwise.
         """
         
         try:
-            ast = self.compile_ast(block)
+            is_complete = codeop.compile_command(block.rstrip() + '\n\n',
+                            "<string>", "exec") 
         except:
             return False
         
         lines = block.split('\n')
-        return (len(lines)==1 or str(lines[-1])=='')
-    
-    
-    def compile_ast(self, block):
-        """Compile block to an AST
-        
-        Parameters:
-            block : str
-        
-        Result:
-            AST
-        
-        Throws:
-            Exception if block cannot be compiled
-        """
-        
-        return compile(block, "<string>", "exec", _ast.PyCF_ONLY_AST)
+        return ((is_complete is not None) 
+                    and (len(lines)==1 or str(lines[-1])==''))
     
     
     def execute(self, block, blockID=None):
         """Execute the block and return the result.
         
         Parameters:
             block : {str, AST}
             blockID : any
                 Caller may provide an ID to identify this block. 
                 result['blockID'] := blockID
         
         Result:
             Deferred result of self.interpreter.execute
         """
         
         if(not self.is_complete(block)):
             raise Exception("Block is not compilable")
         
         if(blockID == None):
-            blockID = uuid.uuid4() #random UUID
+            blockID = guid.generate()
         
         try:
             result = self.shell.execute(block)
         except Exception,e:
             e = self._add_block_id_for_failure(e, blockID=blockID)
             e = self.update_cell_prompt(e, blockID=blockID)
             e = self.render_error(e)
         else:
             result = self._add_block_id_for_result(result, blockID=blockID)
             result = self.update_cell_prompt(result, blockID=blockID)
             result = self.render_result(result)
         
         return result
     
     
     def _add_block_id_for_result(self, result, blockID):
         """Add the blockID to result or failure. Unfortunatley, we have to 
         treat failures differently than result dicts.
         """
         
         result['blockID'] = blockID
         
         return result
     
     def _add_block_id_for_failure(self, failure, blockID):
         """_add_block_id_for_failure"""
         failure.blockID = blockID
         return failure
     
     
     def _add_history(self, result, block=None):
         """Add block to the history"""
         
         assert(block != None)
         self.history.add_items([block])
         self.history_cursor += 1
         
         return result
     
     
     def get_history_previous(self, current_block):
         """ Returns previous history string and decrement history cursor.
         """
         command = self.history.get_history_item(self.history_cursor - 1)
         
         if command is not None:
             if(self.history_cursor+1 == len(self.history.input_cache)):
                 self.history.input_cache[self.history_cursor] = current_block
             self.history_cursor -= 1
         return command
     
     
     def get_history_next(self):
         """ Returns next history string and increment history cursor.
         """
         command = self.history.get_history_item(self.history_cursor+1)
         
         if command is not None:
             self.history_cursor += 1
         return command
     
     ###
     # Subclasses probably want to override these methods...
     ###
     
     def update_cell_prompt(self, result, blockID=None):
         """Subclass may override to update the input prompt for a block. 
 
         This method only really makes sens in asyncrhonous frontend.
         Since this method will be called as a 
         twisted.internet.defer.Deferred's callback, implementations should 
         return result when finished.
         """
         
         raise NotImplementedError
     
     
     def render_result(self, result):
         """Subclasses must override to render result. 
         
         In asynchronous frontends, this method will be called as a
         twisted.internet.defer.Deferred's callback. Implementations
         should thus return result when finished.
         """
         
         raise NotImplementedError
     
     
     def render_error(self, failure):
         """Subclasses must override to render the failure. 
         
         In asynchronous frontends, this method will be called as a
         twisted.internet.defer.Deferred's callback. Implementations
         should thus return result when finished.
         """
         
         raise NotImplementedError
 

IPython/frontend/tests/test_frontendbase.py

@@ -1,155 +1,32 b''
 # encoding: utf-8
-
-"""This file contains unittests for the frontendbase module."""
+"""
+Test the basic functionality of frontendbase. 
+"""
 
 __docformat__ = "restructuredtext en"
 
-#---------------------------------------------------------------------------
+#-------------------------------------------------------------------------------
 #  Copyright (C) 2008  The IPython Development Team                         
 #                                                                           
-#  Distributed under the terms of the BSD License.  The full license is in  
-#  the file COPYING, distributed as part of this software.                  
-#---------------------------------------------------------------------------
-                                                                            
-#---------------------------------------------------------------------------
-# Imports                                                                   
-#---------------------------------------------------------------------------
-
-import unittest
-
-try:
-    from IPython.frontend.asyncfrontendbase import AsyncFrontEndBase
-    from IPython.frontend import frontendbase 
-    from IPython.kernel.engineservice import EngineService
-except ImportError:
-    import nose
-    raise nose.SkipTest("This test requires zope.interface, Twisted and Foolscap")
-
-from IPython.testing.decorators import skip
-
-class FrontEndCallbackChecker(AsyncFrontEndBase):
-    """FrontEndBase subclass for checking callbacks"""
-    def __init__(self, engine=None, history=None):
-        super(FrontEndCallbackChecker, self).__init__(engine=engine, 
-                                                    history=history)
-        self.updateCalled = False
-        self.renderResultCalled = False
-        self.renderErrorCalled = False
-   
-    def update_cell_prompt(self, result, blockID=None):
-        self.updateCalled = True
-        return result
-   
-    def render_result(self, result):
-        self.renderResultCalled = True
-        return result
-    
-    
-    def render_error(self, failure):
-        self.renderErrorCalled = True
-        return failure
-    
-
-
-    
-class TestAsyncFrontendBase(unittest.TestCase):
-    def setUp(self):
-        """Setup the EngineService and FrontEndBase"""
-        
-        self.fb = FrontEndCallbackChecker(engine=EngineService())
-    
-    def test_implements_IFrontEnd(self):
-        assert(frontendbase.IFrontEnd.implementedBy(
-                                    AsyncFrontEndBase))
-    
-    def test_is_complete_returns_False_for_incomplete_block(self):
-        """"""
-        
-        block = """def test(a):"""
-        
-        assert(self.fb.is_complete(block) == False)
-    
-    def test_is_complete_returns_True_for_complete_block(self):
-        """"""
-        
-        block = """def test(a): pass"""
-            
-        assert(self.fb.is_complete(block))
-        
-        block = """a=3"""
+#  Distributed under the terms of the BSD License.  The full license is
+#  in the file COPYING, distributed as part of this software.
+#-------------------------------------------------------------------------------
         
-        assert(self.fb.is_complete(block))
+from IPython.frontend.frontendbase import FrontEndBase
     
-    def test_blockID_added_to_result(self):
-        block = """3+3"""
-        
-        d = self.fb.execute(block, blockID='TEST_ID')
-        
-        d.addCallback(self.checkBlockID, expected='TEST_ID')
-    
-    def test_blockID_added_to_failure(self):
-        block = "raise Exception()"
-        
-        d = self.fb.execute(block,blockID='TEST_ID')
-        d.addErrback(self.checkFailureID, expected='TEST_ID')
-    
-    def checkBlockID(self, result, expected=""):
-        assert(result['blockID'] == expected)
-    
-    
-    def checkFailureID(self, failure, expected=""):
-        assert(failure.blockID == expected)
-    
-    
-    def test_callbacks_added_to_execute(self):
-        """test that
-        update_cell_prompt
-        render_result
-        
-        are added to execute request
+def test_iscomplete():
+    """ Check that is_complete works. 
         """
-        
-        d = self.fb.execute("10+10")
-        d.addCallback(self.checkCallbacks)
-    
-    def checkCallbacks(self, result):
-        assert(self.fb.updateCalled)
-        assert(self.fb.renderResultCalled)
-    
-    @skip("This test fails and lead to an unhandled error in a Deferred.")
-    def test_error_callback_added_to_execute(self):
-        """test that render_error called on execution error"""
-        
-        d = self.fb.execute("raise Exception()")
-        d.addCallback(self.checkRenderError)
-    
-    def checkRenderError(self, result):
-        assert(self.fb.renderErrorCalled)
-    
-    def test_history_returns_expected_block(self):
-        """Make sure history browsing doesn't fail"""
-        
-        blocks = ["a=1","a=2","a=3"]
-        for b in blocks:
-            d = self.fb.execute(b)
-        
-        # d is now the deferred for the last executed block
-        d.addCallback(self.historyTests, blocks)
-        
-    
-    def historyTests(self, result, blocks):
-        """historyTests"""
-        
-        assert(len(blocks) >= 3)
-        assert(self.fb.get_history_previous("") == blocks[-2])
-        assert(self.fb.get_history_previous("") == blocks[-3])
-        assert(self.fb.get_history_next() == blocks[-2])
-    
-    
-    def test_history_returns_none_at_startup(self):
-        """test_history_returns_none_at_startup"""
-        
-        assert(self.fb.get_history_previous("")==None)
-        assert(self.fb.get_history_next()==None)
-    
+    f = FrontEndBase()
+    assert f.is_complete('(a + a)')
+    assert not f.is_complete('(a + a')
+    assert f.is_complete('1')
+    assert not f.is_complete('1 + ')
+    assert not f.is_complete('1 + \n\n')
+    assert f.is_complete('if True:\n  print 1\n')
+    assert not f.is_complete('if True:\n  print 1')
+    assert f.is_complete('def f():\n  print 1\n')
+
+if __name__ == '__main__':
+    test_iscomplete()