Coverage for tests/test_native_handle.py: 94.51%
354 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-01-14 17:01 +1100
1"""Tests for the handling of native pointers via wrappers and utilities."""
3import gc
4import os
5import sys
6from typing import TYPE_CHECKING, Any, Callable, List, Optional
8import pytest
9from cffi import FFI
11if TYPE_CHECKING:
12 from refcount.interop import CffiData
14from refcount.interop import (
15 CffiNativeHandle,
16 CffiWrapperFactory,
17 DeletableCffiNativeHandle,
18 GenericWrapper,
19 OwningCffiNativeHandle,
20 cffi_arg_error_external_obj_type,
21 is_cffi_native_handle,
22 unwrap_cffi_native_handle,
23 wrap_as_pointer_handle,
24 wrap_cffi_native_handle,
25)
26from refcount.putils import library_short_filename
28fname = library_short_filename("test_native_library")
30pkg_dir = os.path.join(os.path.dirname(__file__), "..")
31sys.path.insert(0, pkg_dir)
32if sys.platform == "win32": 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true
33 dir_path = os.path.join(pkg_dir, "tests", "test_native_library", "build", "Debug")
34 if not os.path.exists(os.path.join(dir_path, fname)):
35 # fallback on AppVeyor output location
36 dir_path = os.path.join(pkg_dir, "tests", "test_native_library", "x64", "Debug")
38else:
39 dir_path = os.path.join(pkg_dir, "tests", "test_native_library", "build")
41native_lib_path = os.path.join(dir_path, fname)
43assert os.path.exists(native_lib_path)
45ut_ffi = FFI()
47ut_ffi.cdef(
48 """
49typedef struct _date_time_interop
50{
51 int year;
52 int month;
53 int day;
54 int hour;
55 int minute;
56 int second;
57} date_time_interop;
59typedef struct _interval_interop
60{
61 date_time_interop start;
62 date_time_interop end;
63} interval_interop;
64""",
65)
67ut_ffi.cdef(
68 "extern void create_date(date_time_interop* start, int year, int month, int day, int hour, int min, int sec);",
69)
70ut_ffi.cdef(
71 "extern int test_date(date_time_interop* start, int year, int month, int day, int hour, int min, int sec);",
72)
73ut_ffi.cdef("extern void* create_dog();")
74ut_ffi.cdef("extern int get_dog_refcount( void* obj);")
75ut_ffi.cdef("extern int remove_dog_reference( void* obj);")
76ut_ffi.cdef("extern int add_dog_reference( void* obj);")
77ut_ffi.cdef("extern void* create_croc();")
78ut_ffi.cdef("extern int get_croc_refcount( void* obj);")
79ut_ffi.cdef("extern int remove_croc_reference( void* obj);")
80ut_ffi.cdef("extern int add_croc_reference( void* obj);")
81ut_ffi.cdef("extern void* create_owner( void* d);")
82ut_ffi.cdef("extern int get_owner_refcount( void* obj);")
83ut_ffi.cdef("extern int remove_owner_reference( void* obj);")
84ut_ffi.cdef("extern int add_owner_reference( void* obj);")
85ut_ffi.cdef("extern int num_dogs();")
86ut_ffi.cdef("extern int num_owners();")
87ut_ffi.cdef("extern void say_walk( void* owner);")
88ut_ffi.cdef("extern void release( void* obj);")
90ut_ffi.cdef("extern void register_exception_callback(const void* callback);")
91ut_ffi.cdef("extern void trigger_callback();")
93ut_dll = ut_ffi.dlopen(native_lib_path, ut_ffi.RTLD_LAZY) # Lazy loading
95_message_from_c: str = "<none>"
98@ut_ffi.callback("void(char *)")
99def called_back_from_c(some_string: str) -> None:
100 """This function is called when uchronia raises an exception.
102 It sets the global variable ``_exception_txt_raised_uchronia``.
104 :param cdata exception_string: Exception string.
105 """
106 global _message_from_c # noqa: PLW0603
107 _message_from_c = ut_ffi.string(some_string)
110class CustomCffiNativeHandle(CffiNativeHandle):
111 """a custom native resource handle for testing purposes."""
113 def __init__(self, pointer: "CffiData", type_id: str = "", prior_ref_count: int = 0):
114 """Initialize a reference counter for a resource handle, with an initial reference count."""
115 super(CustomCffiNativeHandle, self).__init__(
116 pointer,
117 type_id=type_id,
118 prior_ref_count=prior_ref_count,
119 )
121 def _release_handle(self) -> bool:
122 ut_dll.release(self.get_handle())
123 return True
126class Dog(CustomCffiNativeHandle):
127 """A custom class for testing purposes."""
129 def __init__(self, pointer: "CffiData" = None):
130 """A custom class for testing purposes."""
131 if pointer is None:
132 pointer = ut_dll.create_dog()
133 super(Dog, self).__init__(pointer, type_id="DOG_PTR")
135 @property
136 def native_reference_count(self) -> int:
137 return ut_dll.get_dog_refcount(self.get_handle())
139 @staticmethod
140 def num_native_instances():
141 return ut_dll.num_dogs()
144class DogOwner(CustomCffiNativeHandle):
145 def __init__(self, dog):
146 super(DogOwner, self).__init__(None, type_id="DOG_OWNER_PTR")
147 self._set_handle(ut_dll.create_owner(dog.get_handle()))
148 self.dog = dog
149 self.dog.add_ref()
151 @property
152 def native_reference_count(self) -> int:
153 return ut_dll.get_owner_refcount(self.get_handle())
155 @staticmethod
156 def num_native_instances():
157 return ut_dll.num_owners()
159 def say_walk(self):
160 ut_dll.say_walk(self.get_handle())
162 def _release_handle(self) -> bool:
163 super(DogOwner, self)._release_handle()
164 # super(DogOwner, self)._release_handle()
165 self.dog.release()
166 return True
169class CrocFiveParameters(CffiNativeHandle):
170 def __init__(
171 self,
172 pointer: Any,
173 release_native: Callable,
174 type_id: str = "",
175 prior_ref_count: int = 0,
176 some_fifth_parameter: float = 0.0,
177 ):
178 super(CrocFiveParameters, self).__init__(
179 pointer,
180 type_id=type_id,
181 prior_ref_count=prior_ref_count,
182 )
183 self._release_native_handle = release_native
184 self.some_fifth_parameter = some_fifth_parameter
186 def _release_handle(self) -> bool:
187 self._release_native_handle(self.get_handle())
188 return True
191class CrocFourParameters(CffiNativeHandle):
192 def __init__(
193 self,
194 pointer: Any,
195 release_native: Callable,
196 type_id: str = "",
197 prior_ref_count: int = 0,
198 ):
199 super(CrocFourParameters, self).__init__(
200 pointer,
201 type_id=type_id,
202 prior_ref_count=prior_ref_count,
203 )
204 self._release_native_handle = release_native
206 def _release_handle(self) -> bool:
207 self._release_native_handle(self.get_handle())
208 return True
211class CrocFourParametersWrongFourthParameter(CffiNativeHandle):
212 def __init__(
213 self,
214 pointer: Any,
215 release_native: Callable,
216 type_id: str = "",
217 unsupported_argument_type: Optional[List] = None,
218 ):
219 super(CrocFourParametersWrongFourthParameter, self).__init__(
220 pointer,
221 type_id=type_id,
222 prior_ref_count=0,
223 )
224 self.unsupported_argument_type = unsupported_argument_type
225 self._release_native_handle = release_native
227 def _release_handle(self) -> bool:
228 self._release_native_handle(self.get_handle())
229 return True
232class CrocThreeParameters(CrocFourParameters):
233 def __init__(self, pointer: Any, release_native: Callable, type_id: str = ""):
234 super(CrocThreeParameters, self).__init__(
235 pointer,
236 release_native=release_native,
237 type_id=type_id,
238 prior_ref_count=0,
239 )
242class CrocTwoParameters(CrocThreeParameters):
243 def __init__(self, pointer: Any, release_native: Callable):
244 super(CrocTwoParameters, self).__init__(
245 pointer,
246 release_native=release_native,
247 type_id="CROC_PTR",
248 )
251class CrocOneParameters(CrocTwoParameters):
252 def __init__(self, pointer: Any):
253 super(CrocOneParameters, self).__init__(pointer, release_native=ut_dll.release)
256class CrocZeroParameters(CrocOneParameters):
257 def __init__(self):
258 raise ValueError(
259 "This class should not have been used to create a wrapper, since it has no constuctor argument.",
260 )
261 super(CrocZeroParameters, self).__init__(None)
264def test_native_obj_ref_counting():
265 dog = Dog()
266 assert dog.reference_count == 1
267 assert dog.native_reference_count == 1
268 dog.add_ref()
269 assert dog.reference_count == 2
270 assert dog.native_reference_count == 1
271 dog.add_ref()
272 assert dog.reference_count == 3
273 assert dog.native_reference_count == 1
274 dog.decrement_ref()
275 assert dog.reference_count == 2
276 assert dog.native_reference_count == 1
277 owner = DogOwner(dog)
278 assert owner.reference_count == 1
279 assert dog.reference_count == 3
280 assert dog.native_reference_count == 1
281 dog.release()
282 assert owner.reference_count == 1
283 assert dog.reference_count == 2
284 assert dog.native_reference_count == 1
285 dog.release()
286 assert owner.reference_count == 1
287 assert owner.native_reference_count == 1
288 assert dog.reference_count == 1
289 assert dog.native_reference_count == 1
290 assert not dog.is_invalid
291 owner.say_walk()
292 owner.release()
293 assert owner.reference_count == 0
294 assert dog.reference_count == 0
295 # Cannot check on the native ref count - deleted objects.
296 # TODO think of a simple way to test these
297 # assert 0, owner.native_reference_count)
298 # assert 0, dog.native_reference_count)
299 assert dog.is_invalid
300 assert owner.is_invalid
303def test_cffi_native_handle_finalizers():
304 init_dog_count = Dog.num_native_instances()
305 dog = Dog()
306 assert (init_dog_count + 1) == Dog.num_native_instances()
307 assert dog.reference_count == 1
308 assert dog.native_reference_count == 1
309 # if dog reference a new instance and we force garbage GC:
310 gc.collect()
311 dog = Dog()
312 gc.collect()
313 gc.collect()
314 assert dog.reference_count == 1
315 assert dog.native_reference_count == 1
316 nn = Dog.num_native_instances()
317 assert (init_dog_count + 1) == nn
318 dog = None
319 gc.collect()
320 assert init_dog_count == Dog.num_native_instances()
323def test_cffi_exceptions():
324 import datetime
326 incorrect_handle = datetime.datetime(2000, 1, 1, 1, 1, 1)
327 with pytest.raises(RuntimeError):
328 x = DeletableCffiNativeHandle(incorrect_handle, release_native=None)
331def test_generic_wrappers():
332 x = ut_ffi.new("char[10]")
333 o_wrapper = OwningCffiNativeHandle(x)
334 assert str(o_wrapper).startswith("CFFI pointer handle to a native pointer")
335 assert o_wrapper.reference_count == 1
336 gw = GenericWrapper(o_wrapper.ptr)
337 assert gw.ptr == o_wrapper.ptr
340def test_str_repr():
341 dog = Dog()
342 assert str(dog).startswith(
343 'CFFI pointer handle to a native pointer of type id "DOG',
344 )
345 assert repr(dog).startswith(
346 'CFFI pointer handle to a native pointer of type id "DOG',
347 )
350def test_cffi_native_handle_dispose():
351 init_dog_count = Dog.num_native_instances()
352 dog = Dog()
353 assert str(dog).startswith("CFFI pointer handle to a native pointer")
354 assert (init_dog_count + 1) == Dog.num_native_instances()
355 assert dog.reference_count == 1
356 assert dog.native_reference_count == 1
357 dog.dispose()
358 assert init_dog_count == Dog.num_native_instances()
359 assert dog.reference_count == 0
360 # assert 0 == dog.native_reference_count
361 # it should be all right to call dispose, even if already called and already zero ref counts.
362 dog.dispose()
365def test_cffi_handle_access():
366 x = ut_ffi.new("char[10]", init=b"foobarbaz0")
367 o_wrapper = OwningCffiNativeHandle(x)
368 assert str(o_wrapper.ptr) == "<cdata 'char[10]' owning 10 bytes>"
369 assert isinstance(o_wrapper.obj, bytes)
370 assert o_wrapper.obj == b"f"
373from datetime import datetime
376def test_wrapper_helper_functions():
377 assert isinstance(wrap_cffi_native_handle(dict()), dict)
378 pointer = ut_dll.create_dog()
379 dog = wrap_cffi_native_handle(pointer, "dog", ut_dll.release)
380 assert isinstance(dog, CffiNativeHandle)
381 assert dog.is_invalid == False
382 assert dog.reference_count == 1
383 assert is_cffi_native_handle(dog, "dog")
384 assert is_cffi_native_handle(dog, "cat") == False
385 assert is_cffi_native_handle(dict()) == False
386 assert is_cffi_native_handle(1) == False
387 assert is_cffi_native_handle(1, "cat") == False
388 assert is_cffi_native_handle(None) == False
389 assert pointer == unwrap_cffi_native_handle(dog, False)
390 assert pointer == unwrap_cffi_native_handle(dog.ptr, False)
391 assert unwrap_cffi_native_handle(None, False) is None
392 assert pointer == unwrap_cffi_native_handle(dog, True)
393 assert pointer == unwrap_cffi_native_handle(dog.ptr, True)
394 assert unwrap_cffi_native_handle(None, True) is None
395 x = datetime(2000, 1, 1, 1, 1)
396 assert unwrap_cffi_native_handle(x, False) == x
397 with pytest.raises(TypeError):
398 assert unwrap_cffi_native_handle(x, True) == x
400 from refcount.interop import (
401 type_error_cffi, # backward compat; maintain unit test coverage
402 )
404 for func in [cffi_arg_error_external_obj_type, type_error_cffi]:
405 msg = func(1, "")
406 assert msg == "Expected a 'CffiNativeHandle' but instead got object of type '<class 'int'>'"
407 msg = func(dog, "cat")
408 assert (
409 msg == "Expected a 'CffiNativeHandle' with underlying type id 'cat' but instead got one with type id 'dog'"
410 )
411 msg = func(None, "cat")
412 assert msg == "Expected a 'CffiNativeHandle' but instead got 'None'"
413 dog = None
414 gc.collect()
417def test_wrap_as_pointer_handle():
418 pointer = ut_dll.create_dog()
419 dog = wrap_cffi_native_handle(pointer, "dog", ut_dll.release)
421 # Allow passing None via a wrapper, to facilitate uniform code generation with c-api-wrapper-generation
422 assert isinstance(wrap_as_pointer_handle(None, False), GenericWrapper)
423 assert isinstance(wrap_as_pointer_handle(None, True), GenericWrapper)
424 assert wrap_as_pointer_handle(None, True).ptr is None
425 assert wrap_as_pointer_handle(None, False).ptr is None
427 x = ut_ffi.new("char[10]", init=b"foobarbaz0")
428 assert isinstance(wrap_as_pointer_handle(x, False), OwningCffiNativeHandle)
429 assert isinstance(wrap_as_pointer_handle(x, True), OwningCffiNativeHandle)
430 assert wrap_as_pointer_handle(dog, False) == dog
431 assert wrap_as_pointer_handle(dog, True) == dog
433 bb = b"foobarbaz0"
434 assert isinstance(wrap_as_pointer_handle(bb, False), GenericWrapper)
435 assert isinstance(wrap_as_pointer_handle(bb, True), GenericWrapper)
437 d = datetime(2000, 1, 1, 1, 1)
438 assert wrap_as_pointer_handle(d, False) == d
439 with pytest.raises(TypeError):
440 assert unwrap_cffi_native_handle(d, True) == d
442 with pytest.raises(
443 TypeError,
444 match="Argument is neither a CffiNativeHandle nor a CFFI external pointer, nor bytes",
445 ):
446 nothing = wrap_as_pointer_handle(d, True)
448 dog = None
449 gc.collect()
452def test_cffi_wrapper_factory():
453 _api_type_wrapper = {
454 "DOG_PTR": Dog,
455 "DOG_OWNER_PTR": DogOwner,
456 }
458 wf_not_strict = CffiWrapperFactory(_api_type_wrapper, False)
459 wf_strict = CffiWrapperFactory(_api_type_wrapper, True)
460 pointer = ut_dll.create_dog()
461 with pytest.raises(ValueError):
462 _ = wf_not_strict.create_wrapper(pointer, None, ut_dll.release)
463 # https://en.wikipedia.org/wiki/The_Thing_(1982_film)
464 x = wf_not_strict.create_wrapper(pointer, "THE_THING_PTR", ut_dll.release)
465 assert isinstance(x, DeletableCffiNativeHandle)
466 assert not isinstance(x, Dog)
467 del x
468 gc.collect()
469 pointer = ut_dll.create_dog()
470 # if strict, we refuse to construct a wrapper outside of the known type identifiers
471 with pytest.raises(ValueError):
472 _ = wf_strict.create_wrapper(pointer, "THE_THING_PTR", ut_dll.release)
473 dog = wf_not_strict.create_wrapper(pointer, "DOG_PTR", ut_dll.release)
474 assert isinstance(dog, Dog)
475 del dog
476 gc.collect()
478 # Test the unexpected cases, where we have a pointer but no
479 # identified native type, and a strict requirement to have one.
480 pointer_croc = ut_dll.create_croc()
481 with pytest.raises(ValueError):
482 wf_strict.create_wrapper(pointer_croc, "CROC_PTR", ut_dll.release)
483 # To increase UT coverage mostly, we will test the case where we have a pointer but no identified python wrapper type.
484 _api_type_wrapper.update({"CROC_PTR": None})
485 with pytest.raises(NotImplementedError):
486 wf_strict.create_wrapper(pointer_croc, "CROC_PTR", ut_dll.release)
487 # Test the case where we have a pointer but no identified python wrapper type, but we are not strict ad use a generic wrapper.
488 anonymous_croc = wf_not_strict.create_wrapper(
489 pointer_croc,
490 "CROC_PTR",
491 ut_dll.release,
492 )
493 assert isinstance(anonymous_croc, DeletableCffiNativeHandle)
494 del anonymous_croc
495 gc.collect()
498def test_cffi_wrapper_factory_various_ctors():
499 """Sweep the various supported wrapper constructors for the wrapper factory"""
500 _api_type_wrapper = {"DOG_PTR": Dog, "DOG_OWNER_PTR": DogOwner, "CROC_PTR": None}
501 wf_strict = CffiWrapperFactory(_api_type_wrapper, True)
502 # we cannot create a wrapper for a type that has no constructor: how would it know the native pointer?
503 _api_type_wrapper.update({"CROC_PTR": CrocZeroParameters})
504 pointer_croc = ut_dll.create_croc()
505 with pytest.raises(TypeError):
506 wf_strict.create_wrapper(pointer_croc, "CROC_PTR", release_native=None)
507 # we can create a wrapper with a constructor that has one argument, the pointer
508 _api_type_wrapper.update({"CROC_PTR": CrocOneParameters})
509 croc_one = wf_strict.create_wrapper(pointer_croc, "CROC_PTR", release_native=None)
510 assert isinstance(croc_one, CrocOneParameters)
511 assert croc_one.type_id == "CROC_PTR"
512 assert croc_one._release_handle is not None
513 del croc_one
514 gc.collect()
516 # two parameters
517 pointer_croc = ut_dll.create_croc()
518 _api_type_wrapper.update({"CROC_PTR": CrocTwoParameters})
519 # if we have two parameters, we need to provide the release function
520 with pytest.raises(
521 ValueError,
522 match="Wrapper class 'CrocTwoParameters' has two constructor arguments; the argument 'release_native' cannot be None",
523 ):
524 _ = wf_strict.create_wrapper(pointer_croc, "CROC_PTR", release_native=None)
525 croc_two = wf_strict.create_wrapper(
526 pointer_croc,
527 "CROC_PTR",
528 release_native=ut_dll.release,
529 )
530 assert isinstance(croc_two, CrocTwoParameters)
531 assert croc_two.type_id == "CROC_PTR"
532 assert croc_two._release_handle is not None
533 # we cannot test the function equality easily SFAIK
534 # assert croc_two._release_handle == ut_dll.release
535 del croc_two
536 gc.collect()
538 # three
539 pointer_croc = ut_dll.create_croc()
540 _api_type_wrapper.update({"CROC_PTR": CrocThreeParameters})
542 # if we have three parameters, we need to provide the release function
543 with pytest.raises(ValueError):
544 _ = wf_strict.create_wrapper(pointer_croc, "CROC_PTR", release_native=None)
546 croc_three = wf_strict.create_wrapper(
547 pointer_croc,
548 "CROC_PTR",
549 release_native=ut_dll.release,
550 )
551 assert isinstance(croc_three, CrocThreeParameters)
552 assert croc_three.type_id == "CROC_PTR"
553 assert croc_three._release_handle is not None
554 # we cannot test the function equality easily SFAIK
555 # assert croc_three._release_handle == ut_dll.release
556 del croc_three
557 gc.collect()
559 # four
560 pointer_croc = ut_dll.create_croc()
561 _api_type_wrapper.update({"CROC_PTR": CrocFourParameters})
562 # This used not to be supported for a few months, but there is a
563 # legacy of classes (in the swift app and more) with an initial ref counter with a zero value default
564 # with pytest.raises(NotImplementedError):
565 # _ = wf_strict.create_wrapper(
566 # pointer_croc, "CROC_PTR", release_native=ut_dll.release
567 # )
568 croc_four = wf_strict.create_wrapper(
569 pointer_croc,
570 "CROC_PTR",
571 release_native=ut_dll.release,
572 )
573 assert isinstance(croc_four, CrocFourParameters)
574 assert croc_four.type_id == "CROC_PTR"
575 assert croc_four._release_handle is not None
576 # we cannot test the function equality easily SFAIK
577 # assert croc_four._release_handle == ut_dll.release
578 del croc_four
579 gc.collect()
581 # four, but not with the expected fourth parameter
582 pointer_croc = ut_dll.create_croc()
583 _api_type_wrapper.update({"CROC_PTR": CrocFourParametersWrongFourthParameter})
584 with pytest.raises(TypeError):
585 _ = wf_strict.create_wrapper(pointer_croc, "CROC_PTR", release_native=ut_dll.release)
586 # manual cleanup for the sake of being pedantic
587 ut_dll.release(pointer_croc)
588 gc.collect()
590 # five, not supported
591 pointer_croc = ut_dll.create_croc()
592 _api_type_wrapper.update({"CROC_PTR": CrocFiveParameters})
593 with pytest.raises(NotImplementedError):
594 _ = wf_strict.create_wrapper(pointer_croc, "CROC_PTR", release_native=ut_dll.release)
596 # manual cleanup for the sake of being pedantic
597 ut_dll.release(pointer_croc)
598 gc.collect()
601def test_nativehandle_default_check_valid() -> None:
602 # mostly added to increase UT coverage
603 # locks in the behavior of the default implementation
604 import pytest
606 from refcount.base import NativeHandle
608 rc = NativeHandle()
609 pointer = ut_dll.create_dog()
610 dog = wrap_cffi_native_handle(pointer, "dog", ut_dll.release)
611 with pytest.raises(NotImplementedError):
612 rc._is_valid_handle(dog)
613 dog = None
614 gc.collect()
617def test_callback_via_cffi() -> None:
618 # https://github.com/csiro-hydroinformatics/uchronia-time-series/issues/1
619 global _message_from_c
620 ut_dll.register_exception_callback(called_back_from_c)
621 ut_dll.trigger_callback()
622 assert _message_from_c != b"<none>"
625if __name__ == "__main__": 625 ↛ 626line 625 didn't jump to line 626 because the condition on line 625 was never true
626 test_callback_via_cffi()