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
|
import py, pytest
import contextlib
from rpython.rtyper.lltypesystem import lltype
from pypy.interpreter.baseobjspace import W_Root
from pypy.module.cpyext.state import State
from pypy.module.cpyext.api import (
slot_function, cpython_api, copy_header_files, INTERPLEVEL_API,
Py_ssize_t, Py_ssize_tP, PyObject, cts)
from pypy.module.cpyext.test.test_cpyext import freeze_refcnts, LeakCheckingTest
from pypy.interpreter.error import OperationError
from rpython.rlib import rawrefcount
import os
@contextlib.contextmanager
def raises_w(space, expected_exc):
with pytest.raises(OperationError) as excinfo:
yield
operror = excinfo.value
assert operror.w_type is getattr(space, 'w_' + expected_exc.__name__)
class BaseApiTest(LeakCheckingTest):
def setup_class(cls):
space = cls.space
# warm up reference counts:
# - the posix module allocates a HCRYPTPROV on Windows
# - writing to stdout and stderr allocates a file lock
space.getbuiltinmodule("cpyext")
space.getbuiltinmodule(os.name)
space.call_function(space.getattr(space.sys.get("stderr"),
space.wrap("write")),
space.wrap(""))
space.call_function(space.getattr(space.sys.get("stdout"),
space.wrap("write")),
space.wrap(""))
class CAPI:
def __getattr__(self, name):
return getattr(cls.space, name)
cls.api = CAPI()
CAPI.__dict__.update(INTERPLEVEL_API)
print 'DONT_FREE_ANY_MORE'
rawrefcount._dont_free_any_more()
def raises(self, space, api, expected_exc, f, *args):
if not callable(f):
raise Exception("%s is not callable" % (f,))
f(*args)
state = space.fromcache(State)
operror = state.operror
if not operror:
raise Exception("DID NOT RAISE")
if getattr(space, 'w_' + expected_exc.__name__) is not operror.w_type:
raise Exception("Wrong exception")
return state.clear_exception()
def setup_method(self, func):
freeze_refcnts(self)
def teardown_method(self, func):
state = self.space.fromcache(State)
try:
state.check_and_raise_exception()
except OperationError as e:
print e.errorstr(self.space)
raise
try:
self.space.getexecutioncontext().cleanup_cpyext_state()
except AttributeError:
pass
if self.check_and_print_leaks():
assert False, "Test leaks or loses object(s)."
@slot_function([PyObject], lltype.Void)
def PyPy_GetWrapped(space, w_arg):
assert isinstance(w_arg, W_Root)
@slot_function([PyObject], lltype.Void)
def PyPy_GetReference(space, arg):
assert lltype.typeOf(arg) == PyObject
@cpython_api([Py_ssize_t], Py_ssize_t, error=-1)
def PyPy_TypedefTest1(space, arg):
assert lltype.typeOf(arg) == Py_ssize_t
return 0
@cpython_api([Py_ssize_tP], Py_ssize_tP)
def PyPy_TypedefTest2(space, arg):
assert lltype.typeOf(arg) == Py_ssize_tP
return None
class TestConversion(BaseApiTest):
def test_conversions(self, space):
PyPy_GetWrapped(space, space.w_None)
PyPy_GetReference(space, space.w_None)
def test_typedef(self, space):
from rpython.translator.c.database import LowLevelDatabase
db = LowLevelDatabase()
assert PyPy_TypedefTest1.api_func.get_c_restype(db) == 'Signed'
assert PyPy_TypedefTest1.api_func.get_c_args(db) == 'Signed arg0'
assert PyPy_TypedefTest2.api_func.get_c_restype(db) == 'Signed *'
assert PyPy_TypedefTest2.api_func.get_c_args(db) == 'Signed *arg0'
PyPy_TypedefTest1(space, 0)
ppos = lltype.malloc(Py_ssize_tP.TO, 1, flavor='raw')
ppos[0] = 0
PyPy_TypedefTest2(space, ppos)
lltype.free(ppos, flavor='raw')
@pytest.mark.skipif(os.environ.get('USER')=='root',
reason='root can write to all files')
def test_copy_header_files(tmpdir):
copy_header_files(cts, tmpdir, True)
def check(name):
f = tmpdir.join(name)
assert f.check(file=True)
py.test.raises(py.error.EACCES, "f.open('w')") # check that it's not writable
check('Python.h')
check('modsupport.h')
check('pypy_decl.h')
|
GitWeb