-
Notifications
You must be signed in to change notification settings - Fork 23
/
test_delete.py
80 lines (72 loc) · 2.17 KB
/
test_delete.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
import time
from pytimber.pagestore import *
from numpy import *
from numpy.random import *
def mkdata(nvar=5, maxvarnames=10,maxrecsize=150,reclen=10,istartmax=1000):
data={}
for vv in range(nvar):
size=[-1,1,maxrecsize,randint(maxrecsize)][randint(4)]
rectype=['U','int','float'][randint(3)]
vname='var'+str(randint(maxvarnames))+rectype
if size==-1:
record=[rand(randint(maxrecsize)).astype(rectype)
for x in range(reclen)]
else:
record=[rand(size).astype(rectype) for x in range(reclen)]
istart=randint(istartmax)
idx=range(istart,istart+reclen)
data[vname]=(idx,record)
return data
class DictDB(object):
def get_var(self,name):
return getattr(self,name,([],[]))
def get(self):
return self.__dict__
def store(self,data):
for name,(idx,rec) in data.items():
new=dict(zip(*self.get_var(name)))
for i,v in zip(idx,rec):
new[i]=v
idx,rec=zip(*sorted(new.items()))
idx=array(idx);rec=array(rec)
setattr(self,name,(idx,rec))
def check_data(a,b):
if not set(a.keys())==set(b.keys()):
print(a.keys())
print(b.keys())
return False
for name,(idx,rec) in a.items():
nidx,nrec=b[name]
assert len(idx)==len(nidx)
assert len(rec)==len(rec)
assert list(idx)==list(nidx)
for av,bv in zip(rec,nrec):
if hasattr(av,'all'):
if not (av==bv).all():
print(name,av[0])
print(name,bv[0])
return False
else:
if not av==bv:
print(name,av)
print(name,av)
return False
return True
a=DictDB()
try:
b=PageStore('test.db','testdata',maxpagesize=100,keep_deleted_pages=True)
data=mkdata()
a.store(data)
check_data(data,a.get())
b.store(data)
check_data(data,b.get('%'))
check_data(a.get(),b.get('%'))
for n in range(10):
data=mkdata()
a.store(data)
b.store(data)
check_data(a.get(),b.get('%'))
except Exception as e:
raise e
finally:
b.delete()