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()