Andros Fenollosa
e4c4d0d619
Add a 'Discover on (IP or subnet)' setting (and --discover flag) that announces presence by unicast to the given addresses, since multicast cannot cross a VPN tunnel. Each newly seen peer is answered directly, so one side configuring the subnet is enough for both to appear. Documented in the README.
192 lines
6.5 KiB
Python
192 lines
6.5 KiB
Python
"""Integration tests for the controller wiring, using in-memory doubles."""
|
|
|
|
import base64
|
|
|
|
import pytest
|
|
|
|
from neotalk.core.use_cases.conversation.compose_message import (
|
|
compose_message_use_case,
|
|
)
|
|
from neotalk.infra.app_controller import AppController
|
|
from tests.fakes import (
|
|
FakeConnection,
|
|
FakeCrypto,
|
|
FakeNotifier,
|
|
FakeTransport,
|
|
FixedClock,
|
|
RecordingDiscovery,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def wiring():
|
|
crypto = FakeCrypto()
|
|
discovery = RecordingDiscovery()
|
|
transport = FakeTransport()
|
|
notifier = FakeNotifier()
|
|
clock = FixedClock(100.0)
|
|
controller = AppController(
|
|
crypto, discovery, transport, notifier, clock, alias="ana"
|
|
)
|
|
events: list[dict] = []
|
|
controller.set_listener(events.append)
|
|
controller.start()
|
|
return controller, discovery, transport, notifier, events
|
|
|
|
|
|
def _peer_handshake(peer_id="bob-id", alias="bob", pubkey=b"bob-key"):
|
|
return {
|
|
"type": "handshake",
|
|
"peer_id": peer_id,
|
|
"alias": alias,
|
|
"pubkey": base64.b64encode(pubkey).decode("ascii"),
|
|
}
|
|
|
|
|
|
def test_start_broadcasts_presence(wiring):
|
|
# Given a started controller
|
|
_controller, discovery, _t, _n, _events = wiring
|
|
|
|
# Then it announces itself over discovery
|
|
assert any(d["type"] == "hello" for d in discovery.broadcasts)
|
|
|
|
|
|
def test_hello_datagram_adds_a_peer_and_emits_roster(wiring, hello_datagram):
|
|
# Given a started controller
|
|
controller, _d, _t, _n, events = wiring
|
|
|
|
# When a peer's hello arrives
|
|
controller._on_datagram(hello_datagram, "192.168.0.5")
|
|
|
|
# Then the roster grows and a roster event is emitted
|
|
assert len(controller.roster_snapshot()) == 1
|
|
assert any(e["kind"] == "roster" for e in events)
|
|
|
|
|
|
def test_ignores_our_own_datagrams(wiring):
|
|
# Given our own peer id
|
|
controller, _d, _t, _n, _events = wiring
|
|
own = {"type": "hello", "peer_id": controller.identity.peer_id}
|
|
|
|
# When it loops back to us
|
|
controller._on_datagram(own, "127.0.0.1")
|
|
|
|
# Then it is ignored
|
|
assert controller.roster_snapshot() == []
|
|
|
|
|
|
def test_incoming_conversation_completes_handshake_and_delivers_message(wiring):
|
|
# Given a started controller and an incoming connection
|
|
controller, _d, transport, notifier, events = wiring
|
|
connection = FakeConnection()
|
|
transport.accept(connection)
|
|
|
|
# Then we immediately send our handshake
|
|
assert connection.sent[0]["type"] == "handshake"
|
|
|
|
# When the peer completes the handshake
|
|
connection.deliver(_peer_handshake())
|
|
assert any(e["kind"] == "conversation_ready" for e in events)
|
|
|
|
# And sends an encrypted message (sealed with the shared key we derived)
|
|
shared_key = FakeCrypto().derive_shared_key(b"bob-key")
|
|
frame = compose_message_use_case("hola", FakeCrypto(), shared_key)["data"]["frame"]
|
|
connection.deliver(frame)
|
|
|
|
# Then a message event is emitted for that peer
|
|
message_events = [e for e in events if e["kind"] == "message"]
|
|
assert message_events[-1]["message"].text == "hola"
|
|
assert message_events[-1]["message"].outgoing is False
|
|
# And a notification candidate is raised
|
|
assert any(e["kind"] == "notify_candidate" for e in events)
|
|
|
|
|
|
def test_send_message_after_handshake_writes_a_secure_frame(wiring):
|
|
# Given an established incoming conversation
|
|
controller, _d, transport, _n, events = wiring
|
|
connection = FakeConnection()
|
|
transport.accept(connection)
|
|
connection.deliver(_peer_handshake())
|
|
|
|
# When we send a message to that peer
|
|
response = controller.send_message("bob-id", "buenas")
|
|
|
|
# Then a secure frame is written and an outgoing message is emitted
|
|
assert response["type"] == "Success"
|
|
assert connection.sent[-1]["type"] == "secure"
|
|
outgoing = [e for e in events if e["kind"] == "message" and e["message"].outgoing]
|
|
assert outgoing[-1]["message"].text == "buenas"
|
|
|
|
|
|
def test_open_conversation_rings_the_doorbell_over_udp(wiring, hello_datagram):
|
|
# Given a known peer in the roster
|
|
controller, discovery, _t, _n, _events = wiring
|
|
controller._on_datagram(hello_datagram, "192.168.0.5")
|
|
|
|
# When we open a conversation with them
|
|
controller.open_conversation("remote-id")
|
|
|
|
# Then a ring datagram is broadcast, addressed to that peer
|
|
rings = [d for d in discovery.broadcasts if d["type"] == "ring"]
|
|
assert rings
|
|
assert rings[-1]["to"] == "remote-id"
|
|
|
|
|
|
def test_set_listen_port_rebinds_and_updates_identity(wiring):
|
|
# Given a started controller
|
|
controller, _d, transport, _n, events = wiring
|
|
|
|
# When we choose a fixed port from the settings
|
|
response = controller.set_listen_port(51821)
|
|
|
|
# Then the transport rebinds and the advertised identity reflects it
|
|
assert response["type"] == "Success"
|
|
assert transport.port == 51821
|
|
assert controller.identity.tcp_port == 51821
|
|
assert any(e["kind"] == "identity" for e in events)
|
|
|
|
|
|
def test_set_discovery_targets_announces_by_unicast(wiring):
|
|
# Given a started controller
|
|
controller, discovery, _t, _n, _events = wiring
|
|
discovery.directed.clear()
|
|
|
|
# When we set a small VPN subnet as the discovery target
|
|
response = controller.set_discovery_targets("192.168.0.0/30")
|
|
|
|
# Then it announces our presence by unicast to each host
|
|
assert response["type"] == "Success"
|
|
hosts = [host for host, _dg in discovery.directed]
|
|
assert "192.168.0.1" in hosts
|
|
assert "192.168.0.2" in hosts
|
|
assert all(dg["type"] == "hello" for _h, dg in discovery.directed)
|
|
|
|
|
|
def test_new_peer_is_answered_directly(wiring, hello_datagram):
|
|
# Given a started controller with no targets
|
|
controller, discovery, _t, _n, _events = wiring
|
|
discovery.directed.clear()
|
|
|
|
# When a peer's hello arrives (as over a VPN unicast)
|
|
controller._on_datagram(hello_datagram, "10.0.200.151")
|
|
|
|
# Then we answer that peer directly, so discovery becomes mutual
|
|
hosts = [host for host, _dg in discovery.directed]
|
|
assert "10.0.200.151" in hosts
|
|
|
|
|
|
def test_connect_to_address_dials_directly_and_completes(wiring):
|
|
# Given a started controller (no peer in the roster, as over a VPN)
|
|
controller, _d, transport, _n, events = wiring
|
|
|
|
# When we dial an address directly
|
|
controller.connect_to_address("10.8.0.2", 51820)
|
|
|
|
# Then it opens an outgoing connection and sends our handshake
|
|
assert len(transport.outgoing) == 1
|
|
connection = transport.outgoing[0]
|
|
assert connection.sent[0]["type"] == "handshake"
|
|
|
|
# And when the peer answers, the conversation is ready
|
|
connection.deliver(_peer_handshake())
|
|
assert any(e["kind"] == "conversation_ready" for e in events)
|