有没有办法模拟网络,以便我可以测试用 python 编写的 p2p 网络代码?

我需要能够在网络的每个“节点”上运行 python 代码,以便我可以正确测试代码。我无法使用不同的端口号并运行代码,因为我需要处理各种其他事情,这些事情强制使用唯一的 IP 地址。



青春有我
浏览 108回答 2
2回答

holdtom

在我的 DHT p2p 项目中,我有一个抽象网络通信的特定对象。在测试过程中,我使用在内存中操作的对象来模拟该对象:class MockProtocol:&nbsp; &nbsp; def __init__(self, network, peer):&nbsp; &nbsp; &nbsp; &nbsp; self.network = network&nbsp; &nbsp; &nbsp; &nbsp; self.peer = peer&nbsp; &nbsp; async def rpc(self, address, name, *args):&nbsp; &nbsp; &nbsp; &nbsp; peer = self.network.peers[address[0]]&nbsp; &nbsp; &nbsp; &nbsp; proc = getattr(peer, name)&nbsp; &nbsp; &nbsp; &nbsp; start = time()&nbsp; &nbsp; &nbsp; &nbsp; out = await proc((self.peer._uid, None), *args)&nbsp; &nbsp; &nbsp; &nbsp; delta = time() - start&nbsp; &nbsp; &nbsp; &nbsp; assert delta < 5, "RPCProtocol allows 5s delay only"&nbsp; &nbsp; &nbsp; &nbsp; return outclass MockNetwork:&nbsp; &nbsp; def __init__(self):&nbsp; &nbsp; &nbsp; &nbsp; self.peers = dict()&nbsp; &nbsp; def add(self, peer):&nbsp; &nbsp; &nbsp; &nbsp; peer._protocol = MockProtocol(self, peer)&nbsp; &nbsp; &nbsp; &nbsp; self.peers[peer._uid] = peer&nbsp; &nbsp; def choice(self):&nbsp; &nbsp; &nbsp; &nbsp; return random.choice(list(self.peers.values()))async def simple_network():&nbsp; &nbsp; network = MockNetwork()&nbsp; &nbsp; for i in range(5):&nbsp; &nbsp; &nbsp; &nbsp; peer = make_peer()&nbsp; &nbsp; &nbsp; &nbsp; network.add(peer)&nbsp; &nbsp; bootstrap = peer&nbsp; &nbsp; for peer in network.peers.values():&nbsp; &nbsp; &nbsp; &nbsp; await peer.bootstrap((bootstrap._uid, None))&nbsp; &nbsp; for peer in network.peers.values():&nbsp; &nbsp; &nbsp; &nbsp; await peer.bootstrap((bootstrap._uid, None))&nbsp; &nbsp; # run connect, this simulate the peers connecting to an existing&nbsp; &nbsp; # network.&nbsp; &nbsp; for peer in network.peers.values():&nbsp; &nbsp; &nbsp; &nbsp; await peer.connect()&nbsp; &nbsp; return network@pytest.mark.asyncioasync def test_dict(make_network):&nbsp; &nbsp; network = await make_network()&nbsp; &nbsp; # setup&nbsp; &nbsp; value = b'test value'&nbsp; &nbsp; key = peer.hash(value)&nbsp; &nbsp; # make network and peers&nbsp; &nbsp; one = network.choice()&nbsp; &nbsp; two = network.choice()&nbsp; &nbsp; three = network.choice()&nbsp; &nbsp; four = network.choice()&nbsp; &nbsp; # exec&nbsp; &nbsp; out = await three.set(value)&nbsp; &nbsp; # check&nbsp; &nbsp; assert out == key&nbsp; &nbsp; fallback = list()&nbsp; &nbsp; for xxx in (one, two, three, four):&nbsp; &nbsp; &nbsp; &nbsp; try:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; out = await xxx.get(key)&nbsp; &nbsp; &nbsp; &nbsp; except KeyError:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; fallback.append(xxx)&nbsp; &nbsp; &nbsp; &nbsp; else:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; assert out == value&nbsp; &nbsp; for xxx in fallback:&nbsp; &nbsp; &nbsp; &nbsp; log.warning('fallback for peer %r', xxx)&nbsp; &nbsp; &nbsp; &nbsp; out = await xxx.get_at(key, three._uid)&nbsp; &nbsp; &nbsp; &nbsp; assert out == value

POPMUISE

我认为 vmware 或 virtual box 可以帮助你。
打开App,查看更多内容
随时随地看视频慕课网APP

相关分类

Python