Move most of the logic of TestCore.test_udp() into the TestNetwork

This commit is contained in:
Lee Miller 2023-12-20 04:26:55 +02:00
parent 606b6d7d08
commit c5ced60d5d
Signed by untrusted user: lee.miller
GPG Key ID: 4F97A5EA88F4AB63
2 changed files with 25 additions and 20 deletions

View File

@ -310,26 +310,6 @@ class TestCore(unittest.TestCase):
else:
return self.fail('No Announcer thread found')
for _ in range(20): # wait for UDP socket
for sock in BMConnectionPool().udpSockets.values():
thread.announceSelf()
break
else:
time.sleep(1)
continue
break
else:
self.fail('UDP socket is not started')
for _ in range(20):
if state.discoveredPeers:
peer = state.discoveredPeers.keys()[0]
self.assertEqual(peer.port, 8444)
break
time.sleep(1)
else:
self.fail('No self in discovered peers')
@staticmethod
def _decode_msg(data, pattern):
proto = BMProto()

View File

@ -63,6 +63,31 @@ class TestNetwork(TestPartialRun):
else:
self.fail('Have not started any connection in 30 sec')
def test_udp(self):
"""Invoke AnnounceThread.announceSelf() and check discovered peers"""
for _ in range(20):
if self.pool.udpSockets:
break
time.sleep(1)
else:
self.fail('No UDP sockets found in 20 sec')
for _ in range(10):
try:
self.state.announceThread.announceSelf()
except AttributeError:
self.fail('state.announceThread is not set properly')
time.sleep(1)
try:
peer = self.state.discoveredPeers.popitem()[0]
except KeyError:
continue
else:
self.assertEqual(peer.port, 8444)
break
else:
self.fail('No self in discovered peers')
@classmethod
def tearDownClass(cls):
super(TestNetwork, cls).tearDownClass()