123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966 |
- #
- # BitBake Tests for the Event implementation (event.py)
- #
- # Copyright (C) 2017 Intel Corporation
- #
- # SPDX-License-Identifier: GPL-2.0-only
- #
- import collections
- import importlib
- import logging
- import pickle
- import threading
- import time
- import unittest
- from unittest.mock import Mock
- from unittest.mock import call
- import bb
- import bb.event
- from bb.msg import BBLogFormatter
- class EventQueueStubBase(object):
- """ Base class for EventQueueStub classes """
- def __init__(self):
- self.event_calls = []
- return
- def _store_event_data_string(self, event):
- if isinstance(event, logging.LogRecord):
- formatter = BBLogFormatter("%(levelname)s: %(message)s")
- self.event_calls.append(formatter.format(event))
- else:
- self.event_calls.append(bb.event.getName(event))
- return
- class EventQueueStub(EventQueueStubBase):
- """ Class used as specification for UI event handler queue stub objects """
- def __init__(self):
- super(EventQueueStub, self).__init__()
- def send(self, event):
- super(EventQueueStub, self)._store_event_data_string(event)
- class PickleEventQueueStub(EventQueueStubBase):
- """ Class used as specification for UI event handler queue stub objects
- with sendpickle method """
- def __init__(self):
- super(PickleEventQueueStub, self).__init__()
- def sendpickle(self, pickled_event):
- event = pickle.loads(pickled_event)
- super(PickleEventQueueStub, self)._store_event_data_string(event)
- class UIClientStub(object):
- """ Class used as specification for UI event handler stub objects """
- def __init__(self):
- self.event = None
- class EventHandlingTest(unittest.TestCase):
- """ Event handling test class """
- def setUp(self):
- self._test_process = Mock()
- ui_client1 = UIClientStub()
- ui_client2 = UIClientStub()
- self._test_ui1 = Mock(wraps=ui_client1)
- self._test_ui2 = Mock(wraps=ui_client2)
- importlib.reload(bb.event)
- def _create_test_handlers(self):
- """ Method used to create a test handler ordered dictionary """
- test_handlers = collections.OrderedDict()
- test_handlers["handler1"] = self._test_process.handler1
- test_handlers["handler2"] = self._test_process.handler2
- return test_handlers
- def test_class_handlers(self):
- """ Test set_class_handlers and get_class_handlers methods """
- test_handlers = self._create_test_handlers()
- bb.event.set_class_handlers(test_handlers)
- self.assertEqual(test_handlers,
- bb.event.get_class_handlers())
- def test_handlers(self):
- """ Test set_handlers and get_handlers """
- test_handlers = self._create_test_handlers()
- bb.event.set_handlers(test_handlers)
- self.assertEqual(test_handlers,
- bb.event.get_handlers())
- def test_clean_class_handlers(self):
- """ Test clean_class_handlers method """
- cleanDict = collections.OrderedDict()
- self.assertEqual(cleanDict,
- bb.event.clean_class_handlers())
- def test_register(self):
- """ Test register method for class handlers """
- result = bb.event.register("handler", self._test_process.handler)
- self.assertEqual(result, bb.event.Registered)
- handlers_dict = bb.event.get_class_handlers()
- self.assertIn("handler", handlers_dict)
- def test_already_registered(self):
- """ Test detection of an already registed class handler """
- bb.event.register("handler", self._test_process.handler)
- handlers_dict = bb.event.get_class_handlers()
- self.assertIn("handler", handlers_dict)
- result = bb.event.register("handler", self._test_process.handler)
- self.assertEqual(result, bb.event.AlreadyRegistered)
- def test_register_from_string(self):
- """ Test register method receiving code in string """
- result = bb.event.register("string_handler", " return True")
- self.assertEqual(result, bb.event.Registered)
- handlers_dict = bb.event.get_class_handlers()
- self.assertIn("string_handler", handlers_dict)
- def test_register_with_mask(self):
- """ Test register method with event masking """
- mask = ["bb.event.OperationStarted",
- "bb.event.OperationCompleted"]
- result = bb.event.register("event_handler",
- self._test_process.event_handler,
- mask)
- self.assertEqual(result, bb.event.Registered)
- handlers_dict = bb.event.get_class_handlers()
- self.assertIn("event_handler", handlers_dict)
- def test_remove(self):
- """ Test remove method for class handlers """
- test_handlers = self._create_test_handlers()
- bb.event.set_class_handlers(test_handlers)
- count = len(test_handlers)
- bb.event.remove("handler1", None)
- test_handlers = bb.event.get_class_handlers()
- self.assertEqual(len(test_handlers), count - 1)
- with self.assertRaises(KeyError):
- bb.event.remove("handler1", None)
- def test_execute_handler(self):
- """ Test execute_handler method for class handlers """
- mask = ["bb.event.OperationProgress"]
- result = bb.event.register("event_handler",
- self._test_process.event_handler,
- mask)
- self.assertEqual(result, bb.event.Registered)
- event = bb.event.OperationProgress(current=10, total=100)
- bb.event.execute_handler("event_handler",
- self._test_process.event_handler,
- event,
- None)
- self._test_process.event_handler.assert_called_once_with(event)
- def test_fire_class_handlers(self):
- """ Test fire_class_handlers method """
- mask = ["bb.event.OperationStarted"]
- result = bb.event.register("event_handler1",
- self._test_process.event_handler1,
- mask)
- self.assertEqual(result, bb.event.Registered)
- result = bb.event.register("event_handler2",
- self._test_process.event_handler2,
- "*")
- self.assertEqual(result, bb.event.Registered)
- event1 = bb.event.OperationStarted()
- event2 = bb.event.OperationCompleted(total=123)
- bb.event.fire_class_handlers(event1, None)
- bb.event.fire_class_handlers(event2, None)
- bb.event.fire_class_handlers(event2, None)
- expected_event_handler1 = [call(event1)]
- expected_event_handler2 = [call(event1),
- call(event2),
- call(event2)]
- self.assertEqual(self._test_process.event_handler1.call_args_list,
- expected_event_handler1)
- self.assertEqual(self._test_process.event_handler2.call_args_list,
- expected_event_handler2)
- def test_class_handler_filters(self):
- """ Test filters for class handlers """
- mask = ["bb.event.OperationStarted"]
- result = bb.event.register("event_handler1",
- self._test_process.event_handler1,
- mask)
- self.assertEqual(result, bb.event.Registered)
- result = bb.event.register("event_handler2",
- self._test_process.event_handler2,
- "*")
- self.assertEqual(result, bb.event.Registered)
- bb.event.set_eventfilter(
- lambda name, handler, event, d :
- name == 'event_handler2' and
- bb.event.getName(event) == "OperationStarted")
- event1 = bb.event.OperationStarted()
- event2 = bb.event.OperationCompleted(total=123)
- bb.event.fire_class_handlers(event1, None)
- bb.event.fire_class_handlers(event2, None)
- bb.event.fire_class_handlers(event2, None)
- expected_event_handler1 = []
- expected_event_handler2 = [call(event1)]
- self.assertEqual(self._test_process.event_handler1.call_args_list,
- expected_event_handler1)
- self.assertEqual(self._test_process.event_handler2.call_args_list,
- expected_event_handler2)
- def test_change_handler_event_mapping(self):
- """ Test changing the event mapping for class handlers """
- event1 = bb.event.OperationStarted()
- event2 = bb.event.OperationCompleted(total=123)
- # register handler for all events
- result = bb.event.register("event_handler1",
- self._test_process.event_handler1,
- "*")
- self.assertEqual(result, bb.event.Registered)
- bb.event.fire_class_handlers(event1, None)
- bb.event.fire_class_handlers(event2, None)
- expected = [call(event1), call(event2)]
- self.assertEqual(self._test_process.event_handler1.call_args_list,
- expected)
- # unregister handler and register it only for OperationStarted
- bb.event.remove("event_handler1",
- self._test_process.event_handler1)
- mask = ["bb.event.OperationStarted"]
- result = bb.event.register("event_handler1",
- self._test_process.event_handler1,
- mask)
- self.assertEqual(result, bb.event.Registered)
- bb.event.fire_class_handlers(event1, None)
- bb.event.fire_class_handlers(event2, None)
- expected = [call(event1), call(event2), call(event1)]
- self.assertEqual(self._test_process.event_handler1.call_args_list,
- expected)
- # unregister handler and register it only for OperationCompleted
- bb.event.remove("event_handler1",
- self._test_process.event_handler1)
- mask = ["bb.event.OperationCompleted"]
- result = bb.event.register("event_handler1",
- self._test_process.event_handler1,
- mask)
- self.assertEqual(result, bb.event.Registered)
- bb.event.fire_class_handlers(event1, None)
- bb.event.fire_class_handlers(event2, None)
- expected = [call(event1), call(event2), call(event1), call(event2)]
- self.assertEqual(self._test_process.event_handler1.call_args_list,
- expected)
- def test_register_UIHhandler(self):
- """ Test register_UIHhandler method """
- result = bb.event.register_UIHhandler(self._test_ui1, mainui=True)
- self.assertEqual(result, 1)
- def test_UIHhandler_already_registered(self):
- """ Test registering an UIHhandler already existing """
- result = bb.event.register_UIHhandler(self._test_ui1, mainui=True)
- self.assertEqual(result, 1)
- result = bb.event.register_UIHhandler(self._test_ui1, mainui=True)
- self.assertEqual(result, 2)
- def test_unregister_UIHhandler(self):
- """ Test unregister_UIHhandler method """
- result = bb.event.register_UIHhandler(self._test_ui1, mainui=True)
- self.assertEqual(result, 1)
- result = bb.event.unregister_UIHhandler(1)
- self.assertIs(result, None)
- def test_fire_ui_handlers(self):
- """ Test fire_ui_handlers method """
- self._test_ui1.event = Mock(spec_set=EventQueueStub)
- result = bb.event.register_UIHhandler(self._test_ui1, mainui=True)
- self.assertEqual(result, 1)
- self._test_ui2.event = Mock(spec_set=PickleEventQueueStub)
- result = bb.event.register_UIHhandler(self._test_ui2, mainui=True)
- self.assertEqual(result, 2)
- event1 = bb.event.OperationStarted()
- bb.event.fire_ui_handlers(event1, None)
- expected = [call(event1)]
- self.assertEqual(self._test_ui1.event.send.call_args_list,
- expected)
- expected = [call(pickle.dumps(event1))]
- self.assertEqual(self._test_ui2.event.sendpickle.call_args_list,
- expected)
- def test_ui_handler_mask_filter(self):
- """ Test filters for UI handlers """
- mask = ["bb.event.OperationStarted"]
- debug_domains = {}
- self._test_ui1.event = Mock(spec_set=EventQueueStub)
- result = bb.event.register_UIHhandler(self._test_ui1, mainui=True)
- bb.event.set_UIHmask(result, logging.INFO, debug_domains, mask)
- self._test_ui2.event = Mock(spec_set=PickleEventQueueStub)
- result = bb.event.register_UIHhandler(self._test_ui2, mainui=True)
- bb.event.set_UIHmask(result, logging.INFO, debug_domains, mask)
- event1 = bb.event.OperationStarted()
- event2 = bb.event.OperationCompleted(total=1)
- bb.event.fire_ui_handlers(event1, None)
- bb.event.fire_ui_handlers(event2, None)
- expected = [call(event1)]
- self.assertEqual(self._test_ui1.event.send.call_args_list,
- expected)
- expected = [call(pickle.dumps(event1))]
- self.assertEqual(self._test_ui2.event.sendpickle.call_args_list,
- expected)
- def test_ui_handler_log_filter(self):
- """ Test log filters for UI handlers """
- mask = ["*"]
- debug_domains = {'BitBake.Foo': logging.WARNING}
- self._test_ui1.event = EventQueueStub()
- result = bb.event.register_UIHhandler(self._test_ui1, mainui=True)
- bb.event.set_UIHmask(result, logging.ERROR, debug_domains, mask)
- self._test_ui2.event = PickleEventQueueStub()
- result = bb.event.register_UIHhandler(self._test_ui2, mainui=True)
- bb.event.set_UIHmask(result, logging.ERROR, debug_domains, mask)
- event1 = bb.event.OperationStarted()
- bb.event.fire_ui_handlers(event1, None) # All events match
- event_log_handler = bb.event.LogHandler()
- logger = logging.getLogger("BitBake")
- logger.addHandler(event_log_handler)
- logger1 = logging.getLogger("BitBake.Foo")
- logger1.warning("Test warning LogRecord1") # Matches debug_domains level
- logger1.info("Test info LogRecord") # Filtered out
- logger2 = logging.getLogger("BitBake.Bar")
- logger2.error("Test error LogRecord") # Matches filter base level
- logger2.warning("Test warning LogRecord2") # Filtered out
- logger.removeHandler(event_log_handler)
- expected = ['OperationStarted',
- 'WARNING: Test warning LogRecord1',
- 'ERROR: Test error LogRecord']
- self.assertEqual(self._test_ui1.event.event_calls, expected)
- self.assertEqual(self._test_ui2.event.event_calls, expected)
- def test_fire(self):
- """ Test fire method used to trigger class and ui event handlers """
- mask = ["bb.event.ConfigParsed"]
- result = bb.event.register("event_handler1",
- self._test_process.event_handler1,
- mask)
- self._test_ui1.event = Mock(spec_set=EventQueueStub)
- result = bb.event.register_UIHhandler(self._test_ui1, mainui=True)
- self.assertEqual(result, 1)
- event1 = bb.event.ConfigParsed()
- bb.event.fire(event1, None)
- expected = [call(event1)]
- self.assertEqual(self._test_process.event_handler1.call_args_list,
- expected)
- self.assertEqual(self._test_ui1.event.send.call_args_list,
- expected)
- def test_fire_from_worker(self):
- """ Test fire_from_worker method """
- self._test_ui1.event = Mock(spec_set=EventQueueStub)
- result = bb.event.register_UIHhandler(self._test_ui1, mainui=True)
- self.assertEqual(result, 1)
- event1 = bb.event.ConfigParsed()
- bb.event.fire_from_worker(event1, None)
- expected = [call(event1)]
- self.assertEqual(self._test_ui1.event.send.call_args_list,
- expected)
- def test_worker_fire(self):
- """ Test the triggering of bb.event.worker_fire callback """
- bb.event.worker_fire = Mock()
- event = bb.event.Event()
- bb.event.fire(event, None)
- expected = [call(event, None)]
- self.assertEqual(bb.event.worker_fire.call_args_list, expected)
- def test_print_ui_queue(self):
- """ Test print_ui_queue method """
- event1 = bb.event.OperationStarted()
- event2 = bb.event.OperationCompleted(total=123)
- bb.event.fire(event1, None)
- bb.event.fire(event2, None)
- event_log_handler = bb.event.LogHandler()
- logger = logging.getLogger("BitBake")
- logger.addHandler(event_log_handler)
- logger.info("Test info LogRecord")
- logger.warning("Test warning LogRecord")
- with self.assertLogs("BitBake", level="INFO") as cm:
- bb.event.print_ui_queue()
- logger.removeHandler(event_log_handler)
- self.assertEqual(cm.output,
- ["INFO:BitBake:Test info LogRecord",
- "WARNING:BitBake:Test warning LogRecord"])
- def _set_threadlock_test_mockups(self):
- """ Create UI event handler mockups used in enable and disable
- threadlock tests """
- def ui1_event_send(event):
- if type(event) is bb.event.ConfigParsed:
- self._threadlock_test_calls.append("w1_ui1")
- if type(event) is bb.event.OperationStarted:
- self._threadlock_test_calls.append("w2_ui1")
- time.sleep(2)
- def ui2_event_send(event):
- if type(event) is bb.event.ConfigParsed:
- self._threadlock_test_calls.append("w1_ui2")
- if type(event) is bb.event.OperationStarted:
- self._threadlock_test_calls.append("w2_ui2")
- time.sleep(2)
- self._threadlock_test_calls = []
- self._test_ui1.event = EventQueueStub()
- self._test_ui1.event.send = ui1_event_send
- result = bb.event.register_UIHhandler(self._test_ui1, mainui=True)
- self.assertEqual(result, 1)
- self._test_ui2.event = EventQueueStub()
- self._test_ui2.event.send = ui2_event_send
- result = bb.event.register_UIHhandler(self._test_ui2, mainui=True)
- self.assertEqual(result, 2)
- def _set_and_run_threadlock_test_workers(self):
- """ Create and run the workers used to trigger events in enable and
- disable threadlock tests """
- worker1 = threading.Thread(target=self._thread_lock_test_worker1)
- worker2 = threading.Thread(target=self._thread_lock_test_worker2)
- worker1.start()
- time.sleep(1)
- worker2.start()
- worker1.join()
- worker2.join()
- def _thread_lock_test_worker1(self):
- """ First worker used to fire the ConfigParsed event for enable and
- disable threadlocks tests """
- bb.event.fire(bb.event.ConfigParsed(), None)
- def _thread_lock_test_worker2(self):
- """ Second worker used to fire the OperationStarted event for enable
- and disable threadlocks tests """
- bb.event.fire(bb.event.OperationStarted(), None)
- def test_enable_threadlock(self):
- """ Test enable_threadlock method """
- self._set_threadlock_test_mockups()
- bb.event.enable_threadlock()
- self._set_and_run_threadlock_test_workers()
- # Calls to UI handlers should be in order as all the registered
- # handlers for the event coming from the first worker should be
- # called before processing the event from the second worker.
- self.assertEqual(self._threadlock_test_calls,
- ["w1_ui1", "w1_ui2", "w2_ui1", "w2_ui2"])
- def test_disable_threadlock(self):
- """ Test disable_threadlock method """
- self._set_threadlock_test_mockups()
- bb.event.disable_threadlock()
- self._set_and_run_threadlock_test_workers()
- # Calls to UI handlers should be intertwined together. Thanks to the
- # delay in the registered handlers for the event coming from the first
- # worker, the event coming from the second worker starts being
- # processed before finishing handling the first worker event.
- self.assertEqual(self._threadlock_test_calls,
- ["w1_ui1", "w2_ui1", "w1_ui2", "w2_ui2"])
- class EventClassesTest(unittest.TestCase):
- """ Event classes test class """
- _worker_pid = 54321
- def setUp(self):
- bb.event.worker_pid = EventClassesTest._worker_pid
- def test_Event(self):
- """ Test the Event base class """
- event = bb.event.Event()
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_HeartbeatEvent(self):
- """ Test the HeartbeatEvent class """
- time = 10
- event = bb.event.HeartbeatEvent(time)
- self.assertEqual(event.time, time)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_OperationStarted(self):
- """ Test OperationStarted event class """
- msg = "Foo Bar"
- event = bb.event.OperationStarted(msg)
- self.assertEqual(event.msg, msg)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_OperationCompleted(self):
- """ Test OperationCompleted event class """
- msg = "Foo Bar"
- total = 123
- event = bb.event.OperationCompleted(total, msg)
- self.assertEqual(event.msg, msg)
- self.assertEqual(event.total, total)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_OperationProgress(self):
- """ Test OperationProgress event class """
- msg = "Foo Bar"
- total = 123
- current = 111
- event = bb.event.OperationProgress(current, total, msg)
- self.assertEqual(event.msg, msg + ": %s/%s" % (current, total))
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_ConfigParsed(self):
- """ Test the ConfigParsed class """
- event = bb.event.ConfigParsed()
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_MultiConfigParsed(self):
- """ Test MultiConfigParsed event class """
- mcdata = {"foobar": "Foo Bar"}
- event = bb.event.MultiConfigParsed(mcdata)
- self.assertEqual(event.mcdata, mcdata)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_RecipeEvent(self):
- """ Test RecipeEvent event base class """
- callback = lambda a: 2 * a
- event = bb.event.RecipeEvent(callback)
- self.assertEqual(event.fn(1), callback(1))
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_RecipePreFinalise(self):
- """ Test RecipePreFinalise event class """
- callback = lambda a: 2 * a
- event = bb.event.RecipePreFinalise(callback)
- self.assertEqual(event.fn(1), callback(1))
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_RecipeTaskPreProcess(self):
- """ Test RecipeTaskPreProcess event class """
- callback = lambda a: 2 * a
- tasklist = [("foobar", callback)]
- event = bb.event.RecipeTaskPreProcess(callback, tasklist)
- self.assertEqual(event.fn(1), callback(1))
- self.assertEqual(event.tasklist, tasklist)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_RecipeParsed(self):
- """ Test RecipeParsed event base class """
- callback = lambda a: 2 * a
- event = bb.event.RecipeParsed(callback)
- self.assertEqual(event.fn(1), callback(1))
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_BuildBase(self):
- """ Test base class for bitbake build events """
- name = "foo"
- pkgs = ["bar"]
- failures = 123
- event = bb.event.BuildBase(name, pkgs, failures)
- self.assertEqual(event.name, name)
- self.assertEqual(event.pkgs, pkgs)
- self.assertEqual(event.getFailures(), failures)
- name = event.name = "bar"
- pkgs = event.pkgs = ["foo"]
- self.assertEqual(event.name, name)
- self.assertEqual(event.pkgs, pkgs)
- self.assertEqual(event.getFailures(), failures)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_BuildInit(self):
- """ Test class for bitbake build invocation events """
- event = bb.event.BuildInit()
- self.assertEqual(event.name, None)
- self.assertEqual(event.pkgs, [])
- self.assertEqual(event.getFailures(), 0)
- name = event.name = "bar"
- pkgs = event.pkgs = ["foo"]
- self.assertEqual(event.name, name)
- self.assertEqual(event.pkgs, pkgs)
- self.assertEqual(event.getFailures(), 0)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_BuildStarted(self):
- """ Test class for build started events """
- name = "foo"
- pkgs = ["bar"]
- failures = 123
- event = bb.event.BuildStarted(name, pkgs, failures)
- self.assertEqual(event.name, name)
- self.assertEqual(event.pkgs, pkgs)
- self.assertEqual(event.getFailures(), failures)
- self.assertEqual(event.msg, "Building Started")
- name = event.name = "bar"
- pkgs = event.pkgs = ["foo"]
- msg = event.msg = "foobar"
- self.assertEqual(event.name, name)
- self.assertEqual(event.pkgs, pkgs)
- self.assertEqual(event.getFailures(), failures)
- self.assertEqual(event.msg, msg)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_BuildCompleted(self):
- """ Test class for build completed events """
- total = 1000
- name = "foo"
- pkgs = ["bar"]
- failures = 123
- interrupted = 1
- event = bb.event.BuildCompleted(total, name, pkgs, failures,
- interrupted)
- self.assertEqual(event.name, name)
- self.assertEqual(event.pkgs, pkgs)
- self.assertEqual(event.getFailures(), failures)
- self.assertEqual(event.msg, "Building Failed")
- event2 = bb.event.BuildCompleted(total, name, pkgs)
- self.assertEqual(event2.name, name)
- self.assertEqual(event2.pkgs, pkgs)
- self.assertEqual(event2.getFailures(), 0)
- self.assertEqual(event2.msg, "Building Succeeded")
- self.assertEqual(event2.pid, EventClassesTest._worker_pid)
- def test_DiskFull(self):
- """ Test DiskFull event class """
- dev = "/dev/foo"
- type = "ext4"
- freespace = "104M"
- mountpoint = "/"
- event = bb.event.DiskFull(dev, type, freespace, mountpoint)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_MonitorDiskEvent(self):
- """ Test MonitorDiskEvent class """
- available_bytes = 10000000
- free_bytes = 90000000
- total_bytes = 1000000000
- du = bb.event.DiskUsageSample(available_bytes, free_bytes,
- total_bytes)
- event = bb.event.MonitorDiskEvent(du)
- self.assertEqual(event.disk_usage.available_bytes, available_bytes)
- self.assertEqual(event.disk_usage.free_bytes, free_bytes)
- self.assertEqual(event.disk_usage.total_bytes, total_bytes)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_NoProvider(self):
- """ Test NoProvider event class """
- item = "foobar"
- event1 = bb.event.NoProvider(item)
- self.assertEqual(event1.getItem(), item)
- self.assertEqual(event1.isRuntime(), False)
- self.assertEqual(str(event1), "Nothing PROVIDES 'foobar'")
- runtime = True
- dependees = ["foo", "bar"]
- reasons = None
- close_matches = ["foibar", "footbar"]
- event2 = bb.event.NoProvider(item, runtime, dependees, reasons,
- close_matches)
- self.assertEqual(event2.isRuntime(), True)
- expected = ("Nothing RPROVIDES 'foobar' (but foo, bar RDEPENDS"
- " on or otherwise requires it). Close matches:\n"
- " foibar\n"
- " footbar")
- self.assertEqual(str(event2), expected)
- reasons = ["Item does not exist on database"]
- close_matches = ["foibar", "footbar"]
- event3 = bb.event.NoProvider(item, runtime, dependees, reasons,
- close_matches)
- expected = ("Nothing RPROVIDES 'foobar' (but foo, bar RDEPENDS"
- " on or otherwise requires it)\n"
- "Item does not exist on database")
- self.assertEqual(str(event3), expected)
- self.assertEqual(event3.pid, EventClassesTest._worker_pid)
- def test_MultipleProviders(self):
- """ Test MultipleProviders event class """
- item = "foobar"
- candidates = ["foobarv1", "foobars"]
- event1 = bb.event.MultipleProviders(item, candidates)
- self.assertEqual(event1.isRuntime(), False)
- self.assertEqual(event1.getItem(), item)
- self.assertEqual(event1.getCandidates(), candidates)
- expected = ("Multiple providers are available for foobar (foobarv1,"
- " foobars)\n"
- "Consider defining a PREFERRED_PROVIDER entry to match "
- "foobar")
- self.assertEqual(str(event1), expected)
- runtime = True
- event2 = bb.event.MultipleProviders(item, candidates, runtime)
- self.assertEqual(event2.isRuntime(), runtime)
- expected = ("Multiple providers are available for runtime foobar "
- "(foobarv1, foobars)\n"
- "Consider defining a PREFERRED_RPROVIDER entry to match "
- "foobar")
- self.assertEqual(str(event2), expected)
- self.assertEqual(event2.pid, EventClassesTest._worker_pid)
- def test_ParseStarted(self):
- """ Test ParseStarted event class """
- total = 123
- event = bb.event.ParseStarted(total)
- self.assertEqual(event.msg, "Recipe parsing Started")
- self.assertEqual(event.total, total)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_ParseCompleted(self):
- """ Test ParseCompleted event class """
- cached = 10
- parsed = 13
- skipped = 7
- virtuals = 2
- masked = 1
- errors = 0
- total = 23
- event = bb.event.ParseCompleted(cached, parsed, skipped, masked,
- virtuals, errors, total)
- self.assertEqual(event.msg, "Recipe parsing Completed")
- expected = [cached, parsed, skipped, virtuals, masked, errors,
- cached + parsed, total]
- actual = [event.cached, event.parsed, event.skipped, event.virtuals,
- event.masked, event.errors, event.sofar, event.total]
- self.assertEqual(str(actual), str(expected))
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_ParseProgress(self):
- """ Test ParseProgress event class """
- current = 10
- total = 100
- event = bb.event.ParseProgress(current, total)
- self.assertEqual(event.msg,
- "Recipe parsing" + ": %s/%s" % (current, total))
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_CacheLoadStarted(self):
- """ Test CacheLoadStarted event class """
- total = 123
- event = bb.event.CacheLoadStarted(total)
- self.assertEqual(event.msg, "Loading cache Started")
- self.assertEqual(event.total, total)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_CacheLoadProgress(self):
- """ Test CacheLoadProgress event class """
- current = 10
- total = 100
- event = bb.event.CacheLoadProgress(current, total)
- self.assertEqual(event.msg,
- "Loading cache" + ": %s/%s" % (current, total))
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_CacheLoadCompleted(self):
- """ Test CacheLoadCompleted event class """
- total = 23
- num_entries = 12
- event = bb.event.CacheLoadCompleted(total, num_entries)
- self.assertEqual(event.msg, "Loading cache Completed")
- expected = [total, num_entries]
- actual = [event.total, event.num_entries]
- self.assertEqual(str(actual), str(expected))
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_TreeDataPreparationStarted(self):
- """ Test TreeDataPreparationStarted event class """
- event = bb.event.TreeDataPreparationStarted()
- self.assertEqual(event.msg, "Preparing tree data Started")
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_TreeDataPreparationProgress(self):
- """ Test TreeDataPreparationProgress event class """
- current = 10
- total = 100
- event = bb.event.TreeDataPreparationProgress(current, total)
- self.assertEqual(event.msg,
- "Preparing tree data" + ": %s/%s" % (current, total))
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_TreeDataPreparationCompleted(self):
- """ Test TreeDataPreparationCompleted event class """
- total = 23
- event = bb.event.TreeDataPreparationCompleted(total)
- self.assertEqual(event.msg, "Preparing tree data Completed")
- self.assertEqual(event.total, total)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_DepTreeGenerated(self):
- """ Test DepTreeGenerated event class """
- depgraph = Mock()
- event = bb.event.DepTreeGenerated(depgraph)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_TargetsTreeGenerated(self):
- """ Test TargetsTreeGenerated event class """
- model = Mock()
- event = bb.event.TargetsTreeGenerated(model)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_ReachableStamps(self):
- """ Test ReachableStamps event class """
- stamps = [Mock(), Mock()]
- event = bb.event.ReachableStamps(stamps)
- self.assertEqual(event.stamps, stamps)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_FilesMatchingFound(self):
- """ Test FilesMatchingFound event class """
- pattern = "foo.*bar"
- matches = ["foobar"]
- event = bb.event.FilesMatchingFound(pattern, matches)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_ConfigFilesFound(self):
- """ Test ConfigFilesFound event class """
- variable = "FOO_BAR"
- values = ["foo", "bar"]
- event = bb.event.ConfigFilesFound(variable, values)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_ConfigFilePathFound(self):
- """ Test ConfigFilePathFound event class """
- path = "/foo/bar"
- event = bb.event.ConfigFilePathFound(path)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_message_classes(self):
- """ Test message event classes """
- msg = "foobar foo bar"
- event = bb.event.MsgBase(msg)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- event = bb.event.MsgDebug(msg)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- event = bb.event.MsgNote(msg)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- event = bb.event.MsgWarn(msg)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- event = bb.event.MsgError(msg)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- event = bb.event.MsgFatal(msg)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- event = bb.event.MsgPlain(msg)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_LogExecTTY(self):
- """ Test LogExecTTY event class """
- msg = "foo bar"
- prog = "foo.sh"
- sleep_delay = 10
- retries = 3
- event = bb.event.LogExecTTY(msg, prog, sleep_delay, retries)
- self.assertEqual(event.msg, msg)
- self.assertEqual(event.prog, prog)
- self.assertEqual(event.sleep_delay, sleep_delay)
- self.assertEqual(event.retries, retries)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def _throw_zero_division_exception(self):
- a = 1 / 0
- return
- def _worker_handler(self, event, d):
- self._returned_event = event
- return
- def test_LogHandler(self):
- """ Test LogHandler class """
- logger = logging.getLogger("TestEventClasses")
- logger.propagate = False
- handler = bb.event.LogHandler(logging.INFO)
- logger.addHandler(handler)
- bb.event.worker_fire = self._worker_handler
- try:
- self._throw_zero_division_exception()
- except ZeroDivisionError as ex:
- logger.exception(ex)
- event = self._returned_event
- try:
- pe = pickle.dumps(event)
- newevent = pickle.loads(pe)
- except:
- self.fail('Logged event is not serializable')
- self.assertEqual(event.taskpid, EventClassesTest._worker_pid)
- def test_MetadataEvent(self):
- """ Test MetadataEvent class """
- eventtype = "footype"
- eventdata = {"foo": "bar"}
- event = bb.event.MetadataEvent(eventtype, eventdata)
- self.assertEqual(event.type, eventtype)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_ProcessStarted(self):
- """ Test ProcessStarted class """
- processname = "foo"
- total = 9783128974
- event = bb.event.ProcessStarted(processname, total)
- self.assertEqual(event.processname, processname)
- self.assertEqual(event.total, total)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_ProcessProgress(self):
- """ Test ProcessProgress class """
- processname = "foo"
- progress = 243224
- event = bb.event.ProcessProgress(processname, progress)
- self.assertEqual(event.processname, processname)
- self.assertEqual(event.progress, progress)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_ProcessFinished(self):
- """ Test ProcessFinished class """
- processname = "foo"
- total = 1242342344
- event = bb.event.ProcessFinished(processname)
- self.assertEqual(event.processname, processname)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_SanityCheck(self):
- """ Test SanityCheck class """
- event1 = bb.event.SanityCheck()
- self.assertEqual(event1.generateevents, True)
- self.assertEqual(event1.pid, EventClassesTest._worker_pid)
- generateevents = False
- event2 = bb.event.SanityCheck(generateevents)
- self.assertEqual(event2.generateevents, generateevents)
- self.assertEqual(event2.pid, EventClassesTest._worker_pid)
- def test_SanityCheckPassed(self):
- """ Test SanityCheckPassed class """
- event = bb.event.SanityCheckPassed()
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
- def test_SanityCheckFailed(self):
- """ Test SanityCheckFailed class """
- msg = "The sanity test failed."
- event1 = bb.event.SanityCheckFailed(msg)
- self.assertEqual(event1.pid, EventClassesTest._worker_pid)
- network_error = True
- event2 = bb.event.SanityCheckFailed(msg, network_error)
- self.assertEqual(event2.pid, EventClassesTest._worker_pid)
- def test_network_event_classes(self):
- """ Test network event classes """
- event1 = bb.event.NetworkTest()
- generateevents = False
- self.assertEqual(event1.pid, EventClassesTest._worker_pid)
- event2 = bb.event.NetworkTest(generateevents)
- self.assertEqual(event2.pid, EventClassesTest._worker_pid)
- event3 = bb.event.NetworkTestPassed()
- self.assertEqual(event3.pid, EventClassesTest._worker_pid)
- event4 = bb.event.NetworkTestFailed()
- self.assertEqual(event4.pid, EventClassesTest._worker_pid)
- def test_FindSigInfoResult(self):
- """ Test FindSigInfoResult event class """
- result = [Mock()]
- event = bb.event.FindSigInfoResult(result)
- self.assertEqual(event.result, result)
- self.assertEqual(event.pid, EventClassesTest._worker_pid)
|