我有一个队列对象,它在我的项目中扮演着相当重要的角色,我无法承受其中有任何错误。
它的想法是类似于 builtin Queue
,它的基类,但它将数据或部分数据至少存储在文件中以保留内存。我决定将其中的一些保留在内存中,因为这样可以加快速度。我已经把代码放在这里了,看它可能比我解释它更容易
想做这件事似乎很奇怪,但我需要排队很多工作,排队的速度比我完成它的速度要快得多,而且如果我使用标准的Queue
. 我不能只在 上放一个 maxsizeQueue
并阻止工人将东西放入队列中,因为我想尽快知道要处理的数据总量。我也无法先计算出总数,但不要排队,然后通过输入返回,Queue
因为每次查看数据时总数都会不同,最后总数不会匹配。
我的问题是如何彻底测试这个以确保没有项目丢失或更重要的是当缓冲区或文件中仍有项目时,或者在调用完成并且队列为空之后,getter 没有任何阻塞。
当您知道某些给定输入的输出应该是什么时,有些事情似乎很容易测试和设置单元测试,但是测试这样的事情我并不确定一种有效的方法。是否可以通过单元测试来测试这种东西?
我已经建立了一个测试程序,它以不同数量的项目以不同、不同的速度放入和获取项目,这看起来很好,但我已经看到有证据表明吸气剂阻塞.get
了仍在队列中的项目,所以我m 导致相信有问题。
我可以彻底测试以找到任何剩余的错误或几乎确定它没有错误的最佳方法是什么?
编辑
可以生成一些类似于我使用以下代码的测试数据,我只有在项目中某些条件下的文件的校验和,而其他时候它是None
所以我只是在下面的代码中生成它来尝试和模拟那个
import os
import hashlib
def hash(f_obj):
md5 = hashlib.md5()
while True:
data = f_obj.read(8192)
if not data:
break
md5.update(data)
return md5.hexdigest()
def produce(at_once,total_items):
items=[]
count=0
for dir,folders,files in os.walk("/"):
for f in files:
try:
f_path= os.path.join(dir,f)
f_size= os.path.getsize(f_path)
f_mtime= os.path.getmtime(f_path)
with open(f_path) as file_obj:
f_hash= hash(file_obj) if f_size%2 else None
items.append((f_path,f_size,f_mtime,f_hash))
count+=1
except Exception as err:
print "#####",err,"#####"
if len(items) >= at_once:
yield items
items=[]
if count >= total_items:
break
if items:
yield items