有没有办法模拟网络,以便我可以测试用 python 编写的 p2p 网络代码?
Is there a way to simulate a network so that I can test my p2p network code written in python?
我需要能够在网络的每个“节点”上 运行 python 编码,以便我可以正确地测试代码。我不能使用不同的端口号和 运行 代码,因为我需要使用唯一的 IP 地址来处理其他各种事情。
我觉得vmware或者virtual box可以帮到你
在我的 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
我需要能够在网络的每个“节点”上 运行 python 编码,以便我可以正确地测试代码。我不能使用不同的端口号和 运行 代码,因为我需要使用唯一的 IP 地址来处理其他各种事情。
我觉得vmware或者virtual box可以帮到你
在我的 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