Source code for thriftworker.tests.app.test_hub

from __future__ import absolute_import

from greenlet import GreenletExit

from thriftworker.hub import sleep
from thriftworker.tests.utils import TestCase, CustomAppMixin, \
    start_stop_ctx, GreenTest


[docs]class ExpectedError(Exception): pass
[docs]class TestHub(CustomAppMixin, TestCase):
[docs] def setUp(self): super(TestHub, self).setUp() self.hub = self.app.hub
[docs] def context(self): hub = self.hub return start_stop_ctx(hub)
[docs] def test_start_stop(self): hub = self.hub with self.context(): self.assertIsNotNone(hub._guard) self.assertTrue(hub._guard.active) self.assertTrue(hub._started.is_set()) self.assertTrue(hub._stopped.is_set())
[docs] def test_wakeup(self): hub = self.hub with self.context(): hub.wakeup()
[docs]class TestGreenlet(GreenTest):
[docs] def test_simple_exit(self): link_test = [] def waiting_func(seconds, value): sleep(seconds) print value return value g = self.hub.Greenlet(waiting_func, 0.01, value=5) g.rawlink(lambda x: link_test.append(x)) assert not g, bool(g) assert not g.dead assert not g.ready() assert not g.successful() assert g.value is None assert g.exception is None g.start() assert g # changed assert not g.dead assert not g.ready() assert not g.successful() assert g.value is None assert g.exception is None sleep(0.001) assert g assert not g.dead assert not g.ready() assert not g.successful() assert g.value is None assert g.exception is None assert not link_test sleep(0.02) assert not g assert g.dead assert g.ready() assert g.successful() assert g.value == 5 assert g.exception is None # not changed assert link_test == [g] # changed
[docs] def test_error_exit(self): link_test = [] def func(delay, return_value=4): sleep(delay) error = ExpectedError('test_error_exit') error.myattr = return_value raise error g = self.hub.Greenlet(func, 0.001, return_value=5) g.rawlink(lambda x: link_test.append(x)) g.start() sleep(0.1) assert not g assert g.dead assert g.ready() assert not g.successful() assert g.value is None # not changed assert g.exception.myattr == 5 assert link_test == [g], link_test
def _assertKilled(self, g): assert not g assert g.dead assert g.ready() assert g.successful(), (repr(g), g.value, g.exception) assert isinstance(g.value, GreenletExit), (repr(g), g.value, g.exception) assert g.exception is None
[docs] def assertKilled(self, g): self._assertKilled(g) sleep(0.01) self._assertKilled(g)
def _test_kill(self, g, block): g.kill(block=block) if not block: sleep(0.01) self.assertKilled(g) # kill second time must not hurt g.kill(block=block) self.assertKilled(g) def _test_kill_not_started(self, block): link_test = [] result = [] g = self.hub.Greenlet(lambda: result.append(1)) g.rawlink(lambda x: link_test.append(x)) self._test_kill(g, block=block) assert not result assert link_test == [g]
[docs] def test_kill_not_started_block(self): self._test_kill_not_started(block=True)
[docs] def test_kill_not_started_noblock(self): self._test_kill_not_started(block=False)
def _test_kill_just_started(self, block): result = [] link_test = [] g = self.hub.Greenlet(lambda: result.append(1)) g.rawlink(lambda x: link_test.append(x)) g.start() self._test_kill(g, block=block) assert not result, result assert link_test == [g]
[docs] def test_kill_just_started_block(self): self._test_kill_just_started(block=True)
[docs] def test_kill_just_started_noblock(self): self._test_kill_just_started(block=False)
def _test_kill_running(self, block): link_test = [] g = self.hub.spawn(sleep, 10) g.rawlink(lambda x: link_test.append(x)) self._test_kill(g, block=block) sleep(0.01) assert link_test == [g]
[docs] def test_kill_running_block(self): self._test_kill_running(block=True)
[docs] def test_kill_running_noblock(self): self._test_kill_running(block=False)
Read the Docs v: latest
Versions
latest
Downloads
PDF
HTML
Epub
On Read the Docs
Project Home
Builds

Free document hosting provided by Read the Docs.