Add tests for things in utils

IPython/utils/tests/test_decorators.py

@@ -0,0 +1,10 b''
+from IPython.utils import decorators
+
+def test_flag_calls():
+    @decorators.flag_calls
+    def f():
+        pass
+    
+    assert not f.called
+    f()
+    assert f.called
  No newline at end of file

IPython/utils/tests/test_pickleshare.py

@@ -0,0 +1,59 b''
+import os
+from unittest import TestCase
+
+from IPython.testing.decorators import skip
+from IPython.utils.tempdir import TemporaryDirectory
+from IPython.utils.pickleshare import PickleShareDB
+
+
+class PickleShareDBTestCase(TestCase):
+    def setUp(self):
+        self.tempdir = TemporaryDirectory()
+    
+    def tearDown(self):
+        self.tempdir.cleanup()
+
+    def test_picklesharedb(self):
+        db = PickleShareDB(self.tempdir.name)
+        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)
+        self.assertEqual(db.hget('hash','aku'), 12)
+        self.assertEqual(db.hget('hash','ankka'), 313)
+        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
+        self.assertEqual(lnk.bar, 7)
+
+    @skip("Too slow for regular running.")
+    def test_stress(self):
+        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, end=' ')
+            sys.stdout.flush()
+            if i % 10 == 0:
+                db.uncache()
  No newline at end of file

IPython/utils/pickleshare.py

@@ -1,370 +1,325 b''
 #!/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 __future__ import print_function
 
 from IPython.external.path import path as Path
 import os,stat,time
 import collections
 try:
     import cPickle as pickle
 except ImportError:
     import 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
             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.
         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, end=' ')
-        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/tests/test_openpy.py

@@ -1,23 +1,39 b''
 import io
 import os.path
 import nose.tools as nt
 
 from IPython.utils import openpy
 
 mydir = os.path.dirname(__file__)
 nonascii_path = os.path.join(mydir, '../../core/tests/nonascii.py')
 
 def test_detect_encoding():
     f = open(nonascii_path, 'rb')
     enc, lines = openpy.detect_encoding(f.readline)
     nt.assert_equal(enc, 'iso-8859-5')
 
 def test_read_file():
     read_specified_enc = io.open(nonascii_path, encoding='iso-8859-5').read()
     read_detected_enc = openpy.read_py_file(nonascii_path, skip_encoding_cookie=False)
     nt.assert_equal(read_detected_enc, read_specified_enc)
     assert u'coding: iso-8859-5' in read_detected_enc
     
     read_strip_enc_cookie = openpy.read_py_file(nonascii_path, skip_encoding_cookie=True)
     assert u'coding: iso-8859-5' not in read_strip_enc_cookie
-    
+
+def test_source_to_unicode():
+    with io.open(nonascii_path, 'rb') as f:
+        source_bytes = f.read()
+    nt.assert_equal(openpy.source_to_unicode(source_bytes, skip_encoding_cookie=False),
+                    source_bytes.decode('iso-8859-5'))
+
+    source_no_cookie = openpy.source_to_unicode(source_bytes, skip_encoding_cookie=True)
+    nt.assert_not_in(u'coding: iso-8859-5', source_no_cookie)
+
+def test_list_readline():
+    l = ['a', 'b']
+    readline = openpy._list_readline(l)
+    nt.assert_equal(readline(), 'a')
+    nt.assert_equal(readline(), 'b')
+    with nt.assert_raises(StopIteration):
+        readline()
  No newline at end of file

IPython/utils/tests/test_text.py

@@ -1,177 +1,190 b''
 # encoding: utf-8
 """Tests for IPython.utils.text"""
 from __future__ import print_function
 
 #-----------------------------------------------------------------------------
 #  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
 #-----------------------------------------------------------------------------
 
 import os
 import math
 import random
 import sys
 
 import nose.tools as nt
 
 from IPython.utils import text
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 
 def test_columnize():
     """Basic columnize tests."""
     size = 5
     items = [l*size for l in 'abc']
     out = text.columnize(items, displaywidth=80)
     nt.assert_equal(out, 'aaaaa  bbbbb  ccccc\n')
     out = text.columnize(items, displaywidth=12)
     nt.assert_equal(out, 'aaaaa  ccccc\nbbbbb\n')
     out = text.columnize(items, displaywidth=10)
     nt.assert_equal(out, 'aaaaa\nbbbbb\nccccc\n')
 
 def test_columnize_random():
     """Test with random input to hopfully catch edge case """
     for nitems in [random.randint(2,70) for i in range(2,20)]:
         displaywidth = random.randint(20,200)
         rand_len = [random.randint(2,displaywidth) for i in range(nitems)]
         items = ['x'*l for l in rand_len]
         out = text.columnize(items, displaywidth=displaywidth)
         longer_line = max([len(x) for x in out.split('\n')])
         longer_element = max(rand_len)
         if longer_line > displaywidth:
             print("Columnize displayed something lager than displaywidth : %s " % longer_line)
             print("longer element : %s " % longer_element)
             print("displaywidth : %s " % displaywidth)
             print("number of element : %s " % nitems)
             print("size of each element :\n %s" % rand_len)
             assert False
 
 def test_columnize_medium():
     """Test with inputs than shouldn't be wider tahn 80 """
     size = 40
     items = [l*size for l in 'abc']
     out = text.columnize(items, displaywidth=80)
     nt.assert_equal(out, '\n'.join(items+['']))
 
 def test_columnize_long():
     """Test columnize with inputs longer than the display window"""
     size = 11
     items = [l*size for l in 'abc']
     out = text.columnize(items, displaywidth=size-1)
     nt.assert_equal(out, '\n'.join(items+['']))
 
 def eval_formatter_check(f):
     ns = dict(n=12, pi=math.pi, stuff='hello there', os=os, u=u"café", b="café")
     s = f.format("{n} {n//4} {stuff.split()[0]}", **ns)
     nt.assert_equal(s, "12 3 hello")
     s = f.format(' '.join(['{n//%i}'%i for i in range(1,8)]), **ns)
     nt.assert_equal(s, "12 6 4 3 2 2 1")
     s = f.format('{[n//i for i in range(1,8)]}', **ns)
     nt.assert_equal(s, "[12, 6, 4, 3, 2, 2, 1]")
     s = f.format("{stuff!s}", **ns)
     nt.assert_equal(s, ns['stuff'])
     s = f.format("{stuff!r}", **ns)
     nt.assert_equal(s, repr(ns['stuff']))
     
     # Check with unicode:
     s = f.format("{u}", **ns)
     nt.assert_equal(s, ns['u'])
     # This decodes in a platform dependent manner, but it shouldn't error out
     s = f.format("{b}", **ns)
         
     nt.assert_raises(NameError, f.format, '{dne}', **ns)
 
 def eval_formatter_slicing_check(f):
     ns = dict(n=12, pi=math.pi, stuff='hello there', os=os)
     s = f.format(" {stuff.split()[:]} ", **ns)
     nt.assert_equal(s, " ['hello', 'there'] ")
     s = f.format(" {stuff.split()[::-1]} ", **ns)
     nt.assert_equal(s, " ['there', 'hello'] ")
     s = f.format("{stuff[::2]}", **ns)
     nt.assert_equal(s, ns['stuff'][::2])
     
     nt.assert_raises(SyntaxError, f.format, "{n:x}", **ns)
 
 def eval_formatter_no_slicing_check(f):
     ns = dict(n=12, pi=math.pi, stuff='hello there', os=os)
     
     s = f.format('{n:x} {pi**2:+f}', **ns)
     nt.assert_equal(s, "c +9.869604")
     
     s = f.format('{stuff[slice(1,4)]}', **ns)
     nt.assert_equal(s, 'ell')
     
     if sys.version_info >= (3, 4):
         # String formatting has changed in Python 3.4, so this now works.
         s = f.format("{a[:]}", a=[1, 2])
         nt.assert_equal(s, "[1, 2]")
     else:
         nt.assert_raises(SyntaxError, f.format, "{a[:]}")
 
 def test_eval_formatter():
     f = text.EvalFormatter()
     eval_formatter_check(f)
     eval_formatter_no_slicing_check(f)
 
 def test_full_eval_formatter():
     f = text.FullEvalFormatter()
     eval_formatter_check(f)
     eval_formatter_slicing_check(f)
 
 def test_dollar_formatter():
     f = text.DollarFormatter()
     eval_formatter_check(f)
     eval_formatter_slicing_check(f)
     
     ns = dict(n=12, pi=math.pi, stuff='hello there', os=os)
     s = f.format("$n", **ns)
     nt.assert_equal(s, "12")
     s = f.format("$n.real", **ns)
     nt.assert_equal(s, "12")
     s = f.format("$n/{stuff[:5]}", **ns)
     nt.assert_equal(s, "12/hello")
     s = f.format("$n $$HOME", **ns)
     nt.assert_equal(s, "12 $HOME")
     s = f.format("${foo}", foo="HOME")
     nt.assert_equal(s, "$HOME")
 
 
 def test_long_substr():
     data = ['hi']
     nt.assert_equal(text.long_substr(data), 'hi')
 
 
 def test_long_substr2():
     data = ['abc', 'abd', 'abf', 'ab']
     nt.assert_equal(text.long_substr(data), 'ab')
 
 def test_long_substr_empty():
     data = []
     nt.assert_equal(text.long_substr(data), '')
 
 def test_strip_email():
     src = """\
         >> >>> def f(x):
         >> ...   return x+1
         >> ... 
         >> >>> zz = f(2.5)"""
     cln = """\
 >>> def f(x):
 ...   return x+1
 ... 
 >>> zz = f(2.5)"""
     nt.assert_equal(text.strip_email_quotes(src), cln)
 
 
 def test_strip_email2():
     src = '> > > list()'
     cln = 'list()'
     nt.assert_equal(text.strip_email_quotes(src), cln)
+
+def test_LSString():
+    lss = text.LSString("abc\ndef")
+    nt.assert_equal(lss.l, ['abc', 'def'])
+    nt.assert_equal(lss.s, 'abc def')
+
+def test_SList():
+    sl = text.SList(['a 11', 'b 1', 'a 2'])
+    nt.assert_equal(sl.n, 'a 11\nb 1\na 2')
+    nt.assert_equal(sl.s, 'a 11 b 1 a 2')
+    nt.assert_equal(sl.grep(lambda x: x.startswith('a')), text.SList(['a 11', 'a 2']))
+    nt.assert_equal(sl.fields(0), text.SList(['a', 'b', 'a']))
+    nt.assert_equal(sl.sort(field=1, nums=True), text.SList(['b 1', 'a 2', 'a 11']))
  No newline at end of file