From 613b03d2778c9e0aef115f3e543c4a07aa8fa423 Mon Sep 17 00:00:00 2001 From: Wojtek Porczyk Date: Mon, 5 Jan 2015 15:39:14 +0100 Subject: [PATCH] qubes/tests: Common TestEmitter class qubes.tests.TestEmitter is intended to check whether specific event fired on given emitter. --- qubes/tests/__init__.py | 82 +++++++++++++++++++++++++++++++++++++++++ qubes/tests/vm/init.py | 49 ++++++------------------ 2 files changed, 93 insertions(+), 38 deletions(-) diff --git a/qubes/tests/__init__.py b/qubes/tests/__init__.py index 193619df..9fec048a 100644 --- a/qubes/tests/__init__.py +++ b/qubes/tests/__init__.py @@ -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 diff --git a/qubes/tests/vm/init.py b/qubes/tests/vm/init.py index fc9f87b4..1608f189 100644 --- a/qubes/tests/vm/init.py +++ b/qubes/tests/vm/init.py @@ -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):