1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
|
"""
transport/stun_multi — the node must query every configured STUN server, not
just the first (which is all aiortc/aioice keep on their own).
"""
import asyncio
import time
import aioice.ice as _ice
import pytest
from meshbay_node.transport import stun_multi
@pytest.fixture(autouse=True)
def _restore():
"""Each test gets a clean, un-patched module."""
saved = (_ice.server_reflexive_candidate, stun_multi._installed,
list(stun_multi._servers))
stun_multi._installed = False
stun_multi._servers = []
yield
(_ice.server_reflexive_candidate, stun_multi._installed,
stun_multi._servers) = saved[0], saved[1], saved[2]
@pytest.fixture
def _identity_dns(monkeypatch):
monkeypatch.setattr(stun_multi.socket, "gethostbyname", lambda host: host)
class _Local:
host, port, component, transport = "192.168.1.50", 54000, 1, "udp"
class _Response:
def __init__(self, ip, port):
self.attributes = {"XOR-MAPPED-ADDRESS": (ip, port)}
class _Protocol:
"""A fake StunProtocol whose .request() behaviour is scripted per address."""
def __init__(self, behaviour):
self.local_candidate = _Local()
self._behaviour = behaviour # {(host, port): callable | value}
self.cancelled = 0
async def request(self, message, addr):
action = self._behaviour.get(addr)
try:
if action is None:
raise OSError(f"unreachable {addr}")
if callable(action):
return await action(addr)
return action, addr
except asyncio.CancelledError:
self.cancelled += 1
raise
@pytest.mark.parametrize("url,expected", [
("stun:stun.l.google.com:19302", ("stun.l.google.com", 19302)),
("stun:stun.cloudflare.com:3478", ("stun.cloudflare.com", 3478)),
("stun:example.org", ("example.org", 3478)),
("stun:example.org?transport=udp", ("example.org", 3478)),
])
def test_parse(url, expected):
assert stun_multi._parse(url) == expected
def test_set_servers_parses_and_drops_empty():
stun_multi.set_servers(["stun:a:1", "stun:b", "stun:", "stun:c:3478?transport=udp"])
assert stun_multi._servers == [("a", 1), ("b", 3478), ("c", 3478)]
def test_install_patches_and_is_idempotent():
stun_multi.install(["stun:a:3478"])
assert _ice.server_reflexive_candidate is stun_multi._fanout_server_reflexive_candidate
first = _ice.server_reflexive_candidate
stun_multi.install(["stun:b:3478", "stun:c:3478"])
assert _ice.server_reflexive_candidate is first # not re-wrapped
assert stun_multi._servers == [("b", 3478), ("c", 3478)] # list still updated
async def test_fanout_first_answer_wins(_identity_dns):
async def slow(addr):
await asyncio.sleep(30)
proto = _Protocol({
("slow", 3478): slow,
("fast", 19302): _Response("203.0.113.7", 5555),
# ("broken", 3478) -> not in map -> OSError
})
stun_multi.set_servers(["stun:slow:3478", "stun:fast:19302", "stun:broken:3478"])
t0 = time.monotonic()
cand, extra = await stun_multi._fanout_server_reflexive_candidate(proto, ("x", 1))
elapsed = time.monotonic() - t0
assert extra is None
assert cand.type == "srflx"
assert (cand.host, cand.port) == ("203.0.113.7", 5555)
assert cand.related_address == _Local.host
assert elapsed < 1.0 # did not wait on `slow`
await asyncio.sleep(0) # let cancellations propagate
assert proto.cancelled >= 1 # the slow query was torn down
async def test_all_servers_fail_raises_without_long_hang(_identity_dns, monkeypatch):
monkeypatch.setattr(stun_multi, "_QUERY_TIMEOUT", 0.3)
proto = _Protocol({}) # every addr -> OSError
stun_multi.set_servers(["stun:a:3478", "stun:b:3478"])
t0 = time.monotonic()
with pytest.raises(OSError):
await stun_multi._fanout_server_reflexive_candidate(proto, ("x", 1))
assert time.monotonic() - t0 < 2.0
async def test_empty_list_falls_back_to_passed_server(_identity_dns):
proto = _Protocol({("fallback.example", 3478): _Response("198.51.100.4", 9)})
stun_multi.set_servers([]) # no fan-out configured
cand, _ = await stun_multi._fanout_server_reflexive_candidate(
proto, ("fallback.example", 3478))
assert (cand.host, cand.port) == ("198.51.100.4", 9)
async def test_unresolvable_names_raise(monkeypatch):
def boom(host):
raise OSError("name resolution failed")
monkeypatch.setattr(stun_multi.socket, "gethostbyname", boom)
proto = _Protocol({})
stun_multi.set_servers(["stun:a:3478", "stun:b:3478"])
with pytest.raises(OSError):
await stun_multi._fanout_server_reflexive_candidate(proto, ("x", 1))
|