/usr/lib/python3/dist-packages/twisted/test/test_factories.py is in python3-twisted-experimental 13.2.0-0ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | # Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Test code for basic Factory classes.
"""
from __future__ import division, absolute_import
import pickle
from twisted.trial.unittest import TestCase
from twisted.internet.task import Clock
from twisted.internet.protocol import ReconnectingClientFactory, Protocol
class FakeConnector(object):
"""
A fake connector class, to be used to mock connections failed or lost.
"""
def stopConnecting(self):
pass
def connect(self):
pass
class ReconnectingFactoryTestCase(TestCase):
"""
Tests for L{ReconnectingClientFactory}.
"""
def test_stopTryingWhenConnected(self):
"""
If a L{ReconnectingClientFactory} has C{stopTrying} called while it is
connected, it does not subsequently attempt to reconnect if the
connection is later lost.
"""
class NoConnectConnector(object):
def stopConnecting(self):
raise RuntimeError("Shouldn't be called, we're connected.")
def connect(self):
raise RuntimeError("Shouldn't be reconnecting.")
c = ReconnectingClientFactory()
c.protocol = Protocol
# Let's pretend we've connected:
c.buildProtocol(None)
# Now we stop trying, then disconnect:
c.stopTrying()
c.clientConnectionLost(NoConnectConnector(), None)
self.assertFalse(c.continueTrying)
def test_stopTryingDoesNotReconnect(self):
"""
Calling stopTrying on a L{ReconnectingClientFactory} doesn't attempt a
retry on any active connector.
"""
class FactoryAwareFakeConnector(FakeConnector):
attemptedRetry = False
def stopConnecting(self):
"""
Behave as though an ongoing connection attempt has now
failed, and notify the factory of this.
"""
f.clientConnectionFailed(self, None)
def connect(self):
"""
Record an attempt to reconnect, since this is what we
are trying to avoid.
"""
self.attemptedRetry = True
f = ReconnectingClientFactory()
f.clock = Clock()
# simulate an active connection - stopConnecting on this connector should
# be triggered when we call stopTrying
f.connector = FactoryAwareFakeConnector()
f.stopTrying()
# make sure we never attempted to retry
self.assertFalse(f.connector.attemptedRetry)
self.assertFalse(f.clock.getDelayedCalls())
def test_serializeUnused(self):
"""
A L{ReconnectingClientFactory} which hasn't been used for anything
can be pickled and unpickled and end up with the same state.
"""
original = ReconnectingClientFactory()
reconstituted = pickle.loads(pickle.dumps(original))
self.assertEqual(original.__dict__, reconstituted.__dict__)
def test_serializeWithClock(self):
"""
The clock attribute of L{ReconnectingClientFactory} is not serialized,
and the restored value sets it to the default value, the reactor.
"""
clock = Clock()
original = ReconnectingClientFactory()
original.clock = clock
reconstituted = pickle.loads(pickle.dumps(original))
self.assertIdentical(reconstituted.clock, None)
def test_deserializationResetsParameters(self):
"""
A L{ReconnectingClientFactory} which is unpickled does not have an
L{IConnector} and has its reconnecting timing parameters reset to their
initial values.
"""
factory = ReconnectingClientFactory()
factory.clientConnectionFailed(FakeConnector(), None)
self.addCleanup(factory.stopTrying)
serialized = pickle.dumps(factory)
unserialized = pickle.loads(serialized)
self.assertEqual(unserialized.connector, None)
self.assertEqual(unserialized._callID, None)
self.assertEqual(unserialized.retries, 0)
self.assertEqual(unserialized.delay, factory.initialDelay)
self.assertEqual(unserialized.continueTrying, True)
def test_parametrizedClock(self):
"""
The clock used by L{ReconnectingClientFactory} can be parametrized, so
that one can cleanly test reconnections.
"""
clock = Clock()
factory = ReconnectingClientFactory()
factory.clock = clock
factory.clientConnectionLost(FakeConnector(), None)
self.assertEqual(len(clock.calls), 1)
|