Andros Fenollosa
907a0eeb55
Add a 'Your port' field in the settings panel: empty keeps local-network mode with an ephemeral port, a value rebinds the TCP listener to a stable address for dialing over a VPN. Active conversations survive the rebind. Adds controller.set_listen_port and a rebindable transport.start(port).
142 lines
4.2 KiB
Python
142 lines
4.2 KiB
Python
"""Test doubles that honour the gateway contracts without real I/O."""
|
|
|
|
from typing import Any
|
|
|
|
|
|
class FakeCrypto:
|
|
"""A trivial, reversible stand-in for :class:`CryptoGateway`.
|
|
|
|
It does not encrypt anything: it simply prefixes the plaintext so tests can
|
|
assert that sealing and opening happened, without pulling in real crypto.
|
|
"""
|
|
|
|
_PREFIX = b"SEALED:"
|
|
|
|
def __init__(self, public_key: bytes = b"local-public-key") -> None:
|
|
self._public_key = public_key
|
|
|
|
def public_key_bytes(self) -> bytes:
|
|
return self._public_key
|
|
|
|
def derive_shared_key(self, peer_public_key: bytes) -> bytes:
|
|
return b"shared:" + peer_public_key
|
|
|
|
def encrypt(self, key: bytes, plaintext: bytes) -> bytes:
|
|
return self._PREFIX + plaintext
|
|
|
|
def decrypt(self, key: bytes, ciphertext: bytes) -> bytes:
|
|
if not ciphertext.startswith(self._PREFIX):
|
|
raise ValueError("not sealed by FakeCrypto")
|
|
return ciphertext[len(self._PREFIX) :]
|
|
|
|
|
|
class ExplodingCrypto:
|
|
"""Crypto double whose ``decrypt`` always fails, for error-path tests."""
|
|
|
|
def public_key_bytes(self) -> bytes:
|
|
return b"x"
|
|
|
|
def derive_shared_key(self, peer_public_key: bytes) -> bytes:
|
|
return b"k"
|
|
|
|
def encrypt(self, key: bytes, plaintext: bytes) -> bytes:
|
|
return plaintext
|
|
|
|
def decrypt(self, key: bytes, ciphertext: bytes) -> bytes:
|
|
raise ValueError("boom")
|
|
|
|
|
|
class FakeConnection:
|
|
"""In-memory :class:`ConnectionGateway` for controller tests."""
|
|
|
|
def __init__(self) -> None:
|
|
self.sent: list[dict[str, Any]] = []
|
|
self.closed = False
|
|
self._on_frame = None
|
|
self._on_close = None
|
|
|
|
def set_on_frame(self, callback) -> None:
|
|
self._on_frame = callback
|
|
|
|
def set_on_close(self, callback) -> None:
|
|
self._on_close = callback
|
|
|
|
def send(self, frame: dict[str, Any]) -> None:
|
|
self.sent.append(frame)
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
if self._on_close is not None:
|
|
self._on_close()
|
|
|
|
def deliver(self, frame: dict[str, Any]) -> None:
|
|
"""Simulate the peer sending us ``frame``."""
|
|
if self._on_frame is not None:
|
|
self._on_frame(frame)
|
|
|
|
|
|
class FakeTransport:
|
|
"""In-memory :class:`TransportGateway` for controller tests."""
|
|
|
|
def __init__(self) -> None:
|
|
self.on_connection = None
|
|
self.outgoing: list[FakeConnection] = []
|
|
self.started = False
|
|
self.port = 55555
|
|
|
|
def start(self, on_connection, port: int | None = None) -> int:
|
|
self.on_connection = on_connection
|
|
self.started = True
|
|
if port:
|
|
self.port = port
|
|
return self.port
|
|
|
|
def stop(self) -> None:
|
|
self.started = False
|
|
|
|
def connect(self, host: str, port: int) -> FakeConnection:
|
|
# Outgoing connections are not routed through on_connection, matching
|
|
# the real transport where only the server side is notified.
|
|
connection = FakeConnection()
|
|
self.outgoing.append(connection)
|
|
return connection
|
|
|
|
def accept(self, connection: FakeConnection) -> None:
|
|
"""Simulate an incoming connection from a peer."""
|
|
self.on_connection(connection)
|
|
|
|
|
|
class FakeNotifier:
|
|
def __init__(self) -> None:
|
|
self.notifications: list[tuple[str, str]] = []
|
|
|
|
def notify(self, title: str, body: str) -> None:
|
|
self.notifications.append((title, body))
|
|
|
|
|
|
class FixedClock:
|
|
def __init__(self, value: float = 0.0) -> None:
|
|
self.value = value
|
|
|
|
def now(self) -> float:
|
|
return self.value
|
|
|
|
|
|
class RecordingDiscovery:
|
|
"""Records datagrams instead of sending them over the network."""
|
|
|
|
def __init__(self) -> None:
|
|
self.broadcasts: list[dict[str, Any]] = []
|
|
self.directed: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def start(self, on_datagram) -> None: # pragma: no cover - trivial
|
|
self._on_datagram = on_datagram
|
|
|
|
def stop(self) -> None: # pragma: no cover - trivial
|
|
pass
|
|
|
|
def broadcast(self, datagram: dict[str, Any]) -> None:
|
|
self.broadcasts.append(datagram)
|
|
|
|
def send_to(self, host: str, datagram: dict[str, Any]) -> None:
|
|
self.directed.append((host, datagram))
|