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
|
"""
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 ast
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)))
# The daemon is `daemon.py`, every class NodeDaemon is built from, and the CLI
# once `main()` is split out into a package of its own.
CLI_PACKAGE = SRC / "cli"
def daemon_files() -> list[Path]:
"""Every module of the daemon and its CLI, however they are split up."""
from meshbay_node.daemon import NodeDaemon
files = [SRC / "daemon.py"]
for klass in NodeDaemon.__mro__:
if klass.__module__.startswith("meshbay_node."):
where = Path(inspect.getsourcefile(klass)).resolve()
if where not in files:
files.append(where)
if CLI_PACKAGE.is_dir():
files += sorted(p for p in CLI_PACKAGE.rglob("*.py")
if "__pycache__" not in p.parts)
return files
def daemon_source() -> str:
return "\n".join(p.read_text(encoding="utf-8") for p in daemon_files())
def daemon_class_source() -> str:
"""The body of NodeDaemon, including every class it is built from."""
from meshbay_node.daemon import NodeDaemon
return "\n".join(inspect.getsource(k) for k in NodeDaemon.__mro__
if k.__module__.startswith("meshbay_node."))
def cli_source() -> str:
"""`meshbay-node`'s entry point and whatever it hands the work to."""
from meshbay_node import daemon
files = []
if CLI_PACKAGE.is_dir():
files = sorted(p for p in CLI_PACKAGE.rglob("*.py")
if "__pycache__" not in p.parts)
return "\n".join([inspect.getsource(daemon.main)]
+ [p.read_text(encoding="utf-8") for p in files])
def daemon_call(name: str) -> str:
"""The one call to `name(...)` in the daemon, found by name, not position."""
found = []
for path in daemon_files():
text = path.read_text(encoding="utf-8")
for node in ast.walk(ast.parse(text)):
if (isinstance(node, ast.Call)
and getattr(node.func, "id", getattr(node.func, "attr", None)) == name):
found.append(ast.get_source_segment(text, node))
assert len(found) == 1, f"{len(found)} calls to {name} in the daemon — re-read this test"
return found[0]
|