Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_create_PoisonPillMessage_with_soft_False_set_attribute_is_soft_to_False():
msg = PoisonPillMessage(soft=False)
assert not msg.is_soft
def test_send_PoisonPillMessage_set_actor_alive_to_False(self, init_actor):
init_actor.send_control(PoisonPillMessage())
time.sleep(0.1)
assert not init_actor.is_alive()
def test_hard_kill_method(dummy_actor_mocked):
"""
Call the hard_kill method of and actor
Test if:
- a hard PoisonPillMessage was sent to the actor on the control socket
"""
dummy_actor_mocked.hard_kill()
assert dummy_actor_mocked.socket_interface.send_control.called
msg = dummy_actor_mocked.socket_interface.send_control.call_args[0][0]
assert msg == PoisonPillMessage(soft=False)
def term_handler(_, __):
self.logger.debug("Term handler")
pp_handler = self.state.get_corresponding_handler(PoisonPillMessage())
pp_handler.handle(PoisonPillMessage(soft=False))
self._kill_process()
sys.exit(0)
def term_handler(_, __):
self.logger.debug("Term handler")
pp_handler = self.state.get_corresponding_handler(PoisonPillMessage())
pp_handler.handle(PoisonPillMessage(soft=False))
self._kill_process()
sys.exit(0)
def setup(self):
"""
Define StartMessage, PoisonPillMessage handlers and a handler for
each report type
"""
self.add_handler(PoisonPillMessage, PusherPoisonPillMessageHandler(self.state))
self.add_handler(self.state.report_model.get_type(), ReportHandler(self.state, self.delay, self.max_size))
self.add_handler(StartMessage, PusherStartHandler(self.state))
"""
Initialize the database and connect all dispatcher to the
socket_interface
"""
db_puller_thread = DBPullerThread(self.state, self.timeout)
db_puller_thread.start()
while db_puller_thread.is_alive() and self.state.alive:
time.sleep(0.4)
msg = self.state.actor.receive_control(0.1)
if msg is not None:
self.handle_internal_msg(msg)
self.handle_internal_msg(PoisonPillMessage(soft=False))
def setup(self):
"""
Check if there is a primary group by rule. Set define
StartMessage, PoisonPillMessage and Report handlers
"""
Actor.setup(self)
if self.state.route_table.primary_dispatch_rule is None:
raise NoPrimaryDispatchRuleRuleException()
self.add_handler(Report, FormulaDispatcherReportHandler(self.state))
self.add_handler(PoisonPillMessage, DispatcherPoisonPillMessageHandler(self.state))
self.add_handler(StartMessage, StartHandler(self.state))
def setup(self):
"""
Define StartMessage handler and PoisonPillMessage handler
"""
self.add_handler(PoisonPillMessage, PullerPoisonPillMessageHandler(self.state))
self.add_handler(StartMessage, PullerStartHandler(self.state, 0.1))