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
|
"""Shared fixtures and helpers for node tests."""
import sys
from pathlib import Path
import pytest
from meshbay_node.roots import RootSet
# ffmpeg / ffprobe run via asyncio.create_subprocess_exec, which needs the
# ProactorEventLoop — but the repo-root conftest forces the SelectorEventLoop
# on win32 so aiortc's ICE stack works there (see docs/MESHBAY_DESIGN.md §11.2).
# The two are mutually exclusive on one Windows asyncio loop; until the media
# path gets a thread-based subprocess runner, these tests can't run on win32.
needs_subprocess = pytest.mark.skipif(
sys.platform == "win32",
reason="ffmpeg subprocess needs ProactorEventLoop; win32 conftest forces "
"SelectorEventLoop for aiortc",
)
@pytest.fixture(autouse=True)
def _restore_media_tool_paths():
"""Put `platform`'s resolved ffmpeg/ffprobe paths back after every test.
`check_media_tools()` writes two module globals. `monkeypatch` restores what
a test patched, and knows nothing about what the code under test then wrote
— so a test that patched `shutil.which` to a Windows path and called
`check_media_tools()` left `_ffprobe_path` at "/opt/bin/ffprobe.exe" for the
rest of the session. Seven tests in two files about video transcoding then
died on FileNotFoundError, for a reason nowhere near themselves, and only
when the whole suite ran: run those two files alone and they passed.
The instance is fixed at the call site as well; this closes the class. Any
future test that resolves media tools is undone here whether it remembers to
or not, which is the only way an order-dependent suite stops being one.
"""
from meshbay_node import platform as _plat
before = (_plat._ffmpeg_path, _plat._ffprobe_path)
yield
_plat._ffmpeg_path, _plat._ffprobe_path = before
# Windows-only gaps still to close (see docs/MESHBAY_DESIGN.md §11.2).
win32_todo = pytest.mark.skipif(
sys.platform == "win32",
reason="Windows behaviour not implemented yet (W3 / platform specifics)",
)
def one_root(path: Path, *, name: str = "", kind: str = "generic",
writable: bool = True) -> RootSet:
"""
A RootSet with a single writable root over `path`.
The equivalent of the old `shared_dir`. Note what it implies for assertions:
a file directly in `path` now has `entry.path == <basename of path>`, not
`""` — every index path carries its root name, in a group with one root as
much as in a group with five.
Writable by default because most callers are testing something else and
want a root an upload can reach. `writable=False` is the read-only group.
"""
return RootSet.build([{"path": str(path), "name": name, "kind": kind,
"writable": writable}])
def sealed_upload(session, *, filename: str, data: bytes,
chunk_index: int = 0, total_chunks: int = 1,
dir: str = "", root: str = "",
upload_id: str = "up-test") -> dict:
"""
A `file_upload` message as the shipping client builds one (MNP 2.0).
Built through `file_upload_wire`, not by hand: a test that assembles the
wire shape itself is a second encoder, and a second encoder is how
`file_chunk` and `index_sync` forked between the transports (finding C6)
with nobody noticing. The key and the AAD are taken off the session, so
these agree with the handler by construction rather than by copying.
"""
from meshbay_common.protocol import file_upload_wire
ctx = session._group_ctx()
return file_upload_wire(
ctx["gek"], session._group_id or "",
upload_id=upload_id, chunk_index=chunk_index, total_chunks=total_chunks,
filename=filename, data=data, dir=dir, root=root,
)
def opened_ack(session, msg: dict) -> dict:
"""The payload of a `file_upload_ack` the node sent, opened as a client would."""
from meshbay_common.protocol import file_upload_ack_payload
ctx = session._group_ctx()
return file_upload_ack_payload(ctx["gek"], session._group_id or "", msg)
def patch_cli(monkeypatch, name: str, value) -> None:
"""Replace `name` wherever `meshbay-node`'s command line reads it.
Each module of `meshbay_node.cli` imports what it uses by name, so a helper
like `_daemon_api` is bound once in every module that calls it, and patching
the one that defines it leaves the verbs calling the real thing — which
talks to whatever node is running on this machine. `daemon` is included:
its `main()` still reads the config path when no verb is given.
"""
import importlib
import pkgutil
import meshbay_node.cli as cli
from meshbay_node import daemon
modules = [daemon] + [importlib.import_module(f"meshbay_node.cli.{m.name}")
for m in pkgutil.iter_modules(cli.__path__)]
holders = [m for m in modules if hasattr(m, name)]
assert holders, f"no module of the CLI has {name!r} to patch"
for module in holders:
monkeypatch.setattr(module, name, value)
|