varun-r-mallya/py-libp2p
1
0
Fork 0
mirror of https://github.com/varun-r-mallya/py-libp2p.git synced 2026-02-12 16:10:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

modify factories to fix tests

Browse Source
This commit is contained in:
Alex Stokes
2019-10-28 20:38:50 +09:00
parent 3deccac2da
commit a07c3b0fb0
2 changed files with 30 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
tests/factories.py
View File

@ -37,8 +37,9 @@ def security_transport_factory(
return {secio.ID: secio.Transport(key_pair)} return {secio.ID: secio.Transport(key_pair)}
def SwarmFactory(is_secure: bool, muxer_opt: TMuxerOptions = None) -> Swarm: def SwarmFactory(
key_pair = generate_new_rsa_identity() is_secure: bool, key_pair: KeyPair, muxer_opt: TMuxerOptions = None
) -> Swarm:
sec_opt = security_transport_factory(is_secure, key_pair) sec_opt = security_transport_factory(is_secure, key_pair)
return initialize_default_swarm(key_pair, sec_opt=sec_opt, muxer_opt=muxer_opt) return initialize_default_swarm(key_pair, sec_opt=sec_opt, muxer_opt=muxer_opt)
@ -50,15 +51,16 @@ class ListeningSwarmFactory(factory.Factory):
@classmethod @classmethod
async def create_and_listen( async def create_and_listen(
cls, is_secure: bool, muxer_opt: TMuxerOptions = None cls, is_secure: bool, muxer_opt: TMuxerOptions = None
) -> Swarm: ) -> Tuple[Swarm, KeyPair]:
swarm = SwarmFactory(is_secure, muxer_opt=muxer_opt) key_pair = generate_new_rsa_identity()
swarm = SwarmFactory(is_secure, key_pair, muxer_opt=muxer_opt)
await swarm.listen(LISTEN_MADDR) await swarm.listen(LISTEN_MADDR)
return swarm return swarm, key_pair
@classmethod @classmethod
async def create_batch_and_listen( async def create_batch_and_listen(
cls, is_secure: bool, number: int, muxer_opt: TMuxerOptions = None cls, is_secure: bool, number: int, muxer_opt: TMuxerOptions = None
) -> Tuple[Swarm, ...]: ) -> Tuple[Tuple[Swarm, KeyPair], ...]:
return await asyncio.gather( return await asyncio.gather(
*[ *[
cls.create_and_listen(is_secure, muxer_opt=muxer_opt) cls.create_and_listen(is_secure, muxer_opt=muxer_opt)
@ -74,19 +76,26 @@ class HostFactory(factory.Factory):
class Params: class Params:
is_secure = False is_secure = False
network = factory.LazyAttribute(lambda o: SwarmFactory(o.is_secure)) network = factory.LazyAttribute(lambda o: SwarmFactory(o.is_secure, o.key_pair))
@classmethod @classmethod
async def create_and_listen(cls, is_secure: bool) -> BasicHost: async def create_and_listen(cls, is_secure: bool) -> BasicHost:
swarms = await ListeningSwarmFactory.create_batch_and_listen(is_secure, 1) swarms_and_keys = await ListeningSwarmFactory.create_batch_and_listen(
return BasicHost(swarms[0]) is_secure, 1
)
swarm, key_pair = swarms_and_keys[0]
return BasicHost(key_pair.public_key, swarm)
@classmethod @classmethod
async def create_batch_and_listen( async def create_batch_and_listen(
cls, is_secure: bool, number: int cls, is_secure: bool, number: int
) -> Tuple[BasicHost, ...]: ) -> Tuple[BasicHost, ...]:
swarms = await ListeningSwarmFactory.create_batch_and_listen(is_secure, number) swarms_and_keys = await ListeningSwarmFactory.create_batch_and_listen(
return tuple(BasicHost(swarm) for swarm in range(swarms)) is_secure, number
)
return tuple(
BasicHost(key_pair.public_key, swarm) for swarm, key_pair in swarms_and_keys
)
class FloodsubFactory(factory.Factory): class FloodsubFactory(factory.Factory):
@ -123,9 +132,10 @@ class PubsubFactory(factory.Factory):
async def swarm_pair_factory( async def swarm_pair_factory(
is_secure: bool, muxer_opt: TMuxerOptions = None is_secure: bool, muxer_opt: TMuxerOptions = None
) -> Tuple[Swarm, Swarm]: ) -> Tuple[Swarm, Swarm]:
swarms = await ListeningSwarmFactory.create_batch_and_listen( swarms_and_keys = await ListeningSwarmFactory.create_batch_and_listen(
is_secure, 2, muxer_opt=muxer_opt is_secure, 2, muxer_opt=muxer_opt
) )
swarms = tuple(swarm for swarm, _key_pair in swarms_and_keys)
await connect_swarm(swarms[0], swarms[1]) await connect_swarm(swarms[0], swarms[1])
return swarms[0], swarms[1] return swarms[0], swarms[1]

10
tests/network/test_swarm.py
View File

@ -9,7 +9,10 @@ from tests.utils import connect_swarm
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_swarm_dial_peer(is_host_secure): async def test_swarm_dial_peer(is_host_secure):
swarms = await ListeningSwarmFactory.create_batch_and_listen(is_host_secure, 3) swarms_and_keys = await ListeningSwarmFactory.create_batch_and_listen(
is_host_secure, 3
)
swarms = tuple(swarm for swarm, _key_pair in swarms_and_keys)
# Test: No addr found. # Test: No addr found.
with pytest.raises(SwarmException): with pytest.raises(SwarmException):
await swarms[0].dial_peer(swarms[1].get_peer_id()) await swarms[0].dial_peer(swarms[1].get_peer_id())
@ -41,7 +44,10 @@ async def test_swarm_dial_peer(is_host_secure):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_swarm_close_peer(is_host_secure): async def test_swarm_close_peer(is_host_secure):
swarms = await ListeningSwarmFactory.create_batch_and_listen(is_host_secure, 3) swarms_and_keys = await ListeningSwarmFactory.create_batch_and_listen(
is_host_secure, 3
)
swarms = tuple(swarm for swarm, _key_pair in swarms_and_keys)
# 0 <> 1 <> 2 # 0 <> 1 <> 2
await connect_swarm(swarms[0], swarms[1]) await connect_swarm(swarms[0], swarms[1])
await connect_swarm(swarms[1], swarms[2]) await connect_swarm(swarms[1], swarms[2])