update and fix hash announcer test
This commit is contained in:
parent
d02ed29e50
commit
c521120b17
3 changed files with 26 additions and 25 deletions
|
@ -50,10 +50,10 @@ class DHTHashAnnouncer(object):
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def immediate_announce(self, blob_hashes):
|
def immediate_announce(self, blob_hashes):
|
||||||
self.hash_queue.extend(b for b in blob_hashes if b not in self.hash_queue)
|
self.hash_queue.extend(b for b in blob_hashes if b not in self.hash_queue)
|
||||||
|
|
||||||
log.info("Announcing %i blobs", len(self.hash_queue))
|
log.info("Announcing %i blobs", len(self.hash_queue))
|
||||||
start = self.clock.seconds()
|
start = self.clock.seconds()
|
||||||
progress_lc = task.LoopingCall(self._show_announce_progress, len(self.hash_queue), start)
|
progress_lc = task.LoopingCall(self._show_announce_progress, len(self.hash_queue), start)
|
||||||
|
progress_lc.clock = self.clock
|
||||||
progress_lc.start(60, now=False)
|
progress_lc.start(60, now=False)
|
||||||
s = defer.DeferredSemaphore(self.concurrent_announcers)
|
s = defer.DeferredSemaphore(self.concurrent_announcers)
|
||||||
results = yield utils.DeferredDict({blob_hash: s.run(self.do_store, blob_hash) for blob_hash in blob_hashes})
|
results = yield utils.DeferredDict({blob_hash: s.run(self.do_store, blob_hash) for blob_hash in blob_hashes})
|
||||||
|
|
|
@ -250,9 +250,6 @@ class Announcer(object):
|
||||||
def hash_queue_size(self):
|
def hash_queue_size(self):
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
def add_supplier(self, supplier):
|
|
||||||
pass
|
|
||||||
|
|
||||||
def immediate_announce(self, *args):
|
def immediate_announce(self, *args):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
|
@ -1,55 +1,59 @@
|
||||||
from twisted.trial import unittest
|
from twisted.trial import unittest
|
||||||
from twisted.internet import defer, task
|
from twisted.internet import defer, task
|
||||||
|
from lbrynet import conf
|
||||||
from lbrynet.core import utils
|
from lbrynet.core import utils
|
||||||
|
from lbrynet.dht.hashannouncer import DHTHashAnnouncer
|
||||||
from lbrynet.tests.util import random_lbry_hash
|
from lbrynet.tests.util import random_lbry_hash
|
||||||
|
|
||||||
|
|
||||||
class MocDHTNode(object):
|
class MocDHTNode(object):
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self.blobs_announced = 0
|
self.blobs_announced = 0
|
||||||
|
self.clock = task.Clock()
|
||||||
|
self.peerPort = 3333
|
||||||
|
|
||||||
def announceHaveBlob(self, blob):
|
def announceHaveBlob(self, blob):
|
||||||
self.blobs_announced += 1
|
self.blobs_announced += 1
|
||||||
return defer.succeed(True)
|
d = defer.Deferred()
|
||||||
|
self.clock.callLater(1, d.callback, ['fake'])
|
||||||
|
return d
|
||||||
|
|
||||||
class MocSupplier(object):
|
|
||||||
|
class MocStorage(object):
|
||||||
def __init__(self, blobs_to_announce):
|
def __init__(self, blobs_to_announce):
|
||||||
self.blobs_to_announce = blobs_to_announce
|
self.blobs_to_announce = blobs_to_announce
|
||||||
self.announced = False
|
self.announced = False
|
||||||
def hashes_to_announce(self):
|
|
||||||
|
def get_blobs_to_announce(self):
|
||||||
if not self.announced:
|
if not self.announced:
|
||||||
self.announced = True
|
self.announced = True
|
||||||
return defer.succeed(self.blobs_to_announce)
|
return defer.succeed(self.blobs_to_announce)
|
||||||
else:
|
else:
|
||||||
return defer.succeed([])
|
return defer.succeed([])
|
||||||
|
|
||||||
|
def update_last_announced_blob(self, blob_hash, now):
|
||||||
|
return defer.succeed(None)
|
||||||
|
|
||||||
|
|
||||||
class DHTHashAnnouncerTest(unittest.TestCase):
|
class DHTHashAnnouncerTest(unittest.TestCase):
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
|
conf.initialize_settings(False)
|
||||||
self.num_blobs = 10
|
self.num_blobs = 10
|
||||||
self.blobs_to_announce = []
|
self.blobs_to_announce = []
|
||||||
for i in range(0, self.num_blobs):
|
for i in range(0, self.num_blobs):
|
||||||
self.blobs_to_announce.append(random_lbry_hash())
|
self.blobs_to_announce.append(random_lbry_hash())
|
||||||
self.clock = task.Clock()
|
|
||||||
self.dht_node = MocDHTNode()
|
self.dht_node = MocDHTNode()
|
||||||
|
self.clock = self.dht_node.clock
|
||||||
utils.call_later = self.clock.callLater
|
utils.call_later = self.clock.callLater
|
||||||
from lbrynet.core.server.DHTHashAnnouncer import DHTHashAnnouncer
|
self.storage = MocStorage(self.blobs_to_announce)
|
||||||
self.announcer = DHTHashAnnouncer(self.dht_node, peer_port=3333)
|
self.announcer = DHTHashAnnouncer(self.dht_node, self.storage)
|
||||||
self.supplier = MocSupplier(self.blobs_to_announce)
|
|
||||||
self.announcer.add_supplier(self.supplier)
|
|
||||||
|
|
||||||
def test_basic(self):
|
@defer.inlineCallbacks
|
||||||
self.announcer._announce_available_hashes()
|
def test_immediate_announce(self):
|
||||||
self.assertEqual(self.announcer.hash_queue_size(), self.announcer.CONCURRENT_ANNOUNCERS)
|
announce_d = self.announcer.immediate_announce(self.blobs_to_announce)
|
||||||
|
self.assertEqual(self.announcer.hash_queue_size(), self.num_blobs)
|
||||||
self.clock.advance(1)
|
self.clock.advance(1)
|
||||||
|
yield announce_d
|
||||||
self.assertEqual(self.dht_node.blobs_announced, self.num_blobs)
|
self.assertEqual(self.dht_node.blobs_announced, self.num_blobs)
|
||||||
self.assertEqual(self.announcer.hash_queue_size(), 0)
|
self.assertEqual(self.announcer.hash_queue_size(), 0)
|
||||||
|
|
||||||
def test_immediate_announce(self):
|
|
||||||
# Test that immediate announce puts a hash at the front of the queue
|
|
||||||
self.announcer._announce_available_hashes()
|
|
||||||
blob_hash = random_lbry_hash()
|
|
||||||
self.announcer.immediate_announce([blob_hash])
|
|
||||||
self.assertEqual(self.announcer.hash_queue_size(), self.announcer.CONCURRENT_ANNOUNCERS+1)
|
|
||||||
self.assertEqual(blob_hash, self.announcer.hash_queue[0][0])
|
|
||||||
|
|
Loading…
Reference in a new issue