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
|
"""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 devel/windows-devel.md §5).
# 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",
)
# Windows-only gaps still to close (see devel/windows-devel.md §5/§6).
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)
|