← Back to branch summary

~justin-fathomdb/nova/justinsb-openstack-api-volumes

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/test/test_factories.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_factories.py
 
1
# Copyright (c) 2001-2009 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
 
4
"""
 
5
Test code for basic Factory classes.
 
6
"""
 
7
 
 
8
import pickle
 
9
 
 
10
from twisted.trial.unittest import TestCase
 
11
 
 
12
from twisted.internet import reactor, defer
 
13
from twisted.internet.task import Clock
 
14
from twisted.internet.protocol import Factory, ReconnectingClientFactory
 
15
from twisted.protocols.basic import Int16StringReceiver
 
16
 
 
17
 
 
18
 
 
19
class In(Int16StringReceiver):
 
20
    def __init__(self):
 
21
        self.msgs = {}
 
22
 
 
23
    def connectionMade(self):
 
24
        self.factory.connections += 1
 
25
 
 
26
    def stringReceived(self, msg):
 
27
        n, msg = pickle.loads(msg)
 
28
        self.msgs[n] = msg
 
29
        self.sendString(pickle.dumps(n))
 
30
 
 
31
    def connectionLost(self, reason):
 
32
        self.factory.allMessages.append(self.msgs)
 
33
        if len(self.factory.allMessages) >= self.factory.goal:
 
34
            self.factory.d.callback(None)
 
35
 
 
36
 
 
37
 
 
38
class Out(Int16StringReceiver):
 
39
    msgs = dict([(x, 'X' * x) for x in range(10)])
 
40
 
 
41
    def __init__(self):
 
42
        self.msgs = Out.msgs.copy()
 
43
 
 
44
    def connectionMade(self):
 
45
        for i in self.msgs.keys():
 
46
            self.sendString(pickle.dumps( (i, self.msgs[i])))
 
47
 
 
48
    def stringReceived(self, msg):
 
49
        n = pickle.loads(msg)
 
50
        del self.msgs[n]
 
51
        if not self.msgs:
 
52
            self.transport.loseConnection()
 
53
            self.factory.howManyTimes -= 1
 
54
            if self.factory.howManyTimes <= 0:
 
55
                self.factory.stopTrying()
 
56
 
 
57
 
 
58
 
 
59
class FakeConnector(object):
 
60
    """
 
61
    A fake connector class, to be used to mock connections failed or lost.
 
62
    """
 
63
 
 
64
    def stopConnecting(self):
 
65
        pass
 
66
 
 
67
 
 
68
    def connect(self):
 
69
        pass
 
70
 
 
71
 
 
72
 
 
73
class ReconnectingFactoryTestCase(TestCase):
 
74
    """
 
75
    Tests for L{ReconnectingClientFactory}.
 
76
    """
 
77
 
 
78
    def testStopTrying(self):
 
79
        f = Factory()
 
80
        f.protocol = In
 
81
        f.connections = 0
 
82
        f.allMessages = []
 
83
        f.goal = 2
 
84
        f.d = defer.Deferred()
 
85
 
 
86
        c = ReconnectingClientFactory()
 
87
        c.initialDelay = c.delay = 0.2
 
88
        c.protocol = Out
 
89
        c.howManyTimes = 2
 
90
 
 
91
        port = reactor.listenTCP(0, f)
 
92
        self.addCleanup(port.stopListening)
 
93
        PORT = port.getHost().port
 
94
        reactor.connectTCP('127.0.0.1', PORT, c)
 
95
 
 
96
        f.d.addCallback(self._testStopTrying_1, f, c)
 
97
        return f.d
 
98
    testStopTrying.timeout = 10
 
99
 
 
100
 
 
101
    def _testStopTrying_1(self, res, f, c):
 
102
        self.assertEquals(len(f.allMessages), 2,
 
103
                          "not enough messages -- %s" % f.allMessages)
 
104
        self.assertEquals(f.connections, 2,
 
105
                          "Number of successful connections incorrect %d" %
 
106
                          f.connections)
 
107
        self.assertEquals(f.allMessages, [Out.msgs] * 2)
 
108
        self.failIf(c.continueTrying, "stopTrying never called or ineffective")
 
109
 
 
110
 
 
111
    def test_serializeUnused(self):
 
112
        """
 
113
        A L{ReconnectingClientFactory} which hasn't been used for anything
 
114
        can be pickled and unpickled and end up with the same state.
 
115
        """
 
116
        original = ReconnectingClientFactory()
 
117
        reconstituted = pickle.loads(pickle.dumps(original))
 
118
        self.assertEqual(original.__dict__, reconstituted.__dict__)
 
119
 
 
120
 
 
121
    def test_serializeWithClock(self):
 
122
        """
 
123
        The clock attribute of L{ReconnectingClientFactory} is not serialized,
 
124
        and the restored value sets it to the default value, the reactor.
 
125
        """
 
126
        clock = Clock()
 
127
        original = ReconnectingClientFactory()
 
128
        original.clock = clock
 
129
        reconstituted = pickle.loads(pickle.dumps(original))
 
130
        self.assertIdentical(reconstituted.clock, None)
 
131
 
 
132
 
 
133
    def test_deserializationResetsParameters(self):
 
134
        """
 
135
        A L{ReconnectingClientFactory} which is unpickled does not have an
 
136
        L{IConnector} and has its reconnecting timing parameters reset to their
 
137
        initial values.
 
138
        """
 
139
        factory = ReconnectingClientFactory()
 
140
        factory.clientConnectionFailed(FakeConnector(), None)
 
141
        self.addCleanup(factory.stopTrying)
 
142
 
 
143
        serialized = pickle.dumps(factory)
 
144
        unserialized = pickle.loads(serialized)
 
145
        self.assertEqual(unserialized.connector, None)
 
146
        self.assertEqual(unserialized._callID, None)
 
147
        self.assertEqual(unserialized.retries, 0)
 
148
        self.assertEqual(unserialized.delay, factory.initialDelay)
 
149
        self.assertEqual(unserialized.continueTrying, True)
 
150
 
 
151
 
 
152
    def test_parametrizedClock(self):
 
153
        """
 
154
        The clock used by L{ReconnectingClientFactory} can be parametrized, so
 
155
        that one can cleanly test reconnections.
 
156
        """
 
157
        clock = Clock()
 
158
        factory = ReconnectingClientFactory()
 
159
        factory.clock = clock
 
160
 
 
161
        factory.clientConnectionLost(FakeConnector(), None)
 
162
        self.assertEquals(len(clock.calls), 1)