services/nodemanager/tests/test_failure.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import absolute_import, print_function
   4
   5 import errno
   6 import logging
   7 import threading
   8 import unittest
   9
  10 import mock
  11 import pykka
  12
  13 from . import testutil
  14
  15 import arvnodeman.baseactor
  16
  17 class BogusActor(arvnodeman.baseactor.BaseNodeManagerActor):
  18     def __init__(self, e):
  19         super(BogusActor, self).__init__()
  20         self.exp = e
  21
  22     def doStuff(self):
  23         raise self.exp
  24
  25 class ActorUnhandledExceptionTest(unittest.TestCase):
  26     def test_fatal_error(self):
  27         for e in (MemoryError(), threading.ThreadError(), OSError(errno.ENOMEM, "")):
  28             with mock.patch('os.killpg') as killpg_mock:
  29                 act = BogusActor.start(e).tell_proxy()
  30                 act.doStuff()
  31                 act.actor_ref.stop(block=True)
  32                 self.assertTrue(killpg_mock.called)
  33
  34     def test_nonfatal_error(self):
  35         with mock.patch('os.killpg') as killpg_mock:
  36             act = BogusActor.start(OSError(errno.ENOENT, "")).tell_proxy()
  37             act.doStuff()
  38             act.actor_ref.stop(block=True)
  39             self.assertFalse(killpg_mock.called)