Allow IPython to run without sqlite3

The History will always appear empty, so history magics, etc. don't do anything.  Readline history for a single session works fine.

Test suite should now pass without sqlite

IPython/core/history.py

@@ -1,909 +1,948 b''
 """ History related magics and functionality """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010 The IPython Development Team.
 #
 #  Distributed under the terms of the BSD License.
 #
 #  The full license is in the file COPYING.txt, distributed with this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Stdlib imports
 import atexit
 import datetime
 import os
 import re
-import sqlite3
+try:
+    import sqlite3
+except ImportError:
+    sqlite3 = None
 import threading
 
 # Our own packages
 from IPython.config.configurable import Configurable
-
+from IPython.external.decorator import decorator
 from IPython.testing.skipdoctest import skip_doctest
 from IPython.utils import io
 from IPython.utils.path import locate_profile
 from IPython.utils.traitlets import Bool, Dict, Instance, Int, CInt, List, Unicode
 from IPython.utils.warn import warn
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
+class DummyDB(object):
+    """Dummy DB that will act as a black hole for history.
+    
+    Only used in the absence of sqlite"""
+    def execute(*args, **kwargs):
+        return []
+    
+    def commit(self, *args, **kwargs):
+        pass
+    
+    def __enter__(self, *args, **kwargs):
+        pass
+    
+    def __exit__(self, *args, **kwargs):
+        pass
+    
+@decorator
+def needs_sqlite(f,*a,**kw):
+    """return an empty list in the absence of sqlite"""
+    if sqlite3 is None:
+        return []
+    else:
+        return f(*a,**kw)
+
 class HistoryAccessor(Configurable):
     """Access the history database without adding to it.
     
     This is intended for use by standalone history tools. IPython shells use
     HistoryManager, below, which is a subclass of this."""
     # String holding the path to the history file
     hist_file = Unicode(config=True)
 
     # The SQLite database
-    db = Instance(sqlite3.Connection)
+    if sqlite3:
+        db = Instance(sqlite3.Connection)
+    else:
+        db = Instance(DummyDB)
     
     def __init__(self, profile='default', hist_file=u'', shell=None, config=None, **traits):
         """Create a new history accessor.
         
         Parameters
         ----------
         profile : str
           The name of the profile from which to open history.
         hist_file : str
           Path to an SQLite history database stored by IPython. If specified,
           hist_file overrides profile.
         shell :
           InteractiveShell object, for use by HistoryManager subclass
         config :
           Config object. hist_file can also be set through this.
         """
         # We need a pointer back to the shell for various tasks.
         super(HistoryAccessor, self).__init__(shell=shell, config=config,
                                               hist_file=hist_file, **traits)
 
         if self.hist_file == u'':
             # No one has set the hist_file, yet.
             self.hist_file = self._get_hist_file_name(profile)
 
+        if sqlite3 is None:
+            warn("IPython History requires SQLite, your history will not be saved\n")
+            self.db = DummyDB()
+            return
+        
         try:
             self.init_db()
         except sqlite3.DatabaseError:
             if os.path.isfile(self.hist_file):
                 # Try to move the file out of the way.
                 newpath = os.path.join(self.shell.profile_dir.location, "hist-corrupt.sqlite")
                 os.rename(self.hist_file, newpath)
                 print("ERROR! History file wasn't a valid SQLite database.",
                 "It was moved to %s" % newpath, "and a new file created.")
                 self.init_db()
             else:
                 # The hist_file is probably :memory: or something else.
                 raise
     
     def _get_hist_file_name(self, profile='default'):
         """Find the history file for the given profile name.
         
         This is overridden by the HistoryManager subclass, to use the shell's
         active profile.
         
         Parameters
         ----------
         profile : str
           The name of a profile which has a history file.
         """
         return os.path.join(locate_profile(profile), 'history.sqlite')
     
     def init_db(self):
         """Connect to the database, and create tables if necessary."""
         # use detect_types so that timestamps return datetime objects
         self.db = sqlite3.connect(self.hist_file, detect_types=sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES)
         self.db.execute("""CREATE TABLE IF NOT EXISTS sessions (session integer
                         primary key autoincrement, start timestamp,
                         end timestamp, num_cmds integer, remark text)""")
         self.db.execute("""CREATE TABLE IF NOT EXISTS history
                 (session integer, line integer, source text, source_raw text,
                 PRIMARY KEY (session, line))""")
         # Output history is optional, but ensure the table's there so it can be
         # enabled later.
         self.db.execute("""CREATE TABLE IF NOT EXISTS output_history
                         (session integer, line integer, output text,
                         PRIMARY KEY (session, line))""")
         self.db.commit()
 
     def writeout_cache(self):
         """Overridden by HistoryManager to dump the cache before certain
         database lookups."""
         pass
 
     ## -------------------------------
     ## Methods for retrieving history:
     ## -------------------------------
     def _run_sql(self, sql, params, raw=True, output=False):
         """Prepares and runs an SQL query for the history database.
 
         Parameters
         ----------
         sql : str
           Any filtering expressions to go after SELECT ... FROM ...
         params : tuple
           Parameters passed to the SQL query (to replace "?")
         raw, output : bool
           See :meth:`get_range`
 
         Returns
         -------
         Tuples as :meth:`get_range`
         """
         toget = 'source_raw' if raw else 'source'
         sqlfrom = "history"
         if output:
             sqlfrom = "history LEFT JOIN output_history USING (session, line)"
             toget = "history.%s, output_history.output" % toget
         cur = self.db.execute("SELECT session, line, %s FROM %s " %\
                                 (toget, sqlfrom) + sql, params)
         if output:    # Regroup into 3-tuples, and parse JSON
             return ((ses, lin, (inp, out)) for ses, lin, inp, out in cur)
         return cur
 
+    @needs_sqlite
     def get_session_info(self, session=0):
         """get info about a session
 
         Parameters
         ----------
 
         session : int
             Session number to retrieve. The current session is 0, and negative
             numbers count back from current session, so -1 is previous session.
 
         Returns
         -------
 
         (session_id [int], start [datetime], end [datetime], num_cmds [int], remark [unicode])
 
         Sessions that are running or did not exit cleanly will have `end=None`
         and `num_cmds=None`.
 
         """
 
         if session <= 0:
             session += self.session_number
 
         query = "SELECT * from sessions where session == ?"
         return self.db.execute(query, (session,)).fetchone()
 
     def get_tail(self, n=10, raw=True, output=False, include_latest=False):
         """Get the last n lines from the history database.
 
         Parameters
         ----------
         n : int
           The number of lines to get
         raw, output : bool
           See :meth:`get_range`
         include_latest : bool
           If False (default), n+1 lines are fetched, and the latest one
           is discarded. This is intended to be used where the function
           is called by a user command, which it should not return.
 
         Returns
         -------
         Tuples as :meth:`get_range`
         """
         self.writeout_cache()
         if not include_latest:
             n += 1
         cur = self._run_sql("ORDER BY session DESC, line DESC LIMIT ?",
                                 (n,), raw=raw, output=output)
         if not include_latest:
             return reversed(list(cur)[1:])
         return reversed(list(cur))
 
     def search(self, pattern="*", raw=True, search_raw=True,
                                                         output=False):
         """Search the database using unix glob-style matching (wildcards
         * and ?).
 
         Parameters
         ----------
         pattern : str
           The wildcarded pattern to match when searching
         search_raw : bool
           If True, search the raw input, otherwise, the parsed input
         raw, output : bool
           See :meth:`get_range`
 
         Returns
         -------
         Tuples as :meth:`get_range`
         """
         tosearch = "source_raw" if search_raw else "source"
         if output:
             tosearch = "history." + tosearch
         self.writeout_cache()
         return self._run_sql("WHERE %s GLOB ?" % tosearch, (pattern,),
                                     raw=raw, output=output)
     
     def get_range(self, session, start=1, stop=None, raw=True,output=False):
         """Retrieve input by session.
 
         Parameters
         ----------
         session : int
             Session number to retrieve.
         start : int
             First line to retrieve.
         stop : int
             End of line range (excluded from output itself). If None, retrieve
             to the end of the session.
         raw : bool
             If True, return untranslated input
         output : bool
             If True, attempt to include output. This will be 'real' Python
             objects for the current session, or text reprs from previous
             sessions if db_log_output was enabled at the time. Where no output
             is found, None is used.
 
         Returns
         -------
         An iterator over the desired lines. Each line is a 3-tuple, either
         (session, line, input) if output is False, or
         (session, line, (input, output)) if output is True.
         """
         if stop:
             lineclause = "line >= ? AND line < ?"
             params = (session, start, stop)
         else:
             lineclause = "line>=?"
             params = (session, start)
 
         return self._run_sql("WHERE session==? AND %s""" % lineclause,
                                     params, raw=raw, output=output)
 
     def get_range_by_str(self, rangestr, raw=True, output=False):
         """Get lines of history from a string of ranges, as used by magic
         commands %hist, %save, %macro, etc.
 
         Parameters
         ----------
         rangestr : str
           A string specifying ranges, e.g. "5 ~2/1-4". See
           :func:`magic_history` for full details.
         raw, output : bool
           As :meth:`get_range`
 
         Returns
         -------
         Tuples as :meth:`get_range`
         """
         for sess, s, e in extract_hist_ranges(rangestr):
             for line in self.get_range(sess, s, e, raw=raw, output=output):
                 yield line
 
 
 class HistoryManager(HistoryAccessor):
     """A class to organize all history-related functionality in one place.
     """
     # Public interface
 
     # An instance of the IPython shell we are attached to
     shell = Instance('IPython.core.interactiveshell.InteractiveShellABC')
     # Lists to hold processed and raw history. These start with a blank entry
     # so that we can index them starting from 1
     input_hist_parsed = List([""])
     input_hist_raw = List([""])
     # A list of directories visited during session
     dir_hist = List()
     def _dir_hist_default(self):
         try:
             return [os.getcwdu()]
         except OSError:
             return []
 
     # A dict of output history, keyed with ints from the shell's
     # execution count.
     output_hist = Dict()
     # The text/plain repr of outputs.
     output_hist_reprs = Dict()
 
     # The number of the current session in the history database
     session_number = CInt()
     # Should we log output to the database? (default no)
     db_log_output = Bool(False, config=True)
     # Write to database every x commands (higher values save disk access & power)
     #  Values of 1 or less effectively disable caching. 
     db_cache_size = Int(0, config=True)
     # The input and output caches
     db_input_cache = List()
     db_output_cache = List()
     
     # History saving in separate thread
     save_thread = Instance('IPython.core.history.HistorySavingThread')
     try:               # Event is a function returning an instance of _Event...
         save_flag = Instance(threading._Event)
     except AttributeError:         # ...until Python 3.3, when it's a class.
         save_flag = Instance(threading.Event)
     
     # Private interface
     # Variables used to store the three last inputs from the user.  On each new
     # history update, we populate the user's namespace with these, shifted as
     # necessary.
     _i00 = Unicode(u'')
     _i = Unicode(u'')
     _ii = Unicode(u'')
     _iii = Unicode(u'')
 
     # A regex matching all forms of the exit command, so that we don't store
     # them in the history (it's annoying to rewind the first entry and land on
     # an exit call).
     _exit_re = re.compile(r"(exit|quit)(\s*\(.*\))?$")
 
     def __init__(self, shell=None, config=None, **traits):
         """Create a new history manager associated with a shell instance.
         """
         # We need a pointer back to the shell for various tasks.
         super(HistoryManager, self).__init__(shell=shell, config=config,
             **traits)
         self.save_flag = threading.Event()
         self.db_input_cache_lock = threading.Lock()
         self.db_output_cache_lock = threading.Lock()
         self.save_thread = HistorySavingThread(self)
         self.save_thread.start()
 
         self.new_session()
-    
+
     def _get_hist_file_name(self, profile=None):
         """Get default history file name based on the Shell's profile.
         
         The profile parameter is ignored, but must exist for compatibility with
         the parent class."""
         profile_dir = self.shell.profile_dir.location
         return os.path.join(profile_dir, 'history.sqlite')
     
+    @needs_sqlite
     def new_session(self, conn=None):
         """Get a new session number."""
         if conn is None:
             conn = self.db
         
         with conn:
             cur = conn.execute("""INSERT INTO sessions VALUES (NULL, ?, NULL,
                             NULL, "") """, (datetime.datetime.now(),))
             self.session_number = cur.lastrowid
             
     def end_session(self):
         """Close the database session, filling in the end time and line count."""
         self.writeout_cache()
         with self.db:
             self.db.execute("""UPDATE sessions SET end=?, num_cmds=? WHERE
                             session==?""", (datetime.datetime.now(),
                             len(self.input_hist_parsed)-1, self.session_number))
         self.session_number = 0
                             
     def name_session(self, name):
         """Give the current session a name in the history database."""
         with self.db:
             self.db.execute("UPDATE sessions SET remark=? WHERE session==?",
                             (name, self.session_number))
                             
     def reset(self, new_session=True):
         """Clear the session history, releasing all object references, and
         optionally open a new session."""
         self.output_hist.clear()
         # The directory history can't be completely empty
         self.dir_hist[:] = [os.getcwdu()]
         
         if new_session:
             if self.session_number:
                 self.end_session()
             self.input_hist_parsed[:] = [""]
             self.input_hist_raw[:] = [""]
             self.new_session()
     
     # ------------------------------
     # Methods for retrieving history
     # ------------------------------
     def _get_range_session(self, start=1, stop=None, raw=True, output=False):
         """Get input and output history from the current session. Called by
         get_range, and takes similar parameters."""
         input_hist = self.input_hist_raw if raw else self.input_hist_parsed
             
         n = len(input_hist)
         if start < 0:
             start += n
         if not stop or (stop > n):
             stop = n
         elif stop < 0:
             stop += n
         
         for i in range(start, stop):
             if output:
                 line = (input_hist[i], self.output_hist_reprs.get(i))
             else:
                 line = input_hist[i]
             yield (0, i, line)
     
     def get_range(self, session=0, start=1, stop=None, raw=True,output=False):
         """Retrieve input by session.
         
         Parameters
         ----------
         session : int
             Session number to retrieve. The current session is 0, and negative
             numbers count back from current session, so -1 is previous session.
         start : int
             First line to retrieve.
         stop : int
             End of line range (excluded from output itself). If None, retrieve
             to the end of the session.
         raw : bool
             If True, return untranslated input
         output : bool
             If True, attempt to include output. This will be 'real' Python
             objects for the current session, or text reprs from previous
             sessions if db_log_output was enabled at the time. Where no output
             is found, None is used.
             
         Returns
         -------
         An iterator over the desired lines. Each line is a 3-tuple, either
         (session, line, input) if output is False, or
         (session, line, (input, output)) if output is True.
         """
         if session <= 0:
             session += self.session_number
         if session==self.session_number:          # Current session
             return self._get_range_session(start, stop, raw, output)
         return super(HistoryManager, self).get_range(session, start, stop, raw, output)
 
     ## ----------------------------
     ## Methods for storing history:
     ## ----------------------------
     def store_inputs(self, line_num, source, source_raw=None):
         """Store source and raw input in history and create input cache
         variables _i*.
 
         Parameters
         ----------
         line_num : int
           The prompt number of this input.
 
         source : str
           Python input.
 
         source_raw : str, optional
           If given, this is the raw input without any IPython transformations
           applied to it.  If not given, ``source`` is used.
         """
         if source_raw is None:
             source_raw = source
         source = source.rstrip('\n')
         source_raw = source_raw.rstrip('\n')
 
         # do not store exit/quit commands
         if self._exit_re.match(source_raw.strip()):
             return
 
         self.input_hist_parsed.append(source)
         self.input_hist_raw.append(source_raw)
 
         with self.db_input_cache_lock:
             self.db_input_cache.append((line_num, source, source_raw))
             # Trigger to flush cache and write to DB.
             if len(self.db_input_cache) >= self.db_cache_size:
                 self.save_flag.set()
 
         # update the auto _i variables
         self._iii = self._ii
         self._ii = self._i
         self._i = self._i00
         self._i00 = source_raw
 
         # hackish access to user namespace to create _i1,_i2... dynamically
         new_i = '_i%s' % line_num
         to_main = {'_i': self._i,
                    '_ii': self._ii,
                    '_iii': self._iii,
                    new_i : self._i00 }
         self.shell.user_ns.update(to_main)
 
     def store_output(self, line_num):
         """If database output logging is enabled, this saves all the
         outputs from the indicated prompt number to the database. It's
         called by run_cell after code has been executed.
 
         Parameters
         ----------
         line_num : int
           The line number from which to save outputs
         """
         if (not self.db_log_output) or (line_num not in self.output_hist_reprs):
             return
         output = self.output_hist_reprs[line_num]
 
         with self.db_output_cache_lock:
             self.db_output_cache.append((line_num, output))
         if self.db_cache_size <= 1:
             self.save_flag.set()
 
     def _writeout_input_cache(self, conn):
         with conn:
             for line in self.db_input_cache:
                 conn.execute("INSERT INTO history VALUES (?, ?, ?, ?)",
                                 (self.session_number,)+line)
 
     def _writeout_output_cache(self, conn):
         with conn:
             for line in self.db_output_cache:
                 conn.execute("INSERT INTO output_history VALUES (?, ?, ?)",
                                 (self.session_number,)+line)
 
+    @needs_sqlite
     def writeout_cache(self, conn=None):
         """Write any entries in the cache to the database."""
         if conn is None:
             conn = self.db
 
         with self.db_input_cache_lock:
             try:
                 self._writeout_input_cache(conn)
             except sqlite3.IntegrityError:
                 self.new_session(conn)
                 print("ERROR! Session/line number was not unique in",
                       "database. History logging moved to new session",
                                                 self.session_number)
                 try: # Try writing to the new session. If this fails, don't recurse
                     self._writeout_input_cache(conn)
                 except sqlite3.IntegrityError:
                     pass
             finally:
                 self.db_input_cache = []
 
         with self.db_output_cache_lock:
             try:
                 self._writeout_output_cache(conn)
             except sqlite3.IntegrityError:
                 print("!! Session/line number for output was not unique",
                       "in database. Output will not be stored.")
             finally:
                 self.db_output_cache = []
 
 
 class HistorySavingThread(threading.Thread):
     """This thread takes care of writing history to the database, so that
     the UI isn't held up while that happens.
 
     It waits for the HistoryManager's save_flag to be set, then writes out
     the history cache. The main thread is responsible for setting the flag when
     the cache size reaches a defined threshold."""
     daemon = True
     stop_now = False
     def __init__(self, history_manager):
         super(HistorySavingThread, self).__init__()
         self.history_manager = history_manager
         atexit.register(self.stop)
 
+    @needs_sqlite
     def run(self):
         # We need a separate db connection per thread:
         try:
             self.db = sqlite3.connect(self.history_manager.hist_file)
             while True:
                 self.history_manager.save_flag.wait()
                 if self.stop_now:
                     return
                 self.history_manager.save_flag.clear()
                 self.history_manager.writeout_cache(self.db)
         except Exception as e:
             print(("The history saving thread hit an unexpected error (%s)."
                    "History will not be written to the database.") % repr(e))
 
     def stop(self):
         """This can be called from the main thread to safely stop this thread.
 
         Note that it does not attempt to write out remaining history before
         exiting. That should be done by calling the HistoryManager's
         end_session method."""
         self.stop_now = True
         self.history_manager.save_flag.set()
         self.join()
 
 
 # To match, e.g. ~5/8-~2/3
 range_re = re.compile(r"""
 ((?P<startsess>~?\d+)/)?
 (?P<start>\d+)                    # Only the start line num is compulsory
 ((?P<sep>[\-:])
  ((?P<endsess>~?\d+)/)?
  (?P<end>\d+))?
 $""", re.VERBOSE)
 
 def extract_hist_ranges(ranges_str):
     """Turn a string of history ranges into 3-tuples of (session, start, stop).
 
     Examples
     --------
     list(extract_input_ranges("~8/5-~7/4 2"))
     [(-8, 5, None), (-7, 1, 4), (0, 2, 3)]
     """
     for range_str in ranges_str.split():
         rmatch = range_re.match(range_str)
         if not rmatch:
             continue
         start = int(rmatch.group("start"))
         end = rmatch.group("end")
         end = int(end) if end else start+1   # If no end specified, get (a, a+1)
         if rmatch.group("sep") == "-":       # 1-3 == 1:4 --> [1, 2, 3]
             end += 1
         startsess = rmatch.group("startsess") or "0"
         endsess = rmatch.group("endsess") or startsess
         startsess = int(startsess.replace("~","-"))
         endsess = int(endsess.replace("~","-"))
         assert endsess >= startsess
 
         if endsess == startsess:
             yield (startsess, start, end)
             continue
         # Multiple sessions in one range:
         yield (startsess, start, None)
         for sess in range(startsess+1, endsess):
             yield (sess, 1, None)
         yield (endsess, 1, end)
 
 def _format_lineno(session, line):
     """Helper function to format line numbers properly."""
     if session == 0:
         return str(line)
     return "%s#%s" % (session, line)
 
 @skip_doctest
 def magic_history(self, parameter_s = ''):
     """Print input history (_i<n> variables), with most recent last.
 
     %history       -> print at most 40 inputs (some may be multi-line)\\
     %history n     -> print at most n inputs\\
     %history n1 n2 -> print inputs between n1 and n2 (n2 not included)\\
 
     By default, input history is printed without line numbers so it can be
     directly pasted into an editor. Use -n to show them.
 
     Ranges of history can be indicated using the syntax:
     4      : Line 4, current session
     4-6    : Lines 4-6, current session
     243/1-5: Lines 1-5, session 243
     ~2/7   : Line 7, session 2 before current
     ~8/1-~6/5 : From the first line of 8 sessions ago, to the fifth line
                 of 6 sessions ago.
     Multiple ranges can be entered, separated by spaces
 
     The same syntax is used by %macro, %save, %edit, %rerun
 
     Options:
 
       -n: print line numbers for each input.
       This feature is only available if numbered prompts are in use.
 
       -o: also print outputs for each input.
 
       -p: print classic '>>>' python prompts before each input.  This is useful
        for making documentation, and in conjunction with -o, for producing
        doctest-ready output.
 
       -r: (default) print the 'raw' history, i.e. the actual commands you typed.
 
       -t: print the 'translated' history, as IPython understands it.  IPython
       filters your input and converts it all into valid Python source before
       executing it (things like magics or aliases are turned into function
       calls, for example). With this option, you'll see the native history
       instead of the user-entered version: '%cd /' will be seen as
       'get_ipython().magic("%cd /")' instead of '%cd /'.
 
       -g: treat the arg as a pattern to grep for in (full) history.
       This includes the saved history (almost all commands ever written).
       Use '%hist -g' to show full saved history (may be very long).
 
       -l: get the last n lines from all sessions. Specify n as a single arg, or
       the default is the last 10 lines.
 
       -f FILENAME: instead of printing the output to the screen, redirect it to
        the given file.  The file is always overwritten, though IPython asks for
        confirmation first if it already exists.
 
     Examples
     --------
     ::
 
       In [6]: %hist -n 4 6
       4:a = 12
       5:print a**2
 
     """
 
     if not self.shell.displayhook.do_full_cache:
         print('This feature is only available if numbered prompts are in use.')
         return
     opts,args = self.parse_options(parameter_s,'noprtglf:',mode='string')
 
     # For brevity
     history_manager = self.shell.history_manager
 
     def _format_lineno(session, line):
         """Helper function to format line numbers properly."""
         if session in (0, history_manager.session_number):
             return str(line)
         return "%s/%s" % (session, line)
 
     # Check if output to specific file was requested.
     try:
         outfname = opts['f']
     except KeyError:
         outfile = io.stdout  # default
         # We don't want to close stdout at the end!
         close_at_end = False
     else:
         if os.path.exists(outfname):
             if not io.ask_yes_no("File %r exists. Overwrite?" % outfname):
                 print('Aborting.')
                 return
 
         outfile = open(outfname,'w')
         close_at_end = True
 
     print_nums = 'n' in opts
     get_output = 'o' in opts
     pyprompts = 'p' in opts
     # Raw history is the default
     raw = not('t' in opts)
 
     default_length = 40
     pattern = None
 
     if 'g' in opts:         # Glob search
         pattern = "*" + args + "*" if args else "*"
         hist = history_manager.search(pattern, raw=raw, output=get_output)
         print_nums = True
     elif 'l' in opts:       # Get 'tail'
         try:
             n = int(args)
         except ValueError, IndexError:
             n = 10
         hist = history_manager.get_tail(n, raw=raw, output=get_output)
     else:
         if args:            # Get history by ranges
             hist = history_manager.get_range_by_str(args, raw, get_output)
         else:               # Just get history for the current session
             hist = history_manager.get_range(raw=raw, output=get_output)
 
     # We could be displaying the entire history, so let's not try to pull it
     # into a list in memory. Anything that needs more space will just misalign.
     width = 4
 
     for session, lineno, inline in hist:
         # Print user history with tabs expanded to 4 spaces.  The GUI clients
         # use hard tabs for easier usability in auto-indented code, but we want
         # to produce PEP-8 compliant history for safe pasting into an editor.
         if get_output:
             inline, output = inline
         inline = inline.expandtabs(4).rstrip()
 
         multiline = "\n" in inline
         line_sep = '\n' if multiline else ' '
         if print_nums:
             print('%s:%s' % (_format_lineno(session, lineno).rjust(width),
                     line_sep),  file=outfile, end='')
         if pyprompts:
             print(">>> ", end="", file=outfile)
             if multiline:
                 inline = "\n... ".join(inline.splitlines()) + "\n..."
         print(inline, file=outfile)
         if get_output and output:
             print(output, file=outfile)
 
     if close_at_end:
         outfile.close()
 
 
 def magic_rep(self, arg):
     r"""Repeat a command, or get command to input line for editing. %recall and
     %rep are equivalent.
 
     - %recall (no arguments):
 
     Place a string version of last computation result (stored in the special '_'
     variable) to the next input prompt. Allows you to create elaborate command
     lines without using copy-paste::
 
          In[1]: l = ["hei", "vaan"]
          In[2]: "".join(l)
         Out[2]: heivaan
          In[3]: %rep
          In[4]: heivaan_ <== cursor blinking
 
     %recall 45
 
     Place history line 45 on the next input prompt. Use %hist to find
     out the number.
 
     %recall 1-4
 
     Combine the specified lines into one cell, and place it on the next
     input prompt. See %history for the slice syntax.
 
     %recall foo+bar
 
     If foo+bar can be evaluated in the user namespace, the result is
     placed at the next input prompt. Otherwise, the history is searched
     for lines which contain that substring, and the most recent one is
     placed at the next input prompt.
     """
     if not arg:                 # Last output
         self.set_next_input(str(self.shell.user_ns["_"]))
         return
                                 # Get history range
     histlines = self.history_manager.get_range_by_str(arg)
     cmd = "\n".join(x[2] for x in histlines)
     if cmd:
         self.set_next_input(cmd.rstrip())
         return
 
     try:                        # Variable in user namespace
         cmd = str(eval(arg, self.shell.user_ns))
     except Exception:           # Search for term in history
         histlines = self.history_manager.search("*"+arg+"*")
         for h in reversed([x[2] for x in histlines]):
             if 'rep' in h:
                 continue
             self.set_next_input(h.rstrip())
             return
     else:
         self.set_next_input(cmd.rstrip())
     print("Couldn't evaluate or find in history:", arg)
 
 def magic_rerun(self, parameter_s=''):
     """Re-run previous input
 
     By default, you can specify ranges of input history to be repeated
     (as with %history). With no arguments, it will repeat the last line.
 
     Options:
 
       -l <n> : Repeat the last n lines of input, not including the
       current command.
 
       -g foo : Repeat the most recent line which contains foo
     """
     opts, args = self.parse_options(parameter_s, 'l:g:', mode='string')
     if "l" in opts:         # Last n lines
         n = int(opts['l'])
         hist = self.history_manager.get_tail(n)
     elif "g" in opts:       # Search
         p = "*"+opts['g']+"*"
         hist = list(self.history_manager.search(p))
         for l in reversed(hist):
             if "rerun" not in l[2]:
                 hist = [l]     # The last match which isn't a %rerun
                 break
         else:
             hist = []          # No matches except %rerun
     elif args:              # Specify history ranges
         hist = self.history_manager.get_range_by_str(args)
     else:                   # Last line
         hist = self.history_manager.get_tail(1)
     hist = [x[2] for x in hist]
     if not hist:
         print("No lines in history match specification")
         return
     histlines = "\n".join(hist)
     print("=== Executing: ===")
     print(histlines)
     print("=== Output: ===")
     self.run_cell("\n".join(hist), store_history=False)
 
 
 def init_ipython(ip):
     ip.define_magic("rep", magic_rep)
     ip.define_magic("recall", magic_rep)
     ip.define_magic("rerun", magic_rerun)
     ip.define_magic("hist",magic_history)    # Alternative name
     ip.define_magic("history",magic_history)
 
     # XXX - ipy_completers are in quarantine, need to be updated to new apis
     #import ipy_completers
     #ipy_completers.quick_completer('%hist' ,'-g -t -r -n')

IPython/core/tests/test_magic.py

@@ -1,452 +1,460 b''
 """Tests for various magic functions.
 
 Needs to be run by nose (to make ipython session available).
 """
 from __future__ import absolute_import
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import os
 import sys
 import tempfile
 import types
 from StringIO import StringIO
 
 import nose.tools as nt
 
 from IPython.utils.path import get_long_path_name
 from IPython.testing import decorators as dec
 from IPython.testing import tools as tt
 from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # Test functions begin
 #-----------------------------------------------------------------------------
 def test_rehashx():
     # clear up everything
     _ip = get_ipython()
     _ip.alias_manager.alias_table.clear()
     del _ip.db['syscmdlist']
     
     _ip.magic('rehashx')
     # Practically ALL ipython development systems will have more than 10 aliases
 
     yield (nt.assert_true, len(_ip.alias_manager.alias_table) > 10)
     for key, val in _ip.alias_manager.alias_table.iteritems():
         # we must strip dots from alias names
         nt.assert_true('.' not in key)
 
     # rehashx must fill up syscmdlist
     scoms = _ip.db['syscmdlist']
     yield (nt.assert_true, len(scoms) > 10)
 
 
 def test_magic_parse_options():
     """Test that we don't mangle paths when parsing magic options."""
     ip = get_ipython()
     path = 'c:\\x'
     opts = ip.parse_options('-f %s' % path,'f:')[0]
     # argv splitting is os-dependent
     if os.name == 'posix':
         expected = 'c:x'
     else:
         expected = path
     nt.assert_equals(opts['f'], expected)
 
-    
+
[e-mail]_without('sqlite3')
 def doctest_hist_f():
     """Test %hist -f with temporary filename.
 
     In [9]: import tempfile
 
     In [10]: tfile = tempfile.mktemp('.py','tmp-ipython-')
 
     In [11]: %hist -nl -f $tfile 3
 
     In [13]: import os; os.unlink(tfile)
     """
 
 
[e-mail]_without('sqlite3')
 def doctest_hist_r():
     """Test %hist -r
 
     XXX - This test is not recording the output correctly.  For some reason, in
     testing mode the raw history isn't getting populated.  No idea why.
     Disabling the output checking for now, though at least we do run it.
 
     In [1]: 'hist' in _ip.lsmagic()
     Out[1]: True
 
     In [2]: x=1
 
     In [3]: %hist -rl 2
     x=1 # random
     %hist -r 2
     """
 
+
[e-mail]_without('sqlite3')
 def doctest_hist_op():
     """Test %hist -op
 
     In [1]: class b(float):
        ...:     pass
        ...: 
 
     In [2]: class s(object):
        ...:     def __str__(self):
        ...:         return 's'
        ...: 
 
     In [3]: 
 
     In [4]: class r(b):
        ...:     def __repr__(self):
        ...:         return 'r'
        ...: 
 
     In [5]: class sr(s,r): pass
        ...: 
 
     In [6]: 
 
     In [7]: bb=b()
 
     In [8]: ss=s()
 
     In [9]: rr=r()
 
     In [10]: ssrr=sr()
 
     In [11]: 4.5
     Out[11]: 4.5
 
     In [12]: str(ss)
     Out[12]: 's'
 
     In [13]: 
 
     In [14]: %hist -op
     >>> class b:
     ...     pass
     ... 
     >>> class s(b):
     ...     def __str__(self):
     ...         return 's'
     ... 
     >>> 
     >>> class r(b):
     ...     def __repr__(self):
     ...         return 'r'
     ... 
     >>> class sr(s,r): pass
     >>> 
     >>> bb=b()
     >>> ss=s()
     >>> rr=r()
     >>> ssrr=sr()
     >>> 4.5
     4.5
     >>> str(ss)
     's'
     >>> 
     """
-  
+
+
[e-mail]_without('sqlite3')
 def test_macro():
     ip = get_ipython()
     ip.history_manager.reset()   # Clear any existing history.
     cmds = ["a=1", "def b():\n  return a**2", "print(a,b())"]
     for i, cmd in enumerate(cmds, start=1):
         ip.history_manager.store_inputs(i, cmd)
     ip.magic("macro test 1-3")
     nt.assert_equal(ip.user_ns["test"].value, "\n".join(cmds)+"\n")
     
     # List macros.
     assert "test" in ip.magic("macro")
 
+
[e-mail]_without('sqlite3')
 def test_macro_run():
     """Test that we can run a multi-line macro successfully."""
     ip = get_ipython()
     ip.history_manager.reset()
     cmds = ["a=10", "a+=1", py3compat.doctest_refactor_print("print a"),
                                                             "%macro test 2-3"]
     for cmd in cmds:
         ip.run_cell(cmd, store_history=True)
     nt.assert_equal(ip.user_ns["test"].value,
                             py3compat.doctest_refactor_print("a+=1\nprint a\n"))
     with tt.AssertPrints("12"):
         ip.run_cell("test")
     with tt.AssertPrints("13"):
         ip.run_cell("test")
 
     
 # XXX failing for now, until we get clearcmd out of quarantine.  But we should
 # fix this and revert the skip to happen only if numpy is not around.
 #@dec.skipif_not_numpy
 @dec.skip_known_failure
 def test_numpy_clear_array_undec():
     from IPython.extensions import clearcmd
 
     _ip.ex('import numpy as np')
     _ip.ex('a = np.empty(2)')
     yield (nt.assert_true, 'a' in _ip.user_ns)
     _ip.magic('clear array')
     yield (nt.assert_false, 'a' in _ip.user_ns)
     
 
 # Multiple tests for clipboard pasting
 @dec.parametric
 def test_paste():
     _ip = get_ipython()
     def paste(txt, flags='-q'):
         """Paste input text, by default in quiet mode"""
         hooks.clipboard_get = lambda : txt
         _ip.magic('paste '+flags)
 
     # Inject fake clipboard hook but save original so we can restore it later
     hooks = _ip.hooks
     user_ns = _ip.user_ns
     original_clip = hooks.clipboard_get
 
     try:
         # Run tests with fake clipboard function
         user_ns.pop('x', None)
         paste('x=1')
         yield nt.assert_equal(user_ns['x'], 1)
 
         user_ns.pop('x', None)
         paste('>>> x=2')
         yield nt.assert_equal(user_ns['x'], 2)
 
         paste("""
         >>> x = [1,2,3]
         >>> y = []
         >>> for i in x:
         ...     y.append(i**2)
         ...
         """)
         yield nt.assert_equal(user_ns['x'], [1,2,3])
         yield nt.assert_equal(user_ns['y'], [1,4,9])
 
         # Now, test that paste -r works
         user_ns.pop('x', None)
         yield nt.assert_false('x' in user_ns)
         _ip.magic('paste -r')
         yield nt.assert_equal(user_ns['x'], [1,2,3])
 
         # Also test paste echoing, by temporarily faking the writer
         w = StringIO()
         writer = _ip.write
         _ip.write = w.write
         code = """
         a = 100
         b = 200"""
         try:
             paste(code,'')
             out = w.getvalue()
         finally:
             _ip.write = writer
         yield nt.assert_equal(user_ns['a'], 100)
         yield nt.assert_equal(user_ns['b'], 200)
         yield nt.assert_equal(out, code+"\n## -- End pasted text --\n")
         
     finally:
         # Restore original hook
         hooks.clipboard_get = original_clip
 
 
 def test_time():
     _ip.magic('time None')
 
 
 @py3compat.doctest_refactor_print
 def doctest_time():
     """
     In [10]: %time None
     CPU times: user 0.00 s, sys: 0.00 s, total: 0.00 s
     Wall time: 0.00 s
     
     In [11]: def f(kmjy):
        ....:    %time print 2*kmjy
        
     In [12]: f(3)
     6
     CPU times: user 0.00 s, sys: 0.00 s, total: 0.00 s
     Wall time: 0.00 s
     """
 
 
 def test_doctest_mode():
     "Toggle doctest_mode twice, it should be a no-op and run without error"
     _ip.magic('doctest_mode')
     _ip.magic('doctest_mode')
 
 
 def test_parse_options():
     """Tests for basic options parsing in magics."""
     # These are only the most minimal of tests, more should be added later.  At
     # the very least we check that basic text/unicode calls work OK.
     nt.assert_equal(_ip.parse_options('foo', '')[1], 'foo')
     nt.assert_equal(_ip.parse_options(u'foo', '')[1], u'foo')
 
     
 def test_dirops():
     """Test various directory handling operations."""
     # curpath = lambda :os.path.splitdrive(os.getcwdu())[1].replace('\\','/')
     curpath = os.getcwdu
     startdir = os.getcwdu()
     ipdir = _ip.ipython_dir
     try:
         _ip.magic('cd "%s"' % ipdir)
         nt.assert_equal(curpath(), ipdir)
         _ip.magic('cd -')
         nt.assert_equal(curpath(), startdir)
         _ip.magic('pushd "%s"' % ipdir)
         nt.assert_equal(curpath(), ipdir)
         _ip.magic('popd')
         nt.assert_equal(curpath(), startdir)
     finally:
         os.chdir(startdir)
 
 
 def check_cpaste(code, should_fail=False):
     """Execute code via 'cpaste' and ensure it was executed, unless
     should_fail is set.
     """
     _ip.user_ns['code_ran'] = False
 
     src = StringIO()
     src.encoding = None   # IPython expects stdin to have an encoding attribute
     src.write('\n')
     src.write(code)
     src.write('\n--\n')
     src.seek(0)
 
     stdin_save = sys.stdin
     sys.stdin = src
     
     try:
         context = tt.AssertPrints if should_fail else tt.AssertNotPrints
         with context("Traceback (most recent call last)"):
                 _ip.magic('cpaste')
         
         if not should_fail:
             assert _ip.user_ns['code_ran']
     finally:
         sys.stdin = stdin_save
 
 
 def test_cpaste():
     """Test cpaste magic"""
 
     def run():
         """Marker function: sets a flag when executed.
         """
         _ip.user_ns['code_ran'] = True
         return 'run' # return string so '+ run()' doesn't result in success
 
     tests = {'pass': ["> > > run()",
                       ">>> > run()",
                       "+++ run()",
                       "++ run()",
                       "  >>> run()"],
 
              'fail': ["+ + run()",
                       " ++ run()"]}
 
     _ip.user_ns['run'] = run
 
     for code in tests['pass']:
         check_cpaste(code)
 
     for code in tests['fail']:
         check_cpaste(code, should_fail=True)
 
 def test_xmode():
     # Calling xmode three times should be a no-op
     xmode = _ip.InteractiveTB.mode
     for i in range(3):
         _ip.magic("xmode")
     nt.assert_equal(_ip.InteractiveTB.mode, xmode)
     
 def test_reset_hard():
     monitor = []
     class A(object):
         def __del__(self):
             monitor.append(1)
         def __repr__(self):
             return "<A instance>"
             
     _ip.user_ns["a"] = A()
     _ip.run_cell("a")
     
     nt.assert_equal(monitor, [])
     _ip.magic_reset("-f")
     nt.assert_equal(monitor, [1])
     
 class TestXdel(tt.TempFileMixin):
     def test_xdel(self):
         """Test that references from %run are cleared by xdel."""
         src = ("class A(object):\n"
                "    monitor = []\n"
                "    def __del__(self):\n"
                "        self.monitor.append(1)\n"
                "a = A()\n")
         self.mktmp(src)
         # %run creates some hidden references...
         _ip.magic("run %s" % self.fname)
         # ... as does the displayhook.
         _ip.run_cell("a")
         
         monitor = _ip.user_ns["A"].monitor
         nt.assert_equal(monitor, [])
         
         _ip.magic("xdel a")
         
         # Check that a's __del__ method has been called.
         nt.assert_equal(monitor, [1])
 
 def doctest_who():
     """doctest for %who
     
     In [1]: %reset -f
     
     In [2]: alpha = 123
     
     In [3]: beta = 'beta'
     
     In [4]: %who int
     alpha
     
     In [5]: %who str
     beta
     
     In [6]: %whos
     Variable   Type    Data/Info
     ----------------------------
     alpha      int     123
     beta       str     beta
     
     In [7]: %who_ls
     Out[7]: ['alpha', 'beta']
     """
 
 @py3compat.u_format
 def doctest_precision():
     """doctest for %precision
     
     In [1]: f = get_ipython().shell.display_formatter.formatters['text/plain']
     
     In [2]: %precision 5
     Out[2]: {u}'%.5f'
     
     In [3]: f.float_format
     Out[3]: {u}'%.5f'
     
     In [4]: %precision %e
     Out[4]: {u}'%e'
     
     In [5]: f(3.1415927)
     Out[5]: {u}'3.141593e+00'
     """
 

IPython/core/tests/test_run.py

@@ -1,210 +1,218 b''
 """Tests for code execution (%run and related), which is particularly tricky.
 
 Because of how %run manages namespaces, and the fact that we are trying here to
 verify subtle object deletion and reference counting issues, the %run tests
 will be kept in this separate file.  This makes it easier to aggregate in one
 place the tricks needed to handle it; most other magics are much easier to test
 and we do so in a common test_magic file.
 """
 from __future__ import absolute_import
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import os
 import sys
 import tempfile
 
 import nose.tools as nt
 from nose import SkipTest
 
 from IPython.testing import decorators as dec
 from IPython.testing import tools as tt
 from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # Test functions begin
 #-----------------------------------------------------------------------------
 
 def doctest_refbug():
     """Very nasty problem with references held by multiple runs of a script.
     See: https://github.com/ipython/ipython/issues/141
 
     In [1]: _ip.clear_main_mod_cache()
     # random
 
     In [2]: %run refbug
 
     In [3]: call_f()
     lowercased: hello
 
     In [4]: %run refbug
 
     In [5]: call_f()
     lowercased: hello
     lowercased: hello
     """
 
 
 def doctest_run_builtins():
     r"""Check that %run doesn't damage __builtins__.
 
     In [1]: import tempfile
 
     In [2]: bid1 = id(__builtins__)
 
     In [3]: fname = tempfile.mkstemp('.py')[1]
 
     In [3]: f = open(fname,'w')
 
     In [4]: dummy= f.write('pass\n')
 
     In [5]: f.flush()
 
     In [6]: t1 = type(__builtins__)
 
     In [7]: %run $fname
 
     In [7]: f.close()
 
     In [8]: bid2 = id(__builtins__)
 
     In [9]: t2 = type(__builtins__)
 
     In [10]: t1 == t2
     Out[10]: True
 
     In [10]: bid1 == bid2
     Out[10]: True
 
     In [12]: try:
        ....:     os.unlink(fname)
        ....: except:
        ....:     pass
        ....:
     """
 
 @py3compat.doctest_refactor_print
 def doctest_reset_del():
     """Test that resetting doesn't cause errors in __del__ methods.
 
     In [2]: class A(object):
        ...:     def __del__(self):
        ...:         print str("Hi")
        ...:
 
     In [3]: a = A()
 
     In [4]: get_ipython().reset()
     Hi
 
     In [5]: 1+1
     Out[5]: 2
     """
 
 # For some tests, it will be handy to organize them in a class with a common
 # setup that makes a temp file
 
 class TestMagicRunPass(tt.TempFileMixin):
 
     def setup(self):
         """Make a valid python temp file."""
         self.mktmp('pass\n')
 
     def run_tmpfile(self):
         _ip = get_ipython()
         # This fails on Windows if self.tmpfile.name has spaces or "~" in it.
         # See below and ticket https://bugs.launchpad.net/bugs/366353
         _ip.magic('run %s' % self.fname)
 
     def test_builtins_id(self):
         """Check that %run doesn't damage __builtins__ """
         _ip = get_ipython()
         # Test that the id of __builtins__ is not modified by %run
         bid1 = id(_ip.user_ns['__builtins__'])
         self.run_tmpfile()
         bid2 = id(_ip.user_ns['__builtins__'])
         tt.assert_equals(bid1, bid2)
 
     def test_builtins_type(self):
         """Check that the type of __builtins__ doesn't change with %run.
 
         However, the above could pass if __builtins__ was already modified to
         be a dict (it should be a module) by a previous use of %run.  So we
         also check explicitly that it really is a module:
         """
         _ip = get_ipython()
         self.run_tmpfile()
         tt.assert_equals(type(_ip.user_ns['__builtins__']),type(sys))
 
     def test_prompts(self):
         """Test that prompts correctly generate after %run"""
         self.run_tmpfile()
         _ip = get_ipython()
         p2 = str(_ip.displayhook.prompt2).strip()
         nt.assert_equals(p2[:3], '...')
 
 
 class TestMagicRunSimple(tt.TempFileMixin):
 
     def test_simpledef(self):
         """Test that simple class definitions work."""
         src = ("class foo: pass\n"
                "def f(): return foo()")
         self.mktmp(src)
         _ip.magic('run %s' % self.fname)
         _ip.run_cell('t = isinstance(f(), foo)')
         nt.assert_true(_ip.user_ns['t'])
 
     def test_obj_del(self):
         """Test that object's __del__ methods are called on exit."""
         if sys.platform == 'win32':
             try:
                 import win32api
             except ImportError:
                 raise SkipTest("Test requires pywin32")
         src = ("class A(object):\n"
                "    def __del__(self):\n"
                "        print 'object A deleted'\n"
                "a = A()\n")
         self.mktmp(py3compat.doctest_refactor_print(src))
-        tt.ipexec_validate(self.fname, 'object A deleted')
-
-    @dec.skip_known_failure
+        if dec.module_not_available('sqlite3'):
+            err = 'WARNING: IPython History requires SQLite, your history will not be saved\n'
+        else:
+            err = None
+        tt.ipexec_validate(self.fname, 'object A deleted', err)
+    
+    @dec.skip_known_failure 
     def test_aggressive_namespace_cleanup(self):
         """Test that namespace cleanup is not too aggressive GH-238
 
         Returning from another run magic deletes the namespace"""
         # see ticket https://github.com/ipython/ipython/issues/238
         class secondtmp(tt.TempFileMixin): pass
         empty = secondtmp()
         empty.mktmp('')
         src = ("ip = get_ipython()\n"
                "for i in range(5):\n"
                "   try:\n"
                "       ip.magic('run %s')\n"
                "   except NameError, e:\n"
                "       print i;break\n" % empty.fname)
         self.mktmp(py3compat.doctest_refactor_print(src))
         _ip.magic('run %s' % self.fname)
         _ip.run_cell('ip == get_ipython()')
         tt.assert_equals(_ip.user_ns['i'], 5)
 
     @dec.skip_win32
     def test_tclass(self):
         mydir = os.path.dirname(__file__)
         tc = os.path.join(mydir, 'tclass')
         src = ("%%run '%s' C-first\n"
                "%%run '%s' C-second\n"
                "%%run '%s' C-third\n") % (tc, tc, tc)
         self.mktmp(src, '.ipy')
         out = """\
 ARGV 1-: ['C-first']
 ARGV 1-: ['C-second']
 tclass.py: deleting object: C-first
 ARGV 1-: ['C-third']
 tclass.py: deleting object: C-second
 tclass.py: deleting object: C-third
 """
-        tt.ipexec_validate(self.fname, out)
+        if dec.module_not_available('sqlite3'):
+            err = 'WARNING: IPython History requires SQLite, your history will not be saved\n'
+        else:
+            err = None
+        tt.ipexec_validate(self.fname, out, err)

IPython/parallel/controller/sqlitedb.py

@@ -1,400 +1,408 b''
 """A TaskRecord backend using sqlite3
 
 Authors:
 
 * Min RK
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2011  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.
 #-----------------------------------------------------------------------------
 
 import json
 import os
 import cPickle as pickle
 from datetime import datetime
 
-import sqlite3
+try:
+    import sqlite3
+except ImportError:
+    sqlite3 = None
 
 from zmq.eventloop import ioloop
 
 from IPython.utils.traitlets import Unicode, Instance, List, Dict
 from .dictdb import BaseDB
 from IPython.utils.jsonutil import date_default, extract_dates, squash_dates
 
 #-----------------------------------------------------------------------------
 # SQLite operators, adapters, and converters
 #-----------------------------------------------------------------------------
 
 try:
     buffer
 except NameError:
     # py3k
     buffer = memoryview
 
 operators = {
  '$lt' : "<",
  '$gt' : ">",
  # null is handled weird with ==,!=
  '$eq' : "=",
  '$ne' : "!=",
  '$lte': "<=",
  '$gte': ">=",
  '$in' : ('=', ' OR '),
  '$nin': ('!=', ' AND '),
  # '$all': None,
  # '$mod': None,
  # '$exists' : None
 }
 null_operators = {
 '=' : "IS NULL",
 '!=' : "IS NOT NULL",
 }
 
 def _adapt_dict(d):
     return json.dumps(d, default=date_default)
 
 def _convert_dict(ds):
     if ds is None:
         return ds
     else:
         if isinstance(ds, bytes):
             # If I understand the sqlite doc correctly, this will always be utf8
             ds = ds.decode('utf8')
         return extract_dates(json.loads(ds))
 
 def _adapt_bufs(bufs):
     # this is *horrible*
     # copy buffers into single list and pickle it:
     if bufs and isinstance(bufs[0], (bytes, buffer)):
         return sqlite3.Binary(pickle.dumps(map(bytes, bufs),-1))
     elif bufs:
         return bufs
     else:
         return None
 
 def _convert_bufs(bs):
     if bs is None:
         return []
     else:
         return pickle.loads(bytes(bs))
 
 #-----------------------------------------------------------------------------
 # SQLiteDB class
 #-----------------------------------------------------------------------------
 
 class SQLiteDB(BaseDB):
     """SQLite3 TaskRecord backend."""
 
     filename = Unicode('tasks.db', config=True,
         help="""The filename of the sqlite task database. [default: 'tasks.db']""")
     location = Unicode('', config=True,
         help="""The directory containing the sqlite task database.  The default
         is to use the cluster_dir location.""")
     table = Unicode("", config=True,
         help="""The SQLite Table to use for storing tasks for this session. If unspecified,
         a new table will be created with the Hub's IDENT.  Specifying the table will result
         in tasks from previous sessions being available via Clients' db_query and
         get_result methods.""")
 
-    _db = Instance('sqlite3.Connection')
+    if sqlite3 is not None:
+        _db = Instance('sqlite3.Connection')
+    else:
+        _db = None
     # the ordered list of column names
     _keys = List(['msg_id' ,
             'header' ,
             'content',
             'buffers',
             'submitted',
             'client_uuid' ,
             'engine_uuid' ,
             'started',
             'completed',
             'resubmitted',
             'result_header' ,
             'result_content' ,
             'result_buffers' ,
             'queue' ,
             'pyin' ,
             'pyout',
             'pyerr',
             'stdout',
             'stderr',
         ])
     # sqlite datatypes for checking that db is current format
     _types = Dict({'msg_id' : 'text' ,
             'header' : 'dict text',
             'content' : 'dict text',
             'buffers' : 'bufs blob',
             'submitted' : 'timestamp',
             'client_uuid' : 'text',
             'engine_uuid' : 'text',
             'started' : 'timestamp',
             'completed' : 'timestamp',
             'resubmitted' : 'timestamp',
             'result_header' : 'dict text',
             'result_content' : 'dict text',
             'result_buffers' : 'bufs blob',
             'queue' : 'text',
             'pyin' : 'text',
             'pyout' : 'text',
             'pyerr' : 'text',
             'stdout' : 'text',
             'stderr' : 'text',
         })
 
     def __init__(self, **kwargs):
         super(SQLiteDB, self).__init__(**kwargs)
+        if sqlite3 is None:
+            raise ImportError("SQLiteDB requires sqlite3")
         if not self.table:
             # use session, and prefix _, since starting with # is illegal
             self.table = '_'+self.session.replace('-','_')
         if not self.location:
             # get current profile
             from IPython.core.application import BaseIPythonApplication
             if BaseIPythonApplication.initialized():
                 app = BaseIPythonApplication.instance()
                 if app.profile_dir is not None:
                     self.location = app.profile_dir.location
                 else:
                     self.location = u'.'
             else:
                 self.location = u'.'
         self._init_db()
 
         # register db commit as 2s periodic callback
         # to prevent clogging pipes
         # assumes we are being run in a zmq ioloop app
         loop = ioloop.IOLoop.instance()
         pc = ioloop.PeriodicCallback(self._db.commit, 2000, loop)
         pc.start()
 
     def _defaults(self, keys=None):
         """create an empty record"""
         d = {}
         keys = self._keys if keys is None else keys
         for key in keys:
             d[key] = None
         return d
 
     def _check_table(self):
         """Ensure that an incorrect table doesn't exist
 
         If a bad (old) table does exist, return False
         """
         cursor = self._db.execute("PRAGMA table_info(%s)"%self.table)
         lines = cursor.fetchall()
         if not lines:
             # table does not exist
             return True
         types = {}
         keys = []
         for line in lines:
             keys.append(line[1])
             types[line[1]] = line[2]
         if self._keys != keys:
             # key mismatch
             self.log.warn('keys mismatch')
             return False
         for key in self._keys:
             if types[key] != self._types[key]:
                 self.log.warn(
                     'type mismatch: %s: %s != %s'%(key,types[key],self._types[key])
                 )
                 return False
         return True
 
     def _init_db(self):
         """Connect to the database and get new session number."""
         # register adapters
         sqlite3.register_adapter(dict, _adapt_dict)
         sqlite3.register_converter('dict', _convert_dict)
         sqlite3.register_adapter(list, _adapt_bufs)
         sqlite3.register_converter('bufs', _convert_bufs)
         # connect to the db
         dbfile = os.path.join(self.location, self.filename)
         self._db = sqlite3.connect(dbfile, detect_types=sqlite3.PARSE_DECLTYPES,
             # isolation_level = None)#,
              cached_statements=64)
         # print dir(self._db)
         first_table = self.table
         i=0
         while not self._check_table():
             i+=1
             self.table = first_table+'_%i'%i
             self.log.warn(
                 "Table %s exists and doesn't match db format, trying %s"%
                 (first_table,self.table)
             )
 
         self._db.execute("""CREATE TABLE IF NOT EXISTS %s
                 (msg_id text PRIMARY KEY,
                 header dict text,
                 content dict text,
                 buffers bufs blob,
                 submitted timestamp,
                 client_uuid text,
                 engine_uuid text,
                 started timestamp,
                 completed timestamp,
                 resubmitted timestamp,
                 result_header dict text,
                 result_content dict text,
                 result_buffers bufs blob,
                 queue text,
                 pyin text,
                 pyout text,
                 pyerr text,
                 stdout text,
                 stderr text)
                 """%self.table)
         self._db.commit()
 
     def _dict_to_list(self, d):
         """turn a mongodb-style record dict into a list."""
 
         return [ d[key] for key in self._keys ]
 
     def _list_to_dict(self, line, keys=None):
         """Inverse of dict_to_list"""
         keys = self._keys if keys is None else keys
         d = self._defaults(keys)
         for key,value in zip(keys, line):
             d[key] = value
 
         return d
 
     def _render_expression(self, check):
         """Turn a mongodb-style search dict into an SQL query."""
         expressions = []
         args = []
 
         skeys = set(check.keys())
         skeys.difference_update(set(self._keys))
         skeys.difference_update(set(['buffers', 'result_buffers']))
         if skeys:
             raise KeyError("Illegal testing key(s): %s"%skeys)
 
         for name,sub_check in check.iteritems():
             if isinstance(sub_check, dict):
                 for test,value in sub_check.iteritems():
                     try:
                         op = operators[test]
                     except KeyError:
                         raise KeyError("Unsupported operator: %r"%test)
                     if isinstance(op, tuple):
                         op, join = op
 
                     if value is None and op in null_operators:
                             expr = "%s %s"%null_operators[op]
                     else:
                         expr = "%s %s ?"%(name, op)
                         if isinstance(value, (tuple,list)):
                             if op in null_operators and any([v is None for v in value]):
                                 # equality tests don't work with NULL
                                 raise ValueError("Cannot use %r test with NULL values on SQLite backend"%test)
                             expr = '( %s )'%( join.join([expr]*len(value)) )
                             args.extend(value)
                         else:
                             args.append(value)
                     expressions.append(expr)
             else:
                 # it's an equality check
                 if sub_check is None:
                     expressions.append("%s IS NULL")
                 else:
                     expressions.append("%s = ?"%name)
                     args.append(sub_check)
 
         expr = " AND ".join(expressions)
         return expr, args
 
     def add_record(self, msg_id, rec):
         """Add a new Task Record, by msg_id."""
         d = self._defaults()
         d.update(rec)
         d['msg_id'] = msg_id
         line = self._dict_to_list(d)
         tups = '(%s)'%(','.join(['?']*len(line)))
         self._db.execute("INSERT INTO %s VALUES %s"%(self.table, tups), line)
         # self._db.commit()
 
     def get_record(self, msg_id):
         """Get a specific Task Record, by msg_id."""
         cursor = self._db.execute("""SELECT * FROM %s WHERE msg_id==?"""%self.table, (msg_id,))
         line = cursor.fetchone()
         if line is None:
             raise KeyError("No such msg: %r"%msg_id)
         return self._list_to_dict(line)
 
     def update_record(self, msg_id, rec):
         """Update the data in an existing record."""
         query = "UPDATE %s SET "%self.table
         sets = []
         keys = sorted(rec.keys())
         values = []
         for key in keys:
             sets.append('%s = ?'%key)
             values.append(rec[key])
         query += ', '.join(sets)
         query += ' WHERE msg_id == ?'
         values.append(msg_id)
         self._db.execute(query, values)
         # self._db.commit()
 
     def drop_record(self, msg_id):
         """Remove a record from the DB."""
         self._db.execute("""DELETE FROM %s WHERE msg_id==?"""%self.table, (msg_id,))
         # self._db.commit()
 
     def drop_matching_records(self, check):
         """Remove a record from the DB."""
         expr,args = self._render_expression(check)
         query = "DELETE FROM %s WHERE %s"%(self.table, expr)
         self._db.execute(query,args)
         # self._db.commit()
 
     def find_records(self, check, keys=None):
         """Find records matching a query dict, optionally extracting subset of keys.
 
         Returns list of matching records.
 
         Parameters
         ----------
 
         check: dict
             mongodb-style query argument
         keys: list of strs [optional]
             if specified, the subset of keys to extract.  msg_id will *always* be
             included.
         """
         if keys:
             bad_keys = [ key for key in keys if key not in self._keys ]
             if bad_keys:
                 raise KeyError("Bad record key(s): %s"%bad_keys)
 
         if keys:
             # ensure msg_id is present and first:
             if 'msg_id' in keys:
                 keys.remove('msg_id')
             keys.insert(0, 'msg_id')
             req = ', '.join(keys)
         else:
             req = '*'
         expr,args = self._render_expression(check)
         query = """SELECT %s FROM %s WHERE %s"""%(req, self.table, expr)
         cursor = self._db.execute(query, args)
         matches = cursor.fetchall()
         records = []
         for line in matches:
             rec = self._list_to_dict(line, keys)
             records.append(rec)
         return records
 
     def get_history(self):
         """get all msg_ids, ordered by time submitted."""
         query = """SELECT msg_id FROM %s ORDER by submitted ASC"""%self.table
         cursor = self._db.execute(query)
         # will be a list of length 1 tuples
         return [ tup[0] for tup in cursor.fetchall()]
 
 __all__ = ['SQLiteDB']
  No newline at end of file

IPython/parallel/tests/test_db.py

@@ -1,180 +1,182 b''
 """Tests for db backends
 
 Authors:
 
 * Min RK
 """
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2011  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
 #-------------------------------------------------------------------------------
 
 from __future__ import division
 
 import tempfile
 import time
 
 from datetime import datetime, timedelta
 from unittest import TestCase
 
-from nose import SkipTest
-
 from IPython.parallel import error
 from IPython.parallel.controller.dictdb import DictDB
 from IPython.parallel.controller.sqlitedb import SQLiteDB
 from IPython.parallel.controller.hub import init_record, empty_record
 
+from IPython.testing import decorators as dec
 from IPython.zmq.session import Session
 
 
 #-------------------------------------------------------------------------------
 # TestCases
 #-------------------------------------------------------------------------------
 
 class TestDictBackend(TestCase):
     def setUp(self):
         self.session = Session()
         self.db = self.create_db()
         self.load_records(16)
     
     def create_db(self):
         return DictDB()
     
     def load_records(self, n=1):
         """load n records for testing"""
         #sleep 1/10 s, to ensure timestamp is different to previous calls
         time.sleep(0.1)
         msg_ids = []
         for i in range(n):
             msg = self.session.msg('apply_request', content=dict(a=5))
             msg['buffers'] = []
             rec = init_record(msg)
             msg_id = msg['header']['msg_id']
             msg_ids.append(msg_id)
             self.db.add_record(msg_id, rec)
         return msg_ids
     
     def test_add_record(self):
         before = self.db.get_history()
         self.load_records(5)
         after = self.db.get_history()
         self.assertEquals(len(after), len(before)+5)
         self.assertEquals(after[:-5],before)
         
     def test_drop_record(self):
         msg_id = self.load_records()[-1]
         rec = self.db.get_record(msg_id)
         self.db.drop_record(msg_id)
         self.assertRaises(KeyError,self.db.get_record, msg_id)
     
     def _round_to_millisecond(self, dt):
         """necessary because mongodb rounds microseconds"""
         micro = dt.microsecond
         extra = int(str(micro)[-3:])
         return dt - timedelta(microseconds=extra)
     
     def test_update_record(self):
         now = self._round_to_millisecond(datetime.now())
         # 
         msg_id = self.db.get_history()[-1]
         rec1 = self.db.get_record(msg_id)
         data = {'stdout': 'hello there', 'completed' : now}
         self.db.update_record(msg_id, data)
         rec2 = self.db.get_record(msg_id)
         self.assertEquals(rec2['stdout'], 'hello there')
         self.assertEquals(rec2['completed'], now)
         rec1.update(data)
         self.assertEquals(rec1, rec2)
     
     # def test_update_record_bad(self):
     #     """test updating nonexistant records"""
     #     msg_id = str(uuid.uuid4())
     #     data = {'stdout': 'hello there'}
     #     self.assertRaises(KeyError, self.db.update_record, msg_id, data)
 
     def test_find_records_dt(self):
         """test finding records by date"""
         hist = self.db.get_history()
         middle = self.db.get_record(hist[len(hist)//2])
         tic = middle['submitted']
         before = self.db.find_records({'submitted' : {'$lt' : tic}})
         after = self.db.find_records({'submitted' : {'$gte' : tic}})
         self.assertEquals(len(before)+len(after),len(hist))
         for b in before:
             self.assertTrue(b['submitted'] < tic)
         for a in after:
             self.assertTrue(a['submitted'] >= tic)
         same = self.db.find_records({'submitted' : tic})
         for s in same:
             self.assertTrue(s['submitted'] == tic)
     
     def test_find_records_keys(self):
         """test extracting subset of record keys"""
         found = self.db.find_records({'msg_id': {'$ne' : ''}},keys=['submitted', 'completed'])
         for rec in found:
             self.assertEquals(set(rec.keys()), set(['msg_id', 'submitted', 'completed']))
     
     def test_find_records_msg_id(self):
         """ensure msg_id is always in found records"""
         found = self.db.find_records({'msg_id': {'$ne' : ''}},keys=['submitted', 'completed'])
         for rec in found:
             self.assertTrue('msg_id' in rec.keys())
         found = self.db.find_records({'msg_id': {'$ne' : ''}},keys=['submitted'])
         for rec in found:
             self.assertTrue('msg_id' in rec.keys())
         found = self.db.find_records({'msg_id': {'$ne' : ''}},keys=['msg_id'])
         for rec in found:
             self.assertTrue('msg_id' in rec.keys())
     
     def test_find_records_in(self):
         """test finding records with '$in','$nin' operators"""
         hist = self.db.get_history()
         even = hist[::2]
         odd = hist[1::2]
         recs = self.db.find_records({ 'msg_id' : {'$in' : even}})
         found = [ r['msg_id'] for r in recs ]
         self.assertEquals(set(even), set(found))
         recs = self.db.find_records({ 'msg_id' : {'$nin' : even}})
         found = [ r['msg_id'] for r in recs ]
         self.assertEquals(set(odd), set(found))
     
     def test_get_history(self):
         msg_ids = self.db.get_history()
         latest = datetime(1984,1,1)
         for msg_id in msg_ids:
             rec = self.db.get_record(msg_id)
             newt = rec['submitted']
             self.assertTrue(newt >= latest)
             latest = newt
         msg_id = self.load_records(1)[-1]
         self.assertEquals(self.db.get_history()[-1],msg_id)
     
     def test_datetime(self):
         """get/set timestamps with datetime objects"""
         msg_id = self.db.get_history()[-1]
         rec = self.db.get_record(msg_id)
         self.assertTrue(isinstance(rec['submitted'], datetime))
         self.db.update_record(msg_id, dict(completed=datetime.now()))
         rec = self.db.get_record(msg_id)
         self.assertTrue(isinstance(rec['completed'], datetime))
 
     def test_drop_matching(self):
         msg_ids = self.load_records(10)
         query = {'msg_id' : {'$in':msg_ids}}
         self.db.drop_matching_records(query)
         recs = self.db.find_records(query)
         self.assertEquals(len(recs), 0)
-            
+
+
 class TestSQLiteBackend(TestDictBackend):
+
+    @dec.skip_without('sqlite3')
     def create_db(self):
         return SQLiteDB(location=tempfile.gettempdir())
     
     def tearDown(self):
         self.db._db.close()

IPython/testing/iptest.py

@@ -1,473 +1,476 b''
 # -*- coding: utf-8 -*-
 """IPython Test Suite Runner.
 
 This module provides a main entry point to a user script to test IPython
 itself from the command line. There are two ways of running this script:
 
 1. With the syntax `iptest all`.  This runs our entire test suite by
    calling this script (with different arguments) recursively.  This
    causes modules and package to be tested in different processes, using nose
    or trial where appropriate.
 2. With the regular nose syntax, like `iptest -vvs IPython`.  In this form
    the script simply calls nose, but with special command line flags and
    plugins loaded.
 
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2009  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
 #-----------------------------------------------------------------------------
 
 # Stdlib
 import os
 import os.path as path
 import signal
 import sys
 import subprocess
 import tempfile
 import time
 import warnings
 
 # Note: monkeypatch!
 # We need to monkeypatch a small problem in nose itself first, before importing
 # it for actual use.  This should get into nose upstream, but its release cycle
 # is slow and we need it for our parametric tests to work correctly.
 from IPython.testing import nosepatch
 # Now, proceed to import nose itself
 import nose.plugins.builtin
 from nose.core import TestProgram
 
 # Our own imports
 from IPython.utils.importstring import import_item
 from IPython.utils.path import get_ipython_module_path
 from IPython.utils.process import find_cmd, pycmd2argv
 from IPython.utils.sysinfo import sys_info
 
 from IPython.testing import globalipapp
 from IPython.testing.plugin.ipdoctest import IPythonDoctest
 from IPython.external.decorators import KnownFailure
 
 pjoin = path.join
 
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 
 
 #-----------------------------------------------------------------------------
 # Warnings control
 #-----------------------------------------------------------------------------
 
 # Twisted generates annoying warnings with Python 2.6, as will do other code
 # that imports 'sets' as of today
 warnings.filterwarnings('ignore', 'the sets module is deprecated',
                         DeprecationWarning )
 
 # This one also comes from Twisted
 warnings.filterwarnings('ignore', 'the sha module is deprecated',
                         DeprecationWarning)
 
 # Wx on Fedora11 spits these out
 warnings.filterwarnings('ignore', 'wxPython/wxWidgets release number mismatch',
                         UserWarning)
 
 #-----------------------------------------------------------------------------
 # Logic for skipping doctests
 #-----------------------------------------------------------------------------
 def extract_version(mod):
     return mod.__version__
 
 def test_for(item, min_version=None, callback=extract_version):
     """Test to see if item is importable, and optionally check against a minimum
     version.
 
     If min_version is given, the default behavior is to check against the
     `__version__` attribute of the item, but specifying `callback` allows you to
     extract the value you are interested in. e.g::
 
         In [1]: import sys
 
         In [2]: from IPython.testing.iptest import test_for
 
         In [3]: test_for('sys', (2,6), callback=lambda sys: sys.version_info)
         Out[3]: True
 
     """
     try:
         check = import_item(item)
     except (ImportError, RuntimeError):
         # GTK reports Runtime error if it can't be initialized even if it's
         # importable.
         return False
     else:
         if min_version:
             if callback:
                 # extra processing step to get version to compare
                 check = callback(check)
 
             return check >= min_version
         else:
             return True
 
 # Global dict where we can store information on what we have and what we don't
 # have available at test run time
 have = {}
 
 have['curses'] = test_for('_curses')
 have['matplotlib'] = test_for('matplotlib')
 have['pexpect'] = test_for('IPython.external.pexpect')
 have['pymongo'] = test_for('pymongo')
 have['wx'] = test_for('wx')
 have['wx.aui'] = test_for('wx.aui')
 have['qt'] = test_for('IPython.external.qt')
+have['sqlite3'] = test_for('sqlite3')
 
 have['tornado'] = test_for('tornado.version_info', (2,1,0), callback=None)
 
 if os.name == 'nt':
     min_zmq = (2,1,7)
 else:
     min_zmq = (2,1,4)
 
 def version_tuple(mod):
     "turn '2.1.9' into (2,1,9), and '2.1dev' into (2,1,999)"
     # turn 'dev' into 999, because Python3 rejects str-int comparisons
     vs = mod.__version__.replace('dev', '.999')
     tup = tuple([int(v) for v in vs.split('.') ])
     return tup
 
 have['zmq'] = test_for('zmq', min_zmq, version_tuple)
 
 #-----------------------------------------------------------------------------
 # Functions and classes
 #-----------------------------------------------------------------------------
 
 def report():
     """Return a string with a summary report of test-related variables."""
 
     out = [ sys_info(), '\n']
 
     avail = []
     not_avail = []
 
     for k, is_avail in have.items():
         if is_avail:
             avail.append(k)
         else:
             not_avail.append(k)
 
     if avail:
         out.append('\nTools and libraries available at test time:\n')
         avail.sort()
         out.append('   ' + ' '.join(avail)+'\n')
 
     if not_avail:
         out.append('\nTools and libraries NOT available at test time:\n')
         not_avail.sort()
         out.append('   ' + ' '.join(not_avail)+'\n')
 
     return ''.join(out)
 
 
 def make_exclude():
     """Make patterns of modules and packages to exclude from testing.
 
     For the IPythonDoctest plugin, we need to exclude certain patterns that
     cause testing problems.  We should strive to minimize the number of
     skipped modules, since this means untested code.
 
     These modules and packages will NOT get scanned by nose at all for tests.
     """
     # Simple utility to make IPython paths more readably, we need a lot of
     # these below
     ipjoin = lambda *paths: pjoin('IPython', *paths)
 
     exclusions = [ipjoin('external'),
                   pjoin('IPython_doctest_plugin'),
                   ipjoin('quarantine'),
                   ipjoin('deathrow'),
                   ipjoin('testing', 'attic'),
                   # This guy is probably attic material
                   ipjoin('testing', 'mkdoctests'),
                   # Testing inputhook will need a lot of thought, to figure out
                   # how to have tests that don't lock up with the gui event
                   # loops in the picture
                   ipjoin('lib', 'inputhook'),
                   # Config files aren't really importable stand-alone
                   ipjoin('config', 'default'),
                   ipjoin('config', 'profile'),
                   ]
-
+    if not have['sqlite3']:
+        exclusions.append(ipjoin('core', 'tests', 'test_history'))
+        exclusions.append(ipjoin('core', 'history'))
     if not have['wx']:
         exclusions.append(ipjoin('lib', 'inputhookwx'))
 
     # We do this unconditionally, so that the test suite doesn't import
     # gtk, changing the default encoding and masking some unicode bugs.
     exclusions.append(ipjoin('lib', 'inputhookgtk'))
 
     # These have to be skipped on win32 because the use echo, rm, cd, etc.
     # See ticket https://github.com/ipython/ipython/issues/87
     if sys.platform == 'win32':
         exclusions.append(ipjoin('testing', 'plugin', 'test_exampleip'))
         exclusions.append(ipjoin('testing', 'plugin', 'dtexample'))
 
     if not have['pexpect']:
         exclusions.extend([ipjoin('scripts', 'irunner'),
                            ipjoin('lib', 'irunner'),
                            ipjoin('lib', 'tests', 'test_irunner')])
 
     if not have['zmq']:
         exclusions.append(ipjoin('zmq'))
         exclusions.append(ipjoin('frontend', 'qt'))
         exclusions.append(ipjoin('parallel'))
     elif not have['qt']:
         exclusions.append(ipjoin('frontend', 'qt'))
 
     if not have['pymongo']:
         exclusions.append(ipjoin('parallel', 'controller', 'mongodb'))
         exclusions.append(ipjoin('parallel', 'tests', 'test_mongodb'))
 
     if not have['matplotlib']:
         exclusions.extend([ipjoin('lib', 'pylabtools'),
                            ipjoin('lib', 'tests', 'test_pylabtools')])
 
     if not have['tornado']:
         exclusions.append(ipjoin('frontend', 'html'))
 
     # This is needed for the reg-exp to match on win32 in the ipdoctest plugin.
     if sys.platform == 'win32':
         exclusions = [s.replace('\\','\\\\') for s in exclusions]
 
     return exclusions
 
 
 class IPTester(object):
     """Call that calls iptest or trial in a subprocess.
     """
     #: string, name of test runner that will be called
     runner = None
     #: list, parameters for test runner
     params = None
     #: list, arguments of system call to be made to call test runner
     call_args = None
     #: list, process ids of subprocesses we start (for cleanup)
     pids = None
 
     def __init__(self, runner='iptest', params=None):
         """Create new test runner."""
         p = os.path
         if runner == 'iptest':
             iptest_app = get_ipython_module_path('IPython.testing.iptest')
             self.runner = pycmd2argv(iptest_app) + sys.argv[1:]
         else:
             raise Exception('Not a valid test runner: %s' % repr(runner))
         if params is None:
             params = []
         if isinstance(params, str):
             params = [params]
         self.params = params
 
         # Assemble call
         self.call_args = self.runner+self.params
 
         # Store pids of anything we start to clean up on deletion, if possible
         # (on posix only, since win32 has no os.kill)
         self.pids = []
 
     if sys.platform == 'win32':
         def _run_cmd(self):
             # On Windows, use os.system instead of subprocess.call, because I
             # was having problems with subprocess and I just don't know enough
             # about win32 to debug this reliably.  Os.system may be the 'old
             # fashioned' way to do it, but it works just fine.  If someone
             # later can clean this up that's fine, as long as the tests run
             # reliably in win32.
             # What types of problems are you having. They may be related to
             # running Python in unboffered mode. BG.
             return os.system(' '.join(self.call_args))
     else:
         def _run_cmd(self):
             # print >> sys.stderr, '*** CMD:', ' '.join(self.call_args) # dbg
             subp = subprocess.Popen(self.call_args)
             self.pids.append(subp.pid)
             # If this fails, the pid will be left in self.pids and cleaned up
             # later, but if the wait call succeeds, then we can clear the
             # stored pid.
             retcode = subp.wait()
             self.pids.pop()
             return retcode
 
     def run(self):
         """Run the stored commands"""
         try:
             return self._run_cmd()
         except:
             import traceback
             traceback.print_exc()
             return 1  # signal failure
 
     def __del__(self):
         """Cleanup on exit by killing any leftover processes."""
 
         if not hasattr(os, 'kill'):
             return
 
         for pid in self.pids:
             try:
                 print 'Cleaning stale PID:', pid
                 os.kill(pid, signal.SIGKILL)
             except OSError:
                 # This is just a best effort, if we fail or the process was
                 # really gone, ignore it.
                 pass
 
 
 def make_runners():
     """Define the top-level packages that need to be tested.
     """
 
     # Packages to be tested via nose, that only depend on the stdlib
     nose_pkg_names = ['config', 'core', 'extensions', 'frontend', 'lib',
                      'scripts', 'testing', 'utils', 'nbformat' ]
 
     if have['zmq']:
         nose_pkg_names.append('parallel')
 
     # For debugging this code, only load quick stuff
     #nose_pkg_names = ['core', 'extensions']  # dbg
 
     # Make fully qualified package names prepending 'IPython.' to our name lists
     nose_packages = ['IPython.%s' % m for m in nose_pkg_names ]
 
     # Make runners
     runners = [ (v, IPTester('iptest', params=v)) for v in nose_packages ]
 
     return runners
 
 
 def run_iptest():
     """Run the IPython test suite using nose.
 
     This function is called when this script is **not** called with the form
     `iptest all`.  It simply calls nose with appropriate command line flags
     and accepts all of the standard nose arguments.
     """
 
     warnings.filterwarnings('ignore',
         'This will be removed soon.  Use IPython.testing.util instead')
 
     argv = sys.argv + [ '--detailed-errors',  # extra info in tracebacks
 
                         # Loading ipdoctest causes problems with Twisted, but
                         # our test suite runner now separates things and runs
                         # all Twisted tests with trial.
                         '--with-ipdoctest',
                         '--ipdoctest-tests','--ipdoctest-extension=txt',
 
                         # We add --exe because of setuptools' imbecility (it
                         # blindly does chmod +x on ALL files).  Nose does the
                         # right thing and it tries to avoid executables,
                         # setuptools unfortunately forces our hand here.  This
                         # has been discussed on the distutils list and the
                         # setuptools devs refuse to fix this problem!
                         '--exe',
                         ]
 
     if nose.__version__ >= '0.11':
         # I don't fully understand why we need this one, but depending on what
         # directory the test suite is run from, if we don't give it, 0 tests
         # get run.  Specifically, if the test suite is run from the source dir
         # with an argument (like 'iptest.py IPython.core', 0 tests are run,
         # even if the same call done in this directory works fine).  It appears
         # that if the requested package is in the current dir, nose bails early
         # by default.  Since it's otherwise harmless, leave it in by default
         # for nose >= 0.11, though unfortunately nose 0.10 doesn't support it.
         argv.append('--traverse-namespace')
 
     # use our plugin for doctesting.  It will remove the standard doctest plugin
     # if it finds it enabled
     plugins = [IPythonDoctest(make_exclude()), KnownFailure()]
     # We need a global ipython running in this process
     globalipapp.start_ipython()
     # Now nose can run
     TestProgram(argv=argv, addplugins=plugins)
 
 
 def run_iptestall():
     """Run the entire IPython test suite by calling nose and trial.
 
     This function constructs :class:`IPTester` instances for all IPython
     modules and package and then runs each of them.  This causes the modules
     and packages of IPython to be tested each in their own subprocess using
     nose or twisted.trial appropriately.
     """
 
     runners = make_runners()
 
     # Run the test runners in a temporary dir so we can nuke it when finished
     # to clean up any junk files left over by accident.  This also makes it
     # robust against being run in non-writeable directories by mistake, as the
     # temp dir will always be user-writeable.
     curdir = os.getcwdu()
     testdir = tempfile.gettempdir()
     os.chdir(testdir)
 
     # Run all test runners, tracking execution time
     failed = []
     t_start = time.time()
     try:
         for (name, runner) in runners:
             print '*'*70
             print 'IPython test group:',name
             res = runner.run()
             if res:
                 failed.append( (name, runner) )
     finally:
         os.chdir(curdir)
     t_end = time.time()
     t_tests = t_end - t_start
     nrunners = len(runners)
     nfail = len(failed)
     # summarize results
     print
     print '*'*70
     print 'Test suite completed for system with the following information:'
     print report()
     print 'Ran %s test groups in %.3fs' % (nrunners, t_tests)
     print
     print 'Status:'
     if not failed:
         print 'OK'
     else:
         # If anything went wrong, point out what command to rerun manually to
         # see the actual errors and individual summary
         print 'ERROR - %s out of %s test groups failed.' % (nfail, nrunners)
         for name, failed_runner in failed:
             print '-'*40
             print 'Runner failed:',name
             print 'You may wish to rerun this one individually, with:'
             print ' '.join(failed_runner.call_args)
             print
         # Ensure that our exit code indicates failure
         sys.exit(1)
 
 
 def main():
     for arg in sys.argv[1:]:
         if arg.startswith('IPython'):
             # This is in-process
             run_iptest()
     else:
         # This starts subprocesses
         run_iptestall()
 
 
 if __name__ == '__main__':
     main()