49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
|
import unittest
|
||
|
|
from datetime import timedelta, datetime
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
from silverstream.bittorrent.dht.peer import Peer, PeerStatus, Contact
|
||
|
|
|
||
|
|
|
||
|
|
class TestContact(unittest.TestCase):
|
||
|
|
def test_compact(self):
|
||
|
|
contact = Contact("1.2.3.4", 8080)
|
||
|
|
self.assertEqual(contact, Contact.from_compact(contact.compact))
|
||
|
|
|
||
|
|
|
||
|
|
class TestPeer(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.peer = Peer("1.2.3.4", 8080, b'\x05%Rz\x00\x91\x91*\xd3\x10\x05C\xb8F\xa7\xe8\xab\xfe\xc7W')
|
||
|
|
|
||
|
|
def test_compact(self):
|
||
|
|
self.assertEqual(self.peer.compact, Peer.from_compact(self.peer.compact).compact)
|
||
|
|
|
||
|
|
def test_questionable_by_default(self):
|
||
|
|
self.assertEqual(self.peer.status, PeerStatus.QUESTIONABLE)
|
||
|
|
|
||
|
|
def test_set_questionable(self):
|
||
|
|
self.peer.status = PeerStatus.BAD
|
||
|
|
self.peer.status = PeerStatus.QUESTIONABLE
|
||
|
|
self.assertEqual(self.peer.status, PeerStatus.QUESTIONABLE)
|
||
|
|
|
||
|
|
def test_bad_overrides(self):
|
||
|
|
peer = Peer("1.2.3.4", 8080, b'\x05%Rz\x00\x91\x91*\xd3\x10\x05C\xb8F\xa7\xe8\xab\xfe\xc7W',
|
||
|
|
status=PeerStatus.BAD)
|
||
|
|
self.assertEqual(peer.status, PeerStatus.BAD)
|
||
|
|
|
||
|
|
def test_good_for_15min(self):
|
||
|
|
with patch("silverstream.bittorrent.dht.peer.datetime") as mock_datetime:
|
||
|
|
mock_datetime.utcnow.return_value = datetime(2010, 1, 1, 12, 0, 0)
|
||
|
|
mock_datetime.side_effect = lambda *args, **kw: datetime(*args, **kw)
|
||
|
|
|
||
|
|
self.peer.status = PeerStatus.GOOD
|
||
|
|
self.assertEqual(self.peer.status, PeerStatus.GOOD)
|
||
|
|
|
||
|
|
mock_datetime.utcnow.return_value += timedelta(minutes=15)
|
||
|
|
|
||
|
|
self.assertEqual(self.peer.status, PeerStatus.QUESTIONABLE)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
unittest.main()
|