1from __future__ import print_function 2import os 3 4from pickleshare import PickleShareDB 5 6def test_pickleshare(tmpdir): 7 db = PickleShareDB(tmpdir) 8 db.clear() 9 print("Should be empty:",db.items()) 10 assert len(db) == 0 11 db['hello'] = 15 12 assert db['hello'] == 15 13 db['aku ankka'] = [1,2,313] 14 assert db['aku ankka'] == [1,2,313] 15 db['paths/nest/ok/keyname'] = [1,(5,46)] 16 assert db['paths/nest/ok/keyname'] == [1,(5,46)] 17 18 db.hset('hash', 'aku', 12) 19 db.hset('hash', 'ankka', 313) 20 assert db.hget('hash', 'aku') == 12 21 assert db.hget('hash', 'ankka') == 313 22 23 print("all hashed",db.hdict('hash')) 24 print(db.keys()) 25 print(db.keys('paths/nest/ok/k*')) 26 print(dict(db)) # snapsot of whole db 27 db.uncache() # frees memory, causes re-reads later 28 29 # shorthand for accessing deeply nested files 30 lnk = db.getlink('myobjects/test') 31 lnk.foo = 2 32 lnk.bar = lnk.foo + 5 33 assert lnk.bar == 7 34 35def test_stress(tmpdir): 36 db = PickleShareDB(tmpdir) 37 import time,sys 38 for i in range(100): 39 for j in range(500): 40 if i % 15 == 0 and i < 70: 41 if str(j) in db: 42 del db[str(j)] 43 continue 44 45 if j%33 == 0: 46 time.sleep(0.02) 47 48 db[str(j)] = db.get(str(j), []) + [(i,j,"proc %d" % os.getpid())] 49 db.hset('hash',j, db.hget('hash',j,15) + 1 ) 50 51 print(i, end=' ') 52 sys.stdout.flush() 53 if i % 10 == 0: 54 db.uncache() 55