twisted/internet/test/test_epollreactor.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Tests for L{twisted.internet.epollreactor}.
   6 """
   7
   8 from twisted.trial.unittest import TestCase
   9 try:
  10     from twisted.internet.epollreactor import _ContinuousPolling
  11 except ImportError:
  12     _ContinuousPolling = None
  13 from twisted.internet.task import Clock
  14 from twisted.internet.error import ConnectionDone
  15
  16
  17
  18 class Descriptor(object):
  19     """
  20     Records reads and writes, as if it were a C{FileDescriptor}.
  21     """
  22
  23     def __init__(self):
  24         self.events = []
  25
  26
  27     def fileno(self):
  28         return 1
  29
  30
  31     def doRead(self):
  32         self.events.append("read")
  33
  34
  35     def doWrite(self):
  36         self.events.append("write")
  37
  38
  39     def connectionLost(self, reason):
  40         reason.trap(ConnectionDone)
  41         self.events.append("lost")
  42
  43
  44
  45 class ContinuousPollingTests(TestCase):
  46     """
  47     L{_ContinuousPolling} can be used to read and write from C{FileDescriptor}
  48     objects.
  49     """
  50
  51     def test_addReader(self):
  52         """
  53         Adding a reader when there was previously no reader starts up a
  54         C{LoopingCall}.
  55         """
  56         poller = _ContinuousPolling(Clock())
  57         self.assertEqual(poller._loop, None)
  58         reader = object()
  59         self.assertFalse(poller.isReading(reader))
  60         poller.addReader(reader)
  61         self.assertNotEqual(poller._loop, None)
  62         self.assertTrue(poller._loop.running)
  63         self.assertIdentical(poller._loop.clock, poller._reactor)
  64         self.assertTrue(poller.isReading(reader))
  65
  66
  67     def test_addWriter(self):
  68         """
  69         Adding a writer when there was previously no writer starts up a
  70         C{LoopingCall}.
  71         """
  72         poller = _ContinuousPolling(Clock())
  73         self.assertEqual(poller._loop, None)
  74         writer = object()
  75         self.assertFalse(poller.isWriting(writer))
  76         poller.addWriter(writer)
  77         self.assertNotEqual(poller._loop, None)
  78         self.assertTrue(poller._loop.running)
  79         self.assertIdentical(poller._loop.clock, poller._reactor)
  80         self.assertTrue(poller.isWriting(writer))
  81
  82
  83     def test_removeReader(self):
  84         """
  85         Removing a reader stops the C{LoopingCall}.
  86         """
  87         poller = _ContinuousPolling(Clock())
  88         reader = object()
  89         poller.addReader(reader)
  90         poller.removeReader(reader)
  91         self.assertEqual(poller._loop, None)
  92         self.assertEqual(poller._reactor.getDelayedCalls(), [])
  93         self.assertFalse(poller.isReading(reader))
  94
  95
  96     def test_removeWriter(self):
  97         """
  98         Removing a writer stops the C{LoopingCall}.
  99         """
 100         poller = _ContinuousPolling(Clock())
 101         writer = object()
 102         poller.addWriter(writer)
 103         poller.removeWriter(writer)
 104         self.assertEqual(poller._loop, None)
 105         self.assertEqual(poller._reactor.getDelayedCalls(), [])
 106         self.assertFalse(poller.isWriting(writer))
 107
 108
 109     def test_removeUnknown(self):
 110         """
 111         Removing unknown readers and writers silently does nothing.
 112         """
 113         poller = _ContinuousPolling(Clock())
 114         poller.removeWriter(object())
 115         poller.removeReader(object())
 116
 117
 118     def test_multipleReadersAndWriters(self):
 119         """
 120         Adding multiple readers and writers results in a single
 121         C{LoopingCall}.
 122         """
 123         poller = _ContinuousPolling(Clock())
 124         writer = object()
 125         poller.addWriter(writer)
 126         self.assertNotEqual(poller._loop, None)
 127         poller.addWriter(object())
 128         self.assertNotEqual(poller._loop, None)
 129         poller.addReader(object())
 130         self.assertNotEqual(poller._loop, None)
 131         poller.addReader(object())
 132         poller.removeWriter(writer)
 133         self.assertNotEqual(poller._loop, None)
 134         self.assertTrue(poller._loop.running)
 135         self.assertEqual(len(poller._reactor.getDelayedCalls()), 1)
 136
 137
 138     def test_readerPolling(self):
 139         """
 140         Adding a reader causes its C{doRead} to be called every 1
 141         milliseconds.
 142         """
 143         reactor = Clock()
 144         poller = _ContinuousPolling(reactor)
 145         desc = Descriptor()
 146         poller.addReader(desc)
 147         self.assertEqual(desc.events, [])
 148         reactor.advance(0.00001)
 149         self.assertEqual(desc.events, ["read"])
 150         reactor.advance(0.00001)
 151         self.assertEqual(desc.events, ["read", "read"])
 152         reactor.advance(0.00001)
 153         self.assertEqual(desc.events, ["read", "read", "read"])
 154
 155
 156     def test_writerPolling(self):
 157         """
 158         Adding a writer causes its C{doWrite} to be called every 1
 159         milliseconds.
 160         """
 161         reactor = Clock()
 162         poller = _ContinuousPolling(reactor)
 163         desc = Descriptor()
 164         poller.addWriter(desc)
 165         self.assertEqual(desc.events, [])
 166         reactor.advance(0.001)
 167         self.assertEqual(desc.events, ["write"])
 168         reactor.advance(0.001)
 169         self.assertEqual(desc.events, ["write", "write"])
 170         reactor.advance(0.001)
 171         self.assertEqual(desc.events, ["write", "write", "write"])
 172
 173
 174     def test_connectionLostOnRead(self):
 175         """
 176         If a C{doRead} returns a value indicating disconnection,
 177         C{connectionLost} is called on it.
 178         """
 179         reactor = Clock()
 180         poller = _ContinuousPolling(reactor)
 181         desc = Descriptor()
 182         desc.doRead = lambda: ConnectionDone()
 183         poller.addReader(desc)
 184         self.assertEqual(desc.events, [])
 185         reactor.advance(0.001)
 186         self.assertEqual(desc.events, ["lost"])
 187
 188
 189     def test_connectionLostOnWrite(self):
 190         """
 191         If a C{doWrite} returns a value indicating disconnection,
 192         C{connectionLost} is called on it.
 193         """
 194         reactor = Clock()
 195         poller = _ContinuousPolling(reactor)
 196         desc = Descriptor()
 197         desc.doWrite = lambda: ConnectionDone()
 198         poller.addWriter(desc)
 199         self.assertEqual(desc.events, [])
 200         reactor.advance(0.001)
 201         self.assertEqual(desc.events, ["lost"])
 202
 203
 204     def test_removeAll(self):
 205         """
 206         L{_ContinuousPolling.removeAll} removes all descriptors and returns
 207         the readers and writers.
 208         """
 209         poller = _ContinuousPolling(Clock())
 210         reader = object()
 211         writer = object()
 212         both = object()
 213         poller.addReader(reader)
 214         poller.addReader(both)
 215         poller.addWriter(writer)
 216         poller.addWriter(both)
 217         removed = poller.removeAll()
 218         self.assertEqual(poller.getReaders(), [])
 219         self.assertEqual(poller.getWriters(), [])
 220         self.assertEqual(len(removed), 3)
 221         self.assertEqual(set(removed), set([reader, writer, both]))
 222
 223
 224     def test_getReaders(self):
 225         """
 226         L{_ContinuousPolling.getReaders} returns a list of the read
 227         descriptors.
 228         """
 229         poller = _ContinuousPolling(Clock())
 230         reader = object()
 231         poller.addReader(reader)
 232         self.assertIn(reader, poller.getReaders())
 233
 234
 235     def test_getWriters(self):
 236         """
 237         L{_ContinuousPolling.getWriters} returns a list of the write
 238         descriptors.
 239         """
 240         poller = _ContinuousPolling(Clock())
 241         writer = object()
 242         poller.addWriter(writer)
 243         self.assertIn(writer, poller.getWriters())
 244
 245     if _ContinuousPolling is None:
 246         skip = "epoll not supported in this environment."