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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
import pytest
from pypy.module._hpy_universal.handlemanager import HandleReleaseCallback
from pypy.module._hpy_universal.state import State
# import for the side effect of adding the API functions
from pypy.module._hpy_universal import interp_hpy
from pypy.module._hpy_universal import llapi
class Config(object):
def __init__(self, space):
self.objspace = space
self.translating = True
class FakeSpace(object):
def __init__(self):
self._cache = {}
self.config = Config(self)
def fromcache(self, cls):
if cls not in self._cache:
self._cache[cls] = cls(self)
return self._cache[cls]
def call(self, w_func, w_args, w_kw):
# This is just enough for the debug closing on_invalid_handle callback
if w_args is None and w_kw is None:
return w_func()
else:
raise RuntimeError('space.call not fully implemented')
def __getattr__(self, name):
return '<fakespace.%s>' % name
@pytest.fixture(scope="module")
def fakespace():
return FakeSpace()
def test_fakespace(fakespace):
assert fakespace.w_ValueError == '<fakespace.w_ValueError>'
def x(space):
return object()
assert fakespace.fromcache(x) is fakespace.fromcache(x)
def callback():
# "trick" the debug context into not raising a fatal error when hitting a
# closed handle
pass
@pytest.fixture(scope="module", params=['universal', 'debug'])
def mgr(fakespace, request):
# Do everything in interp_hpy.startup
from pypy.module._hpy_universal.interp_type import setup_hpy_storage
state = fakespace.fromcache(State)
state.setup(fakespace)
setup_hpy_storage()
# end of interp_hpy.startup
debug = request.param == 'debug'
ret = state.get_handle_manager(debug)
return ret
class TestHandleManager(object):
def test_first_handle_is_not_zero(self, mgr):
h = mgr.new('hello')
assert h > 0
def test_new(self, mgr):
h = mgr.new('hello')
assert mgr.deref(h) == 'hello'
def test_close(self, mgr):
h = mgr.new('hello')
assert mgr.close(h) is None
if not mgr.is_debug:
# will crash PyPy on purpose in debug mode
assert mgr.deref(h) is None
def test_deref(self, mgr):
h = mgr.new('hello')
assert mgr.deref(h) == 'hello' # 'hello' is a fake W_Root
assert mgr.deref(h) == 'hello'
def test_consume(self, mgr):
h = mgr.new('hello')
assert mgr.consume(h) == 'hello'
if not mgr.is_debug:
# will crash PyPy on purpose in debug mode
assert mgr.deref(h) is None
def test_freelist(self, mgr):
if mgr.is_debug:
pytest.skip('only for HandleManager')
h0 = mgr.new('hello')
h1 = mgr.new('world')
assert mgr.consume(h0) == 'hello'
assert mgr.free_list == [h0]
h2 = mgr.new('hello2')
assert h2 == h0
assert mgr.free_list == []
def test_dup(self, mgr):
h0 = mgr.new('hello')
h1 = mgr.dup(h0)
assert h1 != h0
assert mgr.consume(h0) == mgr.consume(h1) == 'hello'
class TestReleaseCallback(object):
class MyCallback(HandleReleaseCallback):
def __init__(self, seen, data):
self.seen = seen
self.data = data
def release(self, h, obj):
self.seen.append((h, obj, self.data))
def test_callback(self, mgr):
seen = []
h0 = mgr.new('hello')
h1 = mgr.dup(h0)
h2 = mgr.dup(h0)
mgr.attach_release_callback(h0, self.MyCallback(seen, 'foo'))
mgr.attach_release_callback(h1, self.MyCallback(seen, 'bar'))
assert seen == []
#
mgr.close(h1)
assert seen == [(h1, 'hello', 'bar')]
#
mgr.close(h2)
assert seen == [(h1, 'hello', 'bar')]
#
mgr.close(h0)
assert seen == [(h1, 'hello', 'bar'),
(h0, 'hello', 'foo')]
def test_clear(self, mgr):
seen = []
h0 = mgr.new('hello')
mgr.attach_release_callback(h0, self.MyCallback(seen, 'foo'))
mgr.close(h0)
assert seen == [(h0, 'hello', 'foo')]
#
# check that the releaser array is cleared when we close the handle
# and that we don't run the releaser for a wrong object
h1 = mgr.new('world')
if not mgr.is_debug:
assert h1 == h0
mgr.close(h1)
assert seen == [(h0, 'hello', 'foo')]
def test_multiple_releasers(self, mgr):
seen = []
h0 = mgr.new('hello')
mgr.attach_release_callback(h0, self.MyCallback(seen, 'foo'))
mgr.attach_release_callback(h0, self.MyCallback(seen, 'bar'))
mgr.close(h0)
assert seen == [(h0, 'hello', 'foo'),
(h0, 'hello', 'bar')]
class TestUsing(object):
def test_simple(self, mgr):
with mgr.using('hello') as h:
assert mgr.deref(h) == 'hello'
if not mgr.is_debug:
# will crash PyPy on purpose in debug mode
assert mgr.deref(h) is None
def test_multiple_handles(self, mgr):
with mgr.using('hello', 'world', 'foo') as (h1, h2, h3):
assert mgr.deref(h1) == 'hello'
assert mgr.deref(h2) == 'world'
assert mgr.deref(h3) == 'foo'
if not mgr.is_debug:
# will crash PyPy on purpose in debug mode
assert mgr.deref(h1) is None
assert mgr.deref(h2) is None
assert mgr.deref(h3) is None
|
