Repository navigation
Expand file tree
/
Copy pathtest.py
More file actions
128 lines (104 loc) · 3.71 KB
/
Copy pathtest.py
File metadata and controls
128 lines (104 loc) · 3.71 KB
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
import os
import gc
import ana
import unittest
import pickle
import logging
l = logging.getLogger("ana.test")
class A(ana.Storable):
def __init__(self, n):
self.n = n
l.debug("%s.__init__", self)
def __repr__(self):
return "<A %s>" % str(self.n)
def _ana_getstate(self):
l.debug("%s._ana_getstate", self)
return (self.n,)
def _ana_setstate(self, s):
self.n = s[0]
l.debug("%s._ana_setstate", self)
def _ana_getliteral(self):
return { 'n': self.n }
class TestAna(unittest.TestCase):
def test_simple(self):
ana.set_dl(ana.SimpleDataLayer())
one = A(1)
one.make_uuid()
o = pickle.dumps(one)
one_copy = pickle.loads(o)
assert one is one_copy
two = A(1)
t = pickle.dumps(one)
two_copy = pickle.loads(t)
assert two_copy is not two
assert pickle.load(open(os.path.join(os.path.dirname(__file__), 'test_pickle.p'), 'rb')).n == 1337
def write_a1337(self):
a1337 = A(1337)
a1337.make_uuid()
pickle.dump(a1337, open(os.path.join(os.path.dirname(__file__), 'test_pickle.p'), 'w'))
def test_dict(self):
ana.set_dl(ana.DictDataLayer())
l.debug("Initializing 1")
one = A(1)
l.debug("Initializing 2")
two = A(2)
one.make_uuid()
l.debug("Copying 1")
one_p = pickle.dumps(one)
one_copy = pickle.loads(one_p)
l.debug("Copying 2")
two_p = pickle.dumps(two)
two_copy = pickle.loads(two_p)
self.assertIs(one_copy, one)
self.assertIsNot(two_copy, two)
self.assertEqual(str(two_copy), str(two))
self.assertIs(one, A.ana_load(one.ana_store()))
self.assertIs(two, A.ana_load(two.ana_store()))
two_copy2 = pickle.loads(pickle.dumps(two))
self.assertEqual(str(two_copy2), str(two))
l.debug("Initializing 3")
three = A(3)
three_str = str(three)
l.debug("Storing 3")
three_uuid = three.ana_store()
l.debug("Deleting 3")
del three
gc.collect()
self.assertFalse(three_uuid in ana.get_dl().uuid_cache)
l.debug("Loading 3")
three_copy = A.ana_load(three_uuid)
self.assertEqual(three_copy.ana_uuid, three_uuid) #pylint:disable=no-member
self.assertEqual(str(three_copy), three_str)
known = set()
first_json = three_copy.to_literal(known)
self.assertTrue(three_copy.ana_uuid in first_json['objects'])
self.assertEqual(first_json['objects'][three_copy.ana_uuid]['object']['n'], three_copy.n)
self.assertEqual(first_json['value']['ana_uuid'], three_copy.ana_uuid)
second_json = three_copy.to_literal(known)
self.assertFalse(three_copy.ana_uuid in second_json['objects'])
self.assertEqual(second_json['value']['ana_uuid'], three_copy.ana_uuid)
def test_dir(self):
ana.dl = ana.DirDataLayer(pickle_dir="/tmp/test_ana")
one = A(1)
self.assertIs(one, A.ana_load(one.ana_store()))
self.assertTrue(os.path.exists("/tmp/test_ana/%s.p" % one.ana_uuid))
uuid = one.ana_uuid
old_id = id(one)
del one
gc.collect()
ana.dl = ana.DirDataLayer(pickle_dir="/tmp/test_ana")
two = A.ana_load(uuid)
self.assertEquals(uuid, two.ana_uuid)
# reset the datalayer to make sure we handle it properly
ana.set_dl(ana.DictDataLayer())
try:
two = A.ana_load(uuid)
assert False
except KeyError:
pass
two.ana_store()
del two
three = A.ana_load(uuid)
assert uuid, three.ana_uuid
if __name__ == '__main__':
unittest.main()