Merge pull request #2088 from bfroehle/_2083_py32_test_failures

Fix (some) Python 3.2 ResourceWarnings

IPython/core/tests/test_magic.py

@@ -1,743 +1,744
 # -*- coding: utf-8 -*-
 """Tests for various magic functions.
 
 Needs to be run by nose (to make ipython session available).
 """
 from __future__ import absolute_import
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import io
 import os
 import sys
 from StringIO import StringIO
 from unittest import TestCase
 
 try:
     from importlib import invalidate_caches   # Required from Python 3.3
 except ImportError:
     def invalidate_caches():
         pass
 
 import nose.tools as nt
 
 from IPython.core import magic
 from IPython.core.magic import (Magics, magics_class, line_magic,
                                 cell_magic, line_cell_magic,
                                 register_line_magic, register_cell_magic,
                                 register_line_cell_magic)
 from IPython.core.magics import execution, script
 from IPython.nbformat.v3.tests.nbexamples import nb0
 from IPython.nbformat import current
 from IPython.testing import decorators as dec
 from IPython.testing import tools as tt
 from IPython.utils import py3compat
 from IPython.utils.tempdir import TemporaryDirectory
 from IPython.utils.process import find_cmd
 
 #-----------------------------------------------------------------------------
 # Test functions begin
 #-----------------------------------------------------------------------------
 
 @magic.magics_class
 class DummyMagics(magic.Magics): pass
 
 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'
     m = DummyMagics(ip)
     opts = m.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)
 
 def test_magic_parse_long_options():
     """Magic.parse_options can handle --foo=bar long options"""
     ip = get_ipython()
     m = DummyMagics(ip)
     opts, _ = m.parse_options('--foo --bar=bubble', 'a', 'foo', 'bar=')
     nt.assert_true('foo' in opts)
     nt.assert_true('bar' in opts)
     nt.assert_true(opts['bar'], "bubble")
 
 
 @dec.skip_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)
     """
 
 
 @dec.skip_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
     """
 
 
 @dec.skip_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'
     >>> 
     """
 
 
 @dec.skip_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")
 
 
 @dec.skip_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")
 
 
 @dec.skipif_not_numpy
 def test_numpy_reset_array_undec():
     "Test '%reset array' functionality"
     _ip.ex('import numpy as np')
     _ip.ex('a = np.empty(2)')
     yield (nt.assert_true, 'a' in _ip.user_ns)
     _ip.magic('reset -f array')
     yield (nt.assert_false, 'a' in _ip.user_ns)
 
 def test_reset_out():
     "Test '%reset out' magic"
     _ip.run_cell("parrot = 'dead'", store_history=True)
     # test '%reset -f out', make an Out prompt
     _ip.run_cell("parrot", store_history=True)
     nt.assert_true('dead' in [_ip.user_ns[x] for x in '_','__','___'])
     _ip.magic('reset -f out')
     nt.assert_false('dead' in [_ip.user_ns[x] for x in '_','__','___'])
     nt.assert_true(len(_ip.user_ns['Out']) == 0)
 
 def test_reset_in():
     "Test '%reset in' magic"
     # test '%reset -f in'
     _ip.run_cell("parrot", store_history=True)
     nt.assert_true('parrot' in [_ip.user_ns[x] for x in '_i','_ii','_iii'])
     _ip.magic('%reset -f in')
     nt.assert_false('parrot' in [_ip.user_ns[x] for x in '_i','_ii','_iii'])
     nt.assert_true(len(set(_ip.user_ns['In'])) == 1)
 
 def test_reset_dhist():
     "Test '%reset dhist' magic"
     _ip.run_cell("tmp = [d for d in _dh]") # copy before clearing
     _ip.magic('cd ' + os.path.dirname(nt.__file__))
     _ip.magic('cd -')
     nt.assert_true(len(_ip.user_ns['_dh']) > 0)
     _ip.magic('reset -f dhist')
     nt.assert_true(len(_ip.user_ns['_dh']) == 0)
     _ip.run_cell("_dh = [d for d in tmp]") #restore
 
 def test_reset_in_length():
     "Test that '%reset in' preserves In[] length"
     _ip.run_cell("print 'foo'")
     _ip.run_cell("reset -f in")
     nt.assert_true(len(_ip.user_ns['In']) == _ip.displayhook.prompt_count+1)
 
 def test_time():
     _ip.magic('time None')
 
 def test_tb_syntaxerror():
     """test %tb after a SyntaxError"""
     ip = get_ipython()
     ip.run_cell("for")
     
     # trap and validate stdout
     save_stdout = sys.stdout
     try:
         sys.stdout = StringIO()
         ip.run_cell("%tb")
         out = sys.stdout.getvalue()
     finally:
         sys.stdout = save_stdout
     # trim output, and only check the last line
     last_line = out.rstrip().splitlines()[-1].strip()
     nt.assert_equals(last_line, "SyntaxError: invalid syntax")
 
 
 @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.
     m = DummyMagics(_ip)
     nt.assert_equal(m.parse_options('foo', '')[1], 'foo')
     nt.assert_equal(m.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 = os.path.realpath(_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 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']
     """
 
 def test_whos():
     """Check that whos is protected against objects where repr() fails."""
     class A(object):
         def __repr__(self):
             raise Exception()
     _ip.user_ns['a'] = A()
     _ip.magic("whos")
 
 @py3compat.u_format
 def doctest_precision():
     """doctest for %precision
     
     In [1]: f = get_ipython().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'
     """
 
 def test_psearch():
     with tt.AssertPrints("dict.fromkeys"):
         _ip.run_cell("dict.fr*?")
 
 def test_timeit_shlex():
     """test shlex issues with timeit (#1109)"""
     _ip.ex("def f(*a,**kw): pass")
     _ip.magic('timeit -n1 "this is a bug".count(" ")')
     _ip.magic('timeit -r1 -n1 f(" ", 1)')
     _ip.magic('timeit -r1 -n1 f(" ", 1, " ", 2, " ")')
     _ip.magic('timeit -r1 -n1 ("a " + "b")')
     _ip.magic('timeit -r1 -n1 f("a " + "b")')
     _ip.magic('timeit -r1 -n1 f("a " + "b ")')
 
 
 def test_timeit_arguments():
     "Test valid timeit arguments, should not cause SyntaxError (GH #1269)"
     _ip.magic("timeit ('#')")
 
 
 def test_timeit_special_syntax():
     "Test %%timeit with IPython special syntax"
     from IPython.core.magic import register_line_magic
 
     @register_line_magic
     def lmagic(line):
         ip = get_ipython()
         ip.user_ns['lmagic_out'] = line
 
     # line mode test
     _ip.run_line_magic('timeit', '-n1 -r1 %lmagic my line')
     nt.assert_equal(_ip.user_ns['lmagic_out'], 'my line')
     # cell mode test
     _ip.run_cell_magic('timeit', '-n1 -r1', '%lmagic my line2')
     nt.assert_equal(_ip.user_ns['lmagic_out'], 'my line2')
     
 
 @dec.skipif(execution.profile is None)
 def test_prun_quotes():
     "Test that prun does not clobber string escapes (GH #1302)"
     _ip.magic(r"prun -q x = '\t'")
     nt.assert_equal(_ip.user_ns['x'], '\t')
 
 def test_extension():
     tmpdir = TemporaryDirectory()
     orig_ipython_dir = _ip.ipython_dir
     try:
         _ip.ipython_dir = tmpdir.name
         nt.assert_raises(ImportError, _ip.magic, "load_ext daft_extension")
         url = os.path.join(os.path.dirname(__file__), "daft_extension.py")
         _ip.magic("install_ext %s" % url)
         _ip.user_ns.pop('arq', None)
         invalidate_caches()   # Clear import caches
         _ip.magic("load_ext daft_extension")
         tt.assert_equal(_ip.user_ns['arq'], 185)
         _ip.magic("unload_ext daft_extension")
         assert 'arq' not in _ip.user_ns
     finally:
         _ip.ipython_dir = orig_ipython_dir
+        tmpdir.cleanup()
         
 def test_notebook_export_json():
     with TemporaryDirectory() as td:
         outfile = os.path.join(td, "nb.ipynb")
         _ip.ex(py3compat.u_format(u"u = {u}'héllo'"))
         _ip.magic("notebook -e %s" % outfile)
 
 def test_notebook_export_py():
     with TemporaryDirectory() as td:
         outfile = os.path.join(td, "nb.py")
         _ip.ex(py3compat.u_format(u"u = {u}'héllo'"))
         _ip.magic("notebook -e %s" % outfile)
 
 def test_notebook_reformat_py():
     with TemporaryDirectory() as td:
         infile = os.path.join(td, "nb.ipynb")
         with io.open(infile, 'w', encoding='utf-8') as f:
             current.write(nb0, f, 'json')
             
         _ip.ex(py3compat.u_format(u"u = {u}'héllo'"))
         _ip.magic("notebook -f py %s" % infile)
 
 def test_notebook_reformat_json():
     with TemporaryDirectory() as td:
         infile = os.path.join(td, "nb.py")
         with io.open(infile, 'w', encoding='utf-8') as f:
             current.write(nb0, f, 'py')
             
         _ip.ex(py3compat.u_format(u"u = {u}'héllo'"))
         _ip.magic("notebook -f ipynb %s" % infile)
         _ip.magic("notebook -f json %s" % infile)
 
 def test_env():
     env = _ip.magic("env")
     assert isinstance(env, dict), type(env)
 
 
 class CellMagicTestCase(TestCase):
 
     def check_ident(self, magic):
         # Manually called, we get the result
         out = _ip.run_cell_magic(magic, 'a', 'b')
         nt.assert_equals(out, ('a','b'))
         # Via run_cell, it goes into the user's namespace via displayhook
         _ip.run_cell('%%' + magic +' c\nd')
         nt.assert_equals(_ip.user_ns['_'], ('c','d'))
 
     def test_cell_magic_func_deco(self):
         "Cell magic using simple decorator"
         @register_cell_magic
         def cellm(line, cell):
             return line, cell
 
         self.check_ident('cellm')
 
     def test_cell_magic_reg(self):
         "Cell magic manually registered"
         def cellm(line, cell):
             return line, cell
 
         _ip.register_magic_function(cellm, 'cell', 'cellm2')
         self.check_ident('cellm2')
 
     def test_cell_magic_class(self):
         "Cell magics declared via a class"
         @magics_class
         class MyMagics(Magics):
 
             @cell_magic
             def cellm3(self, line, cell):
                 return line, cell
 
         _ip.register_magics(MyMagics)
         self.check_ident('cellm3')
 
     def test_cell_magic_class2(self):
         "Cell magics declared via a class, #2"
         @magics_class
         class MyMagics2(Magics):
 
             @cell_magic('cellm4')
             def cellm33(self, line, cell):
                 return line, cell
             
         _ip.register_magics(MyMagics2)
         self.check_ident('cellm4')
         # Check that nothing is registered as 'cellm33'
         c33 = _ip.find_cell_magic('cellm33')
         nt.assert_equals(c33, None)
 
 def test_file():
     """Basic %%file"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, 'file1')
         ip.run_cell_magic("file", fname, u'\n'.join([
             'line1',
             'line2',
         ]))
         with open(fname) as f:
             s = f.read()
         nt.assert_in('line1\n', s)
         nt.assert_in('line2', s)
 
 def test_file_unicode():
     """%%file with unicode cell"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, 'file1')
         ip.run_cell_magic("file", fname, u'\n'.join([
             u'liné1',
             u'liné2',
         ]))
         with io.open(fname, encoding='utf-8') as f:
             s = f.read()
         nt.assert_in(u'liné1\n', s)
         nt.assert_in(u'liné2', s)
 
 def test_file_amend():
     """%%file -a amends files"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, 'file2')
         ip.run_cell_magic("file", fname, u'\n'.join([
             'line1',
             'line2',
         ]))
         ip.run_cell_magic("file", "-a %s" % fname, u'\n'.join([
             'line3',
             'line4',
         ]))
         with open(fname) as f:
             s = f.read()
         nt.assert_in('line1\n', s)
         nt.assert_in('line3\n', s)
         
     
 def test_script_config():
     ip = get_ipython()
     ip.config.ScriptMagics.script_magics = ['whoda']
     sm = script.ScriptMagics(shell=ip)
     nt.assert_in('whoda', sm.magics['cell'])
 
 @dec.skip_win32
 def test_script_out():
     ip = get_ipython()
     ip.run_cell_magic("script", "--out output sh", "echo 'hi'")
     nt.assert_equals(ip.user_ns['output'], 'hi\n')
 
 @dec.skip_win32
 def test_script_err():
     ip = get_ipython()
     ip.run_cell_magic("script", "--err error sh", "echo 'hello' >&2")
     nt.assert_equals(ip.user_ns['error'], 'hello\n')
 
 @dec.skip_win32
 def test_script_out_err():
     ip = get_ipython()
     ip.run_cell_magic("script", "--out output --err error sh", "echo 'hi'\necho 'hello' >&2")
     nt.assert_equals(ip.user_ns['output'], 'hi\n')
     nt.assert_equals(ip.user_ns['error'], 'hello\n')
 
 @dec.skip_win32
 def test_script_bg_out():
     ip = get_ipython()
     ip.run_cell_magic("script", "--bg --out output sh", "echo 'hi'")
     nt.assert_equals(ip.user_ns['output'].read(), b'hi\n')
 
 @dec.skip_win32
 def test_script_bg_err():
     ip = get_ipython()
     ip.run_cell_magic("script", "--bg --err error sh", "echo 'hello' >&2")
     nt.assert_equals(ip.user_ns['error'].read(), b'hello\n')
 
 @dec.skip_win32
 def test_script_bg_out_err():
     ip = get_ipython()
     ip.run_cell_magic("script", "--bg --out output --err error sh", "echo 'hi'\necho 'hello' >&2")
     nt.assert_equals(ip.user_ns['output'].read(), b'hi\n')
     nt.assert_equals(ip.user_ns['error'].read(), b'hello\n')
 
 def test_script_defaults():
     ip = get_ipython()
     for cmd in ['sh', 'bash', 'perl', 'ruby']:
         try:
             find_cmd(cmd)
         except Exception:
             pass
         else:
             nt.assert_in(cmd, ip.magics_manager.magics['cell'])
 
 
 @magics_class
 class FooFoo(Magics):
     """class with both %foo and %%foo magics"""
     @line_magic('foo')
     def line_foo(self, line):
         "I am line foo"
         pass
 
     @cell_magic("foo")
     def cell_foo(self, line, cell):
         "I am cell foo, not line foo"
         pass
 
 def test_line_cell_info():
     """%%foo and %foo magics are distinguishable to inspect"""
     ip = get_ipython()
     ip.magics_manager.register(FooFoo)
     oinfo = ip.object_inspect('foo')
     nt.assert_true(oinfo['found'])
     nt.assert_true(oinfo['ismagic'])
     
     oinfo = ip.object_inspect('%%foo')
     nt.assert_true(oinfo['found'])
     nt.assert_true(oinfo['ismagic'])
     nt.assert_equals(oinfo['docstring'], FooFoo.cell_foo.__doc__)
 
     oinfo = ip.object_inspect('%foo')
     nt.assert_true(oinfo['found'])
     nt.assert_true(oinfo['ismagic'])
     nt.assert_equals(oinfo['docstring'], FooFoo.line_foo.__doc__)
 
 def test_multiple_magics():
     ip = get_ipython()
     foo1 = FooFoo(ip)
     foo2 = FooFoo(ip)
     mm = ip.magics_manager
     mm.register(foo1)
     nt.assert_true(mm.magics['line']['foo'].im_self is foo1)
     mm.register(foo2)
     nt.assert_true(mm.magics['line']['foo'].im_self is foo2)
 
 def test_alias_magic():
     """Test %alias_magic."""
     ip = get_ipython()
     mm = ip.magics_manager
 
     # Basic operation: both cell and line magics are created, if possible.
     ip.run_line_magic('alias_magic', 'timeit_alias timeit')
     nt.assert_true('timeit_alias' in mm.magics['line'])
     nt.assert_true('timeit_alias' in mm.magics['cell'])
 
     # --cell is specified, line magic not created.
     ip.run_line_magic('alias_magic', '--cell timeit_cell_alias timeit')
     nt.assert_false('timeit_cell_alias' in mm.magics['line'])
     nt.assert_true('timeit_cell_alias' in mm.magics['cell'])
 
     # Test that line alias is created successfully.
     ip.run_line_magic('alias_magic', '--line env_alias env')
     nt.assert_equal(ip.run_line_magic('env', ''),
                     ip.run_line_magic('env_alias', ''))

IPython/utils/pickleshare.py

@@ -1,364 +1,366
 #!/usr/bin/env python
 
 """ PickleShare - a small 'shelve' like datastore with concurrency support
 
 Like shelve, a PickleShareDB object acts like a normal dictionary. Unlike
 shelve, many processes can access the database simultaneously. Changing a
 value in database is immediately visible to other processes accessing the
 same database.
 
 Concurrency is possible because the values are stored in separate files. Hence
 the "database" is a directory where *all* files are governed by PickleShare.
 
 Example usage::
 
     from pickleshare import *
     db = PickleShareDB('~/testpickleshare')
     db.clear()
     print "Should be empty:",db.items()
     db['hello'] = 15
     db['aku ankka'] = [1,2,313]
     db['paths/are/ok/key'] = [1,(5,46)]
     print db.keys()
     del db['aku ankka']
 
 This module is certainly not ZODB, but can be used for low-load
 (non-mission-critical) situations where tiny code size trumps the
 advanced features of a "real" object database.
 
 Installation guide: easy_install pickleshare
 
 Author: Ville Vainio <[e-mail]>
 License: MIT open source license.
 
 """
 
 from IPython.external.path import path as Path
 import os,stat,time
 import collections
 import cPickle as pickle
 import glob
 
 def gethashfile(key):
     return ("%02x" % abs(hash(key) % 256))[-2:]
 
 _sentinel = object()
 
 class PickleShareDB(collections.MutableMapping):
     """ The main 'connection' object for PickleShare database """
     def __init__(self,root):
         """ Return a db object that will manage the specied directory"""
         self.root = Path(root).expanduser().abspath()
         if not self.root.isdir():
             self.root.makedirs()
         # cache has { 'key' : (obj, orig_mod_time) }
         self.cache = {}
 
 
     def __getitem__(self,key):
         """ db['key'] reading """
         fil = self.root / key
         try:
             mtime = (fil.stat()[stat.ST_MTIME])
         except OSError:
             raise KeyError(key)
 
         if fil in self.cache and mtime == self.cache[fil][1]:
             return self.cache[fil][0]
         try:
             # The cached item has expired, need to read
-            obj = pickle.loads(fil.open("rb").read())
+            with fil.open("rb") as f:
+                obj = pickle.loads(f.read())
         except:
             raise KeyError(key)
 
         self.cache[fil] = (obj,mtime)
         return obj
 
     def __setitem__(self,key,value):
         """ db['key'] = 5 """
         fil = self.root / key
         parent = fil.parent
         if parent and not parent.isdir():
             parent.makedirs()
         # We specify protocol 2, so that we can mostly go between Python 2
         # and Python 3. We can upgrade to protocol 3 when Python 2 is obsolete.
-        pickled = pickle.dump(value,fil.open('wb'), protocol=2)
+        with fil.open('wb') as f:
+            pickled = pickle.dump(value, f, protocol=2)
         try:
             self.cache[fil] = (value,fil.mtime)
         except OSError as e:
             if e.errno != 2:
                 raise
 
     def hset(self, hashroot, key, value):
         """ hashed set """
         hroot = self.root / hashroot
         if not hroot.isdir():
             hroot.makedirs()
         hfile = hroot / gethashfile(key)
         d = self.get(hfile, {})
         d.update( {key : value})
         self[hfile] = d
 
 
 
     def hget(self, hashroot, key, default = _sentinel, fast_only = True):
         """ hashed get """
         hroot = self.root / hashroot
         hfile = hroot / gethashfile(key)
 
         d = self.get(hfile, _sentinel )
         #print "got dict",d,"from",hfile
         if d is _sentinel:
             if fast_only:
                 if default is _sentinel:
                     raise KeyError(key)
 
                 return default
 
             # slow mode ok, works even after hcompress()
             d = self.hdict(hashroot)
 
         return d.get(key, default)
 
     def hdict(self, hashroot):
         """ Get all data contained in hashed category 'hashroot' as dict """
         hfiles = self.keys(hashroot + "/*")
         hfiles.sort()
         last = len(hfiles) and hfiles[-1] or ''
         if last.endswith('xx'):
             # print "using xx"
             hfiles = [last] + hfiles[:-1]
 
         all = {}
 
         for f in hfiles:
             # print "using",f
             try:
                 all.update(self[f])
             except KeyError:
                 print "Corrupt",f,"deleted - hset is not threadsafe!"
                 del self[f]
 
             self.uncache(f)
 
         return all
 
     def hcompress(self, hashroot):
         """ Compress category 'hashroot', so hset is fast again
 
         hget will fail if fast_only is True for compressed items (that were
         hset before hcompress).
 
         """
         hfiles = self.keys(hashroot + "/*")
         all = {}
         for f in hfiles:
             # print "using",f
             all.update(self[f])
             self.uncache(f)
 
         self[hashroot + '/xx'] = all
         for f in hfiles:
             p = self.root / f
             if p.basename() == 'xx':
                 continue
             p.remove()
 
 
 
     def __delitem__(self,key):
         """ del db["key"] """
         fil = self.root / key
         self.cache.pop(fil,None)
         try:
             fil.remove()
         except OSError:
             # notfound and permission denied are ok - we
             # lost, the other process wins the conflict
             pass
 
     def _normalized(self, p):
         """ Make a key suitable for user's eyes """
         return str(self.root.relpathto(p)).replace('\\','/')
 
     def keys(self, globpat = None):
         """ All keys in DB, or all keys matching a glob"""
 
         if globpat is None:
             files = self.root.walkfiles()
         else:
             files = [Path(p) for p in glob.glob(self.root/globpat)]
         return [self._normalized(p) for p in files if p.isfile()]
 
     def __iter__(self):
         return iter(self.keys())
 
     def __len__(self):
         return len(self.keys())
 
     def uncache(self,*items):
         """ Removes all, or specified items from cache
 
         Use this after reading a large amount of large objects
         to free up memory, when you won't be needing the objects
         for a while.
 
         """
         if not items:
             self.cache = {}
         for it in items:
             self.cache.pop(it,None)
 
     def waitget(self,key, maxwaittime = 60 ):
         """ Wait (poll) for a key to get a value
 
         Will wait for `maxwaittime` seconds before raising a KeyError.
         The call exits normally if the `key` field in db gets a value
         within the timeout period.
 
         Use this for synchronizing different processes or for ensuring
         that an unfortunately timed "db['key'] = newvalue" operation
         in another process (which causes all 'get' operation to cause a
         KeyError for the duration of pickling) won't screw up your program
         logic.
         """
 
         wtimes = [0.2] * 3 + [0.5] * 2 + [1]
         tries = 0
         waited = 0
         while 1:
             try:
                 val = self[key]
                 return val
             except KeyError:
                 pass
 
             if waited > maxwaittime:
                 raise KeyError(key)
 
             time.sleep(wtimes[tries])
             waited+=wtimes[tries]
             if tries < len(wtimes) -1:
                 tries+=1
 
     def getlink(self,folder):
         """ Get a convenient link for accessing items  """
         return PickleShareLink(self, folder)
 
     def __repr__(self):
         return "PickleShareDB('%s')" % self.root
 
 
 
 class PickleShareLink:
     """ A shortdand for accessing nested PickleShare data conveniently.
 
     Created through PickleShareDB.getlink(), example::
 
         lnk = db.getlink('myobjects/test')
         lnk.foo = 2
         lnk.bar = lnk.foo + 5
 
     """
     def __init__(self, db, keydir ):
         self.__dict__.update(locals())
 
     def __getattr__(self,key):
         return self.__dict__['db'][self.__dict__['keydir']+'/' + key]
     def __setattr__(self,key,val):
         self.db[self.keydir+'/' + key] = val
     def __repr__(self):
         db = self.__dict__['db']
         keys = db.keys( self.__dict__['keydir'] +"/*")
         return "<PickleShareLink '%s': %s>" % (
             self.__dict__['keydir'],
             ";".join([Path(k).basename() for k in keys]))
 
 
 def test():
     db = PickleShareDB('~/testpickleshare')
     db.clear()
     print "Should be empty:",db.items()
     db['hello'] = 15
     db['aku ankka'] = [1,2,313]
     db['paths/nest/ok/keyname'] = [1,(5,46)]
     db.hset('hash', 'aku', 12)
     db.hset('hash', 'ankka', 313)
     print "12 =",db.hget('hash','aku')
     print "313 =",db.hget('hash','ankka')
     print "all hashed",db.hdict('hash')
     print db.keys()
     print db.keys('paths/nest/ok/k*')
     print dict(db) # snapsot of whole db
     db.uncache() # frees memory, causes re-reads later
 
     # shorthand for accessing deeply nested files
     lnk = db.getlink('myobjects/test')
     lnk.foo = 2
     lnk.bar = lnk.foo + 5
     print lnk.bar # 7
 
 def stress():
     db = PickleShareDB('~/fsdbtest')
     import time,sys
     for i in range(1000):
         for j in range(1000):
             if i % 15 == 0 and i < 200:
                 if str(j) in db:
                     del db[str(j)]
                 continue
 
             if j%33 == 0:
                 time.sleep(0.02)
 
             db[str(j)] = db.get(str(j), []) + [(i,j,"proc %d" % os.getpid())]
             db.hset('hash',j, db.hget('hash',j,15) + 1 )
 
         print i,
         sys.stdout.flush()
         if i % 10 == 0:
             db.uncache()
 
 def main():
     import textwrap
     usage = textwrap.dedent("""\
     pickleshare - manage PickleShare databases
 
     Usage:
 
         pickleshare dump /path/to/db > dump.txt
         pickleshare load /path/to/db < dump.txt
         pickleshare test /path/to/db
     """)
     DB = PickleShareDB
     import sys
     if len(sys.argv) < 2:
         print usage
         return
 
     cmd = sys.argv[1]
     args = sys.argv[2:]
     if cmd == 'dump':
         if not args: args= ['.']
         db = DB(args[0])
         import pprint
         pprint.pprint(db.items())
     elif cmd == 'load':
         cont = sys.stdin.read()
         db = DB(args[0])
         data = eval(cont)
         db.clear()
         for k,v in db.items():
             db[k] = v
     elif cmd == 'testwait':
         db = DB(args[0])
         db.clear()
         print db.waitget('250')
     elif cmd == 'test':
         test()
         stress()
 
 if __name__== "__main__":
     main()
 
 

IPython/utils/py3compat.py

@@ -1,178 +1,179
 # coding: utf-8
 """Compatibility tricks for Python 3. Mainly to do with unicode."""
 import __builtin__
 import functools
 import sys
 import re
 import types
 
 from .encoding import DEFAULT_ENCODING
 
 orig_open = open
 
 def no_code(x, encoding=None):
     return x
 
 def decode(s, encoding=None):
     encoding = encoding or DEFAULT_ENCODING
     return s.decode(encoding, "replace")
 
 def encode(u, encoding=None):
     encoding = encoding or DEFAULT_ENCODING
     return u.encode(encoding, "replace")
 
 
 def cast_unicode(s, encoding=None):
     if isinstance(s, bytes):
         return decode(s, encoding)
     return s
 
 def cast_bytes(s, encoding=None):
     if not isinstance(s, bytes):
         return encode(s, encoding)
     return s
 
 def _modify_str_or_docstring(str_change_func):
     @functools.wraps(str_change_func)
     def wrapper(func_or_str):
         if isinstance(func_or_str, basestring):
             func = None
             doc = func_or_str
         else:
             func = func_or_str
             doc = func.__doc__
         
         doc = str_change_func(doc)
         
         if func:
             func.__doc__ = doc
             return func
         return doc
     return wrapper
 
 if sys.version_info[0] >= 3:
     PY3 = True
     
     input = input
     builtin_mod_name = "builtins"
     
     str_to_unicode = no_code
     unicode_to_str = no_code
     str_to_bytes = encode
     bytes_to_str = decode
     cast_bytes_py2 = no_code
     
     def isidentifier(s, dotted=False):
         if dotted:
             return all(isidentifier(a) for a in s.split("."))
         return s.isidentifier()
     
     open = orig_open
     
     MethodType = types.MethodType
     
     def execfile(fname, glob, loc=None):
         loc = loc if (loc is not None) else glob
-        exec compile(open(fname, 'rb').read(), fname, 'exec') in glob, loc
+        with open(fname, 'rb') as f:
+            exec compile(f.read(), fname, 'exec') in glob, loc
     
     # Refactor print statements in doctests.
     _print_statement_re = re.compile(r"\bprint (?P<expr>.*)$", re.MULTILINE)
     def _print_statement_sub(match):
         expr = match.groups('expr')
         return "print(%s)" % expr
     
     @_modify_str_or_docstring
     def doctest_refactor_print(doc):
         """Refactor 'print x' statements in a doctest to print(x) style. 2to3
         unfortunately doesn't pick up on our doctests.
         
         Can accept a string or a function, so it can be used as a decorator."""
         return _print_statement_re.sub(_print_statement_sub, doc)
     
     # Abstract u'abc' syntax:
     @_modify_str_or_docstring
     def u_format(s):
         """"{u}'abc'" --> "'abc'" (Python 3)
         
         Accepts a string or a function, so it can be used as a decorator."""
         return s.format(u='')
 
 else:
     PY3 = False
     
     input = raw_input
     builtin_mod_name = "__builtin__"
     
     str_to_unicode = decode
     unicode_to_str = encode
     str_to_bytes = no_code
     bytes_to_str = no_code
     cast_bytes_py2 = cast_bytes
     
     import re
     _name_re = re.compile(r"[a-zA-Z_][a-zA-Z0-9_]*$")
     def isidentifier(s, dotted=False):
         if dotted:
             return all(isidentifier(a) for a in s.split("."))
         return bool(_name_re.match(s))
     
     class open(object):
         """Wrapper providing key part of Python 3 open() interface."""
         def __init__(self, fname, mode="r", encoding="utf-8"):
             self.f = orig_open(fname, mode)
             self.enc = encoding
         
         def write(self, s):
             return self.f.write(s.encode(self.enc))
         
         def read(self, size=-1):
             return self.f.read(size).decode(self.enc)
         
         def close(self):
             return self.f.close()
         
         def __enter__(self):
             return self
         
         def __exit__(self, etype, value, traceback):
             self.f.close()
     
     def MethodType(func, instance):
         return types.MethodType(func, instance, type(instance))
     
     # don't override system execfile on 2.x:
     execfile = execfile
     
     def doctest_refactor_print(func_or_str):
         return func_or_str
 
 
     # Abstract u'abc' syntax:
     @_modify_str_or_docstring
     def u_format(s):
         """"{u}'abc'" --> "u'abc'" (Python 2)
         
         Accepts a string or a function, so it can be used as a decorator."""
         return s.format(u='u')
 
     if sys.platform == 'win32':
         def execfile(fname, glob=None, loc=None):
             loc = loc if (loc is not None) else glob
             # The rstrip() is necessary b/c trailing whitespace in files will
             # cause an IndentationError in Python 2.6 (this was fixed in 2.7,
             # but we still support 2.6).  See issue 1027.
             scripttext = __builtin__.open(fname).read().rstrip() + '\n'
             # compile converts unicode filename to str assuming
             # ascii. Let's do the conversion before calling compile
             if isinstance(fname, unicode):
                 filename = unicode_to_str(fname)
             else:
                 filename = fname
             exec compile(scripttext, filename, 'exec') in glob, loc
     else:
         def execfile(fname, *where):
             if isinstance(fname, unicode):
                 filename = fname.encode(sys.getfilesystemencoding())
             else:
                 filename = fname
             __builtin__.execfile(filename, *where)