|
1 import inspect |
1 import io |
2 import io |
|
3 import types |
|
4 |
|
5 |
|
6 def make_cffi(cls): |
|
7 """Decorator to add CFFI versions of each test method.""" |
|
8 |
|
9 try: |
|
10 import zstd_cffi |
|
11 except ImportError: |
|
12 return cls |
|
13 |
|
14 # If CFFI version is available, dynamically construct test methods |
|
15 # that use it. |
|
16 |
|
17 for attr in dir(cls): |
|
18 fn = getattr(cls, attr) |
|
19 if not inspect.ismethod(fn) and not inspect.isfunction(fn): |
|
20 continue |
|
21 |
|
22 if not fn.__name__.startswith('test_'): |
|
23 continue |
|
24 |
|
25 name = '%s_cffi' % fn.__name__ |
|
26 |
|
27 # Replace the "zstd" symbol with the CFFI module instance. Then copy |
|
28 # the function object and install it in a new attribute. |
|
29 if isinstance(fn, types.FunctionType): |
|
30 globs = dict(fn.__globals__) |
|
31 globs['zstd'] = zstd_cffi |
|
32 new_fn = types.FunctionType(fn.__code__, globs, name, |
|
33 fn.__defaults__, fn.__closure__) |
|
34 new_method = new_fn |
|
35 else: |
|
36 globs = dict(fn.__func__.func_globals) |
|
37 globs['zstd'] = zstd_cffi |
|
38 new_fn = types.FunctionType(fn.__func__.func_code, globs, name, |
|
39 fn.__func__.func_defaults, |
|
40 fn.__func__.func_closure) |
|
41 new_method = types.UnboundMethodType(new_fn, fn.im_self, |
|
42 fn.im_class) |
|
43 |
|
44 setattr(cls, name, new_method) |
|
45 |
|
46 return cls |
|
47 |
2 |
48 |
3 class OpCountingBytesIO(io.BytesIO): |
49 class OpCountingBytesIO(io.BytesIO): |
4 def __init__(self, *args, **kwargs): |
50 def __init__(self, *args, **kwargs): |
5 self._read_count = 0 |
51 self._read_count = 0 |
6 self._write_count = 0 |
52 self._write_count = 0 |