Andros Fenollosa
bfd8b510b6
Clean architecture (core use cases + infra gateways), X25519/AES-GCM E2E encryption, tkinter GUI, and a pytest suite. 45 tests passing.
53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
"""Loopback integration test for the TCP transport."""
|
|
|
|
import queue
|
|
|
|
import pytest
|
|
|
|
from neotalk.infra.transport.tcp_transport import TcpConnection, TcpTransport
|
|
|
|
|
|
@pytest.fixture
|
|
def transport():
|
|
server = TcpTransport(host="127.0.0.1")
|
|
yield server
|
|
server.stop()
|
|
|
|
|
|
def test_frames_travel_between_two_connections(transport):
|
|
# Given a listening transport that echoes received frames onto a queue
|
|
incoming: queue.Queue = queue.Queue()
|
|
|
|
def on_connection(connection: TcpConnection) -> None:
|
|
connection.set_on_frame(incoming.put)
|
|
|
|
port = transport.start(on_connection)
|
|
|
|
# When a client connects and sends a framed dict
|
|
client = transport.connect("127.0.0.1", port)
|
|
try:
|
|
client.send({"type": "secure", "body": "hello"})
|
|
|
|
# Then the server receives exactly that frame
|
|
received = incoming.get(timeout=2.0)
|
|
assert received == {"type": "secure", "body": "hello"}
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
def test_close_callback_fires_when_peer_disconnects(transport):
|
|
# Given a server that records connection closes
|
|
closed: queue.Queue = queue.Queue()
|
|
|
|
def on_connection(connection: TcpConnection) -> None:
|
|
connection.set_on_close(lambda: closed.put(True))
|
|
|
|
port = transport.start(on_connection)
|
|
|
|
# When the client connects then closes
|
|
client = transport.connect("127.0.0.1", port)
|
|
client.send({"type": "secure", "body": "x"})
|
|
client.close()
|
|
|
|
# Then the server side observes the close
|
|
assert closed.get(timeout=2.0) is True
|