StarMirror/py-libp2p
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

add asynccontextmanager utility for a pair of connected hosts

Browse Source
This commit is contained in:
Alex Stokes 2019-11-07 20:59:01 -08:00
parent 2a7b43d853
commit 30dee28ef2
No known key found for this signature in database
GPG Key ID: 51CE1721B245C086
1 changed files with 9 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
tests/factories.py
View File

@ -1,4 +1,5 @@
import asyncio import asyncio
from contextlib import asynccontextmanager
from typing import Dict, Tuple from typing import Dict, Tuple
import factory import factory
@ -163,6 +164,14 @@ async def host_pair_factory(is_secure) -> Tuple[BasicHost, BasicHost]:
return hosts[0], hosts[1] return hosts[0], hosts[1]
@asynccontextmanager
async def pair_of_connected_hosts(is_secure=True):
a, b = await host_pair_factory(is_secure)
yield a, b
close_tasks = (a.close(), b.close())
await asyncio.gather(*close_tasks)
async def swarm_conn_pair_factory( async def swarm_conn_pair_factory(
is_secure: bool, muxer_opt: TMuxerOptions = None is_secure: bool, muxer_opt: TMuxerOptions = None
) -> Tuple[SwarmConn, Swarm, SwarmConn, Swarm]: ) -> Tuple[SwarmConn, Swarm, SwarmConn, Swarm]: