qubes/tests: Common TestEmitter class

qubes.tests.TestEmitter is intended to check whether specific event fired on
given emitter.
This commit is contained in:
Wojtek Porczyk 2015-01-05 15:39:14 +01:00
parent a82bf7cc54
commit 613b03d277
2 changed files with 93 additions and 38 deletions

View File

@ -1,10 +1,92 @@
#!/usr/bin/python -O
import collections
import unittest
import qubes.events
class TestEmitter(qubes.events.Emitter):
'''Dummy event emitter which records events fired on it.
Events are counted in :py:attr:`fired_events` attribute, which is
:py:class:`collections.Counter` instance. For each event, ``(event, args,
kwargs)`` object is counted. *event* is event name (a string), *args* is
tuple with positional arguments and *kwargs* is sorted tuple of items from
keyword arguments.
>>> emitter = TestEmitter()
>>> emitter.fired_events
Counter()
>>> emitter.fire_event('event', 1, 2, 3, spam='eggs', foo='bar')
>>> emitter.fired_events
Counter({('event', (1, 2, 3), (('foo', 'bar'), ('spam', 'eggs'))): 1})
'''
def __init__(self, *args, **kwargs):
super(TestEmitter, self).__init__(*args, **kwargs)
#: :py:class:`collections.Counter` instance
self.fired_events = collections.Counter()
def fire_event(self, event, *args, **kwargs):
super(TestEmitter, self).fire_event(event, *args, **kwargs)
self.fired_events[(event, args, tuple(sorted(kwargs.items())))] += 1
def fire_event_pre(self, event, *args, **kwargs):
super(TestEmitter, self).fire_event_pre(event, *args, **kwargs)
self.fired_events[(event, args, tuple(sorted(kwargs.items())))] += 1
class QubesTestCase(unittest.TestCase):
'''Base class for Qubes unit tests.
'''
def __str__(self):
return '{}/{}/{}'.format(
'.'.join(self.__class__.__module__.split('.')[2:]),
self.__class__.__name__,
self._testMethodName)
def assertEventFired(self, emitter, event, args=[], kwargs=[]):
'''Check whether event was fired on given emitter and fail if it did
not.
:param TestEmitter emitter: emitter which is being checked
:param str event: event identifier
:param list args: when given, all items must appear in args passed to event
:param list kwargs: when given, all items must appear in kwargs passed to event
'''
for ev, ev_args, ev_kwargs in emitter.fired_events:
if ev != event:
continue
if any(i not in ev_args for i in args):
continue
if any(i not in ev_kwargs for i in kwargs):
continue
return
self.fail('event {!r} did not fire on {!r}'.format(event, emitter))
def assertEventNotFired(self, emitter, event, args=[], kwargs=[]):
'''Check whether event was fired on given emitter. Fail if it did.
:param TestEmitter emitter: emitter which is being checked
:param str event: event identifier
:param list args: when given, all items must appear in args passed to event
:param list kwargs: when given, all items must appear in kwargs passed to event
'''
for ev, ev_args, ev_kwargs in emitter.fired_events:
if ev != event:
continue
if any(i not in ev_args for i in args):
continue
if any(i not in ev_kwargs for i in kwargs):
continue
self.fail('event {!r} did fire on {!r}'.format(event, emitter))
return

View File

@ -12,36 +12,9 @@ import qubes.vm
import qubes.tests
class TestEmitter(qubes.events.Emitter):
def __init__(self):
super(TestEmitter, self).__init__()
self.device_pre_attached_fired = False
self.device_attached_fired = False
self.device_pre_detached_fired = False
self.device_detached_fired = False
@qubes.events.handler('device-pre-attached:testclass')
def on_device_pre_attached(self, event, dev):
self.device_pre_attached_fired = True
@qubes.events.handler('device-attached:testclass')
def on_device_attached(self, event, dev):
if self.device_pre_attached_fired:
self.device_attached_fired = True
@qubes.events.handler('device-pre-detached:testclass')
def on_device_pre_detached(self, event, dev):
if self.device_attached_fired:
self.device_pre_detached_fired = True
@qubes.events.handler('device-detached:testclass')
def on_device_detached(self, event, dev):
if self.device_pre_detached_fired:
self.device_detached_fired = True
class TC_00_DeviceCollection(qubes.tests.QubesTestCase):
def setUp(self):
self.emitter = TestEmitter()
self.emitter = qubes.tests.TestEmitter()
self.collection = qubes.vm.DeviceCollection(self.emitter, 'testclass')
def test_000_init(self):
@ -49,18 +22,18 @@ class TC_00_DeviceCollection(qubes.tests.QubesTestCase):
def test_001_attach(self):
self.collection.attach('testdev')
self.assertTrue(self.emitter.device_pre_attached_fired)
self.assertTrue(self.emitter.device_attached_fired)
self.assertFalse(self.emitter.device_pre_detached_fired)
self.assertFalse(self.emitter.device_detached_fired)
self.assertEventFired(self.emitter, 'device-pre-attached:testclass')
self.assertEventFired(self.emitter, 'device-attached:testclass')
self.assertEventNotFired(self.emitter, 'device-pre-detached:testclass')
self.assertEventNotFired(self.emitter, 'device-detached:testclass')
def test_002_detach(self):
self.collection.attach('testdev')
self.collection.detach('testdev')
self.assertTrue(self.emitter.device_pre_attached_fired)
self.assertTrue(self.emitter.device_attached_fired)
self.assertTrue(self.emitter.device_pre_detached_fired)
self.assertTrue(self.emitter.device_detached_fired)
self.assertEventFired(self.emitter, 'device-pre-attached:testclass')
self.assertEventFired(self.emitter, 'device-attached:testclass')
self.assertEventFired(self.emitter, 'device-pre-detached:testclass')
self.assertEventFired(self.emitter, 'device-detached:testclass')
def test_010_empty_detach(self):
with self.assertRaises(LookupError):
@ -82,7 +55,7 @@ class TC_00_DeviceCollection(qubes.tests.QubesTestCase):
class TC_01_DeviceManager(qubes.tests.QubesTestCase):
def setUp(self):
self.emitter = TestEmitter()
self.emitter = qubes.tests.TestEmitter()
self.manager = qubes.vm.DeviceManager(self.emitter)
def test_000_init(self):
@ -90,7 +63,7 @@ class TC_01_DeviceManager(qubes.tests.QubesTestCase):
def test_001_missing(self):
self.manager['testclass'].attach('testdev')
self.assertTrue(self.emitter.device_attached_fired)
self.assertEventFired(self.emitter, 'device-attached:testclass')
class TestVM(qubes.vm.BaseVM):