diff options
Diffstat (limited to 'packages/meshbay-cli/tests')
| -rw-r--r-- | packages/meshbay-cli/tests/fake_node.py | 22 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_e2e_local.py | 38 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_interactive.py | 43 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_sctp_fix.py | 89 |
4 files changed, 178 insertions, 14 deletions
diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py index c618ee4..47f82b3 100644 --- a/packages/meshbay-cli/tests/fake_node.py +++ b/packages/meshbay-cli/tests/fake_node.py @@ -79,6 +79,8 @@ class FakeNode: self.members: dict[str, Identity] = {} # sender → the device it signs with self.unresponsive = False # pings go unanswered, as a dead link's self.admin_ops: dict[str, dict] = {} + self.channels: list[FakeChannel] = [] + self.index_syncs = 0 # full indexes asked for def add(self, path: str, name: str, data: bytes, uploader_id: str = "") -> dict: fid = hashlib.blake2b(data + path.encode() + name.encode(), digest_size=32).hexdigest() @@ -92,9 +94,26 @@ class FakeNode: return entry def session(self) -> GroupSession: - return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", NODE_PK, self.gek, + ch = FakeChannel(self) + self.channels.append(ch) + return GroupSession(None, ch, GROUP_ID, "node-1", NODE_PK, self.gek, dict(self.ack), self.identity, self.nonce, self.role) + def push_delta(self, *, additions=(), updates=(), deletions=()) -> None: + """An `index_delta` to every session, sealed as the node seals one. The + entries are applied here first, so a later `index_sync` agrees.""" + for e in (*additions, *updates): + self.files[e["id"]] = e + for fid in deletions: + self.files.pop(fid, None) + wire = lambda es: [{k: v for k, v in e.items() if k != "data"} for e in es] # noqa: E731 + payload = {"base_version": 1, "version": 1, "additions": wire(additions), + "deletions": list(deletions), "updates": wire(updates)} + msg = {"type": "index_delta", "group_id": GROUP_ID, + **seal(self.gek, PURPOSE_INDEX, "index_delta", GROUP_ID, payload)} + for ch in self.channels: + ch._emit(msg) + def list_device(self, sender: str, device: Identity, *, countersigned_by: Identity | None = None, proof: bool = True) -> None: """A roster row for `sender`'s device: a root, or countersigned by another @@ -189,6 +208,7 @@ class FakeChannel: def _index_sync(self, msg): n = self.node + n.index_syncs += 1 payload = {"version": 1, "roots": n.roots, "dirs": sorted(n.dirs), "entries": [{k: v for k, v in e.items() if k != "data"} for e in n.files.values()]} diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py index b131658..385693d 100644 --- a/packages/meshbay-cli/tests/test_e2e_local.py +++ b/packages/meshbay-cli/tests/test_e2e_local.py @@ -196,7 +196,14 @@ def _build(root: Path, procs: list) -> dict: _wait_http(f"http://127.0.0.1:{ui_port}/", timeout=90) # Attaching reloads the node by itself; the key needs the group loaded. - run_node("group", "add", group_id, "--dir", str(share)) + # The control API answers before the hub socket is up, later still now + # that the connection processes start first. + deadline = time.monotonic() + 90 + while "Hub not connected" in (out := _try(run_node, "group", "add", group_id, + "--dir", str(share))): + assert time.monotonic() < deadline, f"group add never succeeded: {out}" + time.sleep(1) + assert "meshbay-node (" not in out, out deadline = time.monotonic() + 90 while "GEK ready" not in (out := _try(run_node, "gek", "init", "--group", group_id)): assert time.monotonic() < deadline, f"gek init never succeeded: {out}" @@ -781,7 +788,11 @@ class _Chrome: target = next(p for p in pages if p["type"] == "page") break except (httpx.HTTPError, StopIteration): - assert time.monotonic() < deadline, "Chrome did not come up" + if time.monotonic() > deadline: + # Not left running: nobody holds this object to close it. + self.proc.kill() + self.proc.wait(10) + raise AssertionError("Chrome did not come up") from None time.sleep(0.2) self.ws = connect(target["webSocketDebuggerUrl"], max_size=None) self.n = 0 @@ -1375,17 +1386,18 @@ def test_tab_completes_remote_paths_from_the_real_node(world): assert complete("ls", "No such group/") == [] # nothing, quietly # bash itself: the function the script installs, on a word typed with its - # spaces escaped, as bash hands it over. - script = cli(world, "completion", "bash").stdout - escaped = group[:10].replace(" ", "\\ ") - bash = subprocess.run( - ["bash", "-c", script + "\nCOMP_WORDS=(meshbay-cli ls " + escaped + ")\n" - "COMP_CWORD=2\n_meshbay_cli\nprintf '%s\\n' \"${COMPREPLY[@]}\"\n"], - env={**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"), - "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring", - "PATH": f"{BIN}{os.pathsep}{os.environ['PATH']}"}, - capture_output=True, text=True, encoding="utf-8", timeout=60) - assert bash.stdout.splitlines() == [group + "/"], (bash.stdout, bash.stderr) + # spaces escaped, as bash hands it over. On Windows, where Git's is on PATH. + if shutil.which("bash"): + script = cli(world, "completion", "bash").stdout + escaped = group[:10].replace(" ", "\\ ") + bash = subprocess.run( + ["bash", "-c", script + "\nCOMP_WORDS=(meshbay-cli ls " + escaped + ")\n" + "COMP_CWORD=2\n_meshbay_cli\nprintf '%s\\n' \"${COMPREPLY[@]}\"\n"], + env={**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"), + "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring", + "PATH": f"{BIN}{os.pathsep}{os.environ['PATH']}"}, + capture_output=True, text=True, encoding="utf-8", timeout=60) + assert bash.stdout.splitlines() == [group + "/"], (bash.stdout, bash.stderr) # PowerShell 7, or Windows PowerShell 5.1, which decodes a program's output # with the console's code page unless told otherwise: an accented name is the diff --git a/packages/meshbay-cli/tests/test_interactive.py b/packages/meshbay-cli/tests/test_interactive.py index 36b7f47..9960533 100644 --- a/packages/meshbay-cli/tests/test_interactive.py +++ b/packages/meshbay-cli/tests/test_interactive.py @@ -444,3 +444,46 @@ async def test_put_sends_a_file_into_the_chat_folder_and_get_saves_one_here( await say("/get ../share/x") await _until(pilot, lambda: "name a file of the chat's folder" in _said(room)) await pilot.press("escape") + + +async def test_index_updates_are_applied_not_asked_for_again(world, monkeypatch): + """ + A library being enriched pushes an `index_delta` about once a second, each + one filling in a file's fields. The panes used to answer every one with the + whole index — 67 full indexes in the 80 s of one download on a real node, + on the event loop the download was being received on. Updates are applied + where they land; only a delta that adds or removes a file asks again, once + for a burst. + """ + from meshbay_cli.tui import screens + monkeypatch.setattr(screens, "REFRESH_AFTER_DELTA_S", 0.2) + hub, node, work = world + film = node.add("share", "film.mkv", b"F" * 5000) + assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME, + "--password", PASSPHRASE, "login"]) == 0 + app = _app() + async with app.run_test(size=(140, 40)) as pilot: + await _until(pilot, lambda: isinstance(app.screen, MainScreen)) + remote = app.screen.query_one(RemotePane) + await _open(pilot, remote, f"Family@{USERNAME}/") + await _open(pilot, remote, "share/") + await _until(pilot, lambda: "film.mkv" in _names(remote)) + asked = node.index_syncs + + for i in range(20): + node.push_delta(updates=[{**film, "duration": 100 + i}]) + await pilot.pause(0.6) + assert node.index_syncs == asked, "an update must not fetch the whole index" + tree = app.screen._trees["g1"] + assert tree.file("share/film.mkv")["duration"] == 119 + + extra = [node.add("share", f"new{i}.txt", b"n%d" % i) for i in range(3)] + node.files = {k: v for k, v in node.files.items() if v not in extra} + for e in extra: + node.push_delta(additions=[e]) + await _until(pilot, lambda: "new2.txt" in _names(remote)) + await pilot.pause(0.6) + assert node.index_syncs == asked + 1, "a burst of additions asks again once" + + node.push_delta(deletions=[film["id"]]) + await _until(pilot, lambda: "film.mkv" not in _names(remote)) diff --git a/packages/meshbay-cli/tests/test_sctp_fix.py b/packages/meshbay-cli/tests/test_sctp_fix.py new file mode 100644 index 0000000..04638db --- /dev/null +++ b/packages/meshbay-cli/tests/test_sctp_fix.py @@ -0,0 +1,89 @@ +"""The client's copy of the node's aiortc fixes (client/sctp_fix.py). + +The copy exists because neither package may depend on the other; it is held to +the node's by comparing the code itself, docstrings aside, and the two effects +that set a download's speed here are replayed against aiortc. +""" + +import ast +import asyncio +from pathlib import Path + +import pytest +from aiortc.rtcsctptransport import ( + SCTP_DATA_FIRST_FRAG, + SCTP_DATA_LAST_FRAG, + DataChunk, + InboundStream, + RTCSctpTransport, + serialize_packet, +) +from meshbay_cli.client import sctp_fix + +NODE_COPY = (Path(__file__).resolve().parents[2] / "meshbay-node" / "src" / "meshbay_node" + / "transport" / "webrtc" / "sctp_fix.py") + + +def _code(path: Path) -> str: + tree = ast.parse(path.read_text(encoding="utf-8")) + tree.body = tree.body[1:] # the module docstring + return ast.dump(tree) + + +@pytest.mark.skipif(not NODE_COPY.exists(), reason="not in the source tree") +def test_the_code_is_the_nodes(): + assert _code(Path(sctp_fix.__file__)) == _code(NODE_COPY) + + +def _chunks(n_fragments: int, count: int, tsn: int = 1000) -> list[DataChunk]: + out = [] + for seq in range(count): + for i in range(n_fragments): + c = DataChunk() + c.flags = ((SCTP_DATA_FIRST_FRAG if i == 0 else 0) + | (SCTP_DATA_LAST_FRAG if i == n_fragments - 1 else 0)) + c.tsn, c.stream_id, c.stream_seq, c.protocol = tsn, 1, seq, 53 + c.user_data = b"x" + tsn += 1 + out.append(c) + return out + + +def test_a_megabyte_is_walked_once(monkeypatch): + walks = [] + original = sctp_fix._pop_messages + monkeypatch.setattr(sctp_fix, "_pop_messages", + lambda s: (walks.append(1), original(s))[1]) + s = InboundStream() + got = 0 + for c in _chunks(870, 3): + s.add_chunk(c) + got += len(list(s.pop_messages())) + assert got == 3 and len(walks) == 3 + + +def test_every_second_packet_is_acknowledged(): + class _Dtls: + state = "connected" + + def _register_data_receiver(self, *a): + pass + + async def go(): + s = RTCSctpTransport(_Dtls()) + s._association_state = s.State.ESTABLISHED + s._last_received_tsn = 999 + sacks = [] + + async def send_sack(): + sacks.append(s._last_received_tsn) + s._sack_needed = False + + async def deliver(*_a): + pass + + s._send_sack, s._receive = send_sack, deliver + for c in _chunks(1, 6): + await s._handle_data(serialize_packet(1, 1, s._local_verification_tag, c)) + return sacks + assert asyncio.run(go()) == [1001, 1003, 1005] |