← Back to branch summary

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

« back to all changes in this revision

Viewing changes to twisted/test/test_factories.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/test/test_factories.py
1
 
# Copyright (c) 2001-2008 Twisted Matrix Laboratories.
 
1
# Copyright (c) 2001-2009 Twisted Matrix Laboratories.
2
2
# See LICENSE for details.
3
3
 
4
4
"""
7
7
 
8
8
import pickle
9
9
 
10
 
from twisted.trial import unittest
 
10
from twisted.trial.unittest import TestCase
11
11
 
12
12
from twisted.internet import reactor, defer
 
13
from twisted.internet.task import Clock
13
14
from twisted.internet.protocol import Factory, ReconnectingClientFactory
14
15
from twisted.protocols.basic import Int16StringReceiver
15
16
 
 
17
 
 
18
 
16
19
class In(Int16StringReceiver):
17
20
    def __init__(self):
18
21
        self.msgs = {}
30
33
        if len(self.factory.allMessages) >= self.factory.goal:
31
34
            self.factory.d.callback(None)
32
35
 
 
36
 
 
37
 
33
38
class Out(Int16StringReceiver):
34
39
    msgs = dict([(x, 'X' * x) for x in range(10)])
35
40
 
51
56
 
52
57
 
53
58
 
54
 
class ReconnectingFactoryTestCase(unittest.TestCase):
 
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):
55
74
    """
56
75
    Tests for L{ReconnectingClientFactory}.
57
76
    """
 
77
 
58
78
    def testStopTrying(self):
59
79
        f = Factory()
60
80
        f.protocol = In
98
118
        self.assertEqual(original.__dict__, reconstituted.__dict__)
99
119
 
100
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
 
101
133
    def test_deserializationResetsParameters(self):
102
134
        """
103
135
        A L{ReconnectingClientFactory} which is unpickled does not have an
104
 
        L{IConnector} and has its reconnectioning timing parameters reset to
105
 
        their initial values.
 
136
        L{IConnector} and has its reconnecting timing parameters reset to their
 
137
        initial values.
106
138
        """
107
 
        class FakeConnector(object):
108
 
            def stopConnecting(self):
109
 
                pass
110
 
 
111
139
        factory = ReconnectingClientFactory()
112
140
        factory.clientConnectionFailed(FakeConnector(), None)
113
 
        try:
114
 
            serialized = pickle.dumps(factory)
115
 
            unserialized = pickle.loads(serialized)
116
 
            self.assertEqual(unserialized.connector, None)
117
 
            self.assertEqual(unserialized._callID, None)
118
 
            self.assertEqual(unserialized.retries, 0)
119
 
            self.assertEqual(unserialized.delay, factory.initialDelay)
120
 
            self.assertEqual(unserialized.continueTrying, True)
121
 
        finally:
122
 
            factory.stopTrying()
 
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)