""" 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]