← Back to branch summary

~certify-web-dev/twisted/certify-staging

« back to all changes in this revision

Viewing changes to twisted/internet/test/test_posixbase.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2010-01-02 19:38:17 UTC
  • mfrom: (2.2.4 sid)
  • Revision ID: james.westby@ubuntu.com-20100102193817-jphp464ppwh7dulg
Tags: 9.0.0-1
http://bugs.debian.org/557781
* python-twisted: Depend on the python-twisted-* 9.0 packages.
* python-twisted: Depend on python-zope.interface only. Closes: #557781.

Show diffs side-by-side

added added

removed removed

Lines of Context:
twisted/internet/test/test_posixbase.py
 
1
# Copyright (c) 2009 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
 
4
"""
 
5
Tests for L{twisted.internet.posixbase} and supporting code.
 
6
"""
 
7
 
 
8
from twisted.python.compat import set
 
9
from twisted.trial.unittest import TestCase
 
10
from twisted.internet.posixbase import PosixReactorBase, _Waker
 
11
from twisted.internet.protocol import ServerFactory
 
12
from twisted.internet.tcp import Port
 
13
from twisted.internet import reactor
 
14
 
 
15
 
 
16
class TrivialReactor(PosixReactorBase):
 
17
    def __init__(self):
 
18
        self._readers = {}
 
19
        self._writers = {}
 
20
        PosixReactorBase.__init__(self)
 
21
 
 
22
 
 
23
    def addReader(self, reader):
 
24
        self._readers[reader] = True
 
25
 
 
26
 
 
27
    def removeReader(self, reader):
 
28
        del self._readers[reader]
 
29
 
 
30
 
 
31
    def addWriter(self, writer):
 
32
        self._writers[writer] = True
 
33
 
 
34
 
 
35
    def removeWriter(self, writer):
 
36
        del self._writers[writer]
 
37
 
 
38
 
 
39
 
 
40
class PosixReactorBaseTests(TestCase):
 
41
    """
 
42
    Tests for L{PosixReactorBase}.
 
43
    """
 
44
 
 
45
    def _checkWaker(self, reactor):
 
46
        self.assertIsInstance(reactor.waker, _Waker)
 
47
        self.assertIn(reactor.waker, reactor._internalReaders)
 
48
        self.assertIn(reactor.waker, reactor._readers)
 
49
 
 
50
 
 
51
    def test_wakerIsInternalReader(self):
 
52
        """
 
53
        When L{PosixReactorBase} is instantiated, it creates a waker and adds
 
54
        it to its internal readers set.
 
55
        """
 
56
        reactor = TrivialReactor()
 
57
        self._checkWaker(reactor)
 
58
 
 
59
 
 
60
    def test_removeAllSkipsInternalReaders(self):
 
61
        """
 
62
        Any L{IReadDescriptors} in L{PosixReactorBase._internalReaders} are
 
63
        left alone by L{PosixReactorBase._removeAll}.
 
64
        """
 
65
        reactor = TrivialReactor()
 
66
        extra = object()
 
67
        reactor._internalReaders.add(extra)
 
68
        reactor.addReader(extra)
 
69
        reactor._removeAll(reactor._readers, reactor._writers)
 
70
        self._checkWaker(reactor)
 
71
        self.assertIn(extra, reactor._internalReaders)
 
72
        self.assertIn(extra, reactor._readers)
 
73
 
 
74
 
 
75
    def test_removeAllReturnsRemovedDescriptors(self):
 
76
        """
 
77
        L{PosixReactorBase._removeAll} returns a list of removed
 
78
        L{IReadDescriptor} and L{IWriteDescriptor} objects.
 
79
        """
 
80
        reactor = TrivialReactor()
 
81
        reader = object()
 
82
        writer = object()
 
83
        reactor.addReader(reader)
 
84
        reactor.addWriter(writer)
 
85
        removed = reactor._removeAll(
 
86
            reactor._readers, reactor._writers)
 
87
        self.assertEqual(set(removed), set([reader, writer]))
 
88
        self.assertNotIn(reader, reactor._readers)
 
89
        self.assertNotIn(writer, reactor._writers)
 
90
 
 
91
 
 
92
 
 
93
class TCPPortTests(TestCase):
 
94
    """
 
95
    Tests for L{twisted.internet.tcp.Port}.
 
96
    """
 
97
 
 
98
    if not isinstance(reactor, PosixReactorBase):
 
99
        skip = "Non-posixbase reactor"
 
100
 
 
101
    def test_connectionLostFailed(self):
 
102
        """
 
103
        L{Port.stopListening} returns a L{Deferred} which errbacks if
 
104
        L{Port.connectionLost} raises an exception.
 
105
        """
 
106
        port = Port(12345, ServerFactory())
 
107
        port.connected = True
 
108
        port.connectionLost = lambda reason: 1 / 0
 
109
        return self.assertFailure(port.stopListening(), ZeroDivisionError)