tests/eventmock.py

   1
   2 import libvirt
   3 import libvirtmod
   4
   5 _add_handle_impl = None
   6 _update_handle_impl = None
   7 _remove_handle_impl = None
   8
   9 _add_timeout_impl = None
  10 _update_timeout_impl = None
  11 _remove_timeout_impl = None
  12
  13 _registered = False
  14
  15 def _add_handle(fd: int, event: int, cb: libvirt._EventCB, opaque: libvirt._T) -> int:
  16     global _add_handle_impl
  17     assert _add_handle_impl != None
  18     return _add_handle_impl(fd, event, cb, opaque)
  19
  20 def _update_handle(watch: int, event: int) -> None:
  21     global _update_handle_impl
  22     assert _update_handle_impl != None
  23     _update_handle_impl(watch, event)
  24
  25 def _remove_handle(watch: int) -> int:
  26     global _remove_handle_impl
  27     assert _remove_handle_impl != None
  28     return _remove_handle_impl(watch)
  29
  30 def _add_timeout(timeout: int, cb: libvirt._TimerCB, opaque: libvirt._T) -> int:
  31     global _add_timeout_impl
  32     assert _add_timeout_impl != None
  33     return _add_timeout_impl(timeout, cb, opaque)
  34
  35 def _update_timeout(timer: int, timeout: int) -> None:
  36     global _update_timeout_impl
  37     assert _update_timeout_impl != None
  38     _update_timeout_impl(timer, timeout)
  39
  40 def _remove_timeout(timer: int) -> int:
  41     global _remove_timeout_impl
  42     assert _remove_timeout_impl != None
  43     return _remove_timeout_impl(timer)
  44
  45 # libvirt.virEventRegisterImpl() is a one time call per process
  46 # This method is intended to be used with mock patching, so that
  47 # tests can get the appearance of being able to call
  48 # virEventRegisterImpl many times.
  49 #
  50 # Note, this relies on the tests closing all connection objects
  51 # and not leaving any handles/timers pending when they stop
  52 # running their event loop impl.
  53
  54 def virEventRegisterImplMock(add_handle_impl,
  55                              update_handle_impl,
  56                              remove_handle_impl,
  57                              add_timeout_impl,
  58                              update_timeout_impl,
  59                              remove_timeout_impl):
  60     global _add_handle_impl
  61     global _update_handle_impl
  62     global _remove_handle_impl
  63     global _add_timeout_impl
  64     global _update_timeout_impl
  65     global _remove_timeout_impl
  66
  67     _add_handle_impl = add_handle_impl
  68     _update_handle_impl = update_handle_impl
  69     _remove_handle_impl = remove_handle_impl
  70     _add_timeout_impl = add_timeout_impl
  71     _update_timeout_impl = update_timeout_impl
  72     _remove_timeout_impl = remove_timeout_impl
  73
  74     global _registered
  75     if not _registered:
  76         libvirtmod.virEventRegisterImpl(_add_handle,
  77                                         _update_handle,
  78                                         _remove_timeout,
  79                                         _add_timeout,
  80                                         _update_timeout,
  81                                         _remove_timeout)
  82         _registered = True