forked from LBRYCommunity/lbry-sdk
172 lines
7.4 KiB
Python
172 lines
7.4 KiB
Python
from twisted.internet import task
|
|
from twisted.trial import unittest
|
|
from lbrynet.core.utils import generate_id
|
|
from lbrynet.dht.contact import ContactManager
|
|
from lbrynet.dht import constants
|
|
|
|
|
|
class ContactOperatorsTest(unittest.TestCase):
|
|
""" Basic tests case for boolean operators on the Contact class """
|
|
def setUp(self):
|
|
self.contact_manager = ContactManager()
|
|
self.node_ids = [generate_id(), generate_id(), generate_id()]
|
|
self.firstContact = self.contact_manager.make_contact(self.node_ids[1], '127.0.0.1', 1000, None, 1)
|
|
self.secondContact = self.contact_manager.make_contact(self.node_ids[0], '192.168.0.1', 1000, None, 32)
|
|
self.secondContactCopy = self.contact_manager.make_contact(self.node_ids[0], '192.168.0.1', 1000, None, 32)
|
|
self.firstContactDifferentValues = self.contact_manager.make_contact(self.node_ids[1], '192.168.1.20',
|
|
1000, None, 50)
|
|
self.assertRaises(ValueError, self.contact_manager.make_contact, self.node_ids[1], '192.168.1.20',
|
|
100000, None)
|
|
self.assertRaises(ValueError, self.contact_manager.make_contact, self.node_ids[1], '192.168.1.20.1',
|
|
1000, None)
|
|
self.assertRaises(ValueError, self.contact_manager.make_contact, self.node_ids[1], 'this is not an ip',
|
|
1000, None)
|
|
self.assertRaises(ValueError, self.contact_manager.make_contact, "this is not a node id", '192.168.1.20.1',
|
|
1000, None)
|
|
|
|
def testNoDuplicateContactObjects(self):
|
|
self.assertTrue(self.secondContact is self.secondContactCopy)
|
|
self.assertTrue(self.firstContact is not self.firstContactDifferentValues)
|
|
|
|
def testBoolean(self):
|
|
""" Test "equals" and "not equals" comparisons """
|
|
self.failIfEqual(
|
|
self.firstContact, self.secondContact,
|
|
'Contacts with different IDs should not be equal.')
|
|
self.failUnlessEqual(
|
|
self.firstContact, self.firstContactDifferentValues,
|
|
'Contacts with same IDs should be equal, even if their other values differ.')
|
|
self.failUnlessEqual(
|
|
self.secondContact, self.secondContactCopy,
|
|
'Different copies of the same Contact instance should be equal')
|
|
|
|
def testIllogicalComparisons(self):
|
|
""" Test comparisons with non-Contact and non-str types """
|
|
msg = '"{}" operator: Contact object should not be equal to {} type'
|
|
for item in (123, [1, 2, 3], {'key': 'value'}):
|
|
self.failIfEqual(
|
|
self.firstContact, item,
|
|
msg.format('eq', type(item).__name__))
|
|
self.failUnless(
|
|
self.firstContact != item,
|
|
msg.format('ne', type(item).__name__))
|
|
|
|
def testCompactIP(self):
|
|
self.assertEqual(self.firstContact.compact_ip(), '\x7f\x00\x00\x01')
|
|
self.assertEqual(self.secondContact.compact_ip(), '\xc0\xa8\x00\x01')
|
|
|
|
|
|
class TestContactLastReplied(unittest.TestCase):
|
|
def setUp(self):
|
|
self.clock = task.Clock()
|
|
self.contact_manager = ContactManager(self.clock.seconds)
|
|
self.contact = self.contact_manager.make_contact(generate_id(), "127.0.0.1", 4444, None)
|
|
self.clock.advance(3600)
|
|
self.assertTrue(self.contact.contact_is_good is None)
|
|
|
|
def test_stale_replied_to_us(self):
|
|
self.contact.update_last_replied()
|
|
self.assertTrue(self.contact.contact_is_good is True)
|
|
|
|
def test_stale_requested_from_us(self):
|
|
self.contact.update_last_requested()
|
|
self.assertTrue(self.contact.contact_is_good is None)
|
|
|
|
def test_stale_then_fail(self):
|
|
self.contact.update_last_failed()
|
|
self.assertTrue(self.contact.contact_is_good is None)
|
|
self.clock.advance(1)
|
|
self.contact.update_last_failed()
|
|
self.assertTrue(self.contact.contact_is_good is False)
|
|
|
|
def test_good_turned_stale(self):
|
|
self.contact.update_last_replied()
|
|
self.assertTrue(self.contact.contact_is_good is True)
|
|
self.clock.advance(constants.checkRefreshInterval - 1)
|
|
self.assertTrue(self.contact.contact_is_good is True)
|
|
self.clock.advance(1)
|
|
self.assertTrue(self.contact.contact_is_good is None)
|
|
|
|
def test_good_then_fail(self):
|
|
self.contact.update_last_replied()
|
|
self.assertTrue(self.contact.contact_is_good is True)
|
|
self.clock.advance(1)
|
|
self.contact.update_last_failed()
|
|
self.assertTrue(self.contact.contact_is_good is True)
|
|
self.clock.advance(59)
|
|
self.assertTrue(self.contact.contact_is_good is True)
|
|
self.contact.update_last_failed()
|
|
self.assertTrue(self.contact.contact_is_good is False)
|
|
for _ in range(7200):
|
|
self.clock.advance(60)
|
|
self.assertTrue(self.contact.contact_is_good is False)
|
|
|
|
def test_good_then_fail_then_good(self):
|
|
# it replies
|
|
self.contact.update_last_replied()
|
|
self.assertTrue(self.contact.contact_is_good is True)
|
|
self.clock.advance(1)
|
|
|
|
# it fails twice in a row
|
|
self.contact.update_last_failed()
|
|
self.clock.advance(1)
|
|
self.contact.update_last_failed()
|
|
self.assertTrue(self.contact.contact_is_good is False)
|
|
self.clock.advance(1)
|
|
|
|
# it replies
|
|
self.contact.update_last_replied()
|
|
self.clock.advance(1)
|
|
self.assertTrue(self.contact.contact_is_good is True)
|
|
|
|
# it goes stale
|
|
self.clock.advance(constants.checkRefreshInterval - 2)
|
|
self.assertTrue(self.contact.contact_is_good is True)
|
|
self.clock.advance(1)
|
|
self.assertTrue(self.contact.contact_is_good is None)
|
|
|
|
|
|
class TestContactLastRequested(unittest.TestCase):
|
|
def setUp(self):
|
|
self.clock = task.Clock()
|
|
self.contact_manager = ContactManager(self.clock.seconds)
|
|
self.contact = self.contact_manager.make_contact(generate_id(), "127.0.0.1", 4444, None)
|
|
self.clock.advance(1)
|
|
self.contact.update_last_replied()
|
|
self.clock.advance(3600)
|
|
self.assertTrue(self.contact.contact_is_good is None)
|
|
|
|
def test_previous_replied_then_requested(self):
|
|
# it requests
|
|
self.contact.update_last_requested()
|
|
self.assertTrue(self.contact.contact_is_good is True)
|
|
|
|
# it goes stale
|
|
self.clock.advance(constants.checkRefreshInterval - 1)
|
|
self.assertTrue(self.contact.contact_is_good is True)
|
|
self.clock.advance(1)
|
|
self.assertTrue(self.contact.contact_is_good is None)
|
|
|
|
def test_previous_replied_then_requested_then_failed(self):
|
|
# it requests
|
|
self.contact.update_last_requested()
|
|
self.assertTrue(self.contact.contact_is_good is True)
|
|
self.clock.advance(1)
|
|
|
|
# it fails twice in a row
|
|
self.contact.update_last_failed()
|
|
self.clock.advance(1)
|
|
self.contact.update_last_failed()
|
|
self.assertTrue(self.contact.contact_is_good is False)
|
|
self.clock.advance(1)
|
|
|
|
# it requests
|
|
self.contact.update_last_requested()
|
|
self.clock.advance(1)
|
|
self.assertTrue(self.contact.contact_is_good is False)
|
|
|
|
# it goes stale
|
|
self.clock.advance((constants.refreshTimeout / 4) - 2)
|
|
self.assertTrue(self.contact.contact_is_good is False)
|
|
self.clock.advance(1)
|
|
self.assertTrue(self.contact.contact_is_good is False)
|