我需要能夠在網(wǎng)絡(luò)的每個(gè)“節(jié)點(diǎn)”上運(yùn)行 python 代碼，以便我可以正確測(cè)試代碼。我無(wú)法使用不同的端口號(hào)并運(yùn)行代碼，因?yàn)槲倚枰幚砀鞣N其他事情，這些事情強(qiáng)制使用唯一的 IP 地址。
2 回答
holdtom
TA貢獻(xiàn)1805條經(jīng)驗(yàn) 獲得超10個(gè)贊
在我的 DHT p2p 項(xiàng)目中，我有一個(gè)抽象網(wǎng)絡(luò)通信的特定對(duì)象。在測(cè)試過(guò)程中，我使用在內(nèi)存中操作的對(duì)象來(lái)模擬該對(duì)象：
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
添加回答
舉報(bào)
0/150
提交
取消