Madiator2011/lbry-sdk
1
0
Fork 0
forked from LBRYCommunity/lbry-sdk
Code Pull requests Activity

add random_lbry_hash() function to tests.util

Browse source
This commit is contained in:
Kay Kurokawa 2017-02-17 19:59:57 -05:00
parent ac33663444
commit eae8f8a00e
2 changed files with 11 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
tests/unit/core/server/test_DHTHashAnnouncer.py
View file

@ -1,9 +1,10 @@
import os
import binascii
from twisted.trial import unittest from twisted.trial import unittest
from twisted.internet import defer,task from twisted.internet import defer,task
from lbrynet.core import log_support, utils
from lbrynet.core.server.DHTHashAnnouncer import DHTHashAnnouncer,DHTHashSupplier
from lbrynet.core.utils import random_string
from lbrynet.core import log_support, utils
from tests.util import random_lbry_hash
class MocDHTNode(object): class MocDHTNode(object):
def __init__(self): def __init__(self):
@ -30,7 +31,7 @@ class DHTHashAnnouncerTest(unittest.TestCase):
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(binascii.b2a_hex(os.urandom(32))) self.blobs_to_announce.append(random_lbry_hash())
self.clock = task.Clock() self.clock = task.Clock()
self.dht_node = MocDHTNode() self.dht_node = MocDHTNode()
utils.call_later = self.clock.callLater utils.call_later = self.clock.callLater
@ -49,7 +50,7 @@ class DHTHashAnnouncerTest(unittest.TestCase):
def test_immediate_announce(self): def test_immediate_announce(self):
# Test that immediate announce puts a hash at the front of the queue # Test that immediate announce puts a hash at the front of the queue
self.announcer._announce_available_hashes() self.announcer._announce_available_hashes()
blob_hash = binascii.b2a_hex(os.urandom(32)) blob_hash = random_lbry_hash()
self.announcer.immediate_announce([blob_hash]) self.announcer.immediate_announce([blob_hash])
self.assertEqual(self.announcer.hash_queue_size(),self.announcer.CONCURRENT_ANNOUNCERS+1) self.assertEqual(self.announcer.hash_queue_size(),self.announcer.CONCURRENT_ANNOUNCERS+1)
self.assertEqual(blob_hash, self.announcer.hash_queue[0][0]) self.assertEqual(blob_hash, self.announcer.hash_queue[0][0])

5
tests/util.py
View file

@ -1,5 +1,7 @@
import datetime import datetime
import time import time
import binascii
import os
import mock import mock
@ -8,6 +10,9 @@ DEFAULT_TIMESTAMP = datetime.datetime(2016, 1, 1)
DEFAULT_ISO_TIME = time.mktime(DEFAULT_TIMESTAMP.timetuple()) DEFAULT_ISO_TIME = time.mktime(DEFAULT_TIMESTAMP.timetuple())
def random_lbry_hash():
return binascii.b2a_hex(os.urandom(48))
def resetTime(test_case, timestamp=DEFAULT_TIMESTAMP): def resetTime(test_case, timestamp=DEFAULT_TIMESTAMP):
iso_time = time.mktime(timestamp.timetuple()) iso_time = time.mktime(timestamp.timetuple())
patcher = mock.patch('time.time') patcher = mock.patch('time.time')