【问题标题】:Is there a way to simulate a network so that I can test my p2p network code written in python?有没有办法模拟网络,以便我可以测试用 python 编写的 p2p 网络代码?
【发布时间】:2020-12-24 15:30:48
【问题描述】:

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

【问题讨论】:

    标签: python networking simulation p2p


    【解决方案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
    

    【讨论】:

      【解决方案2】:

      我觉得vmware或者virtual box可以帮到你。

      【讨论】:

        猜你喜欢
        • 1970-01-01
        • 1970-01-01
        • 2019-05-10
        • 1970-01-01
        • 2015-07-25
        • 1970-01-01
        • 1970-01-01
        • 2021-01-10
        • 1970-01-01
        相关资源
        最近更新 更多