1074 lines
37 KiB
Python
1074 lines
37 KiB
Python
"""Tests for US-017: P2P gossip, relay node, and TLS infrastructure."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# identity tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_load_or_create_identity_generates_peer_id(tmp_path):
|
|
"Load or create identity generates peer id\n\nTags: gossip, network, relay"
|
|
from meshnet_p2p.identity import load_or_create_identity
|
|
|
|
identity = load_or_create_identity(tmp_path / "identity.json")
|
|
assert len(identity["peer_id"]) == 16
|
|
assert "public_key_pem" in identity
|
|
assert "private_key_pem" in identity
|
|
|
|
|
|
def test_identity_is_stable_across_loads(tmp_path):
|
|
"Identity is stable across loads\n\nTags: gossip, network, relay"
|
|
from meshnet_p2p.identity import load_or_create_identity
|
|
|
|
path = tmp_path / "identity.json"
|
|
first = load_or_create_identity(path)
|
|
second = load_or_create_identity(path)
|
|
assert first["peer_id"] == second["peer_id"]
|
|
assert first["public_key_pem"] == second["public_key_pem"]
|
|
|
|
|
|
def test_identity_different_for_different_paths(tmp_path):
|
|
"Identity different for different paths\n\nTags: gossip, network, relay"
|
|
from meshnet_p2p.identity import load_or_create_identity
|
|
|
|
a = load_or_create_identity(tmp_path / "a.json")
|
|
b = load_or_create_identity(tmp_path / "b.json")
|
|
# Extremely unlikely to collide
|
|
assert a["peer_id"] != b["peer_id"]
|
|
|
|
|
|
def test_relay_peer_id_includes_node_name_for_shared_wallet():
|
|
"Relay peer id includes node name for shared wallet\n\nTags: gossip, network, relay, security, wallet"
|
|
from meshnet_node.relay_bridge import peer_id_from_wallet
|
|
|
|
wallet = "5gMLrmyBYTpkFjmyc4eGwcaWhYquyWgCBFFEqHzR5Qur"
|
|
|
|
assert peer_id_from_wallet(wallet, node_name="gpu head") == "5gMLrmyBYTpkFjmy-gpu-head"
|
|
assert peer_id_from_wallet(wallet, node_name="cpu-tail") == "5gMLrmyBYTpkFjmy-cpu-tail"
|
|
|
|
|
|
def test_relay_peer_id_falls_back_to_endpoint_port_integer():
|
|
"Relay peer id falls back to endpoint port integer\n\nTags: gossip, network, relay"
|
|
from meshnet_node.relay_bridge import peer_id_from_wallet
|
|
|
|
wallet = "5gMLrmyBYTpkFjmyc4eGwcaWhYquyWgCBFFEqHzR5Qur"
|
|
|
|
first = peer_id_from_wallet(wallet, advertised_addr="http://192.168.0.20:7001")
|
|
second = peer_id_from_wallet(wallet, advertised_addr="http://192.168.0.20:7002")
|
|
|
|
assert first == "5gMLrmyBYTpkFjmy-7001"
|
|
assert second == "5gMLrmyBYTpkFjmy-7002"
|
|
assert first != second
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TLS / certificate tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_generate_self_signed_cert_creates_files(tmp_path):
|
|
"Generate self signed cert creates files\n\nTags: gossip, network, relay"
|
|
from meshnet_p2p.tls import generate_self_signed_cert
|
|
|
|
cert_p, key_p = generate_self_signed_cert(
|
|
cert_path=tmp_path / "cert.pem",
|
|
key_path=tmp_path / "key.pem",
|
|
common_name="localhost",
|
|
)
|
|
assert cert_p.exists()
|
|
assert key_p.exists()
|
|
assert cert_p.stat().st_size > 100
|
|
assert key_p.stat().st_size > 100
|
|
|
|
|
|
def test_generate_self_signed_cert_is_idempotent(tmp_path):
|
|
"Generate self signed cert is idempotent\n\nTags: gossip, network, relay"
|
|
from meshnet_p2p.tls import generate_self_signed_cert
|
|
|
|
args = dict(cert_path=tmp_path / "cert.pem", key_path=tmp_path / "key.pem", common_name="test")
|
|
generate_self_signed_cert(**args)
|
|
mtime1 = (tmp_path / "cert.pem").stat().st_mtime
|
|
|
|
generate_self_signed_cert(**args)
|
|
mtime2 = (tmp_path / "cert.pem").stat().st_mtime
|
|
assert mtime1 == mtime2 # file not regenerated
|
|
|
|
|
|
def test_cert_fingerprint_returns_sha256_prefix(tmp_path):
|
|
"Cert fingerprint returns sha256 prefix\n\nTags: gossip, network, relay"
|
|
from meshnet_p2p.tls import generate_self_signed_cert, cert_fingerprint
|
|
|
|
cert_p, key_p = generate_self_signed_cert(
|
|
cert_path=tmp_path / "cert.pem",
|
|
key_path=tmp_path / "key.pem",
|
|
common_name="test",
|
|
)
|
|
fp = cert_fingerprint(cert_p)
|
|
assert fp.startswith("sha256:")
|
|
assert len(fp) == len("sha256:") + 64 # 32 bytes hex = 64 chars
|
|
|
|
|
|
def test_make_server_ssl_context_loads_cert(tmp_path):
|
|
"Make server ssl context loads cert\n\nTags: gossip, network, relay"
|
|
import ssl
|
|
from meshnet_p2p.tls import generate_self_signed_cert, make_server_ssl_context
|
|
|
|
cert_p, key_p = generate_self_signed_cert(
|
|
cert_path=tmp_path / "cert.pem",
|
|
key_path=tmp_path / "key.pem",
|
|
common_name="test",
|
|
)
|
|
ctx = make_server_ssl_context(cert_p, key_p)
|
|
assert isinstance(ctx, ssl.SSLContext)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PeerRegistry tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_peer_registry_register_and_list():
|
|
"Peer registry register and list\n\nTags: gossip, network, relay"
|
|
from meshnet_relay.peer_registry import PeerRegistry
|
|
|
|
reg = PeerRegistry()
|
|
ws_mock = MagicMock()
|
|
reg.register("peer1", "http://1.2.3.4:8001", ws_mock)
|
|
|
|
assert len(reg) == 1
|
|
peers = reg.list_peers()
|
|
assert len(peers) == 1
|
|
assert peers[0]["peer_id"] == "peer1"
|
|
|
|
|
|
def test_peer_registry_all_except_excludes_sender():
|
|
"Peer registry all except excludes sender\n\nTags: gossip, network, relay"
|
|
from meshnet_relay.peer_registry import PeerRegistry
|
|
|
|
reg = PeerRegistry()
|
|
reg.register("a", "http://a:8001", MagicMock())
|
|
reg.register("b", "http://b:8001", MagicMock())
|
|
reg.register("c", "http://c:8001", MagicMock())
|
|
|
|
others = reg.all_except("a")
|
|
assert len(others) == 2
|
|
assert all(e.peer_id != "a" for e in others)
|
|
|
|
|
|
def test_peer_registry_unregister_removes_peer():
|
|
"Peer registry unregister removes peer\n\nTags: gossip, network, relay"
|
|
from meshnet_relay.peer_registry import PeerRegistry
|
|
|
|
reg = PeerRegistry()
|
|
reg.register("x", "http://x:8001", MagicMock())
|
|
reg.unregister("x")
|
|
assert len(reg) == 0
|
|
assert reg.get("x") is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GossipClient + RelayServer integration test
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _start_relay(host="127.0.0.1", port=0):
|
|
from meshnet_relay.server import RelayServer
|
|
server = RelayServer(host=host, port=port)
|
|
actual_port = server.start()
|
|
return server, actual_port
|
|
|
|
|
|
def test_gossip_fanout_through_relay():
|
|
"Node B publishes node-join; node A receives it within 2 seconds.\n\nTags: gossip, network, relay"
|
|
|
|
from meshnet_p2p.gossip import GossipClient, TOPIC_NODE_JOIN
|
|
|
|
relay, port = _start_relay()
|
|
relay_url = f"ws://127.0.0.1:{port}/ws"
|
|
|
|
client_a = GossipClient(relay_url=relay_url, peer_id="peer_a")
|
|
client_b = GossipClient(relay_url=relay_url, peer_id="peer_b")
|
|
|
|
received = []
|
|
client_a.subscribe(TOPIC_NODE_JOIN, lambda env: received.append(env))
|
|
|
|
client_a.start()
|
|
client_b.start()
|
|
|
|
assert client_a.wait_connected(timeout=5), "client_a failed to connect to relay"
|
|
assert client_b.wait_connected(timeout=5), "client_b failed to connect to relay"
|
|
|
|
# Give both peers time to register with relay
|
|
time.sleep(0.2)
|
|
|
|
client_b.publish(TOPIC_NODE_JOIN, {"addr": "http://192.168.1.10:8001", "peer_id": "peer_b"})
|
|
|
|
deadline = time.monotonic() + 2.0
|
|
while time.monotonic() < deadline and not received:
|
|
time.sleep(0.05)
|
|
|
|
client_a.stop()
|
|
client_b.stop()
|
|
relay.stop()
|
|
|
|
assert received, "client_a did not receive node-join message from client_b"
|
|
assert received[0]["topic"] == TOPIC_NODE_JOIN
|
|
assert received[0]["from_peer"] == "peer_b"
|
|
|
|
|
|
def test_gossip_dedup_prevents_processing_duplicate_message_ids():
|
|
"A message with a duplicate msg_id is only processed once.\n\nTags: gossip, network, relay"
|
|
|
|
from meshnet_p2p.gossip import GossipClient, TOPIC_NODE_JOIN
|
|
|
|
relay, port = _start_relay()
|
|
relay_url = f"ws://127.0.0.1:{port}/ws"
|
|
|
|
client_a = GossipClient(relay_url=relay_url, peer_id="peer_a2")
|
|
client_b = GossipClient(relay_url=relay_url, peer_id="peer_b2")
|
|
|
|
received = []
|
|
client_a.subscribe(TOPIC_NODE_JOIN, lambda env: received.append(env))
|
|
|
|
client_a.start()
|
|
client_b.start()
|
|
client_a.wait_connected(5)
|
|
client_b.wait_connected(5)
|
|
time.sleep(0.2)
|
|
|
|
# Publish once
|
|
client_b.publish(TOPIC_NODE_JOIN, {"test": "dedup"})
|
|
time.sleep(0.5)
|
|
|
|
count = len(received)
|
|
|
|
client_a.stop()
|
|
client_b.stop()
|
|
relay.stop()
|
|
|
|
# Should have received exactly one message (not duplicated)
|
|
assert count <= 1
|
|
|
|
|
|
def test_relay_server_peer_list_grows_on_connect():
|
|
"Relay registry grows when clients connect.\n\nTags: gossip, network, relay"
|
|
|
|
from meshnet_p2p.gossip import GossipClient
|
|
|
|
relay, port = _start_relay()
|
|
relay_url = f"ws://127.0.0.1:{port}/ws"
|
|
|
|
client = GossipClient(relay_url=relay_url, peer_id="solo_peer")
|
|
client.start()
|
|
client.wait_connected(5)
|
|
time.sleep(0.3) # let peer-register message process
|
|
|
|
peer_count = len(relay.registry)
|
|
client.stop()
|
|
relay.stop()
|
|
|
|
assert peer_count >= 1
|
|
|
|
|
|
def test_relay_circuit_relay_proxies_message():
|
|
"A node behind NAT (client_a) receives a message via circuit relay from client_b.\n\nTags: gossip, network, relay"
|
|
|
|
import websockets.sync.client # type: ignore[import]
|
|
from meshnet_relay.server import RelayServer
|
|
|
|
relay = RelayServer(host="127.0.0.1", port=0)
|
|
port = relay.start()
|
|
|
|
# client_a connects to gossip hub and registers
|
|
received_via_relay = []
|
|
ready = threading.Event()
|
|
|
|
def run_nat_client():
|
|
import websockets.sync.client as wsc
|
|
with wsc.connect(f"ws://127.0.0.1:{port}/ws") as ws:
|
|
ws.send(json.dumps({
|
|
"topic": "peer-register",
|
|
"version": 1,
|
|
"from_peer": "nat_peer",
|
|
"msg_id": "reg-001",
|
|
"timestamp": "2026-06-29T00:00:00Z",
|
|
"ttl": 3,
|
|
"payload": {"peer_id": "nat_peer", "addr": ""},
|
|
}))
|
|
# consume peer-list response
|
|
ws.recv()
|
|
ready.set()
|
|
# wait for a relayed message
|
|
try:
|
|
msg = ws.recv(timeout=3)
|
|
received_via_relay.append(json.loads(msg))
|
|
except Exception:
|
|
pass
|
|
|
|
nat_thread = threading.Thread(target=run_nat_client, daemon=True)
|
|
nat_thread.start()
|
|
ready.wait(timeout=5)
|
|
time.sleep(0.1) # ensure peer is in registry
|
|
|
|
# client_b connects via circuit relay path
|
|
def send_via_relay():
|
|
try:
|
|
import websockets.sync.client as wsc
|
|
with wsc.connect(f"ws://127.0.0.1:{port}/relay/nat_peer") as ws:
|
|
ws.send(json.dumps({"topic": "direct-relay-test", "payload": {"hi": "there"}}))
|
|
time.sleep(0.3)
|
|
except Exception:
|
|
pass
|
|
|
|
relay_thread = threading.Thread(target=send_via_relay, daemon=True)
|
|
relay_thread.start()
|
|
relay_thread.join(timeout=3)
|
|
nat_thread.join(timeout=3)
|
|
relay.stop()
|
|
|
|
assert received_via_relay, "NAT'd peer did not receive message via circuit relay"
|
|
|
|
|
|
def test_relay_rpc_round_trips_http_request_to_peer():
|
|
"Relay /rpc/<peer> sends one HTTP-shaped request to a connected peer.\n\nTags: gossip, network, relay"
|
|
|
|
import websockets.sync.client as wsc # type: ignore[import]
|
|
from meshnet_relay.server import RelayServer
|
|
|
|
relay = RelayServer(host="127.0.0.1", port=0)
|
|
port = relay.start()
|
|
ready = threading.Event()
|
|
|
|
def run_peer():
|
|
with wsc.connect(f"ws://127.0.0.1:{port}/ws") as ws:
|
|
ws.send(json.dumps({
|
|
"topic": "peer-register",
|
|
"version": 1,
|
|
"from_peer": "rpc_peer",
|
|
"msg_id": "rpc-reg-001",
|
|
"payload": {"peer_id": "rpc_peer", "addr": ""},
|
|
}))
|
|
ws.recv()
|
|
ready.set()
|
|
envelope = json.loads(ws.recv(timeout=3))
|
|
payload = envelope["payload"]
|
|
ws.send(json.dumps({
|
|
"topic": "relay-http-response",
|
|
"version": 1,
|
|
"from_peer": "rpc_peer",
|
|
"payload": {
|
|
"request_id": payload["request_id"],
|
|
"status": 200,
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": json.dumps({"ok": True, "path": payload["path"]}),
|
|
},
|
|
}))
|
|
|
|
peer_thread = threading.Thread(target=run_peer, daemon=True)
|
|
peer_thread.start()
|
|
assert ready.wait(timeout=5)
|
|
|
|
with wsc.connect(f"ws://127.0.0.1:{port}/rpc/rpc_peer") as ws:
|
|
ws.send(json.dumps({
|
|
"request_id": "req-1",
|
|
"method": "POST",
|
|
"path": "/v1/chat/completions",
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": "{}",
|
|
}))
|
|
response = json.loads(ws.recv(timeout=5))
|
|
|
|
relay.stop()
|
|
peer_thread.join(timeout=3)
|
|
|
|
assert response["status"] == 200
|
|
assert json.loads(response["body"]) == {"ok": True, "path": "/v1/chat/completions"}
|
|
|
|
|
|
def test_relay_rpc_reuses_connection_for_sequential_requests(monkeypatch):
|
|
"One route session should not repeat the WebSocket handshake per token.\n\nTags: gossip, network, relay"
|
|
|
|
import websockets.sync.client as wsc # type: ignore[import]
|
|
from meshnet_node.relay_bridge import decode_binary_frame, encode_binary_frame
|
|
from meshnet_node.torch_server import _RelayHopClient
|
|
from meshnet_relay.server import RelayServer
|
|
|
|
relay = RelayServer(host="127.0.0.1", port=0)
|
|
port = relay.start()
|
|
ready = threading.Event()
|
|
|
|
def run_peer():
|
|
with wsc.connect(f"ws://127.0.0.1:{port}/ws") as ws:
|
|
ws.send(json.dumps({
|
|
"topic": "peer-register",
|
|
"version": 1,
|
|
"from_peer": "persistent_peer",
|
|
"msg_id": "persistent-reg-001",
|
|
"payload": {"peer_id": "persistent_peer", "addr": ""},
|
|
}))
|
|
ws.recv()
|
|
ready.set()
|
|
for _ in range(2):
|
|
raw = ws.recv(timeout=3)
|
|
header, body = decode_binary_frame(raw)
|
|
ws.send(encode_binary_frame({
|
|
"request_id": header["request_id"],
|
|
"status": 200,
|
|
"headers": {"Content-Type": "application/octet-stream"},
|
|
}, body))
|
|
|
|
peer_thread = threading.Thread(target=run_peer, daemon=True)
|
|
peer_thread.start()
|
|
assert ready.wait(timeout=5)
|
|
|
|
real_connect = wsc.connect
|
|
connection_attempts = 0
|
|
|
|
def tracked_connect(*args, **kwargs):
|
|
nonlocal connection_attempts
|
|
connection_attempts += 1
|
|
return real_connect(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(wsc, "connect", tracked_connect)
|
|
client = _RelayHopClient(f"ws://127.0.0.1:{port}/rpc/persistent_peer", timeout=5)
|
|
try:
|
|
responses = [
|
|
client.request(
|
|
"/forward",
|
|
f"token-{index}".encode(),
|
|
{"X-Meshnet-Session": "route-session"},
|
|
)
|
|
for index in range(2)
|
|
]
|
|
finally:
|
|
client.close()
|
|
relay.stop()
|
|
peer_thread.join(timeout=3)
|
|
|
|
assert [status for status, _, _ in responses] == [200, 200]
|
|
assert [body for _, _, body in responses] == [b"token-0", b"token-1"]
|
|
assert connection_attempts == 1
|
|
|
|
|
|
def test_binary_relay_frame_codecs_interoperate():
|
|
"Node and relay ship the same binary frame format as separate copies.\n\nTags: gossip, network, relay"
|
|
|
|
import os
|
|
|
|
from meshnet_node import relay_bridge
|
|
from meshnet_relay import server as relay_server
|
|
|
|
header = {"request_id": "r-1", "status": 200, "headers": {"X": "y"}}
|
|
body = os.urandom(4096)
|
|
for encoder, decoder in (
|
|
(relay_bridge.encode_binary_frame, relay_server.decode_binary_frame),
|
|
(relay_server.encode_binary_frame, relay_bridge.decode_binary_frame),
|
|
):
|
|
assert decoder(encoder(header, body)) == (header, body)
|
|
|
|
try:
|
|
relay_server.decode_binary_frame(b"not a frame")
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError("garbage bytes must not decode as a binary frame")
|
|
|
|
|
|
def test_activation_compression_round_trips_and_skips_small_bodies():
|
|
"Pipeline hops zstd-compress large activations; tiny decode bodies pass raw.\n\nTags: gossip, network, relay"
|
|
|
|
import os
|
|
|
|
from meshnet_node.server import _decompress_body
|
|
from meshnet_node.torch_server import _maybe_compress_activation
|
|
|
|
small = os.urandom(1024)
|
|
assert _maybe_compress_activation(small) == (small, None)
|
|
|
|
large = os.urandom(96 * 1024) * 2 # repetition so zstd actually shrinks it
|
|
wire, encoding = _maybe_compress_activation(large)
|
|
assert encoding == "zstd"
|
|
assert len(wire) < len(large)
|
|
assert _decompress_body(wire, encoding) == large
|
|
|
|
|
|
def test_relay_rpc_carries_activation_sized_frames():
|
|
"A >1 MiB activation body must survive the full relay round trip.\n\nTags: gossip, network, relay"
|
|
|
|
import http.server
|
|
import os
|
|
|
|
from meshnet_node.relay_bridge import RelayHttpBridge
|
|
from meshnet_node.torch_server import _relay_hop
|
|
from meshnet_relay.server import RelayServer
|
|
|
|
body = os.urandom(2 * 1024 * 1024) # 2 MiB raw, ~2.7 MiB once base64-wrapped
|
|
|
|
class EchoHandler(http.server.BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
data = self.rfile.read(int(self.headers.get("Content-Length", 0)))
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/octet-stream")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
local = http.server.HTTPServer(("127.0.0.1", 0), EchoHandler)
|
|
local_thread = threading.Thread(target=local.serve_forever, daemon=True)
|
|
local_thread.start()
|
|
|
|
relay = RelayServer(host="127.0.0.1", port=0)
|
|
port = relay.start()
|
|
|
|
bridge = RelayHttpBridge(
|
|
relay_url=f"ws://127.0.0.1:{port}/ws",
|
|
peer_id="big_peer",
|
|
local_base_url=f"http://127.0.0.1:{local.server_address[1]}",
|
|
advertised_addr="",
|
|
)
|
|
try:
|
|
bridge.start()
|
|
assert bridge.wait_connected(timeout=5.0)
|
|
time.sleep(0.2) # let peer-register land in the relay registry
|
|
status, headers, resp_body = _relay_hop(
|
|
f"ws://127.0.0.1:{port}/rpc/big_peer",
|
|
"/forward",
|
|
body,
|
|
{"Content-Type": "application/octet-stream"},
|
|
timeout=30.0,
|
|
)
|
|
finally:
|
|
bridge.stop()
|
|
relay.stop()
|
|
local.shutdown()
|
|
local_thread.join(timeout=3)
|
|
|
|
assert status == 200
|
|
assert resp_body == body
|
|
|
|
|
|
def test_node_relay_bridge_reconnects_after_failed_connection(monkeypatch):
|
|
"Node-side relay bridge keeps retrying its outbound WebSocket connection.\n\nTags: gossip, network, relay"
|
|
|
|
import websockets.sync.client as wsc # type: ignore[import]
|
|
from meshnet_node.relay_bridge import RelayHttpBridge
|
|
|
|
attempts = []
|
|
|
|
class FakeWebSocket:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def send(self, _message):
|
|
pass
|
|
|
|
def recv(self, timeout=1):
|
|
raise TimeoutError
|
|
|
|
def fake_connect(url, open_timeout=5, **kwargs):
|
|
attempts.append((url, open_timeout))
|
|
if len(attempts) == 1:
|
|
raise OSError("temporary relay outage")
|
|
return FakeWebSocket()
|
|
|
|
monkeypatch.setattr(wsc, "connect", fake_connect)
|
|
|
|
bridge = RelayHttpBridge(
|
|
relay_url="ws://relay.example/ws",
|
|
peer_id="peer-reconnect",
|
|
local_base_url="http://127.0.0.1:8001",
|
|
advertised_addr="http://127.0.0.1:8001",
|
|
reconnect_interval=0.01,
|
|
)
|
|
try:
|
|
info = bridge.start()
|
|
assert info.relay_addr == "ws://relay.example/rpc/peer-reconnect"
|
|
assert bridge.wait_connected(timeout=1.0)
|
|
finally:
|
|
bridge.stop()
|
|
|
|
assert len(attempts) >= 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tracker gossip fields tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _start_tracker_and_register(extra_fields: dict) -> dict:
|
|
"""Helper: start tracker, register node with extra gossip fields, return response."""
|
|
import http.server
|
|
import json as _json
|
|
import urllib.request
|
|
|
|
from meshnet_tracker.server import TrackerServer
|
|
|
|
tracker = TrackerServer(host="127.0.0.1", port=0)
|
|
port = tracker.start()
|
|
url = f"http://127.0.0.1:{port}"
|
|
|
|
payload = {
|
|
"endpoint": f"http://127.0.0.1:8001",
|
|
"shard_start": 0,
|
|
"shard_end": 7,
|
|
"model": "stub-model",
|
|
"hardware_profile": {},
|
|
"score": 1.0,
|
|
**extra_fields,
|
|
}
|
|
data = _json.dumps(payload).encode()
|
|
req = urllib.request.Request(
|
|
f"{url}/v1/nodes/register",
|
|
data=data,
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(req, timeout=5) as r:
|
|
resp = _json.loads(r.read())
|
|
|
|
tracker.stop()
|
|
return resp
|
|
|
|
|
|
def test_tracker_derives_relay_url_from_public_self_url():
|
|
"Tracker derives relay url from public self url\n\nTags: gossip, network, relay"
|
|
from meshnet_tracker.server import TrackerServer, derive_relay_url_from_public_tracker_url
|
|
|
|
assert derive_relay_url_from_public_tracker_url("https://ai.neuron.d-popov.com") == (
|
|
"wss://ai.neuron.d-popov.com/ws"
|
|
)
|
|
assert derive_relay_url_from_public_tracker_url("http://127.0.0.1:8081") is None
|
|
|
|
tracker = TrackerServer(cluster_self_url="https://ai.neuron.d-popov.com")
|
|
port = tracker.start()
|
|
try:
|
|
import json as _json
|
|
import urllib.request
|
|
|
|
with urllib.request.urlopen(f"http://127.0.0.1:{port}/v1/network/map", timeout=5) as resp:
|
|
body = _json.loads(resp.read())
|
|
assert body["relay_url"] == "wss://ai.neuron.d-popov.com/ws"
|
|
assert body["relay"]["mode"] == "external"
|
|
assert body["relay"]["url"] == "wss://ai.neuron.d-popov.com/ws"
|
|
finally:
|
|
tracker.stop()
|
|
|
|
|
|
def test_tracker_accepts_relay_addr_in_registration():
|
|
"Tracker accepts relay addr in registration\n\nTags: gossip, network, relay"
|
|
resp = _start_tracker_and_register({
|
|
"relay_addr": "ws://relay.meshnet.ai:8765/relay/abc123",
|
|
"cert_fingerprint": "sha256:deadbeef",
|
|
"peer_id": "abc123def456ef01",
|
|
})
|
|
assert "node_id" in resp
|
|
|
|
|
|
def test_tracker_accepts_registration_without_gossip_fields():
|
|
"Existing registrations without P2P fields still work.\n\nTags: gossip, network, relay"
|
|
|
|
resp = _start_tracker_and_register({})
|
|
assert "node_id" in resp
|
|
|
|
|
|
def test_tracker_network_map_exposes_relay_and_registered_peer():
|
|
"Tracker network map exposes relay and registered peer\n\nTags: gossip, network, relay"
|
|
import json as _json
|
|
import urllib.request
|
|
|
|
from meshnet_tracker.server import TrackerServer
|
|
|
|
tracker = TrackerServer(host="127.0.0.1", port=0, relay_url="wss://ai.neuron.d-popov.com/ws")
|
|
port = tracker.start()
|
|
url = f"http://127.0.0.1:{port}"
|
|
try:
|
|
req = urllib.request.Request(
|
|
f"{url}/v1/nodes/register",
|
|
data=_json.dumps({
|
|
"endpoint": "http://192.0.2.10:7000",
|
|
"model": "stub-model",
|
|
"shard_start": 0,
|
|
"shard_end": 31,
|
|
"relay_addr": "wss://ai.neuron.d-popov.com/rpc/peer123",
|
|
"peer_id": "peer123",
|
|
}).encode(),
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(req, timeout=5):
|
|
pass
|
|
with urllib.request.urlopen(f"{url}/v1/network/map", timeout=5) as resp:
|
|
body = _json.loads(resp.read())
|
|
finally:
|
|
tracker.stop()
|
|
|
|
assert body["relay_url"] == "wss://ai.neuron.d-popov.com/ws"
|
|
assert body["relay"]["mode"] == "external"
|
|
assert body["nodes"][0]["relay_addr"] == "wss://ai.neuron.d-popov.com/rpc/peer123"
|
|
assert body["nodes"][0]["peer_id"] == "peer123"
|
|
|
|
|
|
def test_tracker_can_embed_relay_server_and_advertise_it():
|
|
"Tracker can embed relay server and advertise it\n\nTags: gossip, network, relay"
|
|
import json as _json
|
|
import urllib.request
|
|
|
|
import websockets.sync.client as wsc # type: ignore[import]
|
|
from meshnet_tracker.server import TrackerServer
|
|
|
|
tracker = TrackerServer(
|
|
host="127.0.0.1",
|
|
port=0,
|
|
embedded_relay=True,
|
|
embedded_relay_host="127.0.0.1",
|
|
embedded_relay_port=0,
|
|
)
|
|
port = tracker.start()
|
|
try:
|
|
with urllib.request.urlopen(f"http://127.0.0.1:{port}/v1/network/map", timeout=5) as resp:
|
|
body = _json.loads(resp.read())
|
|
assert body["relay"]["mode"] == "embedded"
|
|
assert body["relay_url"].startswith("ws://127.0.0.1:")
|
|
assert body["relay_url"].endswith("/ws")
|
|
with wsc.connect(body["relay_url"], open_timeout=5) as ws:
|
|
ws.send(_json.dumps({
|
|
"topic": "peer-register",
|
|
"version": 1,
|
|
"from_peer": "embedded-peer",
|
|
"msg_id": "embedded-reg-1",
|
|
"payload": {"peer_id": "embedded-peer", "addr": ""},
|
|
}))
|
|
peer_list = _json.loads(ws.recv(timeout=5))
|
|
assert peer_list["topic"] == "peer-list"
|
|
finally:
|
|
tracker.stop()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# mDNS (no-op without zeroconf installed)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_mdns_discovery_is_available_flag():
|
|
"Mdns discovery is available flag\n\nTags: gossip, network, relay"
|
|
from meshnet_p2p.mdns import MdnsDiscovery
|
|
|
|
disc = MdnsDiscovery(peer_id="test", port=8001)
|
|
# is_available() should be bool regardless of zeroconf install status
|
|
assert isinstance(disc.is_available(), bool)
|
|
|
|
|
|
def test_mdns_start_and_stop_without_zeroconf(monkeypatch):
|
|
"Mdns start and stop without zeroconf\n\nTags: gossip, network, relay"
|
|
from meshnet_p2p import mdns as mdns_mod
|
|
monkeypatch.setattr(mdns_mod, "_HAS_ZEROCONF", False)
|
|
from meshnet_p2p.mdns import MdnsDiscovery
|
|
|
|
disc = MdnsDiscovery(peer_id="x", port=8001)
|
|
disc.start() # should not raise
|
|
disc.stop() # should not raise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# US-036: streamed responses over relay RPC / US-037: bridge concurrency
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _start_local_http_server(handler_cls):
|
|
from http.server import ThreadingHTTPServer
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), handler_cls)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
return server, server.server_address[1]
|
|
|
|
|
|
def _run_streaming_peer(port: int, peer_id: str, ready: threading.Event):
|
|
"""Peer that answers one rpc request with a 4-frame streamed response."""
|
|
import websockets.sync.client as wsc # type: ignore[import]
|
|
|
|
with wsc.connect(f"ws://127.0.0.1:{port}/ws") as ws:
|
|
ws.send(json.dumps({
|
|
"topic": "peer-register",
|
|
"version": 1,
|
|
"from_peer": peer_id,
|
|
"msg_id": f"{peer_id}-reg",
|
|
"payload": {"peer_id": peer_id, "addr": ""},
|
|
}))
|
|
ws.recv()
|
|
ready.set()
|
|
envelope = json.loads(ws.recv(timeout=5))
|
|
rid = envelope["payload"]["request_id"]
|
|
|
|
def frame(payload):
|
|
ws.send(json.dumps({
|
|
"topic": "relay-http-response",
|
|
"version": 1,
|
|
"from_peer": peer_id,
|
|
"payload": payload,
|
|
}))
|
|
|
|
frame({"request_id": rid, "status": 200,
|
|
"headers": {"Content-Type": "text/event-stream"},
|
|
"stream": True, "done": False})
|
|
frame({"request_id": rid, "stream": True,
|
|
"chunk": 'data: {"choices": [{"delta": {"content": "hello"}}]}\n\n',
|
|
"done": False})
|
|
frame({"request_id": rid, "stream": True,
|
|
"chunk": "data: [DONE]\n\n", "done": False})
|
|
frame({"request_id": rid, "stream": True, "done": True})
|
|
|
|
|
|
def test_relay_rpc_forwards_streamed_frames_in_order():
|
|
"A streamed response traverses the relay as multiple frames ending with done.\n\nTags: gossip, network, relay, streaming"
|
|
|
|
import websockets.sync.client as wsc # type: ignore[import]
|
|
from meshnet_relay.server import RelayServer
|
|
|
|
relay = RelayServer(host="127.0.0.1", port=0)
|
|
port = relay.start()
|
|
ready = threading.Event()
|
|
peer_thread = threading.Thread(
|
|
target=_run_streaming_peer, args=(port, "stream_peer", ready), daemon=True
|
|
)
|
|
peer_thread.start()
|
|
assert ready.wait(timeout=5)
|
|
|
|
frames = []
|
|
with wsc.connect(f"ws://127.0.0.1:{port}/rpc/stream_peer") as ws:
|
|
ws.send(json.dumps({
|
|
"request_id": "req-stream-1",
|
|
"method": "POST",
|
|
"path": "/v1/chat/completions",
|
|
"headers": {},
|
|
"body": "{}",
|
|
}))
|
|
while True:
|
|
frames.append(json.loads(ws.recv(timeout=5)))
|
|
if not frames[-1].get("stream") or frames[-1].get("done"):
|
|
break
|
|
|
|
relay.stop()
|
|
peer_thread.join(timeout=3)
|
|
|
|
assert len(frames) == 4
|
|
assert frames[0]["status"] == 200 and frames[0]["stream"] is True
|
|
assert frames[1]["chunk"].startswith("data:")
|
|
assert frames[2]["chunk"] == "data: [DONE]\n\n"
|
|
assert frames[-1]["done"] is True
|
|
|
|
|
|
def test_tracker_relay_http_request_collapses_streamed_frames():
|
|
"Buffered wrapper joins chunk frames into one response body.\n\nTags: gossip, network, relay, streaming"
|
|
|
|
from meshnet_relay.server import RelayServer
|
|
from meshnet_tracker.server import _relay_http_request
|
|
|
|
relay = RelayServer(host="127.0.0.1", port=0)
|
|
port = relay.start()
|
|
ready = threading.Event()
|
|
peer_thread = threading.Thread(
|
|
target=_run_streaming_peer, args=(port, "buffer_peer", ready), daemon=True
|
|
)
|
|
peer_thread.start()
|
|
assert ready.wait(timeout=5)
|
|
|
|
response = _relay_http_request(
|
|
f"ws://127.0.0.1:{port}/rpc/buffer_peer",
|
|
path="/v1/chat/completions",
|
|
body=b"{}",
|
|
headers={},
|
|
)
|
|
relay.stop()
|
|
peer_thread.join(timeout=3)
|
|
|
|
assert response is not None
|
|
assert response["status"] == 200
|
|
assert '"content": "hello"' in response["body"]
|
|
assert "data: [DONE]" in response["body"]
|
|
|
|
|
|
def test_stream_line_tokens_accounting():
|
|
"Stream line tokens accounting\n\nTags: gossip, network, relay, streaming"
|
|
from meshnet_tracker.server import _stream_line_tokens
|
|
|
|
assert _stream_line_tokens(b": comment") == (0, None)
|
|
assert _stream_line_tokens(b"data: [DONE]") == (0, None)
|
|
observed, reported = _stream_line_tokens(
|
|
b'data: {"choices": [{"delta": {"content": "two words"}}]}'
|
|
)
|
|
assert observed == 2 and reported is None
|
|
observed, reported = _stream_line_tokens(b'data: {"usage": {"total_tokens": 42}}')
|
|
assert reported == 42
|
|
assert _stream_line_tokens(b"data: not-json") == (1, None)
|
|
|
|
|
|
def test_relay_bridge_streams_sse_as_chunk_frames():
|
|
"Bridge forwards a local SSE response as header + chunk frames + done.\n\nTags: gossip, network, relay, streaming"
|
|
|
|
from http.server import BaseHTTPRequestHandler
|
|
from meshnet_node.relay_bridge import RelayHttpBridge
|
|
|
|
class SSEHandler(BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
self.rfile.read(int(self.headers.get("Content-Length", 0) or 0))
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/event-stream")
|
|
self.end_headers()
|
|
for i in range(3):
|
|
self.wfile.write(f'data: {{"i": {i}}}\n\n'.encode())
|
|
self.wfile.flush()
|
|
self.wfile.write(b"data: [DONE]\n\n")
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
server, port = _start_local_http_server(SSEHandler)
|
|
try:
|
|
bridge = RelayHttpBridge(
|
|
relay_url="ws://unused.example/ws",
|
|
peer_id="peer-sse",
|
|
local_base_url=f"http://127.0.0.1:{port}",
|
|
advertised_addr="",
|
|
)
|
|
frames = []
|
|
bridge._send_response_frame = lambda payload: (frames.append(payload), True)[1]
|
|
bridge._process_request({
|
|
"request_id": "req-sse-1",
|
|
"method": "POST",
|
|
"path": "/v1/chat/completions",
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": "{}",
|
|
})
|
|
finally:
|
|
server.shutdown()
|
|
|
|
assert frames[0]["stream"] is True and frames[0]["status"] == 200
|
|
chunk_frames = [f for f in frames if f.get("chunk")]
|
|
assert len(chunk_frames) >= 2
|
|
assert frames[-1] == {"request_id": "req-sse-1", "stream": True, "done": True}
|
|
assert "".join(f["chunk"] for f in chunk_frames).count("data:") == 4
|
|
|
|
|
|
def test_relay_bridge_non_stream_response_stays_single_frame():
|
|
"Relay bridge non stream response stays single frame\n\nTags: gossip, network, relay, streaming"
|
|
from http.server import BaseHTTPRequestHandler
|
|
from meshnet_node.relay_bridge import RelayHttpBridge
|
|
|
|
class JSONHandler(BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
self.rfile.read(int(self.headers.get("Content-Length", 0) or 0))
|
|
body = json.dumps({"ok": True}).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
server, port = _start_local_http_server(JSONHandler)
|
|
try:
|
|
bridge = RelayHttpBridge(
|
|
relay_url="ws://unused.example/ws",
|
|
peer_id="peer-json",
|
|
local_base_url=f"http://127.0.0.1:{port}",
|
|
advertised_addr="",
|
|
)
|
|
frames = []
|
|
bridge._send_response_frame = lambda payload: (frames.append(payload), True)[1]
|
|
bridge._process_request({
|
|
"request_id": "req-json-1",
|
|
"method": "POST",
|
|
"path": "/v1/chat/completions",
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": "{}",
|
|
})
|
|
finally:
|
|
server.shutdown()
|
|
|
|
assert len(frames) == 1
|
|
assert "stream" not in frames[0]
|
|
assert json.loads(frames[0]["body"]) == {"ok": True}
|
|
|
|
|
|
def test_relay_bridge_serves_concurrent_requests():
|
|
"US-037: a slow relayed request must not block a fast one.\n\nTags: gossip, network, relay"
|
|
|
|
import websockets.sync.client as wsc # type: ignore[import]
|
|
from http.server import BaseHTTPRequestHandler
|
|
from meshnet_node.relay_bridge import RelayHttpBridge
|
|
from meshnet_relay.server import RelayServer
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
self.rfile.read(int(self.headers.get("Content-Length", 0) or 0))
|
|
if self.path == "/slow":
|
|
time.sleep(2.0)
|
|
body = json.dumps({"path": self.path}).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
http_server, http_port = _start_local_http_server(Handler)
|
|
relay = RelayServer(host="127.0.0.1", port=0)
|
|
relay_port = relay.start()
|
|
bridge = RelayHttpBridge(
|
|
relay_url=f"ws://127.0.0.1:{relay_port}/ws",
|
|
peer_id="conc_peer",
|
|
local_base_url=f"http://127.0.0.1:{http_port}",
|
|
advertised_addr="",
|
|
)
|
|
bridge.start()
|
|
results = {}
|
|
try:
|
|
assert bridge.wait_connected(timeout=5)
|
|
deadline = time.time() + 5
|
|
while time.time() < deadline and relay.registry.get("conc_peer") is None:
|
|
time.sleep(0.05)
|
|
assert relay.registry.get("conc_peer") is not None
|
|
|
|
def call(path, key):
|
|
with wsc.connect(f"ws://127.0.0.1:{relay_port}/rpc/conc_peer") as ws:
|
|
ws.send(json.dumps({
|
|
"request_id": f"req-{key}",
|
|
"method": "POST",
|
|
"path": path,
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": "{}",
|
|
}))
|
|
results[key] = (json.loads(ws.recv(timeout=10)), time.monotonic())
|
|
|
|
start = time.monotonic()
|
|
slow_thread = threading.Thread(target=call, args=("/slow", "slow"), daemon=True)
|
|
slow_thread.start()
|
|
time.sleep(0.3) # let the slow request reach the bridge first
|
|
call("/fast", "fast")
|
|
fast_elapsed = results["fast"][1] - start
|
|
slow_thread.join(timeout=10)
|
|
finally:
|
|
bridge.stop()
|
|
relay.stop()
|
|
http_server.shutdown()
|
|
|
|
assert results["fast"][0]["status"] == 200
|
|
assert results["slow"][0]["status"] == 200
|
|
# The fast request finished while the slow one was still in flight.
|
|
assert fast_elapsed < 1.5
|
|
# Responses matched by request_id despite out-of-order completion.
|
|
assert json.loads(results["fast"][0]["body"]) == {"path": "/fast"}
|
|
assert json.loads(results["slow"][0]["body"]) == {"path": "/slow"}
|