Source code for thriftworker.tests.app.test_listener

from __future__ import absolute_import

import socket

from thriftworker.tests.utils import TestCase, StartStopLoopMixin, \
    start_stop_ctx
from thriftworker.listener import Listener, Listeners
from thriftworker.exceptions import BindError


[docs]class ListenerMixin(StartStopLoopMixin): Listener = None
[docs] def setUp(self): super(ListenerMixin, self).setUp() self.Listener = self.app.subclass_with_self(type(self).Listener)
[docs]class TestListener(ListenerMixin, TestCase): Listener = Listener
[docs] def test_start_stop(self): listener = self.Listener('SomeService', ('localhost', 0)) with start_stop_ctx(listener): self.assertLess(0, listener.port) self.assertEqual(socket.gethostbyname('localhost'), listener.host) self.assertFalse(listener.channel.closed)
[docs] def test_bind_error(self): first_listener = self.Listener('SomeService', ('localhost', 59357)) second_listener = self.Listener('SomeService', ('localhost', 59357)) with self.assertRaises(BindError): with start_stop_ctx(first_listener): with start_stop_ctx(second_listener): pass
[docs] def test_bind_from_pool(self): self.app.port_range = (59000, 59100) first_listener = self.Listener('SomeService', ('localhost', None)) second_listener = self.Listener('SomeService', ('localhost', None)) with start_stop_ctx(first_listener): with start_stop_ctx(second_listener): self.assertEqual(1, second_listener.port - first_listener.port)
[docs]class ListenersMixin(StartStopLoopMixin): Listeners = None
[docs] def setUp(self): super(ListenersMixin, self).setUp() self.Listeners = self.app.subclass_with_self(type(self).Listeners)
[docs]class TestListeners(ListenersMixin, TestCase): Listeners = Listeners
[docs] def test_register(self): listeners = self.Listeners() # Add first service. listeners.register('SomeService', 'localhost', None) registered = list(listeners) self.assertEqual(1, len(registered)) listener = listeners[0] self.assertIs(listener, registered[0]) self.assertIn(listener, listeners) self.assertEqual('SomeService', listener.name) self.assertEqual([listener.channel], listeners.channels) self.assertEqual({0: listener}, listeners.enumerated) # Add new service. listeners.register('OtherService', 'localhost', None) self.assertEqual([listeners[0].channel, listeners[1].channel], listeners.channels) self.assertEqual({0: listeners[0], 1: listeners[1]}, listeners.enumerated)
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.