5630 lines
208 KiB
Python
5630 lines
208 KiB
Python
import base64
|
|
import ctypes
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import queue
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
from pathlib import Path
|
|
|
|
import RNS
|
|
|
|
|
|
BRIDGE_PATH = Path(__file__).with_name("presence_bridge.py")
|
|
BASE58_ALPHABET = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
|
|
|
|
|
|
def base58_encode(data):
|
|
leading_zeroes = len(data) - len(data.lstrip(b"\x00"))
|
|
value = int.from_bytes(data, "big")
|
|
encoded = ""
|
|
while value > 0:
|
|
value, remainder = divmod(value, 58)
|
|
encoded = BASE58_ALPHABET[remainder] + encoded
|
|
return ("1" * leading_zeroes) + (encoded or ("1" if not data else ""))
|
|
|
|
|
|
def load_bridge():
|
|
spec = importlib.util.spec_from_file_location("presence_bridge_under_test", BRIDGE_PATH)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader is not None
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class PresenceBridgeOwnerLifecycleTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
|
|
def tearDown(self):
|
|
self.bridge._shutdown.clear()
|
|
|
|
def test_owner_pid_environment_is_strictly_validated(self):
|
|
with mock.patch.dict(os.environ, {"QORTAL_RETICULUM_OWNER_PID": "4321"}):
|
|
self.assertEqual(self.bridge._owner_pid_from_environment(), 4321)
|
|
for invalid in ("", "not-a-pid", "0", "1", "-5"):
|
|
with self.subTest(invalid=invalid), mock.patch.dict(
|
|
os.environ,
|
|
{"QORTAL_RETICULUM_OWNER_PID": invalid},
|
|
):
|
|
self.assertEqual(self.bridge._owner_pid_from_environment(), 0)
|
|
|
|
@unittest.skipIf(os.name == "nt", "POSIX parent-reparenting behavior")
|
|
def test_owner_watchdog_forces_exit_after_owner_is_lost(self):
|
|
owner_pid = 4321
|
|
self.bridge._shutdown.clear()
|
|
close_monitor = mock.Mock()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_open_posix_owner_monitor",
|
|
return_value=(lambda: False, close_monitor, "test"),
|
|
), mock.patch.object(self.bridge.time, "sleep") as sleep_mock, mock.patch.object(
|
|
self.bridge.os, "_exit", side_effect=SystemExit(0)
|
|
) as exit_mock:
|
|
with self.assertRaises(SystemExit):
|
|
self.bridge._owner_watchdog_loop(owner_pid)
|
|
|
|
self.assertTrue(self.bridge._shutdown.is_set())
|
|
close_monitor.assert_called_once_with()
|
|
sleep_mock.assert_called_once_with(self.bridge._OWNER_EXIT_GRACE_SECONDS)
|
|
exit_mock.assert_called_once_with(0)
|
|
|
|
@unittest.skipIf(os.name == "nt", "POSIX parent-reparenting behavior")
|
|
def test_owner_loss_still_forces_exit_when_stdin_already_started_shutdown(self):
|
|
owner_pid = 4321
|
|
self.bridge._shutdown.set()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_open_posix_owner_monitor",
|
|
return_value=(lambda: False, lambda: None, "test"),
|
|
), mock.patch.object(self.bridge.time, "sleep"), mock.patch.object(
|
|
self.bridge.os, "_exit", side_effect=SystemExit(0)
|
|
) as exit_mock:
|
|
with self.assertRaises(SystemExit):
|
|
self.bridge._owner_watchdog_loop(owner_pid)
|
|
|
|
exit_mock.assert_called_once_with(0)
|
|
|
|
@unittest.skipIf(os.name == "nt", "POSIX parent-reparenting behavior")
|
|
def test_owner_watchdog_leaves_a_live_owner_alone_during_shutdown(self):
|
|
owner_pid = 4321
|
|
self.bridge._shutdown.set()
|
|
close_monitor = mock.Mock()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_open_posix_owner_monitor",
|
|
return_value=(lambda: True, close_monitor, "test"),
|
|
), mock.patch.object(self.bridge.os, "_exit") as exit_mock:
|
|
self.bridge._owner_watchdog_loop(owner_pid)
|
|
close_monitor.assert_called_once_with()
|
|
exit_mock.assert_not_called()
|
|
|
|
@unittest.skipIf(os.name == "nt", "POSIX process monitor")
|
|
def test_owner_monitor_accepts_a_live_process_that_is_not_the_direct_parent(self):
|
|
owner_alive, close_monitor, _kind = self.bridge._open_posix_owner_monitor(
|
|
os.getpid()
|
|
)
|
|
try:
|
|
self.assertTrue(owner_alive())
|
|
finally:
|
|
close_monitor()
|
|
|
|
@unittest.skipIf(os.name == "nt", "POSIX process monitor")
|
|
def test_owner_monitor_detects_the_exact_process_exiting(self):
|
|
owner = subprocess.Popen(
|
|
[sys.executable, "-c", "import time; time.sleep(0.1)"]
|
|
)
|
|
owner_alive, close_monitor, _kind = self.bridge._open_posix_owner_monitor(
|
|
owner.pid
|
|
)
|
|
try:
|
|
self.assertTrue(owner_alive())
|
|
owner.wait(timeout=2.0)
|
|
deadline = time.monotonic() + 1.0
|
|
while owner_alive() and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
self.assertFalse(owner_alive())
|
|
finally:
|
|
if owner.poll() is None:
|
|
owner.terminate()
|
|
owner.wait(timeout=2.0)
|
|
close_monitor()
|
|
|
|
@unittest.skipUnless(sys.platform.startswith("linux"), "Linux fallback monitor")
|
|
def test_linux_owner_monitor_fallback_rejects_a_reused_pid(self):
|
|
with mock.patch.object(
|
|
self.bridge.os, "pidfd_open", side_effect=OSError("unsupported")
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_linux_process_start_token",
|
|
side_effect=["original-start", "original-start", "replacement-start"],
|
|
):
|
|
owner_alive, close_monitor, kind = self.bridge._open_posix_owner_monitor(
|
|
4321
|
|
)
|
|
try:
|
|
self.assertEqual(kind, "proc-start")
|
|
self.assertTrue(owner_alive())
|
|
self.assertFalse(owner_alive())
|
|
finally:
|
|
close_monitor()
|
|
|
|
def test_macos_owner_monitor_uses_process_exit_events(self):
|
|
class FakeKqueue:
|
|
def __init__(self):
|
|
self.exited = False
|
|
self.closed = False
|
|
self.changes = []
|
|
|
|
def control(self, changes, _max_events, _timeout):
|
|
if changes is not None:
|
|
self.changes.extend(changes)
|
|
return []
|
|
return ["exit"] if self.exited else []
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
fake_kqueue = FakeKqueue()
|
|
with mock.patch.object(self.bridge.sys, "platform", "darwin"), mock.patch.object(
|
|
self.bridge.select, "kqueue", return_value=fake_kqueue, create=True
|
|
), mock.patch.object(
|
|
self.bridge.select, "kevent", return_value="process-event", create=True
|
|
), mock.patch.object(
|
|
self.bridge.select, "KQ_FILTER_PROC", 1, create=True
|
|
), mock.patch.object(
|
|
self.bridge.select, "KQ_EV_ADD", 2, create=True
|
|
), mock.patch.object(
|
|
self.bridge.select, "KQ_EV_ENABLE", 4, create=True
|
|
), mock.patch.object(
|
|
self.bridge.select, "KQ_EV_CLEAR", 8, create=True
|
|
), mock.patch.object(
|
|
self.bridge.select, "KQ_NOTE_EXIT", 16, create=True
|
|
):
|
|
owner_alive, close_monitor, kind = self.bridge._open_posix_owner_monitor(
|
|
4321
|
|
)
|
|
self.assertEqual(kind, "kqueue")
|
|
self.assertTrue(owner_alive())
|
|
fake_kqueue.exited = True
|
|
self.assertFalse(owner_alive())
|
|
close_monitor()
|
|
|
|
self.assertEqual(fake_kqueue.changes, ["process-event"])
|
|
self.assertTrue(fake_kqueue.closed)
|
|
|
|
def test_windows_owner_watchdog_uses_the_exact_process_handle(self):
|
|
kernel32 = mock.Mock()
|
|
kernel32.OpenProcess.return_value = 99
|
|
kernel32.WaitForSingleObject.return_value = 0
|
|
fake_windll = mock.Mock(kernel32=kernel32)
|
|
self.bridge._shutdown.clear()
|
|
|
|
with mock.patch.object(self.bridge.os, "name", "nt"), mock.patch.object(
|
|
ctypes, "windll", fake_windll, create=True
|
|
), mock.patch.object(
|
|
self.bridge._shutdown, "wait", return_value=False
|
|
), mock.patch.object(self.bridge.time, "sleep"), mock.patch.object(
|
|
self.bridge.os, "_exit", side_effect=SystemExit(0)
|
|
) as exit_mock:
|
|
with self.assertRaises(SystemExit):
|
|
self.bridge._owner_watchdog_loop(4321)
|
|
|
|
kernel32.OpenProcess.assert_called_once_with(0x00100000, False, 4321)
|
|
kernel32.WaitForSingleObject.assert_called_once()
|
|
kernel32.CloseHandle.assert_called_once_with(99)
|
|
exit_mock.assert_called_once_with(0)
|
|
|
|
|
|
class ReticulumPathVisibilityTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
self.destination_hash = bytes.fromhex("11" * 16)
|
|
|
|
def test_shared_daemon_path_is_authoritative_when_local_table_misses(self):
|
|
daemon = mock.Mock()
|
|
daemon.is_connected_to_shared_instance = True
|
|
daemon.get_path_snapshot.return_value = {
|
|
"hops": 2,
|
|
"timestamp": time.time(),
|
|
}
|
|
self.bridge._reticulum = daemon
|
|
|
|
with mock.patch.object(RNS.Transport, "has_path", return_value=False):
|
|
self.assertTrue(self.bridge._reticulum_has_path(self.destination_hash))
|
|
self.assertTrue(self.bridge._reticulum_has_path(self.destination_hash))
|
|
|
|
daemon.get_path_snapshot.assert_called_once_with(self.destination_hash)
|
|
|
|
def test_embedded_instance_uses_local_path_without_rpc(self):
|
|
reticulum = mock.Mock()
|
|
reticulum.is_connected_to_shared_instance = False
|
|
self.bridge._reticulum = reticulum
|
|
|
|
with mock.patch.object(RNS.Transport, "has_path", return_value=False):
|
|
self.assertFalse(self.bridge._reticulum_has_path(self.destination_hash))
|
|
|
|
reticulum.get_path_snapshot.assert_not_called()
|
|
|
|
def test_dropping_path_invalidates_shared_availability_cache(self):
|
|
daemon = mock.Mock()
|
|
daemon.is_connected_to_shared_instance = True
|
|
daemon.get_path_snapshot.side_effect = [
|
|
{"hops": 2, "timestamp": time.time()},
|
|
None,
|
|
]
|
|
daemon.drop_path.return_value = True
|
|
self.bridge._reticulum = daemon
|
|
|
|
with mock.patch.object(RNS.Transport, "has_path", return_value=False), mock.patch.object(
|
|
RNS.Transport, "expire_path", return_value=False
|
|
), mock.patch.object(RNS.Transport, "mark_path_unresponsive"):
|
|
self.assertTrue(self.bridge._reticulum_has_path(self.destination_hash))
|
|
self.assertTrue(self.bridge._drop_reticulum_path(self.destination_hash))
|
|
self.assertFalse(self.bridge._reticulum_has_path(self.destination_hash))
|
|
|
|
self.assertEqual(daemon.get_path_snapshot.call_count, 2)
|
|
|
|
def test_game_discovery_miss_nudges_without_dropping_path(self):
|
|
peer_hash = self.destination_hash.hex()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_nudge_cached_reticulum_path",
|
|
return_value=True,
|
|
) as nudge, mock.patch.object(
|
|
self.bridge,
|
|
"_force_overlay_peer_path_refresh",
|
|
) as force_refresh:
|
|
self.assertTrue(
|
|
self.bridge._refresh_qortalland_game_path(
|
|
peer_hash,
|
|
"game_link_no_path",
|
|
)
|
|
)
|
|
|
|
nudge.assert_called_once()
|
|
force_refresh.assert_not_called()
|
|
|
|
def test_game_failed_link_can_still_replace_bad_path(self):
|
|
peer_hash = self.destination_hash.hex()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_nudge_cached_reticulum_path",
|
|
) as nudge, mock.patch.object(
|
|
self.bridge,
|
|
"_force_overlay_peer_path_refresh",
|
|
return_value=True,
|
|
) as force_refresh:
|
|
self.assertTrue(
|
|
self.bridge._refresh_qortalland_game_path(
|
|
peer_hash,
|
|
"game_link_attempt_closed",
|
|
)
|
|
)
|
|
|
|
force_refresh.assert_called_once_with(
|
|
peer_hash,
|
|
target="qortalland-game",
|
|
reason="game_link_attempt_closed",
|
|
await_seconds=0.0,
|
|
)
|
|
nudge.assert_not_called()
|
|
|
|
def test_recent_media_success_avoids_route_rpc(self):
|
|
peer_hash = self.destination_hash.hex()
|
|
state = self.bridge._get_call_media_state(peer_hash)
|
|
state.update({
|
|
"destination_hash_hex": peer_hash,
|
|
"path_state": "fresh",
|
|
"last_send_ok": time.time(),
|
|
"last_send_fail": None,
|
|
"last_inbound_at": None,
|
|
})
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_reticulum_has_path",
|
|
side_effect=AssertionError("route lookup should not run for recent traffic"),
|
|
):
|
|
self.assertEqual(
|
|
self.bridge._classify_call_media_path_state(
|
|
peer_hash,
|
|
self.destination_hash,
|
|
),
|
|
"fresh",
|
|
)
|
|
|
|
|
|
class FakeLink:
|
|
def __init__(self):
|
|
self.closed_callback = None
|
|
self.packet_callback = None
|
|
self.remote_identified_callback = None
|
|
self.resource_strategy = None
|
|
self.resource_callback = None
|
|
self.resource_started_callback = None
|
|
self.resource_concluded_callback = None
|
|
self.teardown_called = False
|
|
self.remote_identity = None
|
|
|
|
def get_mdu(self):
|
|
return 4096
|
|
|
|
def set_link_closed_callback(self, callback):
|
|
self.closed_callback = callback
|
|
|
|
def set_packet_callback(self, callback):
|
|
self.packet_callback = callback
|
|
|
|
def set_remote_identified_callback(self, callback):
|
|
self.remote_identified_callback = callback
|
|
|
|
def get_remote_identity(self):
|
|
return self.remote_identity
|
|
|
|
def set_resource_strategy(self, strategy):
|
|
self.resource_strategy = strategy
|
|
|
|
def set_resource_callback(self, callback):
|
|
self.resource_callback = callback
|
|
|
|
def set_resource_started_callback(self, callback):
|
|
self.resource_started_callback = callback
|
|
|
|
def set_resource_concluded_callback(self, callback):
|
|
self.resource_concluded_callback = callback
|
|
|
|
def teardown(self):
|
|
self.teardown_called = True
|
|
|
|
|
|
class FakeSessionReceipt:
|
|
FAILED = 0
|
|
SENT = 1
|
|
|
|
def __init__(self):
|
|
self.progress = 0.0
|
|
self.metadata = None
|
|
self.response = None
|
|
self.cancelled = False
|
|
self.status = self.SENT
|
|
self.concluded_at = None
|
|
self.resource = None
|
|
self.link = None
|
|
self.request_id = None
|
|
|
|
def get_progress(self):
|
|
return self.progress
|
|
|
|
def get_response(self):
|
|
return self.response
|
|
|
|
def cancel(self):
|
|
self.cancelled = True
|
|
|
|
|
|
class FakeSessionLink(FakeLink):
|
|
def __init__(self, link_id=None):
|
|
super().__init__()
|
|
self.link_id = link_id or bytes.fromhex("77" * 16)
|
|
self.requests = []
|
|
self.identified_with = None
|
|
self.remote_identity = None
|
|
self.outgoing_resources = []
|
|
|
|
def identify(self, identity):
|
|
self.identified_with = identity
|
|
|
|
def get_remote_identity(self):
|
|
return self.remote_identity
|
|
|
|
def request(self, path, data=None, **callbacks):
|
|
receipt = FakeSessionReceipt()
|
|
self.requests.append((path, data, callbacks, receipt))
|
|
return receipt
|
|
|
|
|
|
class FakePacket:
|
|
def __init__(self, link):
|
|
self.link = link
|
|
|
|
|
|
class FakeRnsPacket:
|
|
MDU = 500
|
|
ENCRYPTED_MDU = 500
|
|
sent_links = []
|
|
sent_payloads = []
|
|
|
|
def __init__(self, link, data, create_receipt=False):
|
|
self.link = link
|
|
self.data = data
|
|
self.create_receipt = create_receipt
|
|
|
|
def send(self):
|
|
self.__class__.sent_links.append(self.link)
|
|
self.__class__.sent_payloads.append(self.data)
|
|
return True
|
|
|
|
|
|
class FakeDestination:
|
|
def __init__(self):
|
|
self.hash = bytes.fromhex("44" * 16)
|
|
|
|
|
|
class PresenceBridgeWireEncodingTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
self.bridge._destination = FakeDestination()
|
|
|
|
def test_group_signal_stamps_transport_sender_without_overwriting_payload(self):
|
|
read_state = {
|
|
"y": "d",
|
|
"q": "Q-peer",
|
|
"u": 123,
|
|
"n": 124,
|
|
"p": "public-key",
|
|
"z": "signature",
|
|
}
|
|
encoded = self.bridge._encode_group_signal_wire(
|
|
{"t": "RCHAT", "k": "read_sync", "w": read_state}
|
|
)
|
|
|
|
self.assertTrue(encoded["ok"])
|
|
wire = json.loads(encoded["wire_bytes"].decode("utf-8"))
|
|
self.assertEqual(wire["r"], "44" * 16)
|
|
self.assertEqual(wire["w"], read_state)
|
|
|
|
def test_group_signal_relay_preserves_original_sender(self):
|
|
encoded = self.bridge._encode_group_signal_wire(
|
|
{"t": "GJ", "r": "11" * 16}
|
|
)
|
|
|
|
self.assertTrue(encoded["ok"])
|
|
wire = json.loads(encoded["wire_bytes"].decode("utf-8"))
|
|
self.assertEqual(wire["r"], "11" * 16)
|
|
|
|
def test_call_signal_relay_preserves_original_sender(self):
|
|
encoded = self.bridge._encode_call_signal_wire(
|
|
{"t": "CR", "r": "22" * 16}
|
|
)
|
|
|
|
self.assertTrue(encoded["ok"])
|
|
wire = json.loads(encoded["wire_bytes"].decode("utf-8"))
|
|
self.assertEqual(wire["r"], "22" * 16)
|
|
|
|
def test_route_bound_presence_fits_encrypted_mdu_direct_and_relayed(self):
|
|
origin_raw = bytes.fromhex("55" * 16)
|
|
origin_route = base64.urlsafe_b64encode(origin_raw).decode("ascii").rstrip("=")
|
|
local_route = base64.urlsafe_b64encode(bytes.fromhex("44" * 16)).decode(
|
|
"ascii"
|
|
).rstrip("=")
|
|
envelope = {
|
|
"type": "PRESENCE_ANNOUNCE",
|
|
"id": "x" * 16,
|
|
"timestamp": 9_999_999_999_999,
|
|
"signature": "z" * 88,
|
|
"payload": {
|
|
"address": "Q" + ("a" * 33),
|
|
"publicKey": "k" * 44,
|
|
"sessionId": "P" + local_route + ("e" * 13),
|
|
"status": "online",
|
|
"clientVersion": "1.0.0",
|
|
},
|
|
}
|
|
|
|
direct = self.bridge.make_presence_wire(envelope, 4)
|
|
relayed_envelope = {
|
|
**envelope,
|
|
"payload": {
|
|
**envelope["payload"],
|
|
"sessionId": "P" + origin_route + ("e" * 13),
|
|
},
|
|
}
|
|
relayed = self.bridge.make_presence_wire(
|
|
relayed_envelope,
|
|
3,
|
|
origin_sender_hash="55" * 16,
|
|
)
|
|
|
|
self.assertLessEqual(len(direct), self.bridge._MAX_ENCRYPTED_WIRE_BYTES)
|
|
self.assertLessEqual(len(relayed), self.bridge._MAX_ENCRYPTED_WIRE_BYTES)
|
|
self.assertNotIn("o", json.loads(relayed.decode("utf-8")))
|
|
|
|
def test_oversized_legacy_presence_is_rejected_before_packet_send(self):
|
|
envelope = {
|
|
"type": "PRESENCE_ANNOUNCE",
|
|
"id": "x" * 36,
|
|
"timestamp": 9_999_999_999_999,
|
|
"signature": "z" * 88,
|
|
"payload": {
|
|
"address": "Q" + ("a" * 33),
|
|
"publicKey": "k" * 44,
|
|
"sessionId": "legacy-session-id".ljust(36, "x"),
|
|
"status": "online",
|
|
"clientVersion": "1.0.0",
|
|
},
|
|
}
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "exceeds encrypted MDU"):
|
|
self.bridge.make_presence_wire(envelope, 4)
|
|
|
|
def test_stale_route_bound_presence_is_not_republished_after_route_change(self):
|
|
stale_route = base64.urlsafe_b64encode(bytes.fromhex("55" * 16)).decode(
|
|
"ascii"
|
|
).rstrip("=")
|
|
envelope = {
|
|
"type": "PRESENCE_HEARTBEAT",
|
|
"id": "x" * 16,
|
|
"timestamp": 9_999_999_999_999,
|
|
"signature": "z" * 88,
|
|
"payload": {
|
|
"address": "Q" + ("a" * 33),
|
|
"publicKey": "k" * 44,
|
|
"sessionId": "P" + stale_route + ("e" * 13),
|
|
"status": "online",
|
|
},
|
|
}
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "local destination"):
|
|
self.bridge.make_presence_wire(envelope, 4)
|
|
|
|
def test_presence_route_binding_decoder_rejects_noncanonical_ids(self):
|
|
route = base64.urlsafe_b64encode(bytes.fromhex("55" * 16)).decode("ascii").rstrip("=")
|
|
session_id = "P" + route + ("e" * 13)
|
|
self.assertEqual(
|
|
self.bridge._presence_route_bound_destination_hash(session_id),
|
|
"55" * 16,
|
|
)
|
|
self.assertIsNone(
|
|
self.bridge._presence_route_bound_destination_hash("P" + ("!" * 35))
|
|
)
|
|
|
|
def test_route_bound_presence_recovers_signed_origin_from_relay(self):
|
|
origin_raw = bytes.fromhex("55" * 16)
|
|
route = base64.urlsafe_b64encode(origin_raw).decode("ascii").rstrip("=")
|
|
emitted = []
|
|
message = {
|
|
"t": "PRESENCE_HEARTBEAT",
|
|
"i": "x" * 16,
|
|
"a": "Q" + ("a" * 33),
|
|
"k": "k" * 44,
|
|
"n": "P" + route + ("e" * 13),
|
|
"m": 9_999_999_999_999,
|
|
"g": "z" * 88,
|
|
"r": "44" * 16,
|
|
"s": "online",
|
|
"q": 3,
|
|
}
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"emit_event",
|
|
side_effect=lambda event, payload: emitted.append((event, payload)),
|
|
):
|
|
accepted = self.bridge._emit_presence_message(message, "relay-link")
|
|
|
|
self.assertTrue(accepted)
|
|
self.assertEqual(emitted[0][0], "presence_message")
|
|
self.assertEqual(
|
|
emitted[0][1]["route"],
|
|
{
|
|
"kind": "reticulum",
|
|
"destinationHash": "55" * 16,
|
|
"viaDestinationHash": "44" * 16,
|
|
"overlayHopsRemaining": 3,
|
|
"linkId": "relay-link",
|
|
},
|
|
)
|
|
|
|
|
|
class PresenceBridgeReticulumChatInboundDedupTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
|
|
def test_identity_request_dedup_ignores_route_fields(self):
|
|
request = {
|
|
"t": "RCHAT",
|
|
"k": "identity_req",
|
|
"d": "dd" * 16,
|
|
"rid": "11" * 12,
|
|
"h": 0,
|
|
"m": 5,
|
|
"x": int((time.time() + 30) * 1000),
|
|
}
|
|
|
|
self.assertFalse(
|
|
self.bridge._should_drop_duplicate_reticulum_chat_inbound(request)
|
|
)
|
|
self.assertTrue(
|
|
self.bridge._should_drop_duplicate_reticulum_chat_inbound(
|
|
{**request, "h": 3, "r": "aa" * 16}
|
|
)
|
|
)
|
|
|
|
def test_typing_dedup_ignores_origin_hops_and_ingress_sender(self):
|
|
typing = {
|
|
"t": "RCHAT",
|
|
"k": "typing",
|
|
"g": 73,
|
|
"c": "general",
|
|
"a": "Qsender",
|
|
"ts": 123_456,
|
|
"active": True,
|
|
"o": "aa" * 16,
|
|
"h": 1,
|
|
"r": "bb" * 16,
|
|
}
|
|
|
|
self.assertFalse(
|
|
self.bridge._should_drop_duplicate_reticulum_chat_inbound(typing)
|
|
)
|
|
self.assertTrue(
|
|
self.bridge._should_drop_duplicate_reticulum_chat_inbound(
|
|
{**typing, "o": "cc" * 16, "h": 4, "r": "dd" * 16}
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
self.bridge._should_drop_duplicate_reticulum_chat_inbound(
|
|
{**typing, "ts": 123_457}
|
|
)
|
|
)
|
|
|
|
|
|
class PresenceBridgeLandStateFastPathTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
self.bridge._destination = FakeDestination()
|
|
self.group_id = 73
|
|
self.author = "Q-land-author"
|
|
self.session_id = "land-session"
|
|
self.source_hash = "11" * 16
|
|
self.target_hash = "22" * 16
|
|
self.origin = "AQIDBAUGBwgJCgsMDQ4PEA"
|
|
self.private_key = RNS.Cryptography.Ed25519PrivateKey.generate()
|
|
public_key = self.private_key.public_key().public_bytes()
|
|
expires_at = int((time.time() + 60) * 1000)
|
|
self.bridge._configure_land_state_forwarding(
|
|
[
|
|
{
|
|
"groupId": self.group_id,
|
|
"targets": [
|
|
{
|
|
"peerPresenceHash": self.target_hash,
|
|
"expiresAt": expires_at,
|
|
}
|
|
],
|
|
}
|
|
],
|
|
[
|
|
{
|
|
"groupId": self.group_id,
|
|
"authorAddress": self.author,
|
|
"sessionId": self.session_id,
|
|
"ephemeralPublicKey": base58_encode(public_key),
|
|
"expiresAt": expires_at,
|
|
}
|
|
],
|
|
7,
|
|
)
|
|
|
|
def state(self, sequence=1, signature=None):
|
|
timestamp = int(time.time() * 1000)
|
|
fields = {
|
|
"afk": True,
|
|
"authorAddress": self.author,
|
|
"direction": "r",
|
|
"dnd": True,
|
|
"groupId": self.group_id,
|
|
"movement": "walk",
|
|
"roomId": "room",
|
|
"sequence": sequence,
|
|
"sessionId": self.session_id,
|
|
"skinId": 4,
|
|
"timestamp": timestamp,
|
|
"type": "QORTAL_LAND_STATE",
|
|
"x": 50,
|
|
"y": 60,
|
|
}
|
|
signed_bytes = json.dumps(
|
|
fields,
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
).encode("utf-8")
|
|
encoded_signature = signature
|
|
if encoded_signature is None:
|
|
encoded_signature = base58_encode(
|
|
self.private_key.sign(signed_bytes)
|
|
)
|
|
return {
|
|
"t": "RCHAT",
|
|
"k": "land_state",
|
|
"g": self.group_id,
|
|
"a": self.author,
|
|
"s": self.session_id,
|
|
"q": sequence,
|
|
"x": 50,
|
|
"y": 60,
|
|
"u": "room",
|
|
"d": "r",
|
|
"m": "walk",
|
|
"v": 3,
|
|
"i": 4,
|
|
"ts": timestamp,
|
|
"z": encoded_signature,
|
|
"o": self.origin,
|
|
"h": 1,
|
|
}
|
|
|
|
def queue_without_scheduler(self, message):
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_enqueue_scheduler_task",
|
|
return_value=True,
|
|
):
|
|
queued = self.bridge._queue_land_state_fast_path(
|
|
message,
|
|
self.source_hash,
|
|
"source-link",
|
|
)
|
|
self.assertTrue(queued)
|
|
return next(reversed(self.bridge._land_state_forward_pending))
|
|
|
|
def test_preserves_full_origin_and_reports_the_forwarding_revision(self):
|
|
message = self.state()
|
|
pending_key = self.queue_without_scheduler(message)
|
|
emitted = []
|
|
sent = []
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_send_wire_to_established_overlay_peer",
|
|
side_effect=lambda peer, wire, _traffic: sent.append((peer, wire)) or True,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_emit_call_bridge_message",
|
|
side_effect=lambda *args, **kwargs: emitted.append((args, kwargs)) or True,
|
|
):
|
|
self.bridge._process_land_state_fast_path(pending_key)
|
|
|
|
self.assertEqual(len(sent), 1)
|
|
forwarded = json.loads(sent[0][1].decode("utf-8"))
|
|
self.assertEqual(forwarded["o"], self.origin)
|
|
self.assertEqual(forwarded["v"], 3)
|
|
self.assertEqual(forwarded["i"], 4)
|
|
self.assertTrue(emitted[0][1]["land_state_fast_forwarded"])
|
|
self.assertEqual(emitted[0][1]["land_state_forwarding_revision"], 7)
|
|
|
|
def test_destination_hash_matching_requires_exact_hash_equivalence(self):
|
|
full_hash = "0102030405060708090a0b0c0d0e0f10"
|
|
same_prefix = "01020304" + ("ff" * 12)
|
|
|
|
self.assertTrue(
|
|
self.bridge._land_state_hash_matches(self.origin, full_hash)
|
|
)
|
|
self.assertFalse(
|
|
self.bridge._land_state_hash_matches(full_hash, same_prefix)
|
|
)
|
|
|
|
def test_invalid_origin_is_replaced_with_the_verified_ingress_peer(self):
|
|
message = self.state()
|
|
message["o"] = "invalid"
|
|
pending_key = self.queue_without_scheduler(message)
|
|
sent = []
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_send_wire_to_established_overlay_peer",
|
|
side_effect=lambda peer, wire, _traffic: sent.append((peer, wire)) or True,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_emit_call_bridge_message",
|
|
return_value=True,
|
|
):
|
|
self.bridge._process_land_state_fast_path(pending_key)
|
|
|
|
self.assertEqual(len(sent), 1)
|
|
forwarded = json.loads(sent[0][1].decode("utf-8"))
|
|
self.assertEqual(forwarded["o"], self.source_hash)
|
|
|
|
def test_route_revision_change_forces_electron_fallback(self):
|
|
pending_key = self.queue_without_scheduler(self.state())
|
|
emitted = []
|
|
|
|
def send_and_change_revision(_peer, _wire, _traffic):
|
|
self.bridge._land_state_forwarding_revision = 8
|
|
return True
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_send_wire_to_established_overlay_peer",
|
|
side_effect=send_and_change_revision,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_emit_call_bridge_message",
|
|
side_effect=lambda *args, **kwargs: emitted.append((args, kwargs)) or True,
|
|
):
|
|
self.bridge._process_land_state_fast_path(pending_key)
|
|
|
|
self.assertFalse(emitted[0][1]["land_state_fast_forwarded"])
|
|
self.assertIsNone(emitted[0][1]["land_state_forwarding_revision"])
|
|
|
|
def test_unverified_higher_sequence_does_not_replace_pending_valid_state(self):
|
|
valid_key = self.queue_without_scheduler(self.state(sequence=1))
|
|
invalid_key = self.queue_without_scheduler(
|
|
self.state(sequence=999, signature="invalid")
|
|
)
|
|
self.assertEqual(len(self.bridge._land_state_forward_pending), 2)
|
|
emitted = []
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_send_wire_to_established_overlay_peer",
|
|
return_value=True,
|
|
) as send, mock.patch.object(
|
|
self.bridge,
|
|
"_emit_call_bridge_message",
|
|
side_effect=lambda *args, **kwargs: emitted.append((args, kwargs)) or True,
|
|
):
|
|
self.bridge._process_land_state_fast_path(valid_key)
|
|
self.bridge._process_land_state_fast_path(invalid_key)
|
|
|
|
self.assertEqual(send.call_count, 1)
|
|
self.assertEqual(len(emitted), 1)
|
|
|
|
def test_oversized_target_plan_falls_back_instead_of_installing_part_of_it(self):
|
|
expires_at = int((time.time() + 60) * 1000)
|
|
targets = [
|
|
{
|
|
"peerPresenceHash": f"{index:032x}",
|
|
"expiresAt": expires_at,
|
|
}
|
|
for index in range(
|
|
1,
|
|
self.bridge._LAND_STATE_FORWARDING_MAX_TARGETS_PER_GROUP + 2,
|
|
)
|
|
]
|
|
self.bridge._configure_land_state_forwarding(
|
|
[{"groupId": self.group_id, "targets": targets}],
|
|
[],
|
|
8,
|
|
)
|
|
self.assertNotIn(
|
|
self.group_id,
|
|
self.bridge._land_state_forwarding_plans,
|
|
)
|
|
|
|
def test_sequence_zero_is_forwarded_only_once(self):
|
|
message = self.state(sequence=0)
|
|
emitted = []
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_send_wire_to_established_overlay_peer",
|
|
return_value=True,
|
|
) as send, mock.patch.object(
|
|
self.bridge,
|
|
"_emit_call_bridge_message",
|
|
side_effect=lambda *args, **kwargs: emitted.append((args, kwargs)) or True,
|
|
):
|
|
self.bridge._process_land_state_fast_path(
|
|
self.queue_without_scheduler(message)
|
|
)
|
|
self.bridge._process_land_state_fast_path(
|
|
self.queue_without_scheduler(message)
|
|
)
|
|
|
|
self.assertEqual(send.call_count, 1)
|
|
self.assertEqual(len(emitted), 1)
|
|
|
|
|
|
class PresenceBridgeAudioForwardFastPathTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
self.room_id = "gcall-qortal-716"
|
|
self.source_hash = "11" * 16
|
|
self.target_hash = "22" * 16
|
|
|
|
def plan(self, ingress_link="source-link", target_link="target-link"):
|
|
return {
|
|
"roomId": self.room_id,
|
|
"topologyEpoch": 7,
|
|
"rules": [
|
|
{
|
|
"sourceAddress": "Q-source",
|
|
"ingress": {
|
|
"address": "Q-source",
|
|
"transport": "link",
|
|
"linkId": ingress_link,
|
|
"peerPresenceHash": self.source_hash,
|
|
"peerDestinationHash": self.source_hash,
|
|
},
|
|
"targets": [
|
|
{
|
|
"address": "Q-target",
|
|
"transport": "link",
|
|
"linkId": target_link,
|
|
"peerPresenceHash": self.target_hash,
|
|
"peerDestinationHash": self.target_hash,
|
|
}
|
|
],
|
|
}
|
|
],
|
|
}
|
|
|
|
def test_exact_verified_ingress_forwards_unchanged_media(self):
|
|
rooms, rules = self.bridge._configure_audio_forwarding_plans([self.plan()])
|
|
self.assertEqual((rooms, rules), (1, 1))
|
|
captured = []
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_audio_data_plane_broadcast_inbound_audio",
|
|
return_value=True,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_put_audio_decoded_batch_keep_newest",
|
|
side_effect=lambda frames: captured.extend(frames) or True,
|
|
):
|
|
handled = self.bridge._try_group_audio_forward_fast_path(
|
|
self.room_id,
|
|
"source-link",
|
|
self.source_hash,
|
|
self.source_hash,
|
|
1234,
|
|
b"encrypted-media",
|
|
b"inbound-batch",
|
|
)
|
|
self.assertTrue(handled)
|
|
self.assertEqual(len(captured), 1)
|
|
self.assertEqual(captured[0][0], "target-link")
|
|
self.assertEqual(captured[0][1], self.room_id)
|
|
self.assertEqual(captured[0][5], b"encrypted-media")
|
|
|
|
def test_link_media_does_not_fall_back_to_hash_matching(self):
|
|
self.bridge._configure_audio_forwarding_plans([self.plan()])
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_audio_data_plane_broadcast_inbound_audio",
|
|
) as broadcast:
|
|
handled = self.bridge._try_group_audio_forward_fast_path(
|
|
self.room_id,
|
|
"unverified-link",
|
|
self.source_hash,
|
|
self.source_hash,
|
|
1234,
|
|
b"encrypted-media",
|
|
b"inbound-batch",
|
|
)
|
|
self.assertFalse(handled)
|
|
broadcast.assert_not_called()
|
|
|
|
def test_packet_media_matches_only_the_configured_peer_hash(self):
|
|
plan = self.plan()
|
|
ingress = plan["rules"][0]["ingress"]
|
|
ingress["transport"] = "packet"
|
|
ingress["linkId"] = ""
|
|
self.bridge._configure_audio_forwarding_plans([plan])
|
|
captured = []
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_audio_data_plane_broadcast_inbound_audio",
|
|
return_value=True,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_put_audio_decoded_batch_keep_newest",
|
|
side_effect=lambda frames: captured.extend(frames) or True,
|
|
):
|
|
handled = self.bridge._try_group_audio_forward_fast_path(
|
|
self.room_id,
|
|
"",
|
|
self.source_hash,
|
|
"",
|
|
1234,
|
|
b"encrypted-media",
|
|
b"inbound-batch",
|
|
)
|
|
rejected = self.bridge._try_group_audio_forward_fast_path(
|
|
self.room_id,
|
|
"",
|
|
"33" * 16,
|
|
"",
|
|
1235,
|
|
b"other-media",
|
|
b"other-batch",
|
|
)
|
|
self.assertTrue(handled)
|
|
self.assertFalse(rejected)
|
|
self.assertEqual(len(captured), 1)
|
|
|
|
def test_forward_queue_rejection_does_not_duplicate_local_delivery(self):
|
|
self.bridge._configure_audio_forwarding_plans([self.plan()])
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_audio_data_plane_broadcast_inbound_audio",
|
|
return_value=True,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_put_audio_decoded_batch_keep_newest",
|
|
return_value=False,
|
|
):
|
|
handled = self.bridge._try_group_audio_forward_fast_path(
|
|
self.room_id,
|
|
"source-link",
|
|
self.source_hash,
|
|
self.source_hash,
|
|
1234,
|
|
b"encrypted-media",
|
|
b"inbound-batch",
|
|
)
|
|
self.assertTrue(handled)
|
|
|
|
def test_plan_replacement_removes_stale_room_and_loop_target(self):
|
|
loop_plan = self.plan(target_link="source-link")
|
|
rooms, rules = self.bridge._configure_audio_forwarding_plans([loop_plan])
|
|
self.assertEqual((rooms, rules), (1, 1))
|
|
stored_rule = self.bridge._audio_forwarding_plans_by_room[self.room_id][
|
|
"rules"
|
|
][0]
|
|
self.assertEqual(stored_rule["targets"], [])
|
|
|
|
rooms, rules = self.bridge._configure_audio_forwarding_plans([])
|
|
self.assertEqual((rooms, rules), (0, 0))
|
|
self.assertNotIn(self.room_id, self.bridge._audio_forwarding_plans_by_room)
|
|
|
|
|
|
class PresenceBridgeOverlayAudioPromotionTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
self.sender_peer_hash = "22" * 16
|
|
self.original_rns_packet = RNS.Packet
|
|
|
|
def tearDown(self):
|
|
RNS.Packet = self.original_rns_packet
|
|
|
|
def drain_audio_queue(self):
|
|
while True:
|
|
try:
|
|
self.bridge._audio_binary_out_queue.get_nowait()
|
|
except queue.Empty:
|
|
return
|
|
|
|
def group_audio_wire(self):
|
|
room = b"gcall-qortal-1"
|
|
sender_hash = bytes.fromhex(self.sender_peer_hash)
|
|
payload = b"opus"
|
|
return (
|
|
self.bridge._GROUP_AUDIO_BINARY_MAGIC
|
|
+ bytes(
|
|
(
|
|
self.bridge._GROUP_AUDIO_BINARY_VERSION,
|
|
len(room),
|
|
len(sender_hash),
|
|
)
|
|
)
|
|
+ len(payload).to_bytes(2, "big")
|
|
+ room
|
|
+ sender_hash
|
|
+ payload
|
|
)
|
|
|
|
def group_audio_heartbeat_wire(self):
|
|
return self.bridge.json.dumps(
|
|
{
|
|
"t": self.bridge._GROUP_AUDIO_HEARTBEAT_WIRE_TYPE,
|
|
"R": "gcall-qortal-1",
|
|
"c": "PING",
|
|
"m": int(time.time() * 1000),
|
|
"r": self.sender_peer_hash,
|
|
}
|
|
).encode("utf-8")
|
|
|
|
def group_audio_heartbeat_wire_without_sender(self):
|
|
return self.bridge.json.dumps(
|
|
{
|
|
"t": self.bridge._GROUP_AUDIO_HEARTBEAT_WIRE_TYPE,
|
|
"R": "gcall-qortal-1",
|
|
"c": "PING",
|
|
"m": int(time.time() * 1000),
|
|
}
|
|
).encode("utf-8")
|
|
|
|
def qchat_file_auth_wire(self, transfer_id="transfer-1", peer_hash=None):
|
|
return self.bridge.json.dumps(
|
|
{
|
|
"type": "QCHAT_FILE_LINK_AUTH",
|
|
"transferId": transfer_id,
|
|
"senderAddress": "Q-sender",
|
|
"downloaderAddress": "Q-downloader",
|
|
"downloaderPublicKey": "pub-downloader",
|
|
"downloaderReticulumDestinationHash": peer_hash
|
|
or self.sender_peer_hash,
|
|
"downloaderReticulumIdentityPublicKeyBase64": "identity",
|
|
"timestamp": int(time.time() * 1000),
|
|
"signature": "sig",
|
|
}
|
|
).encode("utf-8")
|
|
|
|
def install_overlay_state(self, incoming=True):
|
|
link = FakeLink()
|
|
link_id = "overlay-test-link"
|
|
peer_hash = "11" * 16
|
|
now = time.time()
|
|
self.bridge._overlay_links_by_id[link_id] = {
|
|
"link": link,
|
|
"peerPresenceHash": peer_hash,
|
|
"incoming": incoming,
|
|
"established": True,
|
|
"established_at": now,
|
|
"created_at": now,
|
|
"pending_packets": self.bridge.deque(maxlen=4),
|
|
"last_activity_at": now,
|
|
"last_rx_at": None,
|
|
}
|
|
self.bridge._overlay_link_ids_by_object[id(link)] = link_id
|
|
self.bridge._active_overlay_link_id_by_peer_hash[peer_hash] = link_id
|
|
if incoming:
|
|
self.bridge._inbound_overlay_neighbors[peer_hash] = now
|
|
else:
|
|
self.bridge._active_overlay_neighbors[peer_hash] = now
|
|
return link, link_id, peer_hash
|
|
|
|
def install_audio_state(
|
|
self,
|
|
link_id,
|
|
peer_hash=None,
|
|
established=True,
|
|
link=None,
|
|
last_activity_at=None,
|
|
):
|
|
peer_hash = peer_hash or self.sender_peer_hash
|
|
link = link or FakeLink()
|
|
now = time.time()
|
|
self.bridge._audio_links_by_id[link_id] = {
|
|
"link": link,
|
|
"peerPresenceHash": peer_hash,
|
|
"peerDestinationHash": peer_hash,
|
|
"incoming": False,
|
|
"established": established,
|
|
"established_at": now if established else None,
|
|
"created_at": now - 10,
|
|
"last_activity_at": last_activity_at if last_activity_at is not None else now,
|
|
"last_rx_at": None,
|
|
"last_send_ok_at": None,
|
|
"send_lock": self.bridge.threading.RLock(),
|
|
"generation": 0,
|
|
"closing": False,
|
|
}
|
|
self.bridge._audio_link_ids_by_object[id(link)] = link_id
|
|
return link
|
|
|
|
def drain_json_events(self):
|
|
events = []
|
|
for event_queue in (
|
|
self.bridge._json_priority_event_queue,
|
|
self.bridge._json_event_queue,
|
|
):
|
|
while True:
|
|
try:
|
|
frame = event_queue.get_nowait()
|
|
if frame is not None:
|
|
events.append(frame)
|
|
except queue.Empty:
|
|
break
|
|
while True:
|
|
frame = self.bridge._pop_coalesced_json_event_line()
|
|
if frame is None:
|
|
return events
|
|
events.append(frame)
|
|
|
|
def drain_json_responses(self):
|
|
responses = []
|
|
while True:
|
|
try:
|
|
responses.append(self.bridge._json_resp_queue.get_nowait())
|
|
except queue.Empty:
|
|
return responses
|
|
|
|
def install_fake_rns_packet(self):
|
|
FakeRnsPacket.sent_links = []
|
|
FakeRnsPacket.sent_payloads = []
|
|
RNS.Packet = FakeRnsPacket
|
|
self.bridge.RNS.Packet = FakeRnsPacket
|
|
self.bridge._destination = FakeDestination()
|
|
|
|
def test_incoming_overlay_group_audio_promotes_link_without_teardown(self):
|
|
self.drain_audio_queue()
|
|
link, overlay_link_id, overlay_peer_hash = self.install_overlay_state(
|
|
incoming=True
|
|
)
|
|
packet = FakePacket(link)
|
|
self.bridge._known_peers[self.sender_peer_hash] = object()
|
|
self.bridge._audio_link_desired_by_peer_hash[self.sender_peer_hash] = {
|
|
"desired": True,
|
|
}
|
|
|
|
self.bridge.on_overlay_link_packet(self.group_audio_wire(), packet)
|
|
|
|
self.assertNotIn(overlay_link_id, self.bridge._overlay_links_by_id)
|
|
self.assertNotIn(id(link), self.bridge._overlay_link_ids_by_object)
|
|
self.assertNotIn(
|
|
overlay_peer_hash,
|
|
self.bridge._active_overlay_link_id_by_peer_hash,
|
|
)
|
|
self.assertNotIn(overlay_peer_hash, self.bridge._inbound_overlay_neighbors)
|
|
self.assertFalse(link.teardown_called)
|
|
|
|
audio_link_id = self.bridge.get_audio_link_id(link)
|
|
self.assertIsInstance(audio_link_id, str)
|
|
audio_state = self.bridge.get_audio_link_state(audio_link_id)
|
|
self.assertIsNotNone(audio_state)
|
|
self.assertTrue(audio_state["incoming"])
|
|
self.assertEqual(audio_state["peerPresenceHash"], self.sender_peer_hash)
|
|
self.assertEqual(audio_state["peerDestinationHash"], self.sender_peer_hash)
|
|
self.assertEqual(audio_state["promoted_from_overlay_link_id"], overlay_link_id)
|
|
self.assertIs(link.packet_callback, self.bridge.on_audio_link_packet)
|
|
self.assertGreater(self.bridge._audio_binary_out_queue.qsize(), 0)
|
|
|
|
def test_incoming_overlay_group_audio_without_desired_audio_is_not_promoted(self):
|
|
self.drain_audio_queue()
|
|
link, overlay_link_id, _peer_hash = self.install_overlay_state(incoming=True)
|
|
packet = FakePacket(link)
|
|
self.bridge._known_peers[self.sender_peer_hash] = object()
|
|
|
|
self.bridge.on_overlay_link_packet(self.group_audio_wire(), packet)
|
|
|
|
self.assertIn(overlay_link_id, self.bridge._overlay_links_by_id)
|
|
self.assertIsNone(self.bridge.get_audio_link_id(link))
|
|
self.assertFalse(link.teardown_called)
|
|
self.assertEqual(self.bridge._audio_binary_out_queue.qsize(), 0)
|
|
|
|
def test_incoming_overlay_gac_promotes_link_when_audio_is_desired(self):
|
|
self.drain_audio_queue()
|
|
link, overlay_link_id, _overlay_peer_hash = self.install_overlay_state(
|
|
incoming=True
|
|
)
|
|
packet = FakePacket(link)
|
|
self.bridge._known_peers[self.sender_peer_hash] = object()
|
|
self.bridge._audio_link_desired_by_peer_hash[self.sender_peer_hash] = {
|
|
"desired": True,
|
|
}
|
|
|
|
self.bridge.on_overlay_link_packet(self.group_audio_heartbeat_wire(), packet)
|
|
|
|
self.assertNotIn(overlay_link_id, self.bridge._overlay_links_by_id)
|
|
audio_link_id = self.bridge.get_audio_link_id(link)
|
|
self.assertIsInstance(audio_link_id, str)
|
|
audio_state = self.bridge.get_audio_link_state(audio_link_id)
|
|
self.assertIsNotNone(audio_state)
|
|
self.assertTrue(audio_state["incoming"])
|
|
self.assertEqual(audio_state["peerPresenceHash"], self.sender_peer_hash)
|
|
self.assertEqual(audio_state["peerDestinationHash"], self.sender_peer_hash)
|
|
self.assertIs(link.packet_callback, self.bridge.on_audio_link_packet)
|
|
|
|
def test_incoming_overlay_gac_without_desired_audio_is_not_promoted(self):
|
|
self.drain_audio_queue()
|
|
link, overlay_link_id, _peer_hash = self.install_overlay_state(incoming=True)
|
|
packet = FakePacket(link)
|
|
self.bridge._known_peers[self.sender_peer_hash] = object()
|
|
|
|
self.bridge.on_overlay_link_packet(self.group_audio_heartbeat_wire(), packet)
|
|
|
|
self.assertIn(overlay_link_id, self.bridge._overlay_links_by_id)
|
|
self.assertIsNone(self.bridge.get_audio_link_id(link))
|
|
self.assertFalse(link.teardown_called)
|
|
|
|
def test_incoming_overlay_gac_without_sender_is_not_promoted(self):
|
|
self.drain_audio_queue()
|
|
link, overlay_link_id, _peer_hash = self.install_overlay_state(incoming=True)
|
|
packet = FakePacket(link)
|
|
self.bridge._known_peers[self.sender_peer_hash] = object()
|
|
self.bridge._audio_link_desired_by_peer_hash[self.sender_peer_hash] = {
|
|
"desired": True,
|
|
}
|
|
|
|
self.bridge.on_overlay_link_packet(
|
|
self.group_audio_heartbeat_wire_without_sender(),
|
|
packet,
|
|
)
|
|
|
|
self.assertIn(overlay_link_id, self.bridge._overlay_links_by_id)
|
|
self.assertIsNone(self.bridge.get_audio_link_id(link))
|
|
self.assertFalse(link.teardown_called)
|
|
|
|
def test_stale_audio_mapping_does_not_allow_overlay_promotion(self):
|
|
self.drain_audio_queue()
|
|
link, overlay_link_id, _peer_hash = self.install_overlay_state(incoming=True)
|
|
packet = FakePacket(link)
|
|
self.bridge._known_peers[self.sender_peer_hash] = object()
|
|
self.bridge._active_audio_link_id_by_peer_hash[self.sender_peer_hash] = "stale-link"
|
|
|
|
self.bridge.on_overlay_link_packet(self.group_audio_wire(), packet)
|
|
|
|
self.assertIn(overlay_link_id, self.bridge._overlay_links_by_id)
|
|
self.assertIsNone(self.bridge.get_audio_link_id(link))
|
|
self.assertFalse(link.teardown_called)
|
|
self.assertEqual(self.bridge._audio_binary_out_queue.qsize(), 0)
|
|
|
|
def test_audio_send_with_stale_link_id_uses_established_peer_link(self):
|
|
self.install_fake_rns_packet()
|
|
current_link = self.install_audio_state("current-audio-link")
|
|
self.bridge._active_audio_link_id_by_peer_hash[self.sender_peer_hash] = "stale-audio-link"
|
|
self.bridge._outgoing_audio_link_id_by_peer_hash[self.sender_peer_hash] = "stale-audio-link"
|
|
|
|
self.bridge._process_audio_batch(
|
|
[
|
|
(
|
|
"stale-audio-link",
|
|
"gcall-qortal-1",
|
|
self.sender_peer_hash,
|
|
"",
|
|
int(time.time() * 1000),
|
|
b"opus",
|
|
)
|
|
]
|
|
)
|
|
|
|
self.assertEqual(FakeRnsPacket.sent_links, [current_link])
|
|
self.assertEqual(
|
|
self.bridge._active_audio_link_id_by_peer_hash[self.sender_peer_hash],
|
|
"current-audio-link",
|
|
)
|
|
failures = [
|
|
frame
|
|
for frame in self.drain_json_events()
|
|
if frame.get("event") == "group_audio_send_failed"
|
|
]
|
|
self.assertEqual(failures, [])
|
|
|
|
def test_audio_heartbeat_with_stale_link_id_uses_established_peer_link(self):
|
|
self.install_fake_rns_packet()
|
|
current_link = self.install_audio_state("current-audio-link")
|
|
self.bridge._active_audio_link_id_by_peer_hash[self.sender_peer_hash] = "stale-audio-link"
|
|
self.bridge._outgoing_audio_link_id_by_peer_hash[self.sender_peer_hash] = "stale-audio-link"
|
|
|
|
self.bridge.handle_send_group_audio_link_heartbeat(
|
|
"req-1",
|
|
{
|
|
"linkId": "stale-audio-link",
|
|
"peerPresenceHash": self.sender_peer_hash,
|
|
"roomId": "gcall-qortal-1",
|
|
"command": "PING",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(FakeRnsPacket.sent_links, [current_link])
|
|
responses = self.drain_json_responses()
|
|
self.assertEqual(len(responses), 1)
|
|
self.assertTrue(responses[0].get("ok"))
|
|
self.assertEqual(
|
|
responses[0].get("payload", {}).get("linkId"),
|
|
"current-audio-link",
|
|
)
|
|
|
|
def test_audio_rtt_probe_uses_established_audio_link(self):
|
|
self.install_fake_rns_packet()
|
|
current_link = self.install_audio_state("current-audio-link")
|
|
state = self.bridge.get_audio_link_state("current-audio-link")
|
|
self.bridge._ensure_audio_link_lifecycle_fields(state)
|
|
|
|
self.bridge._process_audio_rtt_probe("current-audio-link", 0)
|
|
|
|
self.assertEqual(FakeRnsPacket.sent_links, [current_link])
|
|
wire = json.loads(FakeRnsPacket.sent_payloads[0].decode("utf-8"))
|
|
self.assertEqual(wire.get("t"), self.bridge._GROUP_AUDIO_RTT_WIRE_TYPE)
|
|
self.assertEqual(wire.get("c"), self.bridge._GROUP_AUDIO_RTT_PROBE_COMMAND)
|
|
self.assertRegex(str(wire.get("q") or ""), r"^[0-9a-f]{16}$")
|
|
self.assertIn(wire["q"], state["rtt_pending"])
|
|
|
|
def test_audio_rtt_ack_resolves_probe_with_monotonic_clock(self):
|
|
self.install_audio_state("current-audio-link")
|
|
state = self.bridge.get_audio_link_state("current-audio-link")
|
|
self.bridge._ensure_audio_link_lifecycle_fields(state)
|
|
probe_id = "a1" * 8
|
|
state["rtt_pending"][probe_id] = {"sent_ns": 1_000_000_000}
|
|
|
|
with mock.patch.object(
|
|
self.bridge.time,
|
|
"monotonic_ns",
|
|
return_value=1_025_000_000,
|
|
):
|
|
rtt_ms = self.bridge._resolve_audio_rtt_probe(
|
|
"current-audio-link",
|
|
state,
|
|
probe_id,
|
|
)
|
|
|
|
self.assertEqual(rtt_ms, 25.0)
|
|
self.assertEqual(state.get("rtt_latest_ms"), 25.0)
|
|
self.assertEqual(state.get("rtt_median_ms"), 25.0)
|
|
self.assertNotIn(probe_id, state["rtt_pending"])
|
|
|
|
def test_audio_rtt_probe_is_acknowledged_inside_bridge(self):
|
|
self.install_fake_rns_packet()
|
|
current_link = self.install_audio_state("current-audio-link")
|
|
probe_id = "b2" * 8
|
|
wire = json.dumps(
|
|
{
|
|
"t": self.bridge._GROUP_AUDIO_RTT_WIRE_TYPE,
|
|
"c": self.bridge._GROUP_AUDIO_RTT_PROBE_COMMAND,
|
|
"q": probe_id,
|
|
"r": self.sender_peer_hash,
|
|
}
|
|
).encode("utf-8")
|
|
|
|
def run_immediately(_lane, _name, func, *args, **kwargs):
|
|
func(*args, **kwargs)
|
|
return True
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_enqueue_scheduler_task",
|
|
side_effect=run_immediately,
|
|
):
|
|
self.bridge.on_audio_link_packet(wire, FakePacket(current_link))
|
|
|
|
self.assertEqual(FakeRnsPacket.sent_links, [current_link])
|
|
ack = json.loads(FakeRnsPacket.sent_payloads[0].decode("utf-8"))
|
|
self.assertEqual(ack.get("t"), self.bridge._GROUP_AUDIO_RTT_WIRE_TYPE)
|
|
self.assertEqual(ack.get("c"), self.bridge._GROUP_AUDIO_RTT_ACK_COMMAND)
|
|
self.assertEqual(ack.get("q"), probe_id)
|
|
|
|
def test_audio_rtt_commands_do_not_intercept_call_heartbeat(self):
|
|
current_link = self.install_audio_state("current-audio-link")
|
|
heartbeat = json.dumps(
|
|
{
|
|
"t": self.bridge._GROUP_AUDIO_HEARTBEAT_WIRE_TYPE,
|
|
"R": "gcall-qortal-1",
|
|
"c": "PING",
|
|
"m": int(time.time() * 1000),
|
|
"r": self.sender_peer_hash,
|
|
}
|
|
).encode("utf-8")
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_emit_call_bridge_message",
|
|
return_value=True,
|
|
) as emit_call_message:
|
|
self.bridge.on_audio_link_packet(heartbeat, FakePacket(current_link))
|
|
|
|
emit_call_message.assert_called_once()
|
|
|
|
def test_removing_audio_link_discards_pending_rtt_probe(self):
|
|
self.install_audio_state("current-audio-link")
|
|
state = self.bridge.get_audio_link_state("current-audio-link")
|
|
self.bridge._ensure_audio_link_lifecycle_fields(state)
|
|
state["rtt_probe_queued"] = True
|
|
state["rtt_pending"]["c3" * 8] = {"sent_ns": time.monotonic_ns()}
|
|
|
|
removed = self.bridge.remove_audio_link("current-audio-link")
|
|
|
|
self.assertIsNotNone(removed)
|
|
self.assertFalse(removed.get("rtt_probe_queued"))
|
|
self.assertEqual(removed.get("rtt_pending"), {})
|
|
|
|
def test_audio_open_stops_after_max_establish_attempts(self):
|
|
self.bridge._destination = FakeDestination()
|
|
self.bridge._audio_link_desired_by_peer_hash[self.sender_peer_hash] = {
|
|
"desired": True,
|
|
"attempts": self.bridge._AUDIO_LINK_MAX_ESTABLISH_ATTEMPTS,
|
|
"retry_delay": self.bridge._AUDIO_LINK_RETRY_MIN_SECONDS,
|
|
"retry_timer": None,
|
|
"last_failure_reason": "establish_timeout",
|
|
}
|
|
|
|
ok, payload, error = self.bridge._open_group_audio_link_for_peer(
|
|
self.sender_peer_hash,
|
|
retry_reason="establish_timeout",
|
|
)
|
|
|
|
self.assertFalse(ok)
|
|
self.assertEqual(payload.get("code"), "max_establish_attempts")
|
|
self.assertEqual(error, "Max group audio link establish attempts reached")
|
|
events = [
|
|
frame
|
|
for frame in self.drain_json_events()
|
|
if frame.get("event") == "group_audio_send_failed"
|
|
]
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(events[0].get("payload", {}).get("reason"), "max_establish_attempts")
|
|
self.assertEqual(events[0].get("payload", {}).get("code"), "max_establish_attempts")
|
|
|
|
def test_audio_retry_not_scheduled_after_max_establish_attempts(self):
|
|
self.bridge._audio_link_desired_by_peer_hash[self.sender_peer_hash] = {
|
|
"desired": True,
|
|
"attempts": self.bridge._AUDIO_LINK_MAX_ESTABLISH_ATTEMPTS,
|
|
"retry_delay": self.bridge._AUDIO_LINK_RETRY_MIN_SECONDS,
|
|
"retry_timer": None,
|
|
"last_failure_reason": "establish_timeout",
|
|
}
|
|
|
|
self.bridge._schedule_audio_link_retry(self.sender_peer_hash, "establish_timeout")
|
|
self.bridge._schedule_audio_link_retry(self.sender_peer_hash, "establish_timeout")
|
|
|
|
desired = self.bridge._audio_link_desired_by_peer_hash[self.sender_peer_hash]
|
|
self.assertIsNone(desired.get("retry_timer"))
|
|
self.assertEqual(desired.get("last_failure_reason"), "max_establish_attempts")
|
|
events = [
|
|
frame
|
|
for frame in self.drain_json_events()
|
|
if frame.get("event") == "group_audio_send_failed"
|
|
]
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(events[0].get("payload", {}).get("reason"), "max_establish_attempts")
|
|
self.assertEqual(events[0].get("payload", {}).get("code"), "max_establish_attempts")
|
|
|
|
def test_audio_retry_timer_callback_does_not_enqueue_after_max_attempts(self):
|
|
enqueued = []
|
|
original_enqueue = self.bridge._enqueue_scheduler_task
|
|
original_timer = self.bridge.threading.Timer
|
|
|
|
class FakeTimer:
|
|
def __init__(self, delay, function):
|
|
self.delay = delay
|
|
self.function = function
|
|
self.daemon = False
|
|
self.started = False
|
|
|
|
def start(self):
|
|
self.started = True
|
|
|
|
def cancel(self):
|
|
self.started = False
|
|
|
|
try:
|
|
self.bridge._enqueue_scheduler_task = lambda *args, **kwargs: enqueued.append(
|
|
(args, kwargs)
|
|
)
|
|
self.bridge.threading.Timer = FakeTimer
|
|
self.bridge._audio_link_desired_by_peer_hash[self.sender_peer_hash] = {
|
|
"desired": True,
|
|
"attempts": self.bridge._AUDIO_LINK_MAX_ESTABLISH_ATTEMPTS - 1,
|
|
"retry_delay": self.bridge._AUDIO_LINK_RETRY_MIN_SECONDS,
|
|
"retry_timer": None,
|
|
"last_failure_reason": "establish_timeout",
|
|
}
|
|
|
|
self.bridge._schedule_audio_link_retry(
|
|
self.sender_peer_hash,
|
|
"establish_timeout",
|
|
immediate=True,
|
|
)
|
|
desired = self.bridge._audio_link_desired_by_peer_hash[self.sender_peer_hash]
|
|
timer = desired.get("retry_timer")
|
|
self.assertIsNotNone(timer)
|
|
desired["attempts"] = self.bridge._AUDIO_LINK_MAX_ESTABLISH_ATTEMPTS
|
|
|
|
timer.function()
|
|
|
|
self.assertEqual(enqueued, [])
|
|
events = [
|
|
frame
|
|
for frame in self.drain_json_events()
|
|
if frame.get("event") == "group_audio_send_failed"
|
|
]
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(
|
|
events[0].get("payload", {}).get("code"),
|
|
"max_establish_attempts",
|
|
)
|
|
finally:
|
|
self.bridge._enqueue_scheduler_task = original_enqueue
|
|
self.bridge.threading.Timer = original_timer
|
|
|
|
def test_audio_established_resets_establish_attempts(self):
|
|
link = self.install_audio_state("current-audio-link", established=False)
|
|
self.bridge._audio_link_desired_by_peer_hash[self.sender_peer_hash] = {
|
|
"desired": True,
|
|
"attempts": self.bridge._AUDIO_LINK_MAX_ESTABLISH_ATTEMPTS,
|
|
"retry_delay": self.bridge._AUDIO_LINK_RETRY_MAX_SECONDS,
|
|
"retry_timer": None,
|
|
"last_failure_reason": "establish_timeout",
|
|
"max_attempts_emitted": True,
|
|
}
|
|
|
|
self.bridge.on_outgoing_audio_link_established(link)
|
|
|
|
desired = self.bridge._audio_link_desired_by_peer_hash[self.sender_peer_hash]
|
|
self.assertEqual(desired.get("attempts"), 0)
|
|
self.assertEqual(desired.get("retry_delay"), self.bridge._AUDIO_LINK_RETRY_MIN_SECONDS)
|
|
self.assertEqual(desired.get("last_failure_reason"), "")
|
|
self.assertFalse(desired.get("max_attempts_emitted"))
|
|
|
|
def test_outbound_overlay_group_audio_is_not_promoted(self):
|
|
self.drain_audio_queue()
|
|
link, overlay_link_id, _peer_hash = self.install_overlay_state(incoming=False)
|
|
packet = FakePacket(link)
|
|
self.bridge._known_peers[self.sender_peer_hash] = object()
|
|
self.bridge._audio_link_desired_by_peer_hash[self.sender_peer_hash] = {
|
|
"desired": True,
|
|
}
|
|
|
|
self.bridge.on_overlay_link_packet(self.group_audio_wire(), packet)
|
|
|
|
self.assertIn(overlay_link_id, self.bridge._overlay_links_by_id)
|
|
self.assertIsNone(self.bridge.get_audio_link_id(link))
|
|
self.assertFalse(link.teardown_called)
|
|
self.assertEqual(self.bridge._audio_binary_out_queue.qsize(), 0)
|
|
|
|
def test_incoming_overlay_qchat_auth_promotes_when_transfer_is_pending(self):
|
|
link, overlay_link_id, overlay_peer_hash = self.install_overlay_state(
|
|
incoming=True
|
|
)
|
|
packet = FakePacket(link)
|
|
self.bridge._qchat_file_pending_sends_by_transfer["transfer-1"] = {
|
|
"expires_at": time.time() + 60,
|
|
}
|
|
|
|
self.bridge.on_overlay_link_packet(self.qchat_file_auth_wire(), packet)
|
|
|
|
self.assertNotIn(overlay_link_id, self.bridge._overlay_links_by_id)
|
|
self.assertNotIn(id(link), self.bridge._overlay_link_ids_by_object)
|
|
self.assertNotIn(
|
|
overlay_peer_hash,
|
|
self.bridge._active_overlay_link_id_by_peer_hash,
|
|
)
|
|
self.assertNotIn(overlay_peer_hash, self.bridge._inbound_overlay_neighbors)
|
|
self.assertFalse(link.teardown_called)
|
|
|
|
file_link_id = self.bridge.get_qchat_file_link_id(link)
|
|
self.assertIsInstance(file_link_id, str)
|
|
file_state = self.bridge.get_qchat_file_link_state(file_link_id)
|
|
self.assertIsNotNone(file_state)
|
|
self.assertTrue(file_state["incoming"])
|
|
self.assertEqual(file_state["peerPresenceHash"], self.sender_peer_hash)
|
|
self.assertEqual(file_state["transferId"], "transfer-1")
|
|
self.assertIs(link.packet_callback, self.bridge.on_qchat_file_link_packet)
|
|
|
|
def test_incoming_overlay_qchat_auth_without_pending_transfer_is_not_promoted(self):
|
|
link, overlay_link_id, _peer_hash = self.install_overlay_state(incoming=True)
|
|
packet = FakePacket(link)
|
|
|
|
self.bridge.on_overlay_link_packet(self.qchat_file_auth_wire(), packet)
|
|
|
|
self.assertIn(overlay_link_id, self.bridge._overlay_links_by_id)
|
|
self.assertIsNone(self.bridge.get_qchat_file_link_id(link))
|
|
self.assertFalse(link.teardown_called)
|
|
|
|
def test_incoming_overlay_qchat_auth_with_expired_transfer_is_not_promoted(self):
|
|
link, overlay_link_id, _peer_hash = self.install_overlay_state(incoming=True)
|
|
packet = FakePacket(link)
|
|
self.bridge._qchat_file_pending_sends_by_transfer["transfer-1"] = {
|
|
"expires_at": time.time() - 1,
|
|
}
|
|
|
|
self.bridge.on_overlay_link_packet(self.qchat_file_auth_wire(), packet)
|
|
|
|
self.assertIn(overlay_link_id, self.bridge._overlay_links_by_id)
|
|
self.assertIsNone(self.bridge.get_qchat_file_link_id(link))
|
|
self.assertFalse(link.teardown_called)
|
|
|
|
def test_incoming_overlay_qchat_auth_with_invalid_peer_hash_is_not_promoted(self):
|
|
link, overlay_link_id, _peer_hash = self.install_overlay_state(incoming=True)
|
|
packet = FakePacket(link)
|
|
self.bridge._qchat_file_pending_sends_by_transfer["transfer-1"] = {
|
|
"expires_at": time.time() + 60,
|
|
}
|
|
|
|
self.bridge.on_overlay_link_packet(
|
|
self.qchat_file_auth_wire(peer_hash="not-a-hash"),
|
|
packet,
|
|
)
|
|
|
|
self.assertIn(overlay_link_id, self.bridge._overlay_links_by_id)
|
|
self.assertIsNone(self.bridge.get_qchat_file_link_id(link))
|
|
self.assertFalse(link.teardown_called)
|
|
|
|
def test_outbound_overlay_qchat_auth_is_not_promoted(self):
|
|
link, overlay_link_id, _peer_hash = self.install_overlay_state(incoming=False)
|
|
packet = FakePacket(link)
|
|
self.bridge._qchat_file_pending_sends_by_transfer["transfer-1"] = {
|
|
"expires_at": time.time() + 60,
|
|
}
|
|
|
|
self.bridge.on_overlay_link_packet(self.qchat_file_auth_wire(), packet)
|
|
|
|
self.assertIn(overlay_link_id, self.bridge._overlay_links_by_id)
|
|
self.assertIsNone(self.bridge.get_qchat_file_link_id(link))
|
|
self.assertFalse(link.teardown_called)
|
|
|
|
|
|
class PresenceBridgeOverlayRouteMigrationTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
self.bridge._destination = FakeDestination()
|
|
self.peer_hash = "ab" * 16
|
|
self.active_link = FakeLink()
|
|
self.active_link_id = "active-overlay"
|
|
self.active_state = {
|
|
"linkId": self.active_link_id,
|
|
"link": self.active_link,
|
|
"peerPresenceHash": self.peer_hash,
|
|
"incoming": False,
|
|
"established": True,
|
|
"established_at": time.time() - 120,
|
|
"created_at": time.time() - 120,
|
|
"overlay_transport_admitted": True,
|
|
"manager_kind": "overlay",
|
|
"manager_state": self.bridge._LINK_STATE_ESTABLISHED,
|
|
"generation": 0,
|
|
"peer_capabilities": {
|
|
self.bridge._OVERLAY_ROUTE_MIGRATION_CAPABILITY,
|
|
},
|
|
}
|
|
self.bridge._overlay_links_by_id[self.active_link_id] = self.active_state
|
|
self.bridge._overlay_link_ids_by_object[id(self.active_link)] = self.active_link_id
|
|
self.bridge._active_overlay_link_id_by_peer_hash[self.peer_hash] = self.active_link_id
|
|
|
|
def test_migration_uses_real_bounded_scheduler_lanes(self):
|
|
for shard in range(self.bridge._SCHEDULER_OVERLAY_MIGRATION_SHARDS):
|
|
lane = f"overlay-migration-{shard}"
|
|
self.assertIn(lane, self.bridge._SCHEDULER_QUEUE_MAX_BY_LANE)
|
|
self.assertGreater(self.bridge._SCHEDULER_QUEUE_MAX_BY_LANE[lane], 0)
|
|
|
|
def test_hello_advertises_route_migration_and_marks_only_candidates(self):
|
|
regular = json.loads(
|
|
self.bridge._make_overlay_transport_wire(
|
|
self.bridge._OVERLAY_HELLO_WIRE_TYPE,
|
|
).decode("utf-8")
|
|
)
|
|
candidate = json.loads(
|
|
self.bridge._make_overlay_transport_wire(
|
|
self.bridge._OVERLAY_HELLO_WIRE_TYPE,
|
|
migration_candidate=True,
|
|
).decode("utf-8")
|
|
)
|
|
|
|
self.assertIn(
|
|
self.bridge._OVERLAY_ROUTE_MIGRATION_CAPABILITY,
|
|
regular["c"],
|
|
)
|
|
self.assertNotIn("m", regular)
|
|
self.assertEqual(
|
|
candidate["m"],
|
|
self.bridge._OVERLAY_ROUTE_MIGRATION_MARKER,
|
|
)
|
|
|
|
def test_rtt_probe_resolution_requires_the_exact_nonce(self):
|
|
event = threading.Event()
|
|
state = {
|
|
"rtt_pending": {
|
|
"11" * 8: {
|
|
"sent_ns": 1_000_000_000,
|
|
"event": event,
|
|
"rtt_ms": None,
|
|
}
|
|
}
|
|
}
|
|
|
|
with mock.patch.object(
|
|
self.bridge.time,
|
|
"monotonic_ns",
|
|
return_value=1_025_000_000,
|
|
):
|
|
self.assertIsNone(
|
|
self.bridge._resolve_overlay_rtt_probe(state, "22" * 8)
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._resolve_overlay_rtt_probe(state, "11" * 8),
|
|
25.0,
|
|
)
|
|
|
|
self.assertTrue(event.is_set())
|
|
self.assertEqual(list(state["rtt_samples_ms"]), [25.0])
|
|
|
|
def test_ping_echoes_the_same_rtt_nonce(self):
|
|
nonce = "33" * 8
|
|
wire = {
|
|
"t": self.bridge._OVERLAY_PING_WIRE_TYPE,
|
|
"r": self.peer_hash,
|
|
"q": nonce,
|
|
}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_admit_overlay_peer_from_transport",
|
|
return_value=True,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_send_overlay_transport_control",
|
|
return_value=True,
|
|
) as send, mock.patch.object(
|
|
self.bridge,
|
|
"_overlay_enqueue_dedup",
|
|
):
|
|
handled = self.bridge._handle_overlay_transport_control(
|
|
wire,
|
|
self.active_link,
|
|
self.active_link_id,
|
|
self.active_state,
|
|
)
|
|
|
|
self.assertTrue(handled)
|
|
self.assertEqual(
|
|
send.call_args.kwargs["correlation_id"],
|
|
nonce,
|
|
)
|
|
|
|
def test_quality_gate_requires_lower_hops_and_clear_rtt_gain(self):
|
|
accepted, result = self.bridge._overlay_migration_quality_acceptable(
|
|
[120, 125, 130, 135, 140],
|
|
[55, 60, 65, 70, 75],
|
|
8,
|
|
3,
|
|
)
|
|
self.assertTrue(accepted)
|
|
self.assertEqual(result["active_median_ms"], 130.0)
|
|
self.assertEqual(result["candidate_median_ms"], 65.0)
|
|
|
|
cases = (
|
|
([120] * 5, [60] * 5, 4, 4),
|
|
([100] * 5, [85] * 5, 4, 2),
|
|
([120] * 5, [60] * 3, 4, 2),
|
|
)
|
|
for active, candidate, active_hops, candidate_hops in cases:
|
|
with self.subTest(
|
|
active_hops=active_hops,
|
|
candidate_hops=candidate_hops,
|
|
candidate_samples=len(candidate),
|
|
):
|
|
accepted, _result = self.bridge._overlay_migration_quality_acceptable(
|
|
active,
|
|
candidate,
|
|
active_hops,
|
|
candidate_hops,
|
|
)
|
|
self.assertFalse(accepted)
|
|
|
|
def test_rtt_rounds_alternate_probe_order(self):
|
|
calls = []
|
|
|
|
def probe(link_id, purpose):
|
|
calls.append((link_id, purpose))
|
|
event = threading.Event()
|
|
event.set()
|
|
return {
|
|
"event": event,
|
|
"rtt_ms": 100.0 if link_id == self.active_link_id else 40.0,
|
|
}
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_send_overlay_rtt_probe",
|
|
side_effect=probe,
|
|
):
|
|
active, candidate = self.bridge._collect_overlay_migration_rtt_samples(
|
|
self.active_link_id,
|
|
"candidate-overlay",
|
|
)
|
|
|
|
self.assertEqual(
|
|
[link_id for link_id, _purpose in calls[:4]],
|
|
[
|
|
self.active_link_id,
|
|
"candidate-overlay",
|
|
"candidate-overlay",
|
|
self.active_link_id,
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
len(active),
|
|
self.bridge._OVERLAY_ROUTE_MIGRATION_PROBE_SAMPLES,
|
|
)
|
|
self.assertEqual(
|
|
len(candidate),
|
|
self.bridge._OVERLAY_ROUTE_MIGRATION_PROBE_SAMPLES,
|
|
)
|
|
|
|
def test_only_active_link_initiator_schedules_a_better_route_probe(self):
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_reticulum_path_snapshot",
|
|
return_value={"has_path": True, "hops": 2},
|
|
) as path_snapshot, mock.patch.object(
|
|
self.bridge,
|
|
"_reticulum_link_route_snapshot",
|
|
return_value={"remote_hops": 7},
|
|
) as route_snapshot, mock.patch.object(
|
|
self.bridge,
|
|
"_enqueue_scheduler_task",
|
|
return_value=True,
|
|
) as enqueue:
|
|
self.assertTrue(
|
|
self.bridge._maybe_schedule_overlay_route_migration(
|
|
self.peer_hash,
|
|
"test_announce",
|
|
)
|
|
)
|
|
|
|
self.assertEqual(enqueue.call_count, 1)
|
|
self.assertIs(
|
|
enqueue.call_args.args[2],
|
|
self.bridge._overlay_route_migration_inspection_job,
|
|
)
|
|
path_snapshot.assert_not_called()
|
|
route_snapshot.assert_not_called()
|
|
self.assertIn(
|
|
self.peer_hash,
|
|
self.bridge._overlay_route_migration_pending_by_peer_hash,
|
|
)
|
|
|
|
self.bridge._overlay_route_migration_pending_by_peer_hash.clear()
|
|
self.bridge._overlay_route_migration_last_attempt_at_by_peer_hash.clear()
|
|
self.active_state["incoming"] = True
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_enqueue_scheduler_task",
|
|
return_value=True,
|
|
) as enqueue:
|
|
self.assertFalse(
|
|
self.bridge._maybe_schedule_overlay_route_migration(
|
|
self.peer_hash,
|
|
"test_announce",
|
|
)
|
|
)
|
|
enqueue.assert_not_called()
|
|
|
|
def test_route_inspection_starts_migration_only_for_a_better_path(self):
|
|
self.bridge._overlay_route_migration_pending_by_peer_hash.add(self.peer_hash)
|
|
target_path = {"has_path": True, "hops": 2}
|
|
active_route = {"remote_hops": 7}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_reticulum_path_snapshot",
|
|
return_value=target_path,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_reticulum_link_route_snapshot",
|
|
return_value=active_route,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_overlay_route_migration_job",
|
|
) as migrate:
|
|
self.bridge._overlay_route_migration_inspection_job(
|
|
self.peer_hash,
|
|
self.active_link_id,
|
|
"test_announce",
|
|
)
|
|
|
|
migrate.assert_called_once_with(
|
|
self.peer_hash,
|
|
self.active_link_id,
|
|
target_path,
|
|
active_route,
|
|
)
|
|
self.assertIn(
|
|
self.peer_hash,
|
|
self.bridge._overlay_route_migration_last_attempt_at_by_peer_hash,
|
|
)
|
|
self.assertNotIn(
|
|
self.peer_hash,
|
|
self.bridge._overlay_route_migration_pending_by_peer_hash,
|
|
)
|
|
|
|
def test_dedup_does_not_select_or_close_a_migration_candidate(self):
|
|
candidate_link_id = "candidate-overlay"
|
|
candidate_state = dict(self.active_state)
|
|
candidate_state.update(
|
|
{
|
|
"linkId": candidate_link_id,
|
|
"link": FakeLink(),
|
|
"migration_candidate": True,
|
|
}
|
|
)
|
|
self.bridge._overlay_links_by_id[candidate_link_id] = candidate_state
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_overlay_duplicate_close",
|
|
) as close:
|
|
kept = self.bridge._dedup_overlay_links_for_peer(self.peer_hash)
|
|
|
|
self.assertIs(kept, self.active_state)
|
|
self.assertEqual(
|
|
self.bridge._active_overlay_link_id_by_peer_hash[self.peer_hash],
|
|
self.active_link_id,
|
|
)
|
|
close.assert_not_called()
|
|
|
|
def test_promoted_candidate_keeps_old_link_as_non_reclaiming_backup(self):
|
|
candidate_link_id = "candidate-overlay"
|
|
transaction_id = "44" * 8
|
|
candidate_state = dict(self.active_state)
|
|
candidate_state.update(
|
|
{
|
|
"linkId": candidate_link_id,
|
|
"link": FakeLink(),
|
|
"migration_candidate": True,
|
|
"migration_source_link_id": self.active_link_id,
|
|
"overlay_transport_admitted": True,
|
|
"migration_transaction_id": transaction_id,
|
|
}
|
|
)
|
|
self.bridge._overlay_links_by_id[candidate_link_id] = candidate_state
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"emit_overlay_link_state",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_delayed_presence_announce_replay",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_flush_overlay_link_pending",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_overlay_duplicate_close",
|
|
) as delayed_close:
|
|
self.assertTrue(
|
|
self.bridge._promote_overlay_migration_candidate(
|
|
self.peer_hash,
|
|
candidate_link_id,
|
|
"test_commit",
|
|
)
|
|
)
|
|
|
|
self.assertEqual(
|
|
self.bridge._active_overlay_link_id_by_peer_hash[self.peer_hash],
|
|
candidate_link_id,
|
|
)
|
|
self.assertTrue(self.active_state["migration_draining"])
|
|
delayed_close.assert_not_called()
|
|
self.assertIs(
|
|
self.bridge._dedup_overlay_links_for_peer(self.peer_hash),
|
|
candidate_state,
|
|
)
|
|
delayed_close.assert_not_called()
|
|
|
|
self.assertTrue(
|
|
self.bridge._finalize_overlay_migration(
|
|
self.peer_hash,
|
|
candidate_link_id,
|
|
transaction_id,
|
|
"test_finalize",
|
|
)
|
|
)
|
|
delayed_close.assert_called_once_with(
|
|
self.peer_hash,
|
|
candidate_link_id,
|
|
self.active_link_id,
|
|
"route_migrated",
|
|
)
|
|
|
|
kept = self.bridge._register_active_overlay_for_peer(
|
|
self.peer_hash,
|
|
self.active_link_id,
|
|
)
|
|
self.assertIs(kept, candidate_state)
|
|
self.assertEqual(
|
|
self.bridge._active_overlay_link_id_by_peer_hash[self.peer_hash],
|
|
candidate_link_id,
|
|
)
|
|
|
|
removed = self.bridge.remove_overlay_link(candidate_link_id)
|
|
self.assertIs(removed, candidate_state)
|
|
self.assertEqual(
|
|
self.bridge._active_overlay_link_id_by_peer_hash[self.peer_hash],
|
|
self.active_link_id,
|
|
)
|
|
self.assertNotIn("migration_draining", self.active_state)
|
|
|
|
def test_candidate_send_failure_does_not_refresh_active_peer_path(self):
|
|
state = {
|
|
"linkId": "candidate-overlay",
|
|
"link": FakeLink(),
|
|
"peerPresenceHash": self.peer_hash,
|
|
"migration_candidate": True,
|
|
}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_send_packet_on_link",
|
|
return_value=False,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_overlay_enqueue_close",
|
|
return_value=True,
|
|
) as close, mock.patch.object(
|
|
self.bridge,
|
|
"_force_overlay_peer_path_refresh",
|
|
) as refresh:
|
|
self.assertFalse(
|
|
self.bridge._send_overlay_transport_control(
|
|
state["link"],
|
|
state,
|
|
self.bridge._OVERLAY_PING_WIRE_TYPE,
|
|
"candidate_test",
|
|
)
|
|
)
|
|
|
|
close.assert_called_once_with(
|
|
"candidate-overlay",
|
|
"overlay_transport_packet_send_false",
|
|
)
|
|
refresh.assert_not_called()
|
|
|
|
def test_candidate_teardown_leaves_active_link_and_peer_state_untouched(self):
|
|
candidate_link = FakeLink()
|
|
candidate_link_id = "candidate-overlay"
|
|
self.bridge._overlay_links_by_id[candidate_link_id] = {
|
|
"linkId": candidate_link_id,
|
|
"link": candidate_link,
|
|
"peerPresenceHash": self.peer_hash,
|
|
"migration_candidate": True,
|
|
"manager_kind": "overlay",
|
|
"manager_state": self.bridge._LINK_STATE_CONNECTING,
|
|
"generation": 0,
|
|
}
|
|
self.bridge._overlay_link_ids_by_object[id(candidate_link)] = candidate_link_id
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_run_with_timeout",
|
|
return_value=(True, None, None),
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"emit_overlay_link_state",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_demote_overlay_fanout_peer",
|
|
) as demote, mock.patch.object(
|
|
self.bridge,
|
|
"_overlay_enqueue_peer_recovery",
|
|
) as recovery:
|
|
self.bridge._teardown_overlay_link_id(
|
|
candidate_link_id,
|
|
"overlay_transport_packet_send_false",
|
|
)
|
|
|
|
self.assertEqual(
|
|
self.bridge._active_overlay_link_id_by_peer_hash[self.peer_hash],
|
|
self.active_link_id,
|
|
)
|
|
self.assertIn(self.active_link_id, self.bridge._overlay_links_by_id)
|
|
demote.assert_not_called()
|
|
recovery.assert_not_called()
|
|
|
|
def test_unsolicited_incoming_candidate_is_rejected(self):
|
|
self.bridge._overlay_links_by_id.clear()
|
|
self.bridge._active_overlay_link_id_by_peer_hash.clear()
|
|
incoming = FakeLink()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_enqueue_scheduler_task",
|
|
return_value=True,
|
|
) as enqueue:
|
|
link_id = self.bridge._register_incoming_overlay_link(
|
|
incoming,
|
|
self.peer_hash,
|
|
"test_candidate",
|
|
migration_candidate=True,
|
|
)
|
|
|
|
self.assertEqual(link_id, "")
|
|
self.assertEqual(self.bridge._overlay_links_by_id, {})
|
|
self.assertEqual(enqueue.call_count, 1)
|
|
|
|
def test_capable_active_peer_can_register_protected_incoming_candidate(self):
|
|
self.active_state["incoming"] = True
|
|
self.bridge._inbound_overlay_neighbors[self.peer_hash] = time.time()
|
|
incoming = FakeLink()
|
|
incoming.remote_identity = object()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_send_overlay_hello_for_link",
|
|
) as send_hello, mock.patch.object(
|
|
self.bridge,
|
|
"derive_presence_destination_hash_for_identity",
|
|
return_value=self.peer_hash,
|
|
):
|
|
link_id = self.bridge._register_incoming_overlay_link(
|
|
incoming,
|
|
self.peer_hash,
|
|
"test_candidate",
|
|
migration_candidate=True,
|
|
)
|
|
|
|
self.assertTrue(link_id)
|
|
self.assertTrue(
|
|
self.bridge._overlay_links_by_id[link_id]["migration_candidate"]
|
|
)
|
|
self.assertTrue(
|
|
self.bridge._overlay_links_by_id[link_id]["migration_peer_authenticated"]
|
|
)
|
|
send_hello.assert_called_once_with(link_id, "incoming:test_candidate")
|
|
self.assertEqual(
|
|
self.bridge._active_overlay_link_id_by_peer_hash[self.peer_hash],
|
|
self.active_link_id,
|
|
)
|
|
|
|
def test_unidentified_incoming_candidate_cannot_replace_active_link(self):
|
|
self.active_state["incoming"] = True
|
|
self.bridge._inbound_overlay_neighbors[self.peer_hash] = time.time()
|
|
incoming = FakeLink()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_send_overlay_hello_for_link",
|
|
) as send_hello, mock.patch.object(
|
|
self.bridge,
|
|
"_send_overlay_transport_control",
|
|
return_value=True,
|
|
) as send_control:
|
|
link_id = self.bridge._register_incoming_overlay_link(
|
|
incoming,
|
|
self.peer_hash,
|
|
"test_candidate",
|
|
migration_candidate=True,
|
|
)
|
|
|
|
self.assertTrue(link_id)
|
|
candidate = self.bridge._overlay_links_by_id[link_id]
|
|
self.assertFalse(candidate["migration_peer_authenticated"])
|
|
self.assertFalse(candidate.get("overlay_transport_admitted") is True)
|
|
send_hello.assert_not_called()
|
|
self.assertTrue(
|
|
self.bridge._handle_overlay_transport_control(
|
|
{
|
|
"t": self.bridge._OVERLAY_HELLO_WIRE_TYPE,
|
|
"r": self.peer_hash,
|
|
"c": [self.bridge._OVERLAY_ROUTE_MIGRATION_CAPABILITY],
|
|
"m": self.bridge._OVERLAY_ROUTE_MIGRATION_MARKER,
|
|
},
|
|
incoming,
|
|
link_id,
|
|
candidate,
|
|
)
|
|
)
|
|
send_control.assert_not_called()
|
|
self.assertFalse(candidate["migration_ready_event"].is_set())
|
|
self.assertFalse(
|
|
self.bridge._promote_overlay_migration_candidate(
|
|
self.peer_hash,
|
|
link_id,
|
|
"unauthenticated_commit",
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._active_overlay_link_id_by_peer_hash[self.peer_hash],
|
|
self.active_link_id,
|
|
)
|
|
|
|
def test_mismatched_incoming_candidate_identity_is_rejected(self):
|
|
self.active_state["incoming"] = True
|
|
self.bridge._inbound_overlay_neighbors[self.peer_hash] = time.time()
|
|
incoming = FakeLink()
|
|
incoming.remote_identity = object()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"derive_presence_destination_hash_for_identity",
|
|
return_value="cd" * 16,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_enqueue_scheduler_task",
|
|
return_value=True,
|
|
) as enqueue:
|
|
link_id = self.bridge._register_incoming_overlay_link(
|
|
incoming,
|
|
self.peer_hash,
|
|
"test_candidate",
|
|
migration_candidate=True,
|
|
)
|
|
|
|
self.assertEqual(link_id, "")
|
|
self.assertNotIn(id(incoming), self.bridge._overlay_link_ids_by_object)
|
|
enqueue.assert_called_once()
|
|
|
|
def test_remote_commit_retains_old_link_until_finalize(self):
|
|
self.active_state["incoming"] = True
|
|
self.bridge._inbound_overlay_neighbors[self.peer_hash] = time.time()
|
|
candidate_link_id = "candidate-overlay"
|
|
transaction_id = "55" * 8
|
|
candidate = dict(self.active_state)
|
|
candidate.update(
|
|
{
|
|
"linkId": candidate_link_id,
|
|
"link": FakeLink(),
|
|
"incoming": True,
|
|
"migration_candidate": True,
|
|
"migration_source_link_id": self.active_link_id,
|
|
"migration_peer_authenticated": True,
|
|
"overlay_transport_admitted": True,
|
|
}
|
|
)
|
|
self.bridge._overlay_links_by_id[candidate_link_id] = candidate
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"emit_overlay_link_state",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_delayed_presence_announce_replay",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_flush_overlay_link_pending",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_send_overlay_transport_control",
|
|
return_value=True,
|
|
) as send, mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_overlay_duplicate_close",
|
|
) as delayed_close, mock.patch.object(
|
|
self.bridge,
|
|
"_overlay_enqueue_dedup",
|
|
):
|
|
commit = {
|
|
"t": self.bridge._OVERLAY_MIGRATION_COMMIT_WIRE_TYPE,
|
|
"r": self.peer_hash,
|
|
"q": transaction_id,
|
|
}
|
|
self.assertTrue(
|
|
self.bridge._handle_overlay_transport_control(
|
|
commit,
|
|
candidate["link"],
|
|
candidate_link_id,
|
|
candidate,
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._active_overlay_link_id_by_peer_hash[self.peer_hash],
|
|
candidate_link_id,
|
|
)
|
|
delayed_close.assert_not_called()
|
|
self.assertEqual(
|
|
send.call_args.args[2],
|
|
self.bridge._OVERLAY_MIGRATION_ACK_WIRE_TYPE,
|
|
)
|
|
self.assertTrue(
|
|
self.bridge._handle_overlay_transport_control(
|
|
commit,
|
|
candidate["link"],
|
|
candidate_link_id,
|
|
candidate,
|
|
)
|
|
)
|
|
self.assertEqual(send.call_count, 2)
|
|
delayed_close.assert_not_called()
|
|
|
|
finalize = {
|
|
"t": self.bridge._OVERLAY_MIGRATION_FINALIZE_WIRE_TYPE,
|
|
"r": self.peer_hash,
|
|
"q": transaction_id,
|
|
}
|
|
self.assertTrue(
|
|
self.bridge._handle_overlay_transport_control(
|
|
finalize,
|
|
candidate["link"],
|
|
candidate_link_id,
|
|
candidate,
|
|
)
|
|
)
|
|
delayed_close.assert_called_once_with(
|
|
self.peer_hash,
|
|
candidate_link_id,
|
|
self.active_link_id,
|
|
"route_migrated",
|
|
)
|
|
|
|
def test_delayed_remote_identity_unlocks_incoming_candidate(self):
|
|
self.active_state["incoming"] = True
|
|
self.bridge._inbound_overlay_neighbors[self.peer_hash] = time.time()
|
|
incoming = FakeLink()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_send_overlay_hello_for_link",
|
|
) as send_hello:
|
|
link_id = self.bridge._register_incoming_overlay_link(
|
|
incoming,
|
|
self.peer_hash,
|
|
"test_candidate",
|
|
migration_candidate=True,
|
|
)
|
|
send_hello.assert_not_called()
|
|
|
|
identity = object()
|
|
incoming.remote_identity = identity
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"derive_presence_destination_hash_for_identity",
|
|
return_value=self.peer_hash,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"find_peer_hash_for_identity",
|
|
return_value=self.peer_hash,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"emit_overlay_link_state",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_note_overlay_peer_alive",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_register_active_overlay_for_peer",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_overlay_enqueue_dedup",
|
|
):
|
|
self.bridge.on_overlay_link_remote_identified(incoming, identity)
|
|
|
|
candidate = self.bridge._overlay_links_by_id[link_id]
|
|
self.assertTrue(candidate["migration_peer_authenticated"])
|
|
self.assertTrue(candidate["overlay_transport_admitted"])
|
|
send_hello.assert_called_once_with(link_id, "migration_identity_verified")
|
|
|
|
|
|
class PresenceBridgeResourceSchedulingTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
self.peer_hash = "ab" * 16
|
|
|
|
def test_resource_commands_use_fast_control_lane(self):
|
|
actions = (
|
|
"accept_reticulum_chat_resource",
|
|
"send_reticulum_chat_resource",
|
|
"authorize_reticulum_chat_resource",
|
|
"reject_reticulum_chat_resource",
|
|
"cancel_reticulum_resource",
|
|
)
|
|
for action in actions:
|
|
with self.subTest(action=action):
|
|
self.assertEqual(
|
|
self.bridge._scheduler_lane_for_command(action),
|
|
"resource-control",
|
|
)
|
|
|
|
def test_resource_open_shard_is_stable_for_a_peer(self):
|
|
lane = self.bridge._resource_open_scheduler_lane(self.peer_hash)
|
|
self.assertEqual(
|
|
lane,
|
|
self.bridge._resource_open_scheduler_lane(self.peer_hash.upper()),
|
|
)
|
|
self.assertIn(lane, self.bridge._SCHEDULER_QUEUE_MAX_BY_LANE)
|
|
self.assertTrue(lane.startswith("resource-open-"))
|
|
|
|
def test_only_one_link_handshake_starts_per_peer(self):
|
|
first = {"peerPresenceHash": self.peer_hash, "transferId": "first"}
|
|
second = {"peerPresenceHash": self.peer_hash, "transferId": "second"}
|
|
scheduled = []
|
|
started = []
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_enqueue_scheduler_task",
|
|
side_effect=lambda lane, name, fn, *args, **kwargs: scheduled.append(
|
|
(lane, name, fn, args, kwargs)
|
|
)
|
|
or True,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_run_qchat_file_open_task",
|
|
side_effect=lambda state: started.append(state),
|
|
):
|
|
self.assertTrue(self.bridge._queue_qchat_file_open_state(first))
|
|
self.assertTrue(self.bridge._queue_qchat_file_open_state(second))
|
|
self.assertEqual(len(scheduled), 1)
|
|
|
|
self.bridge._run_qchat_file_peer_open_queue(self.peer_hash)
|
|
self.assertEqual(started, [first])
|
|
|
|
self.bridge._run_qchat_file_peer_open_queue(self.peer_hash)
|
|
self.assertEqual(started, [first])
|
|
|
|
self.bridge._release_qchat_file_open_slot(first)
|
|
self.bridge._run_qchat_file_peer_open_queue(self.peer_hash)
|
|
self.assertEqual(started, [first, second])
|
|
|
|
def test_chat_opens_are_queued_before_bulk_resources(self):
|
|
bulk = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": "bulk",
|
|
"resourceType": "reticulum_group_resource",
|
|
}
|
|
chat = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": "chat",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
):
|
|
self.assertTrue(self.bridge._queue_qchat_file_open_state(bulk))
|
|
self.assertTrue(self.bridge._queue_qchat_file_open_state(chat))
|
|
|
|
queued = list(self.bridge._qchat_file_open_queue_by_peer[self.peer_hash])
|
|
self.assertEqual([item["transferId"] for item in queued], ["chat", "bulk"])
|
|
|
|
def test_active_resource_links_are_bounded_per_peer(self):
|
|
waiting = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": "waiting",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
}
|
|
started = []
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
):
|
|
self.assertTrue(self.bridge._queue_qchat_file_open_state(waiting))
|
|
for index in range(self.bridge._QCHAT_FILE_ACTIVE_OUTGOING_MAX_PER_PEER):
|
|
self.bridge._qchat_file_links_by_id[f"active-{index}"] = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"incoming": False,
|
|
"transferId": f"active-{index}",
|
|
}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_run_qchat_file_open_task",
|
|
side_effect=lambda state: started.append(state),
|
|
):
|
|
self.bridge._run_qchat_file_peer_open_queue(self.peer_hash)
|
|
self.assertEqual(started, [])
|
|
self.bridge._qchat_file_links_by_id.pop("active-0")
|
|
self.bridge._run_qchat_file_peer_open_queue(self.peer_hash)
|
|
self.assertEqual(started, [waiting])
|
|
|
|
def test_chat_can_use_reserved_slot_while_bulk_is_at_its_limit(self):
|
|
bulk_waiting = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": "bulk-waiting",
|
|
"resourceType": "reticulum_group_resource",
|
|
}
|
|
chat_waiting = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": "chat-waiting",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
):
|
|
self.assertTrue(self.bridge._queue_qchat_file_open_state(bulk_waiting))
|
|
self.assertTrue(self.bridge._queue_qchat_file_open_state(chat_waiting))
|
|
for index in range(self.bridge._QCHAT_FILE_BULK_ACTIVE_MAX_PER_PEER):
|
|
self.bridge._qchat_file_links_by_id[f"bulk-active-{index}"] = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"incoming": False,
|
|
"transferId": f"bulk-active-{index}",
|
|
"resourceType": "reticulum_group_resource",
|
|
}
|
|
started = []
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_run_qchat_file_open_task",
|
|
side_effect=lambda state: started.append(state),
|
|
):
|
|
self.bridge._run_qchat_file_peer_open_queue(self.peer_hash)
|
|
|
|
self.assertEqual(started, [chat_waiting])
|
|
self.assertIn(
|
|
bulk_waiting,
|
|
self.bridge._qchat_file_open_queue_by_peer[self.peer_hash],
|
|
)
|
|
|
|
def test_successful_link_removal_wakes_waiting_peer_queue(self):
|
|
waiting = {"peerPresenceHash": self.peer_hash, "transferId": "waiting"}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
):
|
|
self.assertTrue(self.bridge._queue_qchat_file_open_state(waiting))
|
|
link = FakeLink()
|
|
active = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": "completed",
|
|
"incoming": False,
|
|
"link": link,
|
|
}
|
|
self.bridge._qchat_file_links_by_id["completed-link"] = active
|
|
self.bridge._qchat_file_link_ids_by_object[id(link)] = "completed-link"
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
) as schedule, mock.patch.object(
|
|
self.bridge,
|
|
"_teardown_reticulum_link_bounded",
|
|
):
|
|
self.bridge._qchat_file_receiver_transfer_done(
|
|
self.peer_hash,
|
|
"completed",
|
|
)
|
|
|
|
schedule.assert_any_call(self.peer_hash)
|
|
self.assertIn(
|
|
waiting,
|
|
self.bridge._qchat_file_open_queue_by_peer[self.peer_hash],
|
|
)
|
|
|
|
def test_final_unestablished_link_failure_cleans_pending_transfer(self):
|
|
transfer_id = "max-attempts"
|
|
link = FakeLink()
|
|
pending = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
}
|
|
state = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"incoming": False,
|
|
"established": False,
|
|
"open_attempts": self.bridge._QCHAT_FILE_LINK_MAX_OPEN_ATTEMPTS,
|
|
"authMessage": {"type": "test"},
|
|
"receive_root": pending,
|
|
"link": link,
|
|
}
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, pending)
|
|
self.bridge._qchat_file_links_by_id["max-link"] = state
|
|
self.bridge._qchat_file_link_ids_by_object[id(link)] = "max-link"
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge.on_qchat_file_link_closed(link)
|
|
|
|
failed = [
|
|
call
|
|
for call in emit.call_args_list
|
|
if call.args and call.args[0] == "failed"
|
|
]
|
|
self.assertEqual(len(failed), 1)
|
|
self.assertEqual(
|
|
failed[0].args[1]["reason"],
|
|
"file_link_open_attempts_exhausted",
|
|
)
|
|
self.assertNotIn(
|
|
transfer_id,
|
|
self.bridge._qchat_file_accepts_by_transfer,
|
|
)
|
|
self.assertTrue(pending.get("cancelled"))
|
|
|
|
def test_exhausted_parallel_link_keeps_viable_sibling_transfer(self):
|
|
transfer_id = "parallel-transfer"
|
|
failed_link = FakeLink()
|
|
pending = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
}
|
|
failed_state = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"incoming": False,
|
|
"established": False,
|
|
"open_attempts": self.bridge._QCHAT_FILE_LINK_MAX_OPEN_ATTEMPTS,
|
|
"authMessage": {"type": "test"},
|
|
"receive_root": pending,
|
|
"link": failed_link,
|
|
}
|
|
sibling = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"incoming": False,
|
|
"established": True,
|
|
"receive_root": pending,
|
|
}
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, pending)
|
|
self.bridge._qchat_file_links_by_id["failed-link"] = failed_state
|
|
self.bridge._qchat_file_link_ids_by_object[id(failed_link)] = "failed-link"
|
|
self.bridge._qchat_file_links_by_id["sibling-link"] = sibling
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge.on_qchat_file_link_closed(failed_link)
|
|
|
|
self.assertFalse(
|
|
any(call.args and call.args[0] == "failed" for call in emit.call_args_list)
|
|
)
|
|
self.assertIn(
|
|
transfer_id,
|
|
self.bridge._qchat_file_accepts_by_transfer,
|
|
)
|
|
self.assertFalse(pending.get("cancelled", False))
|
|
|
|
def test_path_timeout_cleans_pending_transfer(self):
|
|
transfer_id = "path-timeout"
|
|
pending = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
}
|
|
state = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"incoming": False,
|
|
"receive_root": pending,
|
|
}
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, pending)
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_open_qchat_file_link_for_state",
|
|
side_effect=TimeoutError("no path"),
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge._run_qchat_file_open_task(state)
|
|
|
|
emit.assert_any_call(
|
|
"failed",
|
|
mock.ANY,
|
|
)
|
|
self.assertNotIn(
|
|
transfer_id,
|
|
self.bridge._qchat_file_accepts_by_transfer,
|
|
)
|
|
self.assertTrue(pending.get("cancelled"))
|
|
|
|
def test_path_timeout_keeps_viable_parallel_sibling(self):
|
|
transfer_id = "parallel-path-timeout"
|
|
pending = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
}
|
|
timed_out_state = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"incoming": False,
|
|
"receive_root": pending,
|
|
}
|
|
sibling = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"incoming": False,
|
|
"established": True,
|
|
"receive_root": pending,
|
|
}
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, pending)
|
|
self.bridge._qchat_file_links_by_id["path-sibling"] = sibling
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_open_qchat_file_link_for_state",
|
|
side_effect=TimeoutError("no path"),
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge._run_qchat_file_open_task(timed_out_state)
|
|
|
|
self.assertFalse(
|
|
any(call.args and call.args[0] == "failed" for call in emit.call_args_list)
|
|
)
|
|
self.assertIn(
|
|
transfer_id,
|
|
self.bridge._qchat_file_accepts_by_transfer,
|
|
)
|
|
self.assertFalse(pending.get("cancelled", False))
|
|
|
|
def test_path_timeout_fails_once_when_only_queued_siblings_remain(self):
|
|
transfer_id = "queued-path-timeout"
|
|
pending = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
}
|
|
timed_out_state = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"incoming": False,
|
|
"receive_root": pending,
|
|
}
|
|
queued_sibling = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"incoming": False,
|
|
"receive_root": pending,
|
|
}
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, pending)
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
):
|
|
self.assertTrue(
|
|
self.bridge._queue_qchat_file_open_state(queued_sibling)
|
|
)
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_open_qchat_file_link_for_state",
|
|
side_effect=TimeoutError("no path"),
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge._run_qchat_file_open_task(timed_out_state)
|
|
|
|
failed = [
|
|
call
|
|
for call in emit.call_args_list
|
|
if call.args and call.args[0] == "failed"
|
|
]
|
|
self.assertEqual(len(failed), 1)
|
|
self.assertNotIn(
|
|
transfer_id,
|
|
self.bridge._qchat_file_accepts_by_transfer,
|
|
)
|
|
self.assertTrue(pending.get("cancelled"))
|
|
|
|
def test_terminal_cleanup_does_not_cancel_another_peer_transfer(self):
|
|
current = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": "current-transfer",
|
|
}
|
|
stale_state = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": "stale-transfer",
|
|
"incoming": False,
|
|
}
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, current)
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit"):
|
|
self.bridge._qchat_file_fail_open_state(
|
|
stale_state,
|
|
"link_open_failed",
|
|
force_transfer_failure=True,
|
|
)
|
|
|
|
self.assertIn(
|
|
"current-transfer",
|
|
self.bridge._qchat_file_accepts_by_transfer,
|
|
)
|
|
self.assertFalse(current.get("cancelled", False))
|
|
|
|
def test_cancelled_transfer_is_removed_from_open_queue(self):
|
|
receive_root = {"cancelled": False}
|
|
waiting = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": "cancelled",
|
|
"receive_root": receive_root,
|
|
}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_qchat_file_peer_open_drain",
|
|
return_value=True,
|
|
):
|
|
self.assertTrue(self.bridge._queue_qchat_file_open_state(waiting))
|
|
receive_root["cancelled"] = True
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_run_qchat_file_open_task",
|
|
) as open_task:
|
|
self.bridge._run_qchat_file_peer_open_queue(self.peer_hash)
|
|
|
|
open_task.assert_not_called()
|
|
self.assertNotIn(
|
|
self.peer_hash,
|
|
self.bridge._qchat_file_open_queue_by_peer,
|
|
)
|
|
self.assertNotIn(id(waiting), self.bridge._qchat_file_open_queue_state_ids)
|
|
|
|
def test_waiting_for_path_does_not_consume_link_attempt(self):
|
|
state = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"peerIdentity": object(),
|
|
"transferId": "waiting-for-path",
|
|
}
|
|
outbound = FakeDestination()
|
|
outbound.hash = bytes.fromhex(self.peer_hash)
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"build_outbound_destination",
|
|
return_value=outbound,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_request_qchat_file_path",
|
|
return_value=False,
|
|
):
|
|
with self.assertRaises(self.bridge._QChatFilePathPending):
|
|
self.bridge._open_qchat_file_link_for_state(state)
|
|
|
|
self.assertEqual(int(state.get("open_attempts") or 0), 0)
|
|
self.assertGreater(float(state.get("path_wait_started_at") or 0), 0)
|
|
|
|
def test_failed_cached_path_is_refreshed_only_once_while_polling(self):
|
|
state = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"peerIdentity": object(),
|
|
"transferId": "refresh-once",
|
|
}
|
|
outbound = FakeDestination()
|
|
outbound.hash = bytes.fromhex(self.peer_hash)
|
|
refresh_permissions = []
|
|
|
|
def request_path(_destination_hash, _peer_hash, **kwargs):
|
|
refresh_permissions.append(kwargs.get("allow_failed_path_refresh"))
|
|
return False
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"build_outbound_destination",
|
|
return_value=outbound,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_peer_has_recent_unestablished_link_failure",
|
|
return_value=True,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_request_qchat_file_path",
|
|
side_effect=request_path,
|
|
):
|
|
for _ in range(2):
|
|
with self.assertRaises(self.bridge._QChatFilePathPending):
|
|
self.bridge._open_qchat_file_link_for_state(state)
|
|
|
|
self.assertEqual(refresh_permissions, [True, False])
|
|
self.assertTrue(state.get("failed_path_refresh_requested"))
|
|
self.assertEqual(int(state.get("open_attempts") or 0), 0)
|
|
|
|
|
|
class PresenceBridgeReusableResourceSessionTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
self.peer_hash = "ab" * 16
|
|
|
|
def pending(
|
|
self,
|
|
transfer_id,
|
|
*,
|
|
resource_type=None,
|
|
event_id="",
|
|
sha256="",
|
|
timestamp=None,
|
|
):
|
|
resource_type = resource_type or self.bridge._RETICULUM_CHAT_RESOURCE_TYPE
|
|
auth = {
|
|
"type": "RCR",
|
|
"transferId": transfer_id,
|
|
"ts": timestamp if timestamp is not None else int(time.time() * 1000),
|
|
}
|
|
if event_id:
|
|
auth["eventId"] = event_id
|
|
return {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"savePath": f"/tmp/{transfer_id}.recv",
|
|
"fileName": f"{transfer_id}.bin",
|
|
"size": 128,
|
|
"sha256": sha256,
|
|
"resourceType": resource_type,
|
|
"metadata": {
|
|
"groupId": 716,
|
|
**({"eventId": event_id} if event_id else {}),
|
|
},
|
|
"peerIdentity": object(),
|
|
"authMessage": auth,
|
|
}
|
|
|
|
def session(self, lane="fast", established=True, slot=0):
|
|
session_id = f"session-{lane}-{slot}"
|
|
link = FakeSessionLink()
|
|
state = {
|
|
"linkId": session_id,
|
|
"manager_kind": "resource_session",
|
|
"sessionKey": self.bridge._resource_session_key(self.peer_hash, lane, slot),
|
|
"sessionLane": lane,
|
|
"sessionSlot": slot,
|
|
"peerPresenceHash": self.peer_hash,
|
|
"incoming": False,
|
|
"established": established,
|
|
"remote_ready": established,
|
|
"provider_ready_sent": established,
|
|
"created_at": time.time(),
|
|
"last_used_at": time.time(),
|
|
"pending_jobs": [],
|
|
"active_requests": {},
|
|
"provider_active": 0,
|
|
"link": link,
|
|
"generation": 1,
|
|
"activity_generation": 1,
|
|
}
|
|
self.bridge._qchat_file_links_by_id[session_id] = state
|
|
self.bridge._qchat_file_link_ids_by_object[id(link)] = session_id
|
|
self.bridge._resource_sessions_by_key[state["sessionKey"]] = session_id
|
|
return state, link
|
|
|
|
def test_parallel_peer_lookup_requires_an_exact_transfer_id(self):
|
|
first = self.pending("first-range", resource_type="reticulum_group_resource_range")
|
|
second = self.pending("second-range", resource_type="reticulum_group_resource_range")
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, first)
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, second)
|
|
|
|
self.assertIs(
|
|
self.bridge._qchat_file_get_pending_receive(self.peer_hash, "first-range"),
|
|
first,
|
|
)
|
|
self.assertIs(
|
|
self.bridge._qchat_file_get_pending_receive(self.peer_hash, "second-range"),
|
|
second,
|
|
)
|
|
self.assertIsNone(
|
|
self.bridge._qchat_file_get_pending_receive(self.peer_hash, "unknown-range")
|
|
)
|
|
self.assertIsNone(self.bridge._qchat_file_get_pending_receive(self.peer_hash))
|
|
|
|
self.bridge._qchat_file_remove_pending_receive(self.peer_hash, "second-range")
|
|
self.assertIs(self.bridge._qchat_file_get_pending_receive(self.peer_hash), first)
|
|
|
|
def test_parallel_ranges_with_same_event_have_distinct_semantic_keys(self):
|
|
first = self.pending(
|
|
"first-range",
|
|
resource_type="reticulum_group_resource_range",
|
|
event_id="shared-event",
|
|
)
|
|
second = self.pending(
|
|
"second-range",
|
|
resource_type="reticulum_group_resource_range",
|
|
event_id="shared-event",
|
|
)
|
|
file_hash = "ab" * 32
|
|
first["metadata"].update(
|
|
{"fileHash": file_hash, "byteRanges": [[0, 1048576]]}
|
|
)
|
|
second["metadata"].update(
|
|
{"fileHash": file_hash, "byteRanges": [[1048576, 2097152]]}
|
|
)
|
|
|
|
self.assertNotEqual(
|
|
self.bridge._resource_session_semantic_key(first),
|
|
self.bridge._resource_session_semantic_key(second),
|
|
)
|
|
|
|
def test_session_response_skips_legacy_peer_fallback_callbacks(self):
|
|
_state, link = self.session(lane="bulk")
|
|
first = self.pending("first-range", resource_type="reticulum_group_resource_range")
|
|
second = self.pending("second-range", resource_type="reticulum_group_resource_range")
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, first)
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, second)
|
|
|
|
class ResponseResource:
|
|
request_id = bytes.fromhex("33" * 16)
|
|
|
|
def __init__(self, response_link):
|
|
self.link = response_link
|
|
|
|
resource = ResponseResource(link)
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_qchat_file_get_pending_receive",
|
|
) as get_pending, mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge.on_qchat_file_resource_started(resource)
|
|
self.bridge.on_qchat_file_resource_concluded(resource)
|
|
|
|
get_pending.assert_not_called()
|
|
emit.assert_not_called()
|
|
self.assertIn("first-range", self.bridge._qchat_file_accepts_by_transfer)
|
|
self.assertIn("second-range", self.bridge._qchat_file_accepts_by_transfer)
|
|
|
|
def test_range_response_is_bound_to_its_transfer_and_payload_hash(self):
|
|
contents = b"verified range response"
|
|
payload_hash = self.bridge.hashlib.sha256(contents).hexdigest()
|
|
transfer_id = "verified-range"
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source_path = Path(directory) / "source.bin"
|
|
save_path = Path(directory) / "received.bin"
|
|
source_path.write_bytes(contents)
|
|
response = open(source_path, "rb")
|
|
receipt = FakeSessionReceipt()
|
|
receipt.response = response
|
|
receipt.metadata = {
|
|
"transferId": transfer_id,
|
|
"size": len(contents),
|
|
"sha256": payload_hash,
|
|
}
|
|
pending = self.pending(
|
|
transfer_id,
|
|
resource_type="reticulum_group_resource_range",
|
|
)
|
|
pending.update(
|
|
{
|
|
"savePath": str(save_path),
|
|
"size": len(contents),
|
|
}
|
|
)
|
|
job = {
|
|
"transferId": transfer_id,
|
|
"pending": pending,
|
|
"created_at": time.time(),
|
|
"followers": [],
|
|
}
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, pending)
|
|
|
|
with mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge._resource_session_response_received(job, receipt)
|
|
|
|
self.assertTrue(job["completed"])
|
|
self.assertEqual(save_path.read_bytes(), contents)
|
|
received = [
|
|
call
|
|
for call in emit.call_args_list
|
|
if call.args and call.args[0] == "received"
|
|
]
|
|
self.assertEqual(len(received), 1)
|
|
self.assertEqual(received[0].args[1]["transferId"], transfer_id)
|
|
self.assertEqual(received[0].args[1]["payloadHash"], payload_hash)
|
|
|
|
def test_range_response_from_legacy_peer_uses_request_receipt_identity(self):
|
|
contents = b"legacy range response"
|
|
payload_hash = self.bridge.hashlib.sha256(contents).hexdigest()
|
|
transfer_id = "expected-range"
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source_path = Path(directory) / "source.bin"
|
|
save_path = Path(directory) / "received.bin"
|
|
source_path.write_bytes(contents)
|
|
receipt = FakeSessionReceipt()
|
|
receipt.response = open(source_path, "rb")
|
|
receipt.metadata = {"size": len(contents)}
|
|
pending = self.pending(
|
|
transfer_id,
|
|
resource_type="reticulum_group_resource_range",
|
|
)
|
|
pending.update(
|
|
{
|
|
"savePath": str(save_path),
|
|
"size": len(contents),
|
|
}
|
|
)
|
|
job = {
|
|
"transferId": transfer_id,
|
|
"pending": pending,
|
|
"created_at": time.time(),
|
|
"followers": [],
|
|
}
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, pending)
|
|
|
|
with mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge._resource_session_response_received(job, receipt)
|
|
|
|
self.assertTrue(job["completed"])
|
|
self.assertEqual(save_path.read_bytes(), contents)
|
|
received = [
|
|
call
|
|
for call in emit.call_args_list
|
|
if call.args and call.args[0] == "received"
|
|
]
|
|
self.assertEqual(len(received), 1)
|
|
self.assertEqual(received[0].args[1]["transferId"], transfer_id)
|
|
self.assertEqual(received[0].args[1]["payloadHash"], payload_hash)
|
|
|
|
def test_range_response_rejects_mismatched_metadata_identity(self):
|
|
contents = b"wrong transfer response"
|
|
transfer_id = "expected-range"
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source_path = Path(directory) / "source.bin"
|
|
save_path = Path(directory) / "received.bin"
|
|
source_path.write_bytes(contents)
|
|
receipt = FakeSessionReceipt()
|
|
receipt.response = open(source_path, "rb")
|
|
receipt.metadata = {"transferId": "different-range"}
|
|
pending = self.pending(
|
|
transfer_id,
|
|
resource_type="reticulum_group_resource_range",
|
|
)
|
|
pending.update({"savePath": str(save_path), "size": len(contents)})
|
|
job = {
|
|
"transferId": transfer_id,
|
|
"pending": pending,
|
|
"created_at": time.time(),
|
|
"followers": [],
|
|
}
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, pending)
|
|
|
|
with mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge._resource_session_response_received(job, receipt)
|
|
|
|
self.assertFalse(save_path.exists())
|
|
failures = [
|
|
call
|
|
for call in emit.call_args_list
|
|
if call.args and call.args[0] == "failed"
|
|
]
|
|
self.assertEqual(len(failures), 1)
|
|
self.assertEqual(failures[0].args[1]["reason"], "resource_response_invalid")
|
|
self.assertIn("transfer id mismatch", failures[0].args[1]["error"])
|
|
|
|
def test_managed_accept_uses_reusable_session_instead_of_link_queue(self):
|
|
payload = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"reticulumIdentityPublicKeyBase64": "identity",
|
|
"authMessage": {"type": "RCR", "ts": int(time.time() * 1000)},
|
|
"transferId": "managed",
|
|
"savePath": "/tmp/managed.recv",
|
|
"fileName": "managed.bin",
|
|
"size": 128,
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"expires_at": time.time() + 60,
|
|
}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_parse_qchat_file_peer_identity",
|
|
return_value=object(),
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_accept",
|
|
) as accept_session, mock.patch.object(
|
|
self.bridge,
|
|
"_open_qchat_file_link_async",
|
|
) as open_legacy:
|
|
self.bridge.handle_accept_qchat_file_resource("req", payload)
|
|
|
|
accept_session.assert_called_once()
|
|
open_legacy.assert_not_called()
|
|
|
|
def test_fast_and_bulk_resources_use_separate_peer_sessions(self):
|
|
fast_job = {
|
|
"pending": self.pending("fast"),
|
|
"created_at": time.time(),
|
|
}
|
|
bulk_job = {
|
|
"pending": self.pending(
|
|
"bulk",
|
|
resource_type="reticulum_group_resource_range",
|
|
),
|
|
"created_at": time.time(),
|
|
}
|
|
with mock.patch.object(self.bridge, "_resource_session_poll_path"):
|
|
self.assertTrue(self.bridge._resource_session_enqueue_job(fast_job)[0])
|
|
self.assertTrue(self.bridge._resource_session_enqueue_job(bulk_job)[0])
|
|
|
|
self.assertIn(
|
|
self.bridge._resource_session_key(self.peer_hash, "fast"),
|
|
self.bridge._resource_sessions_by_key,
|
|
)
|
|
self.assertIn(
|
|
self.bridge._resource_session_key(self.peer_hash, "bulk"),
|
|
self.bridge._resource_sessions_by_key,
|
|
)
|
|
self.assertEqual(len(self.bridge._resource_sessions_by_key), 2)
|
|
|
|
def test_live_event_and_history_use_separate_priority_lanes(self):
|
|
history_pending = self.pending("history")
|
|
history_pending["metadata"].update(
|
|
{
|
|
"logicalResourceType": "reticulum_chat_history_page",
|
|
"channelId": "general",
|
|
"direction": "before",
|
|
}
|
|
)
|
|
history_pending["authMessage"].update({"before": {"id": "cursor"}})
|
|
live_pending = self.pending(
|
|
"live",
|
|
event_id="event-live",
|
|
sha256="11" * 32,
|
|
)
|
|
history_job = {"pending": history_pending, "created_at": time.time()}
|
|
live_job = {"pending": live_pending, "created_at": time.time() + 0.01}
|
|
|
|
with mock.patch.object(self.bridge, "_resource_session_poll_path"):
|
|
self.assertTrue(self.bridge._resource_session_enqueue_job(history_job)[0])
|
|
self.assertTrue(self.bridge._resource_session_enqueue_job(live_job)[0])
|
|
|
|
fast_id = self.bridge._resource_sessions_by_key[
|
|
self.bridge._resource_session_key(self.peer_hash, "fast")
|
|
]
|
|
bulk_id = self.bridge._resource_sessions_by_key[
|
|
self.bridge._resource_session_key(self.peer_hash, "bulk")
|
|
]
|
|
self.assertEqual(
|
|
[
|
|
job["pending"]["transferId"]
|
|
for job in self.bridge._qchat_file_links_by_id[fast_id]["pending_jobs"]
|
|
],
|
|
["live"],
|
|
)
|
|
self.assertEqual(
|
|
[
|
|
job["pending"]["transferId"]
|
|
for job in self.bridge._qchat_file_links_by_id[bulk_id]["pending_jobs"]
|
|
],
|
|
["history"],
|
|
)
|
|
|
|
def test_dm_history_page_uses_fast_lane(self):
|
|
self.assertEqual(
|
|
self.bridge._resource_session_lane(
|
|
self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"reticulum_chat_dm_page",
|
|
),
|
|
"fast",
|
|
)
|
|
|
|
def test_prepare_command_reports_fast_lane_for_dm_history(self):
|
|
payload = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"reticulumIdentityPublicKeyBase64": "identity",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"logicalResourceType": "reticulum_chat_dm_page",
|
|
}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_poll_path",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_parse_qchat_file_peer_identity",
|
|
return_value=object(),
|
|
), mock.patch.object(self.bridge, "emit_resp") as emit_resp:
|
|
self.bridge.handle_prepare_reticulum_resource_session("dm", payload)
|
|
|
|
emit_resp.assert_called_once()
|
|
self.assertTrue(emit_resp.call_args.args[1])
|
|
self.assertEqual(emit_resp.call_args.kwargs["payload"]["lane"], "fast")
|
|
|
|
def test_prepare_command_reuses_pending_session_and_reports_state(self):
|
|
peer_identity = object()
|
|
payload = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"reticulumIdentityPublicKeyBase64": "identity",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"logicalResourceType": "reticulum_chat_history_page",
|
|
}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_poll_path",
|
|
) as poll_path, mock.patch.object(self.bridge, "emit_resp") as emit_resp:
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_parse_qchat_file_peer_identity",
|
|
return_value=peer_identity,
|
|
) as parse_identity:
|
|
self.bridge.handle_prepare_reticulum_resource_session("one", payload)
|
|
self.bridge.handle_prepare_reticulum_resource_session("two", payload)
|
|
|
|
poll_path.assert_called_once()
|
|
parse_identity.assert_has_calls(
|
|
[
|
|
mock.call(self.peer_hash, "identity"),
|
|
mock.call(self.peer_hash, "identity"),
|
|
]
|
|
)
|
|
self.assertEqual(len(self.bridge._resource_sessions_by_key), 1)
|
|
session_id = next(iter(self.bridge._resource_sessions_by_key.values()))
|
|
self.assertIs(
|
|
self.bridge._qchat_file_links_by_id[session_id]["peerIdentity"],
|
|
peer_identity,
|
|
)
|
|
self.assertEqual(emit_resp.call_count, 2)
|
|
self.assertTrue(all(call.args[1] for call in emit_resp.call_args_list))
|
|
self.assertTrue(
|
|
all(
|
|
call.kwargs["payload"]["status"] == "pending"
|
|
and call.kwargs["payload"]["lane"] == "bulk"
|
|
for call in emit_resp.call_args_list
|
|
)
|
|
)
|
|
|
|
def test_prepare_command_recalls_identity_when_public_key_is_omitted(self):
|
|
peer_identity = object()
|
|
payload = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"reticulumIdentityPublicKeyBase64": "",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
}
|
|
|
|
def recall(peer_hash, source):
|
|
self.assertEqual(peer_hash, self.peer_hash)
|
|
self.assertEqual(source, "ts_seed")
|
|
self.bridge._known_peers[peer_hash] = peer_identity
|
|
return True
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"ensure_known_peer_from_recall",
|
|
side_effect=recall,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_poll_path",
|
|
) as poll_path, mock.patch.object(self.bridge, "emit_resp") as emit_resp:
|
|
self.bridge.handle_prepare_reticulum_resource_session("req", payload)
|
|
|
|
poll_path.assert_called_once()
|
|
session_id = next(iter(self.bridge._resource_sessions_by_key.values()))
|
|
self.assertIs(
|
|
self.bridge._qchat_file_links_by_id[session_id]["peerIdentity"],
|
|
peer_identity,
|
|
)
|
|
emit_resp.assert_called_once_with(
|
|
"req",
|
|
True,
|
|
payload=mock.ANY,
|
|
)
|
|
|
|
def test_prepare_command_rejects_unverified_peer_identity(self):
|
|
payload = {
|
|
"peerPresenceHash": self.peer_hash,
|
|
"reticulumIdentityPublicKeyBase64": "bad-identity",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_parse_qchat_file_peer_identity",
|
|
side_effect=ValueError("identity mismatch"),
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_poll_path",
|
|
) as poll_path, mock.patch.object(self.bridge, "emit_resp") as emit_resp:
|
|
self.bridge.handle_prepare_reticulum_resource_session("req", payload)
|
|
|
|
poll_path.assert_not_called()
|
|
emit_resp.assert_called_once_with(
|
|
"req",
|
|
False,
|
|
payload={"code": "bad_reticulum_identity"},
|
|
error="identity mismatch",
|
|
)
|
|
|
|
def test_prepare_command_parses_real_reticulum_identity(self):
|
|
peer_identity = RNS.Identity()
|
|
peer_hash = self.bridge.destination_hash_hex(
|
|
self.bridge.build_outbound_destination(peer_identity).hash
|
|
)
|
|
public_key = base64.b64encode(peer_identity.get_public_key()).decode("ascii")
|
|
payload = {
|
|
"peerPresenceHash": peer_hash,
|
|
"reticulumIdentityPublicKeyBase64": public_key,
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
}
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_poll_path",
|
|
) as poll_path, mock.patch.object(self.bridge, "emit_resp") as emit_resp:
|
|
self.bridge.handle_prepare_reticulum_resource_session("req", payload)
|
|
|
|
poll_path.assert_called_once()
|
|
session_id = next(iter(self.bridge._resource_sessions_by_key.values()))
|
|
parsed_identity = self.bridge._qchat_file_links_by_id[session_id]["peerIdentity"]
|
|
self.assertEqual(parsed_identity.get_public_key(), peer_identity.get_public_key())
|
|
emit_resp.assert_called_once_with("req", True, payload=mock.ANY)
|
|
|
|
def test_sessions_are_bounded_per_peer_without_a_global_connection_cap(self):
|
|
with mock.patch.object(self.bridge, "_resource_session_poll_path"):
|
|
for index in range(24):
|
|
peer_hash = f"{index + 1:032x}"
|
|
state, reason = self.bridge._resource_session_get_or_create(
|
|
peer_hash,
|
|
object(),
|
|
"fast",
|
|
)
|
|
self.assertEqual(reason, "")
|
|
self.assertIsInstance(state, dict)
|
|
|
|
self.assertEqual(len(self.bridge._resource_sessions_by_key), 24)
|
|
|
|
def test_bulk_pool_uses_one_resource_per_link_and_reserves_history(self):
|
|
sessions = [
|
|
self.session(lane="bulk", slot=index)
|
|
for index in range(self.bridge._RESOURCE_SESSION_BULK_POOL_SIZE)
|
|
]
|
|
for index, (state, _link) in enumerate(sessions[:5]):
|
|
state["pending_jobs"] = [
|
|
{
|
|
"pending": self.pending(
|
|
f"range-{index}",
|
|
resource_type="reticulum_group_resource_range",
|
|
),
|
|
"created_at": time.time() + index / 1000,
|
|
"followers": [],
|
|
}
|
|
]
|
|
history = self.pending("visible-history")
|
|
history["metadata"]["logicalResourceType"] = "reticulum_chat_history_page"
|
|
sessions[5][0]["pending_jobs"] = [
|
|
{
|
|
"pending": history,
|
|
"created_at": time.time() + 1,
|
|
"followers": [],
|
|
}
|
|
]
|
|
|
|
for state, _link in sessions:
|
|
self.bridge._resource_session_dispatch_pending(state)
|
|
|
|
active = [
|
|
job
|
|
for state, _link in sessions
|
|
for job in state["active_requests"].values()
|
|
]
|
|
self.assertEqual(sum(len(link.requests) for _state, link in sessions), 6)
|
|
self.assertTrue(all(len(link.requests) == 1 for _state, link in sessions))
|
|
self.assertEqual(
|
|
sum(
|
|
self.bridge._resource_session_job_class(job) == "attachment"
|
|
for job in active
|
|
),
|
|
5,
|
|
)
|
|
self.assertTrue(
|
|
any(
|
|
job["pending"]["transferId"] == "visible-history"
|
|
for job in active
|
|
)
|
|
)
|
|
|
|
def test_session_waits_for_provider_ready_before_dispatch(self):
|
|
state, link = self.session(established=False)
|
|
self.bridge._destination = FakeDestination()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_send_packet_on_link",
|
|
return_value=True,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_dispatch_pending",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_schedule_idle_close",
|
|
), mock.patch.object(self.bridge, "emit_event") as emit_event:
|
|
self.bridge.on_outgoing_resource_session_established(link)
|
|
self.assertTrue(state["established"])
|
|
self.assertFalse(state["remote_ready"])
|
|
self.assertFalse(
|
|
any(
|
|
call.args[0] == "reticulum_resource_session"
|
|
and call.args[1].get("status") == "ready"
|
|
for call in emit_event.call_args_list
|
|
)
|
|
)
|
|
self.bridge._handle_qchat_file_link_packet(
|
|
json.dumps(
|
|
{
|
|
"type": self.bridge._RESOURCE_SESSION_READY_TYPE,
|
|
"r": self.peer_hash,
|
|
"lane": "fast",
|
|
"providerIdleMs": 180000,
|
|
}
|
|
).encode("utf-8"),
|
|
FakePacket(link),
|
|
)
|
|
|
|
self.assertTrue(state["remote_ready"])
|
|
self.assertEqual(state["remote_provider_idle_seconds"], 180.0)
|
|
emit_event.assert_any_call(
|
|
"reticulum_resource_session",
|
|
{
|
|
"status": "ready",
|
|
"peerPresenceHash": self.peer_hash,
|
|
"lane": "fast",
|
|
"linkId": state["linkId"],
|
|
},
|
|
)
|
|
|
|
def test_first_packet_classifies_incoming_resource_session_without_overlay(self):
|
|
link = FakeSessionLink()
|
|
remote_identity = object()
|
|
link.remote_identity = remote_identity
|
|
self.bridge._destination = FakeDestination()
|
|
hello = json.dumps(
|
|
{
|
|
"type": self.bridge._RESOURCE_SESSION_HELLO_TYPE,
|
|
"r": self.peer_hash,
|
|
"lane": "fast",
|
|
}
|
|
).encode("utf-8")
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_schedule_inbound_classify_fallback",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_destination_hash_for_identity",
|
|
return_value=self.peer_hash,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_send_packet_on_link",
|
|
return_value=True,
|
|
) as send_packet, mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_schedule_idle_close",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_teardown_reticulum_link_bounded",
|
|
):
|
|
self.bridge.on_incoming_unified_link_established(link)
|
|
link.packet_callback(hello, FakePacket(link))
|
|
|
|
link_id = self.bridge.get_qchat_file_link_id(link)
|
|
self.assertIsInstance(link_id, str)
|
|
self.assertIsNone(self.bridge.get_overlay_link_id(link))
|
|
state = self.bridge.get_qchat_file_link_state(link_id)
|
|
self.assertEqual(state["linkId"], link_id)
|
|
self.assertEqual(state["manager_kind"], "resource_session")
|
|
self.assertEqual(state["peerPresenceHash"], self.peer_hash)
|
|
self.assertTrue(state["provider_ready_sent"])
|
|
send_packet.assert_called_once()
|
|
ready_wire = json.loads(send_packet.call_args.args[1].decode("utf-8"))
|
|
self.assertEqual(
|
|
ready_wire["providerIdleMs"],
|
|
int(
|
|
self.bridge._RESOURCE_SESSION_INCOMING_IDLE_TIMEOUT_SECONDS
|
|
* 1000
|
|
),
|
|
)
|
|
|
|
link.closed_callback(link)
|
|
|
|
self.assertIsNone(self.bridge.get_qchat_file_link_id(link))
|
|
self.assertNotIn(link_id, self.bridge._qchat_file_links_by_id)
|
|
|
|
def test_unidentified_incoming_resource_session_gets_idle_cleanup(self):
|
|
link = FakeSessionLink()
|
|
link.remote_identity = None
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_schedule_idle_close",
|
|
) as schedule_idle, mock.patch.object(
|
|
self.bridge,
|
|
"_send_packet_on_link",
|
|
) as send_packet:
|
|
link_id = self.bridge._register_incoming_resource_session(
|
|
link,
|
|
self.peer_hash,
|
|
"bulk",
|
|
)
|
|
|
|
self.assertIsInstance(link_id, str)
|
|
state = self.bridge.get_qchat_file_link_state(link_id)
|
|
self.assertIsInstance(state, dict)
|
|
self.assertFalse(state.get("provider_ready_sent", False))
|
|
send_packet.assert_not_called()
|
|
schedule_idle.assert_called_once_with(state)
|
|
|
|
def test_parallel_dispatch_never_exceeds_lane_limit(self):
|
|
for lane, limit in (
|
|
("fast", self.bridge._RESOURCE_SESSION_FAST_CONCURRENCY),
|
|
("bulk", self.bridge._RESOURCE_SESSION_BULK_CONCURRENCY),
|
|
):
|
|
with self.subTest(lane=lane):
|
|
state, link = self.session(lane=lane)
|
|
job_count = limit + 6
|
|
state["pending_jobs"] = [
|
|
{
|
|
"pending": self.pending(f"{lane}-parallel-{index}"),
|
|
"created_at": time.time(),
|
|
"followers": [],
|
|
}
|
|
for index in range(job_count)
|
|
]
|
|
|
|
threads = [
|
|
threading.Thread(
|
|
target=self.bridge._resource_session_dispatch_pending,
|
|
args=(state,),
|
|
)
|
|
for _ in range(4)
|
|
]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join(timeout=2)
|
|
|
|
self.assertEqual(len(state["active_requests"]), limit)
|
|
self.assertEqual(len(link.requests), limit)
|
|
self.assertEqual(len(state["pending_jobs"]), job_count - limit)
|
|
|
|
def test_bulk_jobs_are_balanced_across_independent_session_pool(self):
|
|
selected_states = []
|
|
with mock.patch.object(self.bridge, "_resource_session_poll_path"):
|
|
for index in range(self.bridge._RESOURCE_SESSION_BULK_POOL_SIZE * 2):
|
|
state, reason = self.bridge._resource_session_get_or_create(
|
|
self.peer_hash,
|
|
object(),
|
|
"bulk",
|
|
)
|
|
self.assertEqual(reason, "")
|
|
self.assertIsInstance(state, dict)
|
|
state["pending_jobs"].append({"index": index})
|
|
selected_states.append(state)
|
|
|
|
unique_states = {state["linkId"]: state for state in selected_states}
|
|
self.assertEqual(
|
|
len(unique_states),
|
|
self.bridge._RESOURCE_SESSION_BULK_POOL_SIZE,
|
|
)
|
|
self.assertEqual(
|
|
sorted(len(state["pending_jobs"]) for state in unique_states.values()),
|
|
[2] * self.bridge._RESOURCE_SESSION_BULK_POOL_SIZE,
|
|
)
|
|
self.assertEqual(
|
|
sorted(state["sessionSlot"] for state in unique_states.values()),
|
|
list(range(self.bridge._RESOURCE_SESSION_BULK_POOL_SIZE)),
|
|
)
|
|
|
|
def test_identical_event_downloads_share_one_network_job(self):
|
|
first = self.pending("first", event_id="event-1", sha256="22" * 32)
|
|
second = self.pending("second", event_id="event-1", sha256="22" * 32)
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, first)
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, second)
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_enqueue_job",
|
|
return_value=(True, ""),
|
|
) as enqueue, mock.patch.object(self.bridge, "emit_resp"):
|
|
self.bridge._resource_session_accept("req-1", first)
|
|
self.bridge._resource_session_accept("req-2", second)
|
|
|
|
self.assertEqual(enqueue.call_count, 1)
|
|
canonical = self.bridge._resource_session_jobs_by_transfer["first"]
|
|
self.assertEqual(len(canonical["followers"]), 1)
|
|
self.assertIs(
|
|
self.bridge._resource_session_jobs_by_transfer["second"],
|
|
canonical["followers"][0],
|
|
)
|
|
|
|
def test_stale_authorization_is_not_dispatched(self):
|
|
state, link = self.session()
|
|
job = {
|
|
"pending": self.pending(
|
|
"stale",
|
|
timestamp=int(
|
|
(time.time() - self.bridge._RESOURCE_SESSION_AUTH_MAX_QUEUE_SECONDS - 1)
|
|
* 1000
|
|
),
|
|
),
|
|
"created_at": time.time() - 100,
|
|
"followers": [],
|
|
}
|
|
with mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
dispatched = self.bridge._resource_session_dispatch_job(state, job)
|
|
|
|
self.assertFalse(dispatched)
|
|
self.assertEqual(link.requests, [])
|
|
self.assertTrue(job["completed"])
|
|
self.assertTrue(
|
|
any(
|
|
call.args[0] == "failed"
|
|
and call.args[1]["reason"] == "resource_auth_refresh_required"
|
|
for call in emit.call_args_list
|
|
)
|
|
)
|
|
|
|
def test_request_exception_fails_job_instead_of_leaving_it_active(self):
|
|
state, link = self.session()
|
|
job = {
|
|
"pending": self.pending("request-exception"),
|
|
"created_at": time.time(),
|
|
"followers": [],
|
|
"session": state,
|
|
}
|
|
with mock.patch.object(
|
|
link,
|
|
"request",
|
|
side_effect=RuntimeError("request failed"),
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
dispatched = self.bridge._resource_session_dispatch_job(state, job)
|
|
|
|
self.assertFalse(dispatched)
|
|
self.assertTrue(job["completed"])
|
|
self.assertNotIn("request-exception", state["active_requests"])
|
|
self.assertTrue(
|
|
any(
|
|
call.args[0] == "failed"
|
|
and call.args[1]["reason"] == "resource_request_send_failed"
|
|
for call in emit.call_args_list
|
|
)
|
|
)
|
|
|
|
def test_late_response_after_cancellation_closes_response_file(self):
|
|
job = {
|
|
"completed": True,
|
|
"pending": self.pending("late-response"),
|
|
}
|
|
receipt = FakeSessionReceipt()
|
|
receipt.response = tempfile.TemporaryFile()
|
|
|
|
self.bridge._resource_session_response_received(job, receipt)
|
|
|
|
self.assertTrue(receipt.response.closed)
|
|
|
|
def test_establishment_timeout_fails_jobs_once_and_refreshes_path(self):
|
|
state, link = self.session(established=False)
|
|
state["link_created_at"] = time.time() - 31
|
|
jobs = [
|
|
{
|
|
"pending": self.pending(f"job-{index}"),
|
|
"created_at": time.time(),
|
|
"followers": [],
|
|
"session": state,
|
|
}
|
|
for index in range(2)
|
|
]
|
|
state["pending_jobs"] = jobs
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_force_overlay_peer_path_refresh",
|
|
) as refresh, mock.patch.object(
|
|
self.bridge,
|
|
"_teardown_reticulum_link_bounded",
|
|
) as teardown, mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge._resource_session_open_timeout(state)
|
|
|
|
refresh.assert_called_once()
|
|
teardown.assert_called_once_with(link, mock.ANY)
|
|
self.assertTrue(all(job["completed"] for job in jobs))
|
|
self.assertEqual(
|
|
len([call for call in emit.call_args_list if call.args[0] == "failed"]),
|
|
2,
|
|
)
|
|
self.assertNotIn(state["sessionKey"], self.bridge._resource_sessions_by_key)
|
|
|
|
def test_session_failure_requeues_jobs_that_were_never_dispatched(self):
|
|
failed, failed_link = self.session(lane="bulk", slot=0)
|
|
surviving, _surviving_link = self.session(lane="bulk", slot=1)
|
|
job = {
|
|
"pending": self.pending(
|
|
"queued-range",
|
|
resource_type="reticulum_group_resource_range",
|
|
),
|
|
"created_at": time.time(),
|
|
"followers": [],
|
|
"session": failed,
|
|
}
|
|
failed["pending_jobs"] = [job]
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_teardown_reticulum_link_bounded",
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge._resource_session_fail_state(
|
|
failed,
|
|
"resource_session_link_closed",
|
|
)
|
|
|
|
self.assertFalse(job.get("completed", False))
|
|
self.assertIs(job.get("session"), surviving)
|
|
self.assertIs(surviving["active_requests"].get("queued-range"), job)
|
|
self.assertFalse(
|
|
any(call.args and call.args[0] == "failed" for call in emit.call_args_list)
|
|
)
|
|
self.assertNotIn(failed["sessionKey"], self.bridge._resource_sessions_by_key)
|
|
self.assertFalse(failed_link.teardown_called)
|
|
|
|
def test_provider_session_outlives_requester_idle_leases(self):
|
|
primary, _ = self.session(lane="bulk", slot=0)
|
|
overflow, _ = self.session(lane="bulk", slot=1)
|
|
incoming = dict(primary)
|
|
incoming["incoming"] = True
|
|
primary["remote_provider_idle_seconds"] = (
|
|
self.bridge._RESOURCE_SESSION_INCOMING_IDLE_TIMEOUT_SECONDS
|
|
)
|
|
overflow["remote_provider_idle_seconds"] = (
|
|
self.bridge._RESOURCE_SESSION_INCOMING_IDLE_TIMEOUT_SECONDS
|
|
)
|
|
|
|
self.assertEqual(
|
|
self.bridge._resource_session_idle_timeout_seconds(primary),
|
|
self.bridge._RESOURCE_SESSION_PRIMARY_IDLE_TIMEOUT_SECONDS,
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._resource_session_idle_timeout_seconds(overflow),
|
|
self.bridge._RESOURCE_SESSION_OVERFLOW_IDLE_TIMEOUT_SECONDS,
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._resource_session_idle_timeout_seconds(incoming),
|
|
self.bridge._RESOURCE_SESSION_INCOMING_IDLE_TIMEOUT_SECONDS,
|
|
)
|
|
self.assertGreater(
|
|
self.bridge._RESOURCE_SESSION_INCOMING_IDLE_TIMEOUT_SECONDS,
|
|
self.bridge._RESOURCE_SESSION_PRIMARY_IDLE_TIMEOUT_SECONDS,
|
|
)
|
|
self.assertGreater(
|
|
self.bridge._RESOURCE_SESSION_INCOMING_IDLE_TIMEOUT_SECONDS,
|
|
self.bridge._RESOURCE_SESSION_OVERFLOW_IDLE_TIMEOUT_SECONDS,
|
|
)
|
|
|
|
def test_legacy_provider_session_closes_before_its_unadvertised_lease(self):
|
|
primary, _ = self.session(lane="bulk", slot=0)
|
|
overflow, _ = self.session(lane="bulk", slot=1)
|
|
|
|
self.assertEqual(
|
|
self.bridge._resource_session_idle_timeout_seconds(primary),
|
|
self.bridge._RESOURCE_SESSION_LEGACY_PROVIDER_IDLE_TIMEOUT_SECONDS
|
|
- self.bridge._RESOURCE_SESSION_PROVIDER_IDLE_GUARD_SECONDS,
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._resource_session_idle_timeout_seconds(overflow),
|
|
self.bridge._RESOURCE_SESSION_OVERFLOW_IDLE_TIMEOUT_SECONDS,
|
|
)
|
|
|
|
def test_provider_request_waits_for_existing_electron_authorization(self):
|
|
state, link = self.session()
|
|
state["incoming"] = True
|
|
transfer_id = "provider"
|
|
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
|
|
temp_file.write(b"resource-response")
|
|
file_path = temp_file.name
|
|
pending = {
|
|
"allowedRecipientAddress": "",
|
|
"transferId": transfer_id,
|
|
"filePath": file_path,
|
|
"fileName": "provider.bin",
|
|
"size": len(b"resource-response"),
|
|
"sha256": "",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"metadata": {"eventId": "event-provider"},
|
|
"expires_at": time.time() + 60,
|
|
}
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id] = pending
|
|
|
|
def authorize_on_event(status, payload):
|
|
if status != "auth":
|
|
return
|
|
key = self.bridge._resource_session_waiter_key(
|
|
state["linkId"],
|
|
transfer_id,
|
|
)
|
|
waiter = self.bridge._resource_session_provider_waiters[key]
|
|
waiter["authorized"] = True
|
|
waiter["event"].set()
|
|
|
|
try:
|
|
remote_identity = object()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_qchat_file_emit",
|
|
side_effect=authorize_on_event,
|
|
), mock.patch.object(self.bridge, "_resource_session_watch_provider_file"):
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_destination_hash_for_identity",
|
|
return_value=self.peer_hash,
|
|
):
|
|
response = self.bridge._resource_session_response_generator(
|
|
self.bridge._RESOURCE_SESSION_REQUEST_PATH,
|
|
{
|
|
"version": 1,
|
|
"transferId": transfer_id,
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"metadata": {"eventId": "event-provider"},
|
|
"authMessage": {"type": "RCR"},
|
|
},
|
|
b"request",
|
|
link.link_id,
|
|
remote_identity,
|
|
time.time(),
|
|
)
|
|
self.assertIsInstance(response, tuple)
|
|
self.assertEqual(response[1]["transferId"], transfer_id)
|
|
self.assertEqual(state["provider_active"], 1)
|
|
self.assertEqual(response[0].read(), b"resource-response")
|
|
response[0].close()
|
|
finally:
|
|
Path(file_path).unlink(missing_ok=True)
|
|
|
|
def test_provider_rejects_unidentified_resource_session(self):
|
|
_state, link = self.session()
|
|
with mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
response = self.bridge._resource_session_response_generator(
|
|
self.bridge._RESOURCE_SESSION_REQUEST_PATH,
|
|
{
|
|
"version": 1,
|
|
"transferId": "unidentified",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"metadata": {"eventId": "event-unidentified"},
|
|
"authMessage": {"type": "RCR"},
|
|
},
|
|
b"request",
|
|
link.link_id,
|
|
None,
|
|
time.time(),
|
|
)
|
|
|
|
self.assertEqual(response["reason"], "resource_peer_unidentified")
|
|
emit.assert_not_called()
|
|
|
|
def test_provider_rejects_request_after_idle_close_commits(self):
|
|
state, link = self.session()
|
|
state["incoming"] = True
|
|
state["closing"] = True
|
|
remote_identity = object()
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_destination_hash_for_identity",
|
|
return_value=self.peer_hash,
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
response = self.bridge._resource_session_response_generator(
|
|
self.bridge._RESOURCE_SESSION_REQUEST_PATH,
|
|
{
|
|
"version": 1,
|
|
"transferId": "closing",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"metadata": {"eventId": "event-closing"},
|
|
"authMessage": {"type": "RCR"},
|
|
},
|
|
b"request",
|
|
link.link_id,
|
|
remote_identity,
|
|
time.time(),
|
|
)
|
|
|
|
self.assertEqual(response["reason"], "resource_session_unavailable")
|
|
self.assertEqual(state["provider_active"], 0)
|
|
emit.assert_not_called()
|
|
|
|
def test_provider_waits_for_previous_resource_before_reusing_link(self):
|
|
state, _link = self.session()
|
|
state["incoming"] = True
|
|
state["provider_active"] = 1
|
|
result = {}
|
|
|
|
def acquire():
|
|
result["reason"] = (
|
|
self.bridge._resource_session_provider_acquire_link_slot(
|
|
state,
|
|
"next-resource",
|
|
self.peer_hash,
|
|
)
|
|
)
|
|
|
|
thread = threading.Thread(target=acquire)
|
|
thread.start()
|
|
deadline = time.time() + 1
|
|
while (
|
|
state.get("provider_link_waiter_transfer") != "next-resource"
|
|
and time.time() < deadline
|
|
):
|
|
time.sleep(0.01)
|
|
self.assertEqual(
|
|
state.get("provider_link_waiter_transfer"),
|
|
"next-resource",
|
|
)
|
|
|
|
with self.bridge._resource_session_provider_capacity_condition:
|
|
state["provider_active"] = 0
|
|
self.bridge._resource_session_provider_capacity_condition.notify_all()
|
|
thread.join(timeout=1)
|
|
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertEqual(result["reason"], "")
|
|
self.assertEqual(state["provider_active"], 1)
|
|
self.assertEqual(state["provider_link_handoffs"], 1)
|
|
self.assertNotIn("provider_link_waiter_transfer", state)
|
|
|
|
def test_provider_response_waits_for_link_handoff_then_starts_resource(self):
|
|
state, link = self.session()
|
|
state["incoming"] = True
|
|
state["provider_active"] = 1
|
|
transfer_id = "handoff-response"
|
|
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
|
|
temp_file.write(b"handoff-resource")
|
|
file_path = temp_file.name
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id] = {
|
|
"allowedRecipientAddress": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"filePath": file_path,
|
|
"fileName": "handoff.bin",
|
|
"size": len(b"handoff-resource"),
|
|
"sha256": "",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"metadata": {"eventId": "event-handoff"},
|
|
"expires_at": time.time() + 60,
|
|
}
|
|
auth_emitted = threading.Event()
|
|
result = {}
|
|
|
|
def authorize_on_event(status, _payload):
|
|
if status != "auth":
|
|
return
|
|
waiter_key = self.bridge._resource_session_waiter_key(
|
|
state["linkId"],
|
|
transfer_id,
|
|
)
|
|
waiter = self.bridge._resource_session_provider_waiters[waiter_key]
|
|
waiter["authorized"] = True
|
|
waiter["event"].set()
|
|
auth_emitted.set()
|
|
|
|
def request():
|
|
result["response"] = self.bridge._resource_session_response_generator(
|
|
self.bridge._RESOURCE_SESSION_REQUEST_PATH,
|
|
{
|
|
"version": 1,
|
|
"transferId": transfer_id,
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"metadata": {"eventId": "event-handoff"},
|
|
"authMessage": {"type": "RCR"},
|
|
},
|
|
b"request",
|
|
link.link_id,
|
|
object(),
|
|
time.time(),
|
|
)
|
|
|
|
try:
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_destination_hash_for_identity",
|
|
return_value=self.peer_hash,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_qchat_file_emit",
|
|
side_effect=authorize_on_event,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_watch_provider_file",
|
|
):
|
|
thread = threading.Thread(target=request)
|
|
thread.start()
|
|
deadline = time.time() + 1
|
|
while (
|
|
state.get("provider_link_waiter_transfer") != transfer_id
|
|
and time.time() < deadline
|
|
):
|
|
time.sleep(0.01)
|
|
self.assertEqual(
|
|
state.get("provider_link_waiter_transfer"),
|
|
transfer_id,
|
|
)
|
|
self.assertFalse(auth_emitted.is_set())
|
|
|
|
with self.bridge._resource_session_provider_capacity_condition:
|
|
state["provider_active"] = 0
|
|
self.bridge._resource_session_provider_capacity_condition.notify_all()
|
|
thread.join(timeout=1)
|
|
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertTrue(auth_emitted.is_set())
|
|
response = result["response"]
|
|
self.assertIsInstance(response, tuple)
|
|
self.assertEqual(response[1]["transferId"], transfer_id)
|
|
self.assertEqual(response[0].read(), b"handoff-resource")
|
|
response[0].close()
|
|
self.assertEqual(state["provider_active"], 1)
|
|
self.assertEqual(state["provider_link_handoffs"], 1)
|
|
finally:
|
|
Path(file_path).unlink(missing_ok=True)
|
|
|
|
def test_provider_allows_only_one_waiting_link_handoff(self):
|
|
state, _link = self.session()
|
|
state["incoming"] = True
|
|
state["provider_active"] = 1
|
|
result = {}
|
|
|
|
def acquire():
|
|
result["reason"] = (
|
|
self.bridge._resource_session_provider_acquire_link_slot(
|
|
state,
|
|
"first-waiter",
|
|
self.peer_hash,
|
|
)
|
|
)
|
|
|
|
thread = threading.Thread(target=acquire)
|
|
thread.start()
|
|
deadline = time.time() + 1
|
|
while (
|
|
state.get("provider_link_waiter_transfer") != "first-waiter"
|
|
and time.time() < deadline
|
|
):
|
|
time.sleep(0.01)
|
|
|
|
self.assertEqual(
|
|
self.bridge._resource_session_provider_acquire_link_slot(
|
|
state,
|
|
"first-waiter",
|
|
self.peer_hash,
|
|
),
|
|
"duplicate_resource_request",
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._resource_session_provider_acquire_link_slot(
|
|
state,
|
|
"second-waiter",
|
|
self.peer_hash,
|
|
),
|
|
"resource_session_busy",
|
|
)
|
|
|
|
with self.bridge._resource_session_provider_capacity_condition:
|
|
state["provider_active"] = 0
|
|
self.bridge._resource_session_provider_capacity_condition.notify_all()
|
|
thread.join(timeout=1)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertEqual(result["reason"], "")
|
|
self.assertEqual(state["provider_active"], 1)
|
|
|
|
def test_provider_cancel_wakes_waiting_link_handoff(self):
|
|
state, _link = self.session()
|
|
state["incoming"] = True
|
|
state["provider_active"] = 1
|
|
result = {}
|
|
|
|
def acquire():
|
|
result["reason"] = (
|
|
self.bridge._resource_session_provider_acquire_link_slot(
|
|
state,
|
|
"cancelled-waiter",
|
|
self.peer_hash,
|
|
)
|
|
)
|
|
|
|
thread = threading.Thread(target=acquire)
|
|
thread.start()
|
|
deadline = time.time() + 1
|
|
while (
|
|
state.get("provider_link_waiter_transfer") != "cancelled-waiter"
|
|
and time.time() < deadline
|
|
):
|
|
time.sleep(0.01)
|
|
|
|
self.bridge._resource_session_cancel_provider_transfer(
|
|
state,
|
|
"cancelled-waiter",
|
|
)
|
|
thread.join(timeout=1)
|
|
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertEqual(result["reason"], "resource_requester_cancelled")
|
|
self.assertEqual(state["provider_active"], 1)
|
|
self.assertNotIn("provider_link_waiter_transfer", state)
|
|
|
|
def test_provider_times_out_when_previous_resource_never_releases(self):
|
|
state, link = self.session()
|
|
state["incoming"] = True
|
|
state["provider_active"] = 1
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_destination_hash_for_identity",
|
|
return_value=self.peer_hash,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_RESOURCE_SESSION_PROVIDER_LINK_HANDOFF_WAIT_SECONDS",
|
|
0.02,
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
response = self.bridge._resource_session_response_generator(
|
|
self.bridge._RESOURCE_SESSION_REQUEST_PATH,
|
|
{
|
|
"version": 1,
|
|
"transferId": "same-link-second-resource",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"metadata": {"eventId": "event-second-resource"},
|
|
"authMessage": {"type": "RCR"},
|
|
},
|
|
b"request",
|
|
link.link_id,
|
|
object(),
|
|
time.time(),
|
|
)
|
|
|
|
self.assertEqual(response["reason"], "resource_session_busy")
|
|
self.assertEqual(state["provider_active"], 1)
|
|
self.assertNotIn("provider_link_waiter_transfer", state)
|
|
emit.assert_not_called()
|
|
|
|
def test_provider_pending_auth_limit_rejects_without_emitting_auth(self):
|
|
state, link = self.session()
|
|
state["incoming"] = True
|
|
for index in range(self.bridge._RESOURCE_SESSION_PROVIDER_PENDING_AUTH_MAX):
|
|
self.bridge._resource_session_provider_waiters[f"existing:{index}"] = {
|
|
"peerPresenceHash": f"{index:032x}",
|
|
}
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_destination_hash_for_identity",
|
|
return_value=self.peer_hash,
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
response = self.bridge._resource_session_response_generator(
|
|
self.bridge._RESOURCE_SESSION_REQUEST_PATH,
|
|
{
|
|
"version": 1,
|
|
"transferId": "capacity",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"metadata": {"eventId": "event-capacity"},
|
|
"authMessage": {"type": "RCR"},
|
|
},
|
|
b"request",
|
|
link.link_id,
|
|
object(),
|
|
time.time(),
|
|
)
|
|
self.assertEqual(response["reason"], "resource_provider_busy")
|
|
self.assertEqual(state["provider_active"], 0)
|
|
emit.assert_not_called()
|
|
|
|
def test_provider_handoff_shares_pending_authorization_limit(self):
|
|
state, _link = self.session()
|
|
state["incoming"] = True
|
|
state["provider_active"] = 1
|
|
for index in range(self.bridge._RESOURCE_SESSION_PROVIDER_PENDING_AUTH_MAX):
|
|
self.bridge._resource_session_provider_waiters[f"existing:{index}"] = {
|
|
"peerPresenceHash": f"{index:032x}",
|
|
}
|
|
|
|
self.assertEqual(
|
|
self.bridge._resource_session_provider_acquire_link_slot(
|
|
state,
|
|
"bounded-handoff",
|
|
self.peer_hash,
|
|
),
|
|
"resource_provider_busy",
|
|
)
|
|
self.assertEqual(state["provider_active"], 1)
|
|
self.assertNotIn("provider_link_waiter_transfer", state)
|
|
|
|
def test_provider_waiting_for_auth_does_not_consume_transfer_capacity(self):
|
|
state, link = self.session()
|
|
state["incoming"] = True
|
|
auth_emitted = threading.Event()
|
|
result = {}
|
|
|
|
def on_emit(status, _payload):
|
|
if status == "auth":
|
|
auth_emitted.set()
|
|
|
|
def request():
|
|
result["response"] = self.bridge._resource_session_response_generator(
|
|
self.bridge._RESOURCE_SESSION_REQUEST_PATH,
|
|
{
|
|
"version": 1,
|
|
"transferId": "waiting-auth",
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"metadata": {"eventId": "event-waiting-auth"},
|
|
"authMessage": {"type": "RCR"},
|
|
},
|
|
b"request",
|
|
link.link_id,
|
|
object(),
|
|
time.time(),
|
|
)
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_destination_hash_for_identity",
|
|
return_value=self.peer_hash,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_qchat_file_emit",
|
|
side_effect=on_emit,
|
|
):
|
|
thread = threading.Thread(target=request)
|
|
thread.start()
|
|
self.assertTrue(auth_emitted.wait(1.0))
|
|
self.assertEqual(
|
|
sum(self.bridge._resource_session_provider_active_by_class.values()),
|
|
0,
|
|
)
|
|
waiter = next(
|
|
iter(self.bridge._resource_session_provider_waiters.values())
|
|
)
|
|
waiter["reason"] = "test_rejected"
|
|
waiter["event"].set()
|
|
thread.join(1.0)
|
|
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertEqual(result["response"]["reason"], "test_rejected")
|
|
self.assertEqual(state["provider_active"], 0)
|
|
self.assertFalse(self.bridge._resource_session_provider_waiters)
|
|
self.assertFalse(
|
|
self.bridge._resource_session_provider_pending_auth_by_peer
|
|
)
|
|
|
|
def test_late_authorization_discards_abandoned_registered_send(self):
|
|
state, _link = self.session()
|
|
state["incoming"] = True
|
|
transfer_id = "late-authorization"
|
|
pending = {
|
|
"allowedRecipientAddress": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"fileName": "late.bin",
|
|
"size": 128,
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"expires_at": time.time() + 60,
|
|
}
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id] = pending
|
|
|
|
with mock.patch.object(self.bridge, "_qchat_file_emit") as emit, mock.patch.object(
|
|
self.bridge,
|
|
"emit_resp",
|
|
) as emit_resp:
|
|
self.bridge.handle_authorize_qchat_file_resource(
|
|
"req",
|
|
{"linkId": state["linkId"], "transferId": transfer_id},
|
|
)
|
|
|
|
self.assertNotIn(
|
|
transfer_id,
|
|
self.bridge._qchat_file_pending_sends_by_transfer,
|
|
)
|
|
self.assertTrue(pending["cancelled"])
|
|
emit.assert_called_once_with("failed", mock.ANY)
|
|
self.assertEqual(
|
|
emit.call_args.args[1]["reason"],
|
|
"resource_authorization_no_longer_active",
|
|
)
|
|
emit_resp.assert_called_once_with(
|
|
"req",
|
|
False,
|
|
payload={"code": "unknown_resource_request"},
|
|
error="Unknown resource session request",
|
|
)
|
|
|
|
def test_late_authorization_cannot_discard_inflight_send(self):
|
|
state, _link = self.session()
|
|
state["incoming"] = True
|
|
transfer_id = "inflight-authorization"
|
|
pending = {
|
|
"allowedRecipientAddress": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"fileName": "active.bin",
|
|
"size": 128,
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
}
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id] = pending
|
|
self.bridge._resource_session_provider_inflight_transfers.add(transfer_id)
|
|
|
|
with mock.patch.object(self.bridge, "_qchat_file_emit") as emit, mock.patch.object(
|
|
self.bridge,
|
|
"emit_resp",
|
|
):
|
|
self.bridge.handle_authorize_qchat_file_resource(
|
|
"req",
|
|
{"linkId": state["linkId"], "transferId": transfer_id},
|
|
)
|
|
|
|
self.assertIs(
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id],
|
|
pending,
|
|
)
|
|
self.assertNotIn("cancelled", pending)
|
|
emit.assert_not_called()
|
|
|
|
def test_provider_capacity_preserves_live_and_sync_slots(self):
|
|
active = self.bridge._resource_session_provider_active_by_class
|
|
active["attachment"] = (
|
|
self.bridge._RESOURCE_SESSION_PROVIDER_ATTACHMENT_CONCURRENCY
|
|
)
|
|
self.assertFalse(
|
|
self.bridge._resource_session_provider_can_start_locked("attachment")
|
|
)
|
|
self.assertTrue(
|
|
self.bridge._resource_session_provider_can_start_locked("history")
|
|
)
|
|
active["history"] = 1
|
|
self.assertFalse(
|
|
self.bridge._resource_session_provider_can_start_locked("history")
|
|
)
|
|
self.assertTrue(
|
|
self.bridge._resource_session_provider_can_start_locked("live")
|
|
)
|
|
active["attachment"] = 0
|
|
active["history"] = 0
|
|
active["metadata"] = (
|
|
self.bridge._RESOURCE_SESSION_PROVIDER_METADATA_CONCURRENCY
|
|
)
|
|
self.assertFalse(
|
|
self.bridge._resource_session_provider_can_start_locked("metadata")
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._resource_session_provider_class(
|
|
"reticulum_chat_calendar",
|
|
"reticulum_chat_calendar",
|
|
),
|
|
"metadata",
|
|
)
|
|
|
|
def test_provider_capacity_waiters_are_prioritized(self):
|
|
live_waiter = {
|
|
"providerClass": "live",
|
|
"peerPresenceHash": self.peer_hash,
|
|
}
|
|
metadata_waiter = {
|
|
"providerClass": "metadata",
|
|
"peerPresenceHash": "cd" * 16,
|
|
}
|
|
history_waiter = {
|
|
"providerClass": "history",
|
|
"peerPresenceHash": "ef" * 16,
|
|
}
|
|
queue = self.bridge._resource_session_provider_capacity_queue
|
|
queue.extend([history_waiter, metadata_waiter, live_waiter])
|
|
self.assertFalse(
|
|
self.bridge._resource_session_provider_can_start_locked(
|
|
"metadata",
|
|
"cd" * 16,
|
|
metadata_waiter,
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
self.bridge._resource_session_provider_can_start_locked(
|
|
"history",
|
|
"ef" * 16,
|
|
history_waiter,
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
self.bridge._resource_session_provider_can_start_locked(
|
|
"live",
|
|
self.peer_hash,
|
|
live_waiter,
|
|
)
|
|
)
|
|
|
|
def test_ineligible_priority_waiter_does_not_block_other_peers(self):
|
|
blocked_peer = self.peer_hash
|
|
available_peer = "cd" * 16
|
|
live_waiter = {
|
|
"providerClass": "live",
|
|
"peerPresenceHash": blocked_peer,
|
|
}
|
|
history_waiter = {
|
|
"providerClass": "history",
|
|
"peerPresenceHash": available_peer,
|
|
}
|
|
self.bridge._resource_session_provider_capacity_queue.extend(
|
|
[live_waiter, history_waiter]
|
|
)
|
|
self.bridge._resource_session_provider_active_by_peer[blocked_peer] = (
|
|
self.bridge._RESOURCE_SESSION_PROVIDER_ACTIVE_MAX_PER_PEER
|
|
)
|
|
|
|
self.assertTrue(
|
|
self.bridge._resource_session_provider_can_start_locked(
|
|
"history",
|
|
available_peer,
|
|
history_waiter,
|
|
)
|
|
)
|
|
|
|
def test_provider_active_capacity_is_bounded_per_peer(self):
|
|
self.bridge._resource_session_provider_active_by_peer[self.peer_hash] = (
|
|
self.bridge._RESOURCE_SESSION_PROVIDER_ACTIVE_MAX_PER_PEER
|
|
)
|
|
self.assertFalse(
|
|
self.bridge._resource_session_provider_can_start_locked(
|
|
"live",
|
|
self.peer_hash,
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
self.bridge._resource_session_provider_can_start_locked(
|
|
"live",
|
|
"cd" * 16,
|
|
)
|
|
)
|
|
|
|
def test_attachment_capacity_reserves_two_slots_for_same_peer_chat(self):
|
|
self.bridge._resource_session_provider_active_by_class["attachment"] = (
|
|
self.bridge._RESOURCE_SESSION_PROVIDER_ATTACHMENT_MAX_PER_PEER
|
|
)
|
|
self.bridge._resource_session_provider_active_by_peer[self.peer_hash] = (
|
|
self.bridge._RESOURCE_SESSION_PROVIDER_ATTACHMENT_MAX_PER_PEER
|
|
)
|
|
self.bridge._resource_session_provider_active_attachments_by_peer[
|
|
self.peer_hash
|
|
] = self.bridge._RESOURCE_SESSION_PROVIDER_ATTACHMENT_MAX_PER_PEER
|
|
|
|
self.assertFalse(
|
|
self.bridge._resource_session_provider_can_start_locked(
|
|
"attachment",
|
|
self.peer_hash,
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
self.bridge._resource_session_provider_can_start_locked(
|
|
"history",
|
|
self.peer_hash,
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
self.bridge._resource_session_provider_can_start_locked(
|
|
"live",
|
|
self.peer_hash,
|
|
)
|
|
)
|
|
|
|
def test_provider_post_auth_wait_queue_is_bounded(self):
|
|
state, _link = self.session()
|
|
self.bridge._resource_session_provider_capacity_queue.extend(
|
|
{
|
|
"providerClass": "attachment",
|
|
"peerPresenceHash": f"{index:032x}",
|
|
"transferId": f"queued-{index}",
|
|
}
|
|
for index in range(
|
|
self.bridge._RESOURCE_SESSION_PROVIDER_CAPACITY_QUEUE_MAX
|
|
)
|
|
)
|
|
|
|
acquired = self.bridge._resource_session_provider_acquire_capacity(
|
|
"live",
|
|
"queue-full",
|
|
{},
|
|
state,
|
|
)
|
|
|
|
self.assertFalse(acquired)
|
|
self.assertFalse(
|
|
self.bridge._resource_session_provider_capacity_waiters_by_peer
|
|
)
|
|
|
|
def test_provider_classifies_attachment_ranges_separately(self):
|
|
self.assertEqual(
|
|
self.bridge._resource_session_provider_class(
|
|
"reticulum_group_resource_range",
|
|
"reticulum_group_resource_range",
|
|
),
|
|
"attachment",
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._resource_session_provider_class(
|
|
self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"reticulum_chat_history_page",
|
|
),
|
|
"history",
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._resource_session_provider_class(
|
|
self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"reticulum_chat_metadata_snapshot",
|
|
),
|
|
"metadata",
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._resource_session_provider_class(
|
|
self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"qortalland_chat",
|
|
),
|
|
"live",
|
|
)
|
|
|
|
def test_provider_uses_registered_resource_for_capacity_class(self):
|
|
state, link = self.session()
|
|
state["incoming"] = True
|
|
transfer_id = "authoritative-class"
|
|
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
|
|
temp_file.write(b"range-response")
|
|
file_path = temp_file.name
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id] = {
|
|
"allowedRecipientAddress": self.peer_hash,
|
|
"transferId": transfer_id,
|
|
"filePath": file_path,
|
|
"fileName": "range.bin",
|
|
"size": len(b"range-response"),
|
|
"sha256": "",
|
|
"resourceType": "reticulum_group_resource_range",
|
|
"metadata": {
|
|
"logicalResourceType": "reticulum_group_resource_range",
|
|
},
|
|
"expires_at": time.time() + 60,
|
|
}
|
|
captured_classes = []
|
|
|
|
def authorize(status, _payload):
|
|
if status != "auth":
|
|
return
|
|
waiter = next(
|
|
iter(self.bridge._resource_session_provider_waiters.values())
|
|
)
|
|
waiter["authorized"] = True
|
|
waiter["event"].set()
|
|
|
|
def watch(
|
|
_file,
|
|
_transfer_id,
|
|
_pending,
|
|
_state,
|
|
_request_id,
|
|
provider_class,
|
|
):
|
|
captured_classes.append(provider_class)
|
|
|
|
try:
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_destination_hash_for_identity",
|
|
return_value=self.peer_hash,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_qchat_file_emit",
|
|
side_effect=authorize,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_watch_provider_file",
|
|
side_effect=watch,
|
|
):
|
|
response = self.bridge._resource_session_response_generator(
|
|
self.bridge._RESOURCE_SESSION_REQUEST_PATH,
|
|
{
|
|
"version": 1,
|
|
"transferId": transfer_id,
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"metadata": {"eventId": "mislabelled-range"},
|
|
"authMessage": {"type": "RCR"},
|
|
},
|
|
b"request",
|
|
link.link_id,
|
|
object(),
|
|
time.time(),
|
|
)
|
|
self.assertIsInstance(response, tuple)
|
|
self.assertEqual(captured_classes, ["attachment"])
|
|
response[0].close()
|
|
finally:
|
|
Path(file_path).unlink(missing_ok=True)
|
|
|
|
def test_provider_slot_is_held_until_progressing_response_completes(self):
|
|
state, link = self.session()
|
|
state["incoming"] = True
|
|
state["provider_active"] = 1
|
|
transfer_id = "provider-progress"
|
|
request_id = bytes.fromhex("88" * 16)
|
|
|
|
class ProviderResource:
|
|
def __init__(self):
|
|
self.request_id = request_id
|
|
self.status = RNS.Resource.TRANSFERRING
|
|
self.progress = 0.0
|
|
|
|
def get_progress(self):
|
|
return self.progress
|
|
|
|
resource = ProviderResource()
|
|
link.outgoing_resources.append(resource)
|
|
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
|
|
temp_file.write(b"progressing-resource")
|
|
file_path = temp_file.name
|
|
file_handle = open(file_path, "rb")
|
|
pending = {
|
|
"transferId": transfer_id,
|
|
"fileName": "progress.bin",
|
|
"size": len(b"progressing-resource"),
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
}
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id] = pending
|
|
self.bridge._resource_session_provider_active_by_class["live"] = 1
|
|
|
|
try:
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_RESOURCE_SESSION_RESPONSE_STALL_TIMEOUT_SECONDS",
|
|
0.25,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_RESOURCE_SESSION_RESPONSE_INITIAL_PROGRESS_TIMEOUT_SECONDS",
|
|
0.25,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_schedule_idle_close",
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge._resource_session_watch_provider_file(
|
|
file_handle,
|
|
transfer_id,
|
|
pending,
|
|
state,
|
|
request_id,
|
|
"live",
|
|
)
|
|
for progress in (0.1, 0.2, 0.3, 0.4):
|
|
time.sleep(0.11)
|
|
resource.progress = progress
|
|
resource.status = RNS.Resource.COMPLETE
|
|
deadline = time.time() + 2
|
|
while state["provider_active"] > 0 and time.time() < deadline:
|
|
time.sleep(0.02)
|
|
|
|
self.assertEqual(state["provider_active"], 0)
|
|
self.assertNotIn(
|
|
transfer_id,
|
|
self.bridge._qchat_file_pending_sends_by_transfer,
|
|
)
|
|
self.assertTrue(
|
|
any(
|
|
call.args[0] == "sent"
|
|
and call.args[1]["transferId"] == transfer_id
|
|
for call in emit.call_args_list
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._resource_session_provider_active_by_class["live"],
|
|
0,
|
|
)
|
|
finally:
|
|
if not file_handle.closed:
|
|
file_handle.close()
|
|
Path(file_path).unlink(missing_ok=True)
|
|
|
|
def test_provider_zero_progress_uses_initial_progress_timeout(self):
|
|
state, link = self.session()
|
|
state["incoming"] = True
|
|
state["provider_active"] = 1
|
|
transfer_id = "provider-no-progress"
|
|
request_id = bytes.fromhex("89" * 16)
|
|
|
|
resource = mock.Mock()
|
|
resource.request_id = request_id
|
|
resource.status = RNS.Resource.TRANSFERRING
|
|
resource.get_progress.return_value = 0.0
|
|
link.outgoing_resources.append(resource)
|
|
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
|
|
temp_file.write(b"no-progress-resource")
|
|
file_path = temp_file.name
|
|
file_handle = open(file_path, "rb")
|
|
pending = {
|
|
"transferId": transfer_id,
|
|
"fileName": "no-progress.bin",
|
|
"size": len(b"no-progress-resource"),
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
}
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id] = pending
|
|
self.bridge._resource_session_provider_active_by_class["live"] = 1
|
|
|
|
try:
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_RESOURCE_SESSION_RESPONSE_INITIAL_PROGRESS_TIMEOUT_SECONDS",
|
|
0.12,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_RESOURCE_SESSION_RESPONSE_STALL_TIMEOUT_SECONDS",
|
|
1.0,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_schedule_idle_close",
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge._resource_session_watch_provider_file(
|
|
file_handle,
|
|
transfer_id,
|
|
pending,
|
|
state,
|
|
request_id,
|
|
"live",
|
|
)
|
|
deadline = time.time() + 1
|
|
while state["provider_active"] > 0 and time.time() < deadline:
|
|
time.sleep(0.01)
|
|
|
|
self.assertEqual(state["provider_active"], 0)
|
|
resource.cancel.assert_called_once_with()
|
|
self.assertTrue(
|
|
any(
|
|
call.args[0] == "failed"
|
|
and call.args[1]["reason"] == "resource_response_not_started"
|
|
for call in emit.call_args_list
|
|
)
|
|
)
|
|
finally:
|
|
if not file_handle.closed:
|
|
file_handle.close()
|
|
Path(file_path).unlink(missing_ok=True)
|
|
|
|
def test_provider_watcher_follows_all_resource_segments(self):
|
|
state, link = self.session()
|
|
state["incoming"] = True
|
|
state["provider_active"] = 1
|
|
transfer_id = "provider-segments"
|
|
request_id = bytes.fromhex("99" * 16)
|
|
|
|
class ProviderSegment:
|
|
def __init__(self, index, status, progress):
|
|
self.request_id = request_id
|
|
self.segment_index = index
|
|
self.total_segments = 2
|
|
self.status = status
|
|
self.progress = progress
|
|
self.next_segment = None
|
|
|
|
def get_progress(self):
|
|
return self.progress
|
|
|
|
second = ProviderSegment(2, RNS.Resource.TRANSFERRING, 0.5)
|
|
first = ProviderSegment(1, RNS.Resource.COMPLETE, 0.5)
|
|
first.next_segment = second
|
|
link.outgoing_resources.append(first)
|
|
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
|
|
temp_file.write(b"segmented-resource")
|
|
file_path = temp_file.name
|
|
file_handle = open(file_path, "rb")
|
|
pending = {
|
|
"transferId": transfer_id,
|
|
"fileName": "segments.bin",
|
|
"size": len(b"segmented-resource"),
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
}
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id] = pending
|
|
self.bridge._resource_session_provider_active_by_class["live"] = 1
|
|
|
|
try:
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_RESOURCE_SESSION_RESPONSE_STALL_TIMEOUT_SECONDS",
|
|
1.0,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_RESOURCE_SESSION_RESPONSE_INITIAL_PROGRESS_TIMEOUT_SECONDS",
|
|
1.0,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_schedule_idle_close",
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
self.bridge._resource_session_watch_provider_file(
|
|
file_handle,
|
|
transfer_id,
|
|
pending,
|
|
state,
|
|
request_id,
|
|
"live",
|
|
)
|
|
time.sleep(0.15)
|
|
self.assertEqual(state["provider_active"], 1)
|
|
self.assertFalse(
|
|
any(call.args[0] == "sent" for call in emit.call_args_list)
|
|
)
|
|
second.progress = 1.0
|
|
second.status = RNS.Resource.COMPLETE
|
|
deadline = time.time() + 2
|
|
while time.time() < deadline:
|
|
sent = any(
|
|
call.args[0] == "sent"
|
|
for call in emit.call_args_list
|
|
)
|
|
if state["provider_active"] == 0 and sent:
|
|
break
|
|
time.sleep(0.02)
|
|
|
|
self.assertEqual(state["provider_active"], 0)
|
|
self.assertTrue(
|
|
any(call.args[0] == "sent" for call in emit.call_args_list)
|
|
)
|
|
finally:
|
|
if not file_handle.closed:
|
|
file_handle.close()
|
|
Path(file_path).unlink(missing_ok=True)
|
|
|
|
def test_provider_watcher_releases_inflight_admission_after_pending_replacement(self):
|
|
state, link = self.session()
|
|
state["incoming"] = True
|
|
state["provider_active"] = 1
|
|
transfer_id = "provider-replaced-pending"
|
|
request_id = bytes.fromhex("aa" * 16)
|
|
resource = mock.Mock()
|
|
resource.request_id = request_id
|
|
resource.status = RNS.Resource.TRANSFERRING
|
|
resource.get_progress.return_value = 0.5
|
|
link.outgoing_resources.append(resource)
|
|
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
|
|
temp_file.write(b"replacement-resource")
|
|
file_path = temp_file.name
|
|
file_handle = open(file_path, "rb")
|
|
pending = {
|
|
"transferId": transfer_id,
|
|
"fileName": "original.bin",
|
|
"size": len(b"replacement-resource"),
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
}
|
|
replacement = {**pending, "fileName": "retry.bin"}
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id] = pending
|
|
self.bridge._resource_session_provider_inflight_transfers.add(transfer_id)
|
|
self.bridge._resource_session_provider_active_by_class["live"] = 1
|
|
|
|
try:
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_schedule_idle_close",
|
|
):
|
|
self.bridge._resource_session_watch_provider_file(
|
|
file_handle,
|
|
transfer_id,
|
|
pending,
|
|
state,
|
|
request_id,
|
|
"live",
|
|
)
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id] = (
|
|
replacement
|
|
)
|
|
resource.status = RNS.Resource.COMPLETE
|
|
deadline = time.time() + 1
|
|
while state["provider_active"] > 0 and time.time() < deadline:
|
|
time.sleep(0.01)
|
|
|
|
self.assertIs(
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id],
|
|
replacement,
|
|
)
|
|
self.assertNotIn(
|
|
transfer_id,
|
|
self.bridge._resource_session_provider_inflight_transfers,
|
|
)
|
|
finally:
|
|
if not file_handle.closed:
|
|
file_handle.close()
|
|
Path(file_path).unlink(missing_ok=True)
|
|
|
|
def test_provider_watcher_preserves_file_during_response_handoff(self):
|
|
state, _link = self.session()
|
|
state["incoming"] = True
|
|
state["provider_active"] = 1
|
|
transfer_id = "provider-handoff-cancel"
|
|
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
|
|
temp_file.write(b"handoff-resource")
|
|
file_path = temp_file.name
|
|
file_handle = open(file_path, "rb")
|
|
pending = {
|
|
"transferId": transfer_id,
|
|
"fileName": "handoff.bin",
|
|
"size": len(b"handoff-resource"),
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"cancelled": True,
|
|
}
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id] = pending
|
|
self.bridge._resource_session_provider_active_by_class["live"] = 1
|
|
|
|
try:
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_RESOURCE_SESSION_PROVIDER_RESPONSE_START_GRACE_SECONDS",
|
|
0.1,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_resource_session_schedule_idle_close",
|
|
):
|
|
self.bridge._resource_session_watch_provider_file(
|
|
file_handle,
|
|
transfer_id,
|
|
pending,
|
|
state,
|
|
b"request",
|
|
"live",
|
|
)
|
|
time.sleep(0.03)
|
|
self.assertFalse(file_handle.closed)
|
|
deadline = time.time() + 1
|
|
while state["provider_active"] > 0 and time.time() < deadline:
|
|
time.sleep(0.01)
|
|
|
|
self.assertTrue(file_handle.closed)
|
|
self.assertEqual(state["provider_active"], 0)
|
|
self.assertEqual(
|
|
self.bridge._resource_session_provider_active_by_class["live"],
|
|
0,
|
|
)
|
|
finally:
|
|
if not file_handle.closed:
|
|
file_handle.close()
|
|
Path(file_path).unlink(missing_ok=True)
|
|
|
|
def test_provider_cancel_wakes_auth_and_marks_registered_send(self):
|
|
state, _link = self.session()
|
|
state["incoming"] = True
|
|
transfer_id = "provider-cancel"
|
|
waiter = {
|
|
"event": threading.Event(),
|
|
"authorized": False,
|
|
"reason": "resource_authorization_timeout",
|
|
}
|
|
waiter_key = self.bridge._resource_session_waiter_key(
|
|
state["linkId"],
|
|
transfer_id,
|
|
)
|
|
pending = {
|
|
"transferId": transfer_id,
|
|
"allowedRecipientAddress": self.peer_hash,
|
|
}
|
|
self.bridge._resource_session_provider_waiters[waiter_key] = waiter
|
|
self.bridge._qchat_file_pending_sends_by_transfer[transfer_id] = pending
|
|
|
|
self.bridge._resource_session_cancel_provider_transfer(
|
|
state,
|
|
transfer_id,
|
|
)
|
|
|
|
self.assertTrue(waiter["event"].is_set())
|
|
self.assertEqual(waiter["reason"], "resource_requester_cancelled")
|
|
self.assertTrue(pending["cancelled"])
|
|
|
|
def test_provider_remembers_cancel_that_arrives_before_request(self):
|
|
state, link = self.session()
|
|
state["incoming"] = True
|
|
transfer_id = "cancel-before-request"
|
|
self.bridge._resource_session_cancel_provider_transfer(
|
|
state,
|
|
transfer_id,
|
|
)
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_destination_hash_for_identity",
|
|
return_value=self.peer_hash,
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
response = self.bridge._resource_session_response_generator(
|
|
self.bridge._RESOURCE_SESSION_REQUEST_PATH,
|
|
{
|
|
"version": 1,
|
|
"transferId": transfer_id,
|
|
"resourceType": self.bridge._RETICULUM_CHAT_RESOURCE_TYPE,
|
|
"metadata": {"eventId": "cancelled-event"},
|
|
"authMessage": {"type": "RCR"},
|
|
},
|
|
b"request",
|
|
link.link_id,
|
|
object(),
|
|
time.time(),
|
|
)
|
|
|
|
self.assertEqual(response["reason"], "resource_requester_cancelled")
|
|
self.assertEqual(state["provider_active"], 0)
|
|
emit.assert_not_called()
|
|
|
|
def test_cancelled_request_does_not_close_reusable_session(self):
|
|
state, link = self.session()
|
|
pending = self.pending("cancel-me")
|
|
receipt = FakeSessionReceipt()
|
|
receipt.resource = mock.Mock()
|
|
job = {
|
|
"pending": pending,
|
|
"created_at": time.time(),
|
|
"followers": [],
|
|
"session": state,
|
|
"semanticKey": "cancel-key",
|
|
"receipt": receipt,
|
|
}
|
|
state["active_requests"]["cancel-me"] = job
|
|
self.bridge._resource_session_jobs_by_transfer["cancel-me"] = job
|
|
self.bridge._resource_session_jobs_by_semantic_key["cancel-key"] = job
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, pending)
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_qchat_file_emit",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_send_packet_on_link",
|
|
return_value=True,
|
|
) as send_packet:
|
|
closed = self.bridge._qchat_file_cancel_transfer(
|
|
"cancel-me",
|
|
self.peer_hash,
|
|
"test-cancel",
|
|
)
|
|
|
|
self.assertEqual(closed, 0)
|
|
self.assertFalse(link.teardown_called)
|
|
receipt.resource.cancel.assert_called_once_with()
|
|
self.assertEqual(receipt.status, FakeSessionReceipt.FAILED)
|
|
cancel_wire = json.loads(send_packet.call_args.args[1].decode("utf-8"))
|
|
self.assertEqual(
|
|
cancel_wire,
|
|
{
|
|
"type": self.bridge._RESOURCE_SESSION_CANCEL_TYPE,
|
|
"transferId": "cancel-me",
|
|
},
|
|
)
|
|
self.assertIn(state["sessionKey"], self.bridge._resource_sessions_by_key)
|
|
self.assertTrue(job["completed"])
|
|
|
|
def test_authorization_timeout_retires_only_the_suspect_session(self):
|
|
state, link = self.session(lane="bulk", slot=0)
|
|
surviving, surviving_link = self.session(lane="bulk", slot=1)
|
|
pending = self.pending(
|
|
"authorization-timeout",
|
|
resource_type="reticulum_group_resource_range",
|
|
)
|
|
receipt = FakeSessionReceipt()
|
|
receipt.resource = mock.Mock()
|
|
job = {
|
|
"pending": pending,
|
|
"created_at": time.time(),
|
|
"followers": [],
|
|
"session": state,
|
|
"semanticKey": "authorization-timeout-key",
|
|
"receipt": receipt,
|
|
}
|
|
queued = {
|
|
"pending": self.pending(
|
|
"queued-after-timeout",
|
|
resource_type="reticulum_group_resource_range",
|
|
),
|
|
"created_at": time.time(),
|
|
"followers": [],
|
|
"session": state,
|
|
"semanticKey": "queued-after-timeout-key",
|
|
}
|
|
state["active_requests"]["authorization-timeout"] = job
|
|
state["pending_jobs"].append(queued)
|
|
self.bridge._resource_session_jobs_by_transfer["authorization-timeout"] = job
|
|
self.bridge._resource_session_jobs_by_transfer["queued-after-timeout"] = queued
|
|
self.bridge._resource_session_jobs_by_semantic_key[
|
|
"authorization-timeout-key"
|
|
] = job
|
|
self.bridge._resource_session_jobs_by_semantic_key[
|
|
"queued-after-timeout-key"
|
|
] = queued
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, pending)
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_qchat_file_emit",
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_send_packet_on_link",
|
|
return_value=True,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_teardown_reticulum_link_bounded",
|
|
) as teardown:
|
|
closed = self.bridge._qchat_file_cancel_transfer(
|
|
"authorization-timeout",
|
|
self.peer_hash,
|
|
"resource_authorization_timeout",
|
|
)
|
|
|
|
self.assertEqual(closed, 1)
|
|
self.assertTrue(state["closing"])
|
|
self.assertNotIn(state["sessionKey"], self.bridge._resource_sessions_by_key)
|
|
teardown.assert_called_once()
|
|
self.assertIs(teardown.call_args.args[0], link)
|
|
receipt.resource.cancel.assert_called_once_with()
|
|
self.assertEqual(receipt.status, FakeSessionReceipt.FAILED)
|
|
self.assertFalse(surviving.get("closing", False))
|
|
self.assertFalse(surviving_link.teardown_called)
|
|
self.assertIs(queued.get("session"), surviving)
|
|
self.assertIs(
|
|
surviving["active_requests"].get("queued-after-timeout"),
|
|
queued,
|
|
)
|
|
|
|
def test_cancel_during_request_handoff_cancels_late_receipt(self):
|
|
state, link = self.session()
|
|
pending = self.pending("handoff-cancel")
|
|
job = {
|
|
"pending": pending,
|
|
"created_at": time.time(),
|
|
"followers": [],
|
|
"session": state,
|
|
"semanticKey": "handoff-key",
|
|
}
|
|
self.bridge._resource_session_jobs_by_transfer["handoff-cancel"] = job
|
|
self.bridge._resource_session_jobs_by_semantic_key["handoff-key"] = job
|
|
self.bridge._qchat_file_store_pending_receive(self.peer_hash, pending)
|
|
receipt = FakeSessionReceipt()
|
|
receipt.resource = mock.Mock()
|
|
|
|
def request_then_cancel(*_args, **_kwargs):
|
|
self.bridge._qchat_file_cancel_transfer(
|
|
"handoff-cancel",
|
|
self.peer_hash,
|
|
"test-handoff-cancel",
|
|
)
|
|
return receipt
|
|
|
|
with mock.patch.object(
|
|
link,
|
|
"request",
|
|
side_effect=request_then_cancel,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_send_packet_on_link",
|
|
return_value=True,
|
|
), mock.patch.object(self.bridge, "_qchat_file_emit") as emit:
|
|
dispatched = self.bridge._resource_session_dispatch_job(state, job)
|
|
|
|
self.assertFalse(dispatched)
|
|
self.assertTrue(job["completed"])
|
|
self.assertTrue(job["cancelled"])
|
|
self.assertEqual(receipt.status, FakeSessionReceipt.FAILED)
|
|
receipt.resource.cancel.assert_called_once_with()
|
|
self.assertNotIn("receipt", job)
|
|
self.assertFalse(
|
|
any(call.args[0] == "auth_sent" for call in emit.call_args_list)
|
|
)
|
|
|
|
|
|
class PresenceBridgeOverlayGoodOutboundCacheTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.bridge._reticulum_config_dir = self.temp_dir.name
|
|
self.bridge._overlay_good_outbound_cache.clear()
|
|
self.bridge._overlay_good_outbound_cache_loaded = False
|
|
self.bridge._overlay_good_outbound_cache_dirty = False
|
|
self.bridge._overlay_good_outbound_cache_last_write_at = 0.0
|
|
|
|
def tearDown(self):
|
|
self.temp_dir.cleanup()
|
|
self.bridge._shutdown.clear()
|
|
|
|
def _write_cache(self, payload):
|
|
path = self.bridge._overlay_good_outbound_cache_path()
|
|
with open(path, "w", encoding="utf-8") as cache_file:
|
|
json.dump(payload, cache_file)
|
|
return path
|
|
|
|
def test_namespace_fingerprint_covers_all_destination_name_components(self):
|
|
baseline = self.bridge._overlay_good_outbound_cache_namespace_fingerprint()
|
|
for field_name in ("APP_NAMESPACE", "PRESENCE_ASPECT", "PRESENCE_VERSION"):
|
|
with self.subTest(field=field_name), mock.patch.object(
|
|
self.bridge,
|
|
field_name,
|
|
f"{getattr(self.bridge, field_name)}-changed",
|
|
):
|
|
self.assertNotEqual(
|
|
self.bridge._overlay_good_outbound_cache_namespace_fingerprint(),
|
|
baseline,
|
|
)
|
|
|
|
def test_old_cache_is_replaced_without_seeding_old_namespace_peers(self):
|
|
old_peer = "ab" * 16
|
|
path = self._write_cache(
|
|
{
|
|
"version": 1,
|
|
"updatedAt": time.time(),
|
|
"peers": [{"peerHash": old_peer, "lastRxAt": time.time()}],
|
|
}
|
|
)
|
|
|
|
self.bridge._load_overlay_good_outbound_cache()
|
|
|
|
self.assertEqual(self.bridge._overlay_good_outbound_cache, {})
|
|
with open(path, "r", encoding="utf-8") as cache_file:
|
|
replacement = json.load(cache_file)
|
|
self.assertEqual(
|
|
replacement["version"],
|
|
self.bridge._OVERLAY_GOOD_OUTBOUND_CACHE_VERSION,
|
|
)
|
|
self.assertEqual(
|
|
replacement["namespaceFingerprint"],
|
|
self.bridge._overlay_good_outbound_cache_namespace_fingerprint(),
|
|
)
|
|
self.assertEqual(replacement["peers"], [])
|
|
|
|
def test_current_version_cache_from_another_namespace_is_replaced(self):
|
|
path = self._write_cache(
|
|
{
|
|
"version": self.bridge._OVERLAY_GOOD_OUTBOUND_CACHE_VERSION,
|
|
"namespaceFingerprint": "wrong-namespace",
|
|
"updatedAt": time.time(),
|
|
"peers": [{"peerHash": "cd" * 16, "lastRxAt": time.time()}],
|
|
}
|
|
)
|
|
|
|
self.bridge._load_overlay_good_outbound_cache()
|
|
|
|
self.assertEqual(self.bridge._overlay_good_outbound_cache, {})
|
|
with open(path, "r", encoding="utf-8") as cache_file:
|
|
replacement = json.load(cache_file)
|
|
self.assertEqual(
|
|
replacement["namespaceFingerprint"],
|
|
self.bridge._overlay_good_outbound_cache_namespace_fingerprint(),
|
|
)
|
|
|
|
def test_failed_recall_removes_cached_peer_instead_of_marking_candidate(self):
|
|
peer_hash = "ef" * 16
|
|
self.bridge._overlay_good_outbound_cache_loaded = True
|
|
self.bridge._overlay_good_outbound_cache[peer_hash] = {
|
|
"first_rx_at": time.time(),
|
|
"last_rx_at": time.time(),
|
|
"rx_count": 1,
|
|
}
|
|
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"_overlay_peer_available_for_new_outbound",
|
|
return_value=True,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"ensure_known_peer_from_recall",
|
|
return_value=False,
|
|
), mock.patch.object(
|
|
self.bridge,
|
|
"_mark_candidate_peer",
|
|
) as mark_candidate, mock.patch.object(
|
|
self.bridge,
|
|
"_flush_overlay_good_outbound_cache",
|
|
) as flush_cache:
|
|
self.bridge._seed_overlay_good_outbound_cache_candidates()
|
|
|
|
mark_candidate.assert_not_called()
|
|
flush_cache.assert_called_once_with(force=True)
|
|
self.assertNotIn(peer_hash, self.bridge._overlay_good_outbound_cache)
|
|
self.assertNotIn(peer_hash, self.bridge._candidate_peers)
|
|
|
|
|
|
class PresenceBridgeAccountEndpointLeaseTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
self.bridge._destination = FakeDestination()
|
|
|
|
def lease(self, address, destination, session, verification="direct-bound", offset=0):
|
|
now_ms = int(time.time() * 1000)
|
|
return {
|
|
"address": address,
|
|
"destinationHash": destination,
|
|
"sessionId": session,
|
|
"lastSeen": now_ms + offset,
|
|
"expiresAt": now_ms + 45_000 + offset,
|
|
"verification": verification,
|
|
}
|
|
|
|
def test_one_transport_can_serve_multiple_signed_account_leases(self):
|
|
destination = "aa" * 16
|
|
self.bridge._set_verified_overlay_peers(
|
|
[{"destinationHash": destination, "lastSeen": int(time.time() * 1000)}],
|
|
[destination],
|
|
[
|
|
self.lease("Q-account-a", destination, "session-a"),
|
|
self.lease("Q-account-b", destination, "session-b"),
|
|
],
|
|
)
|
|
|
|
self.assertNotIn("address", self.bridge._verified_overlay_peers[destination])
|
|
self.assertEqual(
|
|
self.bridge._resolve_verified_game_peer("Q-account-a", destination),
|
|
destination,
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._resolve_verified_game_peer("Q-account-b", destination),
|
|
destination,
|
|
)
|
|
|
|
def test_account_switch_removes_only_the_ended_lease(self):
|
|
destination = "aa" * 16
|
|
self.bridge._set_verified_overlay_peers(
|
|
[{"destinationHash": destination, "lastSeen": int(time.time() * 1000)}],
|
|
[destination],
|
|
[
|
|
self.lease("Q-account-a", destination, "session-a"),
|
|
self.lease("Q-account-b", destination, "session-b"),
|
|
],
|
|
)
|
|
self.bridge._set_verified_overlay_peers(
|
|
[{"destinationHash": destination, "lastSeen": int(time.time() * 1000)}],
|
|
[destination],
|
|
[self.lease("Q-account-b", destination, "session-b", offset=1)],
|
|
)
|
|
|
|
self.assertIsNone(
|
|
self.bridge._resolve_verified_game_peer("Q-account-a", destination)
|
|
)
|
|
self.assertEqual(
|
|
self.bridge._resolve_verified_game_peer("Q-account-b", destination),
|
|
destination,
|
|
)
|
|
self.assertIn(destination, self.bridge._verified_overlay_peers)
|
|
|
|
def test_expired_account_lease_is_never_resolved(self):
|
|
destination = "aa" * 16
|
|
expired = self.lease("Q-account-a", destination, "session-a")
|
|
expired["expiresAt"] = int(time.time() * 1000) - 1
|
|
self.bridge._set_verified_overlay_peers(
|
|
[{"destinationHash": destination, "lastSeen": int(time.time() * 1000)}],
|
|
[destination],
|
|
[expired],
|
|
)
|
|
|
|
self.assertIsNone(
|
|
self.bridge._resolve_verified_game_peer("Q-account-a", destination)
|
|
)
|
|
self.assertIn(destination, self.bridge._verified_overlay_peers)
|
|
|
|
def test_unpreferred_resolution_uses_strongest_fresh_lease(self):
|
|
relayed = "aa" * 16
|
|
direct = "bb" * 16
|
|
self.bridge._set_verified_overlay_peers(
|
|
[
|
|
{"destinationHash": relayed, "lastSeen": int(time.time() * 1000)},
|
|
{"destinationHash": direct, "lastSeen": int(time.time() * 1000)},
|
|
],
|
|
[relayed, direct],
|
|
[
|
|
self.lease(
|
|
"Q-account", relayed, "session-relayed", "relayed-bound", 10
|
|
),
|
|
self.lease(
|
|
"Q-account", direct, "session-direct", "direct-bound", 0
|
|
),
|
|
],
|
|
)
|
|
|
|
self.assertEqual(
|
|
self.bridge._resolve_verified_game_peer("Q-account"), direct
|
|
)
|
|
|
|
|
|
class PresenceBridgePinnedChatPeersTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bridge = load_bridge()
|
|
|
|
def tearDown(self):
|
|
self.bridge._shutdown.clear()
|
|
|
|
def test_pins_only_current_account_endpoint_leases_and_clears_them(self):
|
|
peer_hash = "aa" * 16
|
|
expires_at = time.time() + 60
|
|
self.bridge._account_endpoint_leases = {
|
|
"Q-account": {
|
|
peer_hash: {
|
|
"destination_hash": peer_hash,
|
|
"expires_at": expires_at,
|
|
}
|
|
}
|
|
}
|
|
responses = []
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"emit_resp",
|
|
side_effect=lambda *args, **kwargs: responses.append((args, kwargs)),
|
|
), mock.patch.object(
|
|
self.bridge, "_enqueue_scheduler_task", return_value=True
|
|
), mock.patch.object(
|
|
self.bridge, "ensure_known_peer_from_recall", return_value=True
|
|
):
|
|
self.bridge.handle_configure_reticulum_chat_pinned_peers(
|
|
"pin",
|
|
{
|
|
"peers": [
|
|
{
|
|
"accountAddress": "Q-account",
|
|
"destinationHash": peer_hash,
|
|
"expiresAt": int((expires_at + 30) * 1000),
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
self.bridge._pinned_chat_overlay_peers,
|
|
{peer_hash: expires_at},
|
|
)
|
|
self.assertTrue(responses[-1][1]["payload"]["maintenanceQueued"])
|
|
|
|
self.bridge.handle_configure_reticulum_chat_pinned_peers(
|
|
"clear", {"peers": []}
|
|
)
|
|
self.assertEqual(self.bridge._pinned_chat_overlay_peers, {})
|
|
|
|
def test_rejects_a_pin_without_a_current_endpoint_lease(self):
|
|
responses = []
|
|
with mock.patch.object(
|
|
self.bridge,
|
|
"emit_resp",
|
|
side_effect=lambda *args, **kwargs: responses.append((args, kwargs)),
|
|
):
|
|
self.bridge.handle_configure_reticulum_chat_pinned_peers(
|
|
"pin",
|
|
{
|
|
"peers": [
|
|
{
|
|
"accountAddress": "Q-account",
|
|
"destinationHash": "bb" * 16,
|
|
"expiresAt": int((time.time() + 60) * 1000),
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
self.assertFalse(responses[-1][0][1])
|
|
self.assertEqual(self.bridge._pinned_chat_overlay_peers, {})
|
|
|
|
def test_expired_pin_keeps_an_otherwise_valid_overlay_neighbor(self):
|
|
peer_hash = "cc" * 16
|
|
self.bridge._pinned_chat_overlay_peers[peer_hash] = time.time() - 1
|
|
self.bridge._active_overlay_neighbors[peer_hash] = time.time()
|
|
self.bridge._candidate_peers[peer_hash] = {
|
|
"last_seen": time.time(),
|
|
"source": "test",
|
|
}
|
|
|
|
self.assertEqual(self.bridge._prune_pinned_chat_overlay_peers(), set())
|
|
self.assertIn(peer_hash, self.bridge._active_overlay_neighbors)
|
|
self.assertIn(peer_hash, self.bridge._candidate_peers)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|