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