← Back to branch summary

~ntt-pf-lab/nova/monkey_patch_notification

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/test/test_stringtransport.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/test/test_stringtransport.py
 
1
# Copyright (c) 2009-2010 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
 
4
"""
 
5
Tests for L{twisted.test.proto_helpers.StringTransport}.
 
6
"""
 
7
 
 
8
from zope.interface.verify import verifyObject
 
9
 
 
10
from twisted.internet.interfaces import ITransport, IPushProducer, IConsumer
 
11
from twisted.trial.unittest import TestCase
 
12
from twisted.test.proto_helpers import StringTransport
 
13
 
 
14
 
 
15
class StringTransportTests(TestCase):
 
16
    """
 
17
    Tests for L{twisted.test.proto_helpers.StringTransport}.
 
18
    """
 
19
    def setUp(self):
 
20
        self.transport = StringTransport()
 
21
 
 
22
 
 
23
    def test_interfaces(self):
 
24
        """
 
25
        L{StringTransport} instances provide L{ITransport}, L{IPushProducer},
 
26
        and L{IConsumer}.
 
27
        """
 
28
        self.assertTrue(verifyObject(ITransport, self.transport))
 
29
        self.assertTrue(verifyObject(IPushProducer, self.transport))
 
30
        self.assertTrue(verifyObject(IConsumer, self.transport))
 
31
 
 
32
 
 
33
    def test_registerProducer(self):
 
34
        """
 
35
        L{StringTransport.registerProducer} records the arguments supplied to
 
36
        it as instance attributes.
 
37
        """
 
38
        producer = object()
 
39
        streaming = object()
 
40
        self.transport.registerProducer(producer, streaming)
 
41
        self.assertIdentical(self.transport.producer, producer)
 
42
        self.assertIdentical(self.transport.streaming, streaming)
 
43
 
 
44
 
 
45
    def test_disallowedRegisterProducer(self):
 
46
        """
 
47
        L{StringTransport.registerProducer} raises L{RuntimeError} if a
 
48
        producer is already registered.
 
49
        """
 
50
        producer = object()
 
51
        self.transport.registerProducer(producer, True)
 
52
        self.assertRaises(
 
53
            RuntimeError, self.transport.registerProducer, object(), False)
 
54
        self.assertIdentical(self.transport.producer, producer)
 
55
        self.assertTrue(self.transport.streaming)
 
56
 
 
57
 
 
58
    def test_unregisterProducer(self):
 
59
        """
 
60
        L{StringTransport.unregisterProducer} causes the transport to forget
 
61
        about the registered producer and makes it possible to register a new
 
62
        one.
 
63
        """
 
64
        oldProducer = object()
 
65
        newProducer = object()
 
66
        self.transport.registerProducer(oldProducer, False)
 
67
        self.transport.unregisterProducer()
 
68
        self.assertIdentical(self.transport.producer, None)
 
69
        self.transport.registerProducer(newProducer, True)
 
70
        self.assertIdentical(self.transport.producer, newProducer)
 
71
        self.assertTrue(self.transport.streaming)
 
72
 
 
73
 
 
74
    def test_invalidUnregisterProducer(self):
 
75
        """
 
76
        L{StringTransport.unregisterProducer} raises L{RuntimeError} if called
 
77
        when no producer is registered.
 
78
        """
 
79
        self.assertRaises(RuntimeError, self.transport.unregisterProducer)
 
80
 
 
81
 
 
82
    def test_initialProducerState(self):
 
83
        """
 
84
        L{StringTransport.producerState} is initially C{'producing'}.
 
85
        """
 
86
        self.assertEqual(self.transport.producerState, 'producing')
 
87
 
 
88
 
 
89
    def test_pauseProducing(self):
 
90
        """
 
91
        L{StringTransport.pauseProducing} changes the C{producerState} of the
 
92
        transport to C{'paused'}.
 
93
        """
 
94
        self.transport.pauseProducing()
 
95
        self.assertEqual(self.transport.producerState, 'paused')
 
96
 
 
97
 
 
98
    def test_resumeProducing(self):
 
99
        """
 
100
        L{StringTransport.resumeProducing} changes the C{producerState} of the
 
101
        transport to C{'producing'}.
 
102
        """
 
103
        self.transport.pauseProducing()
 
104
        self.transport.resumeProducing()
 
105
        self.assertEqual(self.transport.producerState, 'producing')
 
106
 
 
107
 
 
108
    def test_stopProducing(self):
 
109
        """
 
110
        L{StringTransport.stopProducing} changes the C{'producerState'} of the
 
111
        transport to C{'stopped'}.
 
112
        """
 
113
        self.transport.stopProducing()
 
114
        self.assertEqual(self.transport.producerState, 'stopped')
 
115
 
 
116
 
 
117
    def test_stoppedTransportCannotPause(self):
 
118
        """
 
119
        L{StringTransport.pauseProducing} raises L{RuntimeError} if the
 
120
        transport has been stopped.
 
121
        """
 
122
        self.transport.stopProducing()
 
123
        self.assertRaises(RuntimeError, self.transport.pauseProducing)
 
124
 
 
125
 
 
126
    def test_stoppedTransportCannotResume(self):
 
127
        """
 
128
        L{StringTransport.resumeProducing} raises L{RuntimeError} if the
 
129
        transport has been stopped.
 
130
        """
 
131
        self.transport.stopProducing()
 
132
        self.assertRaises(RuntimeError, self.transport.resumeProducing)
 
133
 
 
134
 
 
135
    def test_disconnectingTransportCannotPause(self):
 
136
        """
 
137
        L{StringTransport.pauseProducing} raises L{RuntimeError} if the
 
138
        transport is being disconnected.
 
139
        """
 
140
        self.transport.loseConnection()
 
141
        self.assertRaises(RuntimeError, self.transport.pauseProducing)
 
142
 
 
143
 
 
144
    def test_disconnectingTransportCannotResume(self):
 
145
        """
 
146
        L{StringTransport.resumeProducing} raises L{RuntimeError} if the
 
147
        transport is being disconnected.
 
148
        """
 
149
        self.transport.loseConnection()
 
150
        self.assertRaises(RuntimeError, self.transport.resumeProducing)
 
151
 
 
152
 
 
153
    def test_loseConnectionSetsDisconnecting(self):
 
154
        """
 
155
        L{StringTransport.loseConnection} toggles the C{disconnecting} instance
 
156
        variable to C{True}.
 
157
        """
 
158
        self.assertFalse(self.transport.disconnecting)
 
159
        self.transport.loseConnection()
 
160
        self.assertTrue(self.transport.disconnecting)