我编写了一个小型代理实用程序(tcp 端口转发器),现在尝试测试它。 所以我的测试:
def test_forwarding(self):
route = self.config.routes[0]
q = multiprocessing.JoinableQueue()
proc_listen = multiprocessing.Process(
target=lambda q: q.put(str(
subprocess.Popen(["nc", "-l", str(route.sink[1])], stdout=subprocess.PIPE, shell=False).communicate()[0])),
args=(q,))
proc_listen.start()
proc_write = multiprocessing.Process(
target=lambda: subprocess.Popen("nc %s %s < ~/Workshop/port-forwarder/tests/test_data.txt" % route.source,
stdout=subprocess.PIPE, shell=True))
proc_write.start()
proc_write.join(3)
proc_write.terminate()
proc_listen.join(3)
proc_listen.terminate()
q.join()
self.assertEqual(open("test_data.txt", 'r').readline(), q.get())
但测试卡在断言中的 get()
调用上。完全搞砸了,尝试了所有诸如加入/终止之类的东西。任何帮助将不胜感激。
这里是 github 存储库及其来源: enter link description here
最佳答案
如果你被get屏蔽了,你可以
或使用empty()检查侧面是否有某些元素。如果不为空则获取,否则跳过
关于Python 多进程队列 get() block ,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/42766384/