| Server IP : 74.208.127.88 / Your IP : 18.191.137.190 Web Server : Apache/2.4.41 (Ubuntu) System : Linux ubuntu 5.4.0-163-generic #180-Ubuntu SMP Tue Sep 5 13:21:23 UTC 2023 x86_64 User : www-data ( 33) PHP Version : 7.4.3-4ubuntu2.29 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /lib/python3/dist-packages/twisted/test/ |
Upload File : |
# 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 ReconnectingFactoryTests(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.assertIsNone(reconstituted.clock)
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.assertIsNone(unserialized.connector)
self.assertIsNone(unserialized._callID)
self.assertEqual(unserialized.retries, 0)
self.assertEqual(unserialized.delay, factory.initialDelay)
self.assertTrue(unserialized.continueTrying)
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)