blob: df6377352b04fa110c38a9fe49036ec00aa4b197 (
plain) (
blame)
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
|
"""
The node's source text, for the tests that have to read it.
A test that reads source to prove something is *absent* — no bare
`ensure_future`, no exception text sent to a peer, no GEK unwrapped over MNP —
goes on passing when the code it guards moves to another file: it simply stops
looking at that code. So every such test takes its text from here, and the
file sets below are derived from the tree rather than listed, so that a module
added later is read without anybody having to remember it.
"""
import inspect
import textwrap
from pathlib import Path
SRC = Path(__file__).resolve().parents[1] / "src" / "meshbay_node"
TRANSPORT = SRC / "transport"
# The WebRTC transport is `webrtc_server.py` and whatever has been split out of
# it into `transport/webrtc/`. The other modules beside them are not part of it
# and are not held to its rules by these tests: QUIC is a transport of its own
# and not at parity (docs/MESHBAY_DESIGN.md §15.3), and `wire.py`, `tls_cert.py`
# and the ICE helpers serve both or neither.
WEBRTC_PACKAGE = TRANSPORT / "webrtc"
def webrtc_files() -> list[Path]:
"""Every module of the WebRTC transport, however it is split up."""
files = [TRANSPORT / "webrtc_server.py"]
if WEBRTC_PACKAGE.is_dir():
files += sorted(p for p in WEBRTC_PACKAGE.rglob("*.py")
if "__pycache__" not in p.parts)
return files
def webrtc_source() -> str:
return "\n".join(p.read_text(encoding="utf-8") for p in webrtc_files())
def session_source() -> str:
"""The body of WebRTCPeerSession, including every class it is built from."""
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
return "\n".join(inspect.getsource(k) for k in WebRTCPeerSession.__mro__
if k.__module__.startswith("meshbay_node."))
def session_method(name: str) -> str:
"""One method of WebRTCPeerSession, wherever it is defined."""
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
return textwrap.dedent(inspect.getsource(getattr(WebRTCPeerSession, name)))
|