有没有一种模拟网络的方法，以便可以测试用python编写的p2p网络代码？

问题描述

我需要能够在网络的每个“节点”上运行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