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

Question

我需要能够在网络的每个“节点”上运行 python 代码，以便我可以正确测试代码。我无法使用不同的端口号并运行代码，因为我需要处理各种其他事情，这些事情强制使用唯一的 IP 地址。

Answer 1

在我的 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

Answer 2

我认为 vmware 或 virtual box 可以帮助你。