ExternalVendorCode/tahoe-lafs
1
0
Fork 0
mirror of https://github.com/tahoe-lafs/tahoe-lafs.git synced 2025-02-22 18:22:40 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

tests: exercise connection_status.py

Browse Source
This commit is contained in:
Brian Warner 2016-12-09 11:46:26 -08:00
parent 70a616826c
commit 8b35fe9efe
1 changed files with 117 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

117
src/allmydata/test/test_connections.py
View File

@ -7,6 +7,7 @@ from twisted.internet.interfaces import IStreamClientEndpoint
from ConfigParser import SafeConfigParser from ConfigParser import SafeConfigParser
from foolscap.connections import tcp from foolscap.connections import tcp
from ..node import Node, PrivacyError from ..node import Node, PrivacyError
from ..util import connection_status
class FakeNode(Node): class FakeNode(Node):
def __init__(self, config_str): def __init__(self, config_str):
@ -343,4 +344,120 @@ class Privacy(unittest.TestCase):
None, "hostname:1234") None, "hostname:1234")
self.assertEqual(str(e), "tub.location includes tcp: hint") self.assertEqual(str(e), "tub.location includes tcp: hint")
class Status(unittest.TestCase):
def test_describe(self):
t = connection_status._describe_statuses(["h2","h1"],
{"h1": "hand1"},
{"h1": "st1", "h2": "st2"})
self.assertEqual(t, " h1 via hand1: st1\n h2: st2\n")
def test_reconnector_connected(self):
ci = mock.Mock()
ci.connectorStatuses = {"h1": "st1"}
ci.connectionHandlers = {"h1": "hand1"}
ci.winningHint = "h1"
ci.establishedAt = 120
ri = mock.Mock()
ri.state = "connected"
ri.connectionInfo = ci
rc = mock.Mock
rc.getReconnectionInfo = mock.Mock(return_value=ri)
cs = connection_status.from_foolscap_reconnector(rc, 123)
self.assertEqual(cs.connected, True)
self.assertEqual(cs.last_connection_summary,
"Connected to h1 via hand1")
self.assertEqual(cs.last_connection_description,
"Connection successful to h1 via hand1")
self.assertEqual(cs.last_connection_time, 120)
self.assertEqual(cs.last_received_time, 123)
def test_reconnector_connected_others(self):
ci = mock.Mock()
ci.connectorStatuses = {"h1": "st1", "h2": "st2"}
ci.connectionHandlers = {"h1": "hand1"}
ci.winningHint = "h1"
ci.establishedAt = 120
ri = mock.Mock()
ri.state = "connected"
ri.connectionInfo = ci
rc = mock.Mock
rc.getReconnectionInfo = mock.Mock(return_value=ri)
cs = connection_status.from_foolscap_reconnector(rc, 123)
self.assertEqual(cs.connected, True)
self.assertEqual(cs.last_connection_summary,
"Connected to h1 via hand1")
self.assertEqual(cs.last_connection_description,
"Connection successful to h1 via hand1\n"
"other hints:\n"
" h2: st2\n")
self.assertEqual(cs.last_connection_time, 120)
self.assertEqual(cs.last_received_time, 123)
def test_reconnector_connected_listener(self):
ci = mock.Mock()
ci.connectorStatuses = {"h1": "st1", "h2": "st2"}
ci.connectionHandlers = {"h1": "hand1"}
ci.listenerStatus = ("listener1", "successful")
ci.winningHint = None
ci.establishedAt = 120
ri = mock.Mock()
ri.state = "connected"
ri.connectionInfo = ci
rc = mock.Mock
rc.getReconnectionInfo = mock.Mock(return_value=ri)
cs = connection_status.from_foolscap_reconnector(rc, 123)
self.assertEqual(cs.connected, True)
self.assertEqual(cs.last_connection_summary,
"Connected via listener (listener1)")
self.assertEqual(cs.last_connection_description,
"Connection successful via listener (listener1)\n"
"other hints:\n"
" h1 via hand1: st1\n"
" h2: st2\n")
self.assertEqual(cs.last_connection_time, 120)
self.assertEqual(cs.last_received_time, 123)
def test_reconnector_connecting(self):
ci = mock.Mock()
ci.connectorStatuses = {"h1": "st1", "h2": "st2"}
ci.connectionHandlers = {"h1": "hand1"}
ri = mock.Mock()
ri.state = "connecting"
ri.connectionInfo = ci
rc = mock.Mock
rc.getReconnectionInfo = mock.Mock(return_value=ri)
cs = connection_status.from_foolscap_reconnector(rc, 123)
self.assertEqual(cs.connected, False)
self.assertEqual(cs.last_connection_summary,
"Trying to connect")
self.assertEqual(cs.last_connection_description,
"Trying to connect:\n"
" h1 via hand1: st1\n"
" h2: st2\n")
self.assertEqual(cs.last_connection_time, None)
self.assertEqual(cs.last_received_time, 123)
def test_reconnector_waiting(self):
ci = mock.Mock()
ci.connectorStatuses = {"h1": "st1", "h2": "st2"}
ci.connectionHandlers = {"h1": "hand1"}
ri = mock.Mock()
ri.state = "waiting"
ri.lastAttempt = 10
ri.nextAttempt = 20
ri.connectionInfo = ci
rc = mock.Mock
rc.getReconnectionInfo = mock.Mock(return_value=ri)
with mock.patch("time.time", return_value=12):
cs = connection_status.from_foolscap_reconnector(rc, 5)
self.assertEqual(cs.connected, False)
self.assertEqual(cs.last_connection_summary,
"Reconnecting in 8 seconds")
self.assertEqual(cs.last_connection_description,
"Reconnecting in 8 seconds\n"
"Last attempt 2s ago:\n"
" h1 via hand1: st1\n"
" h2: st2\n")
self.assertEqual(cs.last_connection_time, None)
self.assertEqual(cs.last_received_time, 5)