Commit 7072ceba authored by Jorrit Schaap's avatar Jorrit Schaap

Merge branch 'SW-818' into 'LOFAR-Release-4_0'

SW-818: fixed t_trigger_service.py and made sure that Popen is mocked away so...

See merge request ro/lofar!63
parents d9df1338 b22c35b3
......@@ -60,9 +60,6 @@ validationrpc = ValidationRPC.create(exchange=DEFAULT_BUSNAME, broker=DEFAULT_BR
specificationrpc = SpecificationRPC.create(exchange=DEFAULT_BUSNAME, broker=DEFAULT_BROKER)
translationrpc = TranslationRPC.create(exchange=DEFAULT_BUSNAME, broker=DEFAULT_BROKER)
notification_bus = ToBus(exchange=DEFAULT_BUSNAME, broker=DEFAULT_BROKER)
def _auth_allows_triggers(project):
response = momqueryrpc.allows_triggers(project)
return response['allows']
......@@ -121,7 +118,8 @@ def _send_notification(user, host, project, trigger_id, metadata):
"metadata": metadata
}
msg = EventMessage(subject=DEFAULT_TRIGGER_NOTIFICATION_SUBJECT, content=content)
notification_bus.send(msg)
with ToBus(exchange=DEFAULT_BUSNAME, broker=DEFAULT_BROKER) as notification_bus:
notification_bus.send(msg)
except Exception as err:
logger.error("Could not send notification ->" + str(err))
......@@ -139,8 +137,6 @@ class TriggerServiceMessageHandler(ServiceMessageHandler):
# handle QPID messages
super(TriggerServiceMessageHandler, self).__init__()
notification_bus.open()
self.service2MethodMap = {
'handle_trigger': self.handle_trigger,
}
......
......@@ -46,8 +46,7 @@ class TestTriggerHandler(unittest.TestCase):
with open(TRIGGER_PATH) as f:
cls.trigger_xml = f.read()
with mock.patch('lofar.triggerservices.trigger_service.notification_bus'):
cls.handler = TriggerServiceMessageHandler()
cls.handler = TriggerServiceMessageHandler()
def setUp(self):
logging.info('-----------------')
......@@ -116,13 +115,16 @@ class TestALERTHandler(unittest.TestCase):
cls.voevent_xml = f.read()
cls.voevent_etree = etree.fromstring(cls.voevent_xml)
with mock.patch('lofar.triggerservices.trigger_service.notification_bus'):
cls.handler = ALERTHandler()
cls.handler = ALERTHandler()
def setUp(self):
logging.info('-----------------')
logging.info('Setup test %s' % self._testMethodName)
popen_patcher = mock.patch('lofar.common.subprocess_utils.Popen')
self.addCleanup(popen_patcher.stop)
popen_patcher.start()
def test_valid_voevent_should_invoke_tbb_dump(self):
with mock.patch('lofar.mac.tbbservice.client.tbbservice_rpc.TBBRPC.do_tbb_subband_dump') as dump, \
mock.patch('lofar.triggerservices.voevent_decider.ALERTDecider') as sciencecheck:
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment