written test case randomtrackingdict dict module

This commit is contained in:
navjot 2020-12-28 16:01:23 +05:30
parent 9fe4ad0489
commit a2ab0a02dc
No known key found for this signature in database
GPG Key ID: 9EE70AFD71357F1C
3 changed files with 54 additions and 40 deletions

View File

@ -1,7 +1,6 @@
"""
Track randomize ordered dict
"""
import random
from threading import RLock
from time import time
@ -128,41 +127,3 @@ class RandomTrackingDict(object):
self.pendingLen += 1
self.lastPoll = time()
return retval
if __name__ == '__main__':
# pylint: disable=redefined-outer-name
def randString():
"""helper function for tests, generates a random string"""
retval = b''
for _ in range(32):
retval += chr(random.randint(0, 255))
return retval
a = []
k = RandomTrackingDict()
d = {}
print "populating random tracking dict"
a.append(time())
for i in range(50000):
k[randString()] = True
a.append(time())
print "done"
while k:
retval = k.randomKeys(1000)
if not retval:
print "error getting random keys"
try:
k.randomKeys(100)
print "bad"
except KeyError:
pass
for i in retval:
del k[i]
a.append(time())
for x in range(len(a) - 1):
print "%i: %.3f" % (x, a[x + 1] - a[x])

View File

@ -7,6 +7,7 @@ from binascii import hexlify, unhexlify
from os import listdir, makedirs, path, remove, rmdir
from threading import RLock
from debug import logger
from paths import lookupAppdataFolder
from storage import InventoryItem, InventoryStorage
@ -162,7 +163,7 @@ class FilesystemInventory(InventoryStorage):
newInventory[streamNumber][hashId] = InventoryItem(
objectType, streamNumber, None, expiresTime, tag)
except KeyError:
print "error loading %s" % (hexlify(hashId))
logger.warning('error loading %s', hexlify(hashId))
self._inventory = newInventory
# for i, v in self._inventory.items():
# print "loaded stream: %s, %i items" % (i, len(v))

View File

@ -0,0 +1,52 @@
"""
Tests for RandomTrackingDict class
"""
import random
import unittest
from time import time
class TestRandomTrackingDict(unittest.TestCase):
"""
Main protocol test case
"""
@staticmethod
def randString():
"""helper function for tests, generates a random string"""
retval = b''
for _ in range(32):
retval += chr(random.randint(0, 255))
return retval
def test_check_randomtrackingdict(self):
"""Check the logic of RandomTrackingDict class"""
# from network import RandomTrackingDict
# from pybitmessage.network.randomtrackingdict import RandomTrackingDict
# from network.randomtrackingdict import RandomTrackingDict
from pybitmessage import network
a = []
k = network.randomtrackingdict.RandomTrackingDict()
a.append(time())
for i in range(50000):
k[self.randString()] = True
a.append(time())
while k:
retval = k.randomKeys(1000)
if not retval:
self.fail("error getting random keys")
try:
k.randomKeys(100)
self.fail("bad")
except KeyError:
pass
for i in retval:
del k[i]
a.append(time())
for x in range(len(a) - 1):
self.assertLess(a[x + 1] - a[x], 10)