fixes and tests for history

IPython/frontend/cocoa/cocoa_frontend.py

@@ -1,389 +1,389 b''
 # encoding: utf-8
 # -*- test-case-name: ipython1.frontend.cocoa.tests.test_cocoa_frontend -*-
 
 """PyObjC classes to provide a Cocoa frontend to the ipython1.kernel.engineservice.EngineService.
 
 The Cocoa frontend is divided into two classes:
     - IPythonCocoaController
     - IPythonCLITextViewDelegate
 
 To add an IPython interpreter to a cocoa app, instantiate both of these classes in an XIB...[FINISH]
 """
 
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #       Copyright (C) 2008  Barry Wark <[e-mail]>
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 import objc
 import uuid
 
 from Foundation import NSObject, NSMutableArray, NSMutableDictionary,\
                         NSLog, NSNotificationCenter, NSMakeRange,\
                         NSLocalizedString, NSIntersectionRange
                         
 from AppKit import NSApplicationWillTerminateNotification, NSBeep,\
                     NSTextView, NSRulerView, NSVerticalRuler
 
 from pprint import saferepr
 
 import IPython
 from IPython.kernel.engineservice import EngineService, ThreadedEngineService
 from IPython.frontend.frontendbase import FrontEndBase
 
 from twisted.internet.threads import blockingCallFromThread
 
 #-------------------------------------------------------------------------------
 # 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 IPythonCocoaController(NSObject, FrontEndBase):
     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()
         FrontEndBase.__init__(self, engine=ThreadedEngineService())
         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.nextBlockID()
         self.blockRanges = {} # blockID=>NSRange
     
     
     def awakeFromNib(self):
         """awakeFromNib"""
         
         self._common_init()
         
         # Start the IPython engine
         self.engine.startService()
         NSLog('IPython engine started')
         
         # Register for app termination
         NSNotificationCenter.defaultCenter().addObserver_selector_name_object_(self,
                                                 'appWillTerminate:',
                                                  NSApplicationWillTerminateNotification,
                                                   None)
         
         self.textView.setDelegate_(self)
         self.textView.enclosingScrollView().setHasVerticalRuler_(True)
         self.verticalRulerView = NSRulerView.alloc().initWithScrollView_orientation_(
             self.textView.enclosingScrollView(),
             NSVerticalRuler)
         self.verticalRulerView.setClientView_(self.textView)
         self.startCLIForTextView()
     
     
     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 ipython1.kernel.engineservice.IEngineInteractive.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._engineDone)
         d.addCallback(self._updateUserNS)
         
         return d
     
         
     def _engineDone(self, x):
         self.waitingForEngine = False
         self.didChangeValueForKey_('commandHistory')
         return x
     
     def _updateUserNS(self, result):
         """Update self.userNS from self.engine's namespace"""
         d = self.engine.keys()
         d.addCallback(self._getEngineNamepsaceValuesForKeys)
         
         return result
     
     
     def _getEngineNamepsaceValuesForKeys(self, keys):
         d = self.engine.pull(keys)
         d.addCallback(self._storeEngineNamespaceValues, keys=keys)
     
     
     def _storeEngineNamespaceValues(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 startCLIForTextView(self):
         """Print banner"""
         
         banner = """IPython1 %s -- An enhanced Interactive Python.""" % IPython.__version__
         
         self.insert_text(banner + '\n\n')
     
     # NSTextView/IPythonTextView delegate methods
     def textView_doCommandBySelector_(self, textView, selector):
         assert(textView == self.textView)
         NSLog("textView_doCommandBySelector_: "+selector)
         
         
         if(selector == 'insertNewline:'):
             indent = self.currentIndentString()
             if(indent):
                 line = indent + self.currentLine()
             else:
                 line = self.currentLine()
             
             if(self.is_complete(self.currentBlock())):
                 self.execute(self.currentBlock(),
                                 blockID=self.currentBlockID)
                 self.startNewBlock()
                 
                 return True
             
             return False
         
         elif(selector == 'moveUp:'):
             prevBlock = self.get_history_previous(self.currentBlock())
             if(prevBlock != None):
                 self.replaceCurrentBlockWithString(textView, prevBlock)
             else:
                 NSBeep()
             return True
         
         elif(selector == 'moveDown:'):
-            nextBlock = self.get_history_next(self.currentBlock())
+            nextBlock = self.get_history_next()
             if(nextBlock != None):
                 self.replaceCurrentBlockWithString(textView, nextBlock)
             else:
                 NSBeep()
             return True
         
         elif(selector == 'moveToBeginningOfParagraph:'):
             textView.setSelectedRange_(NSMakeRange(self.currentBlockRange().location, 0))
             return True
         elif(selector == 'moveToEndOfParagraph:'):
             textView.setSelectedRange_(NSMakeRange(self.currentBlockRange().location + \
                                                 self.currentBlockRange().length, 0))
             return True
         elif(selector == 'deleteToEndOfParagraph:'):
             if(textView.selectedRange().location <= self.currentBlockRange().location):
                 # Intersect the selected range with the current line range
                 if(self.currentBlockRange().length < 0):
                     self.blockRanges[self.currentBlockID].length = 0
             
                 r = NSIntersectionRange(textView.rangesForUserTextChange()[0],
                                         self.currentBlockRange())
                 
                 if(r.length > 0): #no intersection
                     textView.setSelectedRange_(r)
             
             return False # don't actually handle the delete
         
         elif(selector == 'insertTab:'):
             if(len(self.currentLine().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.currentBlockRange().location):
                 return True
             else:
                 self.currentBlockRange().length-=1
                 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.
         """
         
         #print 'textView_shouldChangeTextInRanges_replacementStrings_:',ranges,replacementStrings
         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.currentBlockRange().location):
                 self.insert_text(s)
                 allow = False
             
             
             self.blockRanges.setdefault(self.currentBlockID, self.currentBlockRange()).length += len(s)
         
         return allow
     
     def textView_completions_forPartialWordRange_indexOfSelectedItem_(self, textView, words, charRange, index):
         try:
             token = textView.textStorage().string().substringWithRange_(charRange)
             completions = blockingCallFromThread(self.complete, token)
         except:
             completions = objc.nil
             NSBeep()
         
         return (completions,0)
     
     
     def startNewBlock(self):
         """"""
         
         self.currentBlockID = self.nextBlockID()
     
     
     
     def nextBlockID(self):
         
         return uuid.uuid4()
     
     def currentBlockRange(self):
         return self.blockRanges.get(self.currentBlockID, NSMakeRange(self.textView.textStorage().length(), 0))
     
     def currentBlock(self):
         """The current block's text"""
         
         return self.textForRange(self.currentBlockRange())
     
     def textForRange(self, textRange):
         """textForRange"""
         
         return self.textView.textStorage().string().substringWithRange_(textRange)
     
     def currentLine(self):
         block = self.textForRange(self.currentBlockRange())
         block = block.split('\n')
         return block[-1]
     
     def update_cell_prompt(self, result):
         blockID = result['blockID']
         self.insert_text(self.inputPrompt(result=result),
                         textRange=NSMakeRange(self.blockRanges[blockID].location,0),
                         scrollToVisible=False
                         )
         
         return result
     
     
     def render_result(self, result):
         blockID = result['blockID']
         inputRange = self.blockRanges[blockID]
         del self.blockRanges[blockID]
         
         #print inputRange,self.currentBlockRange()
         self.insert_text('\n' +
                         self.outputPrompt(result) +
                         result.get('display',{}).get('pprint','') +
                         '\n\n',
                         textRange=NSMakeRange(inputRange.location+inputRange.length, 0))
         return result
     
         
     def render_error(self, failure):
         self.insert_text('\n\n'+str(failure)+'\n\n')
         self.startNewBlock()
         return failure
     
     
     def insert_text(self, string=None, textRange=None, scrollToVisible=True):
         """Insert text into textView at textRange, updating blockRanges as necessary"""
         
         if(textRange == None):
             textRange = NSMakeRange(self.textView.textStorage().length(), 0) #range for end of text
         
         for r in self.blockRanges.itervalues():
             intersection = NSIntersectionRange(r,textRange)
             if(intersection.length == 0): #ranges don't intersect
                 if r.location >= textRange.location:
                     r.location += len(string)
             else: #ranges intersect
                 if(r.location <= textRange.location):
                     assert(intersection.length == textRange.length)
                     r.length += textRange.length
                 else:
                     r.location += intersection.length
         
         self.textView.replaceCharactersInRange_withString_(textRange, string) #textStorage().string()
         self.textView.setSelectedRange_(NSMakeRange(textRange.location+len(string), 0))
         if(scrollToVisible):
             self.textView.scrollRangeToVisible_(textRange)
         
     
     
     def replaceCurrentBlockWithString(self, textView, string):
         textView.replaceCharactersInRange_withString_(self.currentBlockRange(),
                                                         string)
         self.currentBlockRange().length = len(string)
         r = NSMakeRange(textView.textStorage().length(), 0)
         textView.scrollRangeToVisible_(r)
         textView.setSelectedRange_(r)
     
     
     def currentIndentString(self):
         """returns string for indent or None if no indent"""
         
         if(len(self.currentBlock()) > 0):
             lines = self.currentBlock().split('\n')
             currentIndent = len(lines[-1]) - len(lines[-1])
             if(currentIndent == 0):
                 currentIndent = self.tabSpaces
         
             if(self.tabUsesSpaces):
                 result = ' ' * currentIndent
             else:
                 result = '\t' * (currentIndent/self.tabSpaces)
         else:
             result = None
         
         return result
     
 

IPython/frontend/frontendbase.py

@@ -1,310 +1,311 b''
 # encoding: utf-8
 """
 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
 import uuid
 import _ast
 
 import zope.interface as zi
 
 from IPython.kernel.core.history import FrontEndHistory
 from IPython.kernel.core.util import Bunch
 from IPython.kernel.engineservice import IEngineCore
 
 from twisted.python.failure import Failure
 
 ##############################################################################
 # 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]:  '
 
 ##############################################################################
 
 class IFrontEndFactory(zi.Interface):
     """Factory interface for frontends."""
     
     def __call__(engine=None, history=None):
         """
         Parameters:
         interpreter : IPython.kernel.engineservice.IEngineCore
         """
         
         pass
 
 
 
 class IFrontEnd(zi.Interface):
     """Interface for frontends. All methods return t.i.d.Deferred"""
     
     zi.Attribute("input_prompt_template", "string.Template instance substituteable with execute result.")
     zi.Attribute("output_prompt_template", "string.Template instance substituteable with execute result.")
     zi.Attribute("continuation_prompt_template", "string.Template instance substituteable with execute result.")
     
     def update_cell_prompt(self, result):
         """Subclass may override to update the input prompt for a block. 
         Since this method will be called as a twisted.internet.defer.Deferred's callback,
         implementations should return result when finished."""
         
         pass
     
     def render_result(self, 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 )
         
         Result:
             Output of frontend rendering
         """
         
         pass
     
     def render_error(self, failure):
         """Subclasses must override to render the failure. Since this method will be called as a 
         twisted.internet.defer.Deferred's callback, implementations should return result 
         when finished."""
         
         pass
     
     
     def inputPrompt(result={}):
         """Returns the input prompt by subsituting into self.input_prompt_template"""
         pass
     
     def outputPrompt(result):
         """Returns the output prompt by subsituting into self.output_prompt_template"""
         
         pass
     
     def continuationPrompt():
         """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(currentBlock):
-        """Returns the block previous in  the history."""
+        """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(currentBlock):
+    def get_history_next():
         """Returns the next block in the history."""
         
         pass
     
 
 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?
     """
     
     zi.implements(IFrontEnd)
     zi.classProvides(IFrontEndFactory)
     
     history_cursor = 0
     
     current_indent_level = 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, 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 inputPrompt(self, result={}):
         """Returns the current input prompt
         
         It would be great to use ipython1.core.prompts.Prompt1 here
         """
         
         result.setdefault('number','')
         
         return self.input_prompt_template.safe_substitute(result)
     
     
     def continuationPrompt(self):
         """Returns the current continuation prompt"""
         
         return self.continuation_prompt_template.safe_substitute()
     
     def outputPrompt(self, result):
         """Returns the output prompt for result"""
         
         return self.output_prompt_template.safe_substitute(result)
     
     
     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)
         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)
     
     
     def execute(self, block, blockID=None):
         """Execute the block and return 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
         
         d = self.engine.execute(block)
         d.addCallback(self._add_block_id, blockID)
         d.addCallback(self._add_history, block=block)
         d.addCallback(self.update_cell_prompt)
         d.addCallbacks(self.render_result, errback=self.render_error)
         
         return d
     
     
     def _add_block_id(self, result, blockID):
         """Add the blockID to result"""
         
         result['blockID'] = blockID
         
         return result
     
     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, currentBlock):
         """ Returns previous history string and decrement history cursor.
         """
-        print self.history
         command = self.history.get_history_item(self.history_cursor - 1)
-        print command
+        
         if command is not None:
-            self.history.input_cache[self.history_cursor] = currentBlock
+            if(self.history_cursor == len(self.history.input_cache)):
+                self.history.input_cache[self.history_cursor] = currentBlock
             self.history_cursor -= 1
         return command
     
     
-    def get_history_next(self, currentBlock):
+    def get_history_next(self):
         """ Returns next history string and increment history cursor.
         """
-        command = self.history.get_history_item(self.history_cursor + 1)
+        command = self.history.get_history_item(self.history_cursor+1)
+        
         if command is not None:
-            self.history.input_cache[self.history_cursor] = currentBlock
             self.history_cursor += 1
         return command
     
     ###
     # Subclasses probably want to override these methods...
     ###
     
     def update_cell_prompt(self, result):
         """Subclass may override to update the input prompt for a block. 
         Since this method will be called as a twisted.internet.defer.Deferred's callback,
         implementations should return result when finished."""
         
         return result
     
     
     def render_result(self, result):
         """Subclasses must override to render result. Since this method will be called as a 
         twisted.internet.defer.Deferred's callback, implementations should return result 
         when finished."""
         
         return result
     
     
     def render_error(self, failure):
         """Subclasses must override to render the failure. Since this method will be called as a 
         twisted.internet.defer.Deferred's callback, implementations should return result 
         when finished."""
         
         return failure
     
 
 

IPython/frontend/tests/test_frontendbase.py

@@ -1,114 +1,139 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
 from IPython.frontend import frontendbase
 from IPython.kernel.engineservice import EngineService
 
 class FrontEndCallbackChecker(frontendbase.FrontEndBase):
     """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):
         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 TestFrontendBase(unittest.TestCase):
     def setUp(self):
         """Setup the EngineService and FrontEndBase"""
         
         self.fb = FrontEndCallbackChecker(engine=EngineService())
     
     
     def test_implementsIFrontEnd(self):
         assert(frontendbase.IFrontEnd.implementedBy(frontendbase.FrontEndBase))
     
     
     def test_is_completeReturnsFalseForIncompleteBlock(self):
         """"""
         
         block = """def test(a):"""
         
         assert(self.fb.is_complete(block) == False)
     
     def test_is_completeReturnsTrueForCompleteBlock(self):
         """"""
         
         block = """def test(a): pass"""
             
         assert(self.fb.is_complete(block))
         
         block = """a=3"""
         
         assert(self.fb.is_complete(block))
     
     
     def test_blockIDAddedToResult(self):
         block = """3+3"""
         
         d = self.fb.execute(block, blockID='TEST_ID')
         
         d.addCallback(self.checkBlockID, expected='TEST_ID')
     
     def checkBlockID(self, result, expected=""):
         assert(result['blockID'] == expected)
     
     
     def test_callbacksAddedToExecuteRequest(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)
     
     
     def test_errorCallbackAddedToExecuteRequest(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)
     
-    # TODO: add tests for history
+    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)
+