"""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)