我需要能夠在網絡的每個“節點”上運行 python 代碼，以便我可以正確測試代碼。我無法使用不同的端口號并運行代碼，因為我需要處理各種其他事情，這些事情強制使用唯一的 IP 地址。
2 回答
holdtom
TA貢獻1805條經驗 獲得超10個贊
在我的 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 out
class 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.asyncio
async 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
添加回答
舉報
0/150
提交
取消