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
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
|
import os
from tinydb import TinyDB
from tinydb.middlewares import CachingMiddleware
from tinydb.storages import MemoryStorage, JSONStorage
doc = {'none': [None, None], 'int': 42, 'float': 3.1415899999999999,
'list': ['LITE', 'RES_ACID', 'SUS_DEXT'],
'dict': {'hp': 13, 'sp': 5},
'bool': [True, False, True, False]}
def test_caching(storage):
# Write contents
storage.write(doc)
# Verify contents
assert doc == storage.read()
def test_caching_read():
db = TinyDB(storage=CachingMiddleware(MemoryStorage))
assert db.all() == []
def test_caching_write_many(storage):
storage.WRITE_CACHE_SIZE = 3
# Storage should be still empty
assert storage.memory is None
# Write contents
for x in range(2):
storage.write(doc)
assert storage.memory is None # Still cached
storage.write(doc)
# Verify contents: Cache should be emptied and written to storage
assert storage.memory
def test_caching_flush(storage):
# Write contents
for _ in range(CachingMiddleware.WRITE_CACHE_SIZE - 1):
storage.write(doc)
# Not yet flushed...
assert storage.memory is None
storage.write(doc)
# Verify contents: Cache should be emptied and written to storage
assert storage.memory
def test_caching_flush_manually(storage):
# Write contents
storage.write(doc)
storage.flush()
# Verify contents: Cache should be emptied and written to storage
assert storage.memory
def test_caching_write(storage):
# Write contents
storage.write(doc)
storage.close()
# Verify contents: Cache should be emptied and written to storage
assert storage.storage.memory
def test_nested():
storage = CachingMiddleware(MemoryStorage)
storage() # Initialization
# Write contents
storage.write(doc)
# Verify contents
assert doc == storage.read()
def test_caching_json_write(tmpdir):
path = str(tmpdir.join('test.db'))
with TinyDB(path, storage=CachingMiddleware(JSONStorage)) as db:
db.insert({'key': 'value'})
# Verify database filesize
statinfo = os.stat(path)
assert statinfo.st_size != 0
# Assert JSON file has been closed
assert db._storage._handle.closed
del db
# Reopen database
with TinyDB(path, storage=CachingMiddleware(JSONStorage)) as db:
assert db.all() == [{'key': 'value'}]
|
