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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
"""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 _no_log_file_in_the_developers_profile(monkeypatch):
"""A test that runs the daemon's entry point must not open its log file.
On Windows that file is %LOCALAPPDATA%\\meshbay\\state\\node.log -- the
developer's own node's log. The first run of the suite with it appended
every later test's log lines there, through a handler left on the root
logger. A test about the log file points log_file() somewhere of its own.
"""
import logging
from meshbay_node import platform as _plat
monkeypatch.setattr(_plat, "log_file", lambda: None)
root = logging.getLogger()
before = list(root.handlers)
yield
for h in root.handlers[:]:
if h not in before and getattr(h, "baseFilename", None):
root.removeHandler(h)
h.close()
@pytest.fixture(scope="session")
def _throwaway_profile(tmp_path_factory):
return tmp_path_factory.mktemp("profile")
@pytest.fixture(autouse=True)
def _no_test_touches_the_developers_profile(monkeypatch, _throwaway_profile):
"""Every per-user location, pointed somewhere nobody lives.
Tests redirected HOME and nothing else, which isolates nothing on Windows:
Path.home() reads USERPROFILE there and the node's own directories come from
LOCALAPPDATA. `member invite` and `operator pair`, walked by the CLI tests,
wrote their codes into the developer's real node directory and a
~/.local/share/meshbay nobody had, on every run. A test that sets its own
HOME still can; it just no longer falls through to the real one.
"""
for var in ("HOME", "USERPROFILE", "LOCALAPPDATA", "APPDATA"):
monkeypatch.setenv(var, str(_throwaway_profile / var.lower()))
@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)
|