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
|
"""
The node's WebRTC transport, as source text, for the hub tests that check a
client contract against what the node actually sends.
Read from the tree rather than imported, so these tests keep working in a
checkout without the node installed. The file set is derived — the transport
is `webrtc_server.py` and whatever has been split out of it under
`transport/webrtc/` — so that code moving between those files does not leave a
test reading a file its subject has left. The node suite's `node_source.py`
draws the same line and holds it with `test_node_source_scope.py`.
"""
import ast
from pathlib import Path
# parents[2] is `packages/`: the tests live at packages/meshbay-hub/tests/. A
# wrong index here does not fail anything — every test using it skips, which
# is worse than not having them: a green run that measured nothing. That
# happened once.
TRANSPORT = (Path(__file__).resolve().parents[2] / "meshbay-node" / "src"
/ "meshbay_node" / "transport")
SERVER = TRANSPORT / "webrtc_server.py"
def available() -> bool:
return SERVER.exists()
def webrtc_files() -> list[Path]:
files = [SERVER]
package = TRANSPORT / "webrtc"
if package.is_dir():
files += sorted(p for p in 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 method(name: str) -> str:
"""The text of one method of a class in the transport, found by name."""
found = []
for path in webrtc_files():
text = path.read_text(encoding="utf-8")
for node in ast.walk(ast.parse(text)):
if isinstance(node, ast.ClassDef):
for member in node.body:
if isinstance(member, (ast.FunctionDef, ast.AsyncFunctionDef)) \
and member.name == name:
found.append(ast.get_source_segment(text, member, padded=True))
assert len(found) == 1, f"{name}: expected one definition, found {len(found)}"
return found[0]
def module_files(name: str) -> list[Path]:
"""A node module by name — `ops.py`, or every file of the `ops/` package it
becomes — so a test reading it for an absence goes on reading all of it."""
src = TRANSPORT.parent
files = [src / f"{name}.py"] if (src / f"{name}.py").exists() else []
if (src / name).is_dir():
files += sorted(p for p in (src / name).rglob("*.py")
if "__pycache__" not in p.parts)
assert files, f"no module or package named {name} in the node"
return files
def module_source(name: str) -> str:
return "\n".join(p.read_text(encoding="utf-8") for p in module_files(name))
|