py-libp2p/tests/security/test_security_multistream.py
mhchia c72dfe1dd3
Use factories and fixtures in pubsub tests
Done
- Add factories using factory-boy
- Modify fixtures and tests to use factories
- Modify tests to use fixtures and factories
- Clean up
2019-08-01 13:30:08 +08:00

181 lines
5.6 KiB
Python

import asyncio
import multiaddr
import pytest
from libp2p import new_node
from libp2p.peer.peerinfo import info_from_p2p_addr
from libp2p.protocol_muxer.multiselect_client import MultiselectClientError
from libp2p.security.insecure_security import InsecureTransport
from libp2p.security.simple_security import SimpleSecurityTransport
from tests.utils import cleanup, connect
# TODO: Add tests for multiple streams being opened on different
# protocols through the same connection
def peer_id_for_node(node):
addr = node.get_addrs()[0]
info = info_from_p2p_addr(addr)
return info.peer_id
async def perform_simple_test(
assertion_func, transports_for_initiator, transports_for_noninitiator
):
# Create libp2p nodes and connect them, then secure the connection, then check
# the proper security was chosen
# TODO: implement -- note we need to introduce the notion of communicating over a raw connection
# for testing, we do NOT want to communicate over a stream so we can't just create two nodes
# and use their conn because our mplex will internally relay messages to a stream
sec_opt1 = transports_for_initiator
sec_opt2 = transports_for_noninitiator
node1 = await new_node(transport_opt=["/ip4/127.0.0.1/tcp/0"], sec_opt=sec_opt1)
node2 = await new_node(transport_opt=["/ip4/127.0.0.1/tcp/0"], sec_opt=sec_opt2)
await node1.get_network().listen(multiaddr.Multiaddr("/ip4/127.0.0.1/tcp/0"))
await node2.get_network().listen(multiaddr.Multiaddr("/ip4/127.0.0.1/tcp/0"))
await connect(node1, node2)
# Wait a very short period to allow conns to be stored (since the functions
# storing the conns are async, they may happen at slightly different times
# on each node)
await asyncio.sleep(0.1)
# Get conns
node1_conn = node1.get_network().connections[peer_id_for_node(node2)]
node2_conn = node2.get_network().connections[peer_id_for_node(node1)]
# Perform assertion
assertion_func(node1_conn.secured_conn.get_security_details())
assertion_func(node2_conn.secured_conn.get_security_details())
# Success, terminate pending tasks.
await cleanup()
@pytest.mark.asyncio
async def test_single_insecure_security_transport_succeeds():
transports_for_initiator = {"foo": InsecureTransport("foo")}
transports_for_noninitiator = {"foo": InsecureTransport("foo")}
def assertion_func(details):
assert details["id"] == "foo"
await perform_simple_test(
assertion_func, transports_for_initiator, transports_for_noninitiator
)
@pytest.mark.asyncio
async def test_single_simple_test_security_transport_succeeds():
transports_for_initiator = {"tacos": SimpleSecurityTransport("tacos")}
transports_for_noninitiator = {"tacos": SimpleSecurityTransport("tacos")}
def assertion_func(details):
assert details["key_phrase"] == "tacos"
await perform_simple_test(
assertion_func, transports_for_initiator, transports_for_noninitiator
)
@pytest.mark.asyncio
async def test_two_simple_test_security_transport_for_initiator_succeeds():
transports_for_initiator = {
"tacos": SimpleSecurityTransport("tacos"),
"shleep": SimpleSecurityTransport("shleep"),
}
transports_for_noninitiator = {"shleep": SimpleSecurityTransport("shleep")}
def assertion_func(details):
assert details["key_phrase"] == "shleep"
await perform_simple_test(
assertion_func, transports_for_initiator, transports_for_noninitiator
)
@pytest.mark.asyncio
async def test_two_simple_test_security_transport_for_noninitiator_succeeds():
transports_for_initiator = {"tacos": SimpleSecurityTransport("tacos")}
transports_for_noninitiator = {
"shleep": SimpleSecurityTransport("shleep"),
"tacos": SimpleSecurityTransport("tacos"),
}
def assertion_func(details):
assert details["key_phrase"] == "tacos"
await perform_simple_test(
assertion_func, transports_for_initiator, transports_for_noninitiator
)
@pytest.mark.asyncio
async def test_two_simple_test_security_transport_for_both_succeeds():
transports_for_initiator = {
"a": SimpleSecurityTransport("a"),
"b": SimpleSecurityTransport("b"),
}
transports_for_noninitiator = {
"c": SimpleSecurityTransport("c"),
"b": SimpleSecurityTransport("b"),
}
def assertion_func(details):
assert details["key_phrase"] == "b"
await perform_simple_test(
assertion_func, transports_for_initiator, transports_for_noninitiator
)
@pytest.mark.asyncio
async def test_multiple_security_none_the_same_fails():
transports_for_initiator = {
"a": SimpleSecurityTransport("a"),
"b": SimpleSecurityTransport("b"),
}
transports_for_noninitiator = {
"c": SimpleSecurityTransport("c"),
"d": SimpleSecurityTransport("d"),
}
def assertion_func(_):
assert False
with pytest.raises(MultiselectClientError):
await perform_simple_test(
assertion_func, transports_for_initiator, transports_for_noninitiator
)
await cleanup()
@pytest.mark.asyncio
async def test_default_insecure_security():
transports_for_initiator = None
transports_for_noninitiator = None
details1 = None
details2 = None
def assertion_func(details):
nonlocal details1
nonlocal details2
if not details1:
details1 = details
elif not details2:
details2 = details
else:
assert details1 == details2
await perform_simple_test(
assertion_func, transports_for_initiator, transports_for_noninitiator
)