from __future__ import print_function 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()