diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_ops.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_ops.py | 146 |
1 files changed, 130 insertions, 16 deletions
diff --git a/packages/meshbay-node/tests/test_ops.py b/packages/meshbay-node/tests/test_ops.py index 92e32bf..c118b5a 100644 --- a/packages/meshbay-node/tests/test_ops.py +++ b/packages/meshbay-node/tests/test_ops.py @@ -11,12 +11,14 @@ call them. import asyncio import inspect -from pathlib import Path +from pathlib import Path, PureWindowsPath +from types import SimpleNamespace import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_node import ops from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roots import RootSet from meshbay_node.transport.quic_server import Denylist from conftest import one_root @@ -64,8 +66,8 @@ def test_the_http_adapter_adds_no_logic(): # Every endpoint that performs an operation routes through _op(...). for endpoint in ("operator_pair", "create_invite", "revoke_member", "unpin_member", "init_gek", "attach_group", "delete_file", - "add_root", "remove_root", "set_member_upload", - "reload_config"): + "add_root", "remove_root", "update_root", + "eject_root", "plug_root", "reload_config"): start = source.index(f"async def {endpoint}(") body = source[start:start + 700] assert "_op(" in body.split("\n\n")[0] + body, ( @@ -181,25 +183,103 @@ async def test_an_unhosted_group_offers_what_it_does_host(tmp_path): assert exc.value.extra.get("available") -# ── Upload policy (set_member_upload) ─────────────────────────────────────── +# ── Upload policy (per-root writable) ─────────────────────────────────────── -async def test_set_member_upload_toggles_and_persists(tmp_path): +async def test_the_group_wide_upload_switch_is_gone(tmp_path): + """ + `set_member_upload` was the whole of the old policy, and it is deliberately + not here any more — RO/RW on the root replaced it. A wrapper kept "for + compatibility" would be a second way to decide who writes to the operator's + disk, and two answers to that question is how C1 and C6 both happened. + """ + assert not hasattr(ops, "set_member_upload") + from meshbay_node.roster import Roster + assert not hasattr(Roster, "set_member_upload") + assert not hasattr(Roster, "member_upload_allowed") + + +async def test_eject_and_plug_persist_through_the_roster(tmp_path): + """ + The state has to outlive the process: an operator ejects a drive, unplugs + it, and restarts the node — and the rescan that follows must not read the + empty mount point as an erased library. + """ from meshbay_node.roster import Roster state = _state(tmp_path) + usb = tmp_path / "USB" + usb.mkdir() + state["groups_ctx"]["g" * 32]["roots"] = RootSet.build( + [{"path": str(usb), "removable": True, "writable": True}]) + state["config"] = SimpleNamespace( + groups=[SimpleNamespace(id="g" * 32, roots=[])]) roster = Roster(db_path=tmp_path / "roster.db") await roster.open() state["roster"] = roster state["node_user_id"] = "operator" + try: + out = await ops.eject_root(state, "g" * 32, "USB") + assert out["status"] == "ejected" + assert await roster.ejected_roots("g" * 32) == {"usb"} + assert out["roots"][0]["ejected"] is True + assert out["roots"][0]["available"] is False - out = await ops.set_member_upload(state, "g" * 32, True) + out = await ops.plug_root(state, "g" * 32, "USB") + assert out["status"] == "plugged" + assert await roster.ejected_roots("g" * 32) == set() + finally: + await roster.close() - assert out["allowed"] is True - assert state["groups_ctx"]["g" * 32]["member_upload"] is True - out2 = await ops.set_member_upload(state, "g" * 32, False) +async def test_a_root_that_is_not_removable_cannot_be_ejected(tmp_path): + """ + Eject means "I am about to unplug this". On a directory that is not on a + removable device it would hide a library with no way for the safety net to + notice anything happened, and nothing to plug back in. + """ + from meshbay_node.roster import Roster + state = _state(tmp_path) + fixed = tmp_path / "Fixed" + fixed.mkdir() + state["groups_ctx"]["g" * 32]["roots"] = RootSet.build( + [{"path": str(fixed), "writable": True}]) + state["config"] = SimpleNamespace( + groups=[SimpleNamespace(id="g" * 32, roots=[])]) + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + state["roster"] = roster + try: + with pytest.raises(ops.OpError, match="removable"): + await ops.eject_root(state, "g" * 32, "Fixed") + finally: + await roster.close() - assert out2["allowed"] is False - assert state["groups_ctx"]["g" * 32]["member_upload"] is False + +async def test_plugging_a_drive_that_is_not_there_is_refused(tmp_path): + """ + Clearing the flag while the device is still absent would restart the + watchdog on a missing path and hand the next reconcile an empty directory — + the deletion storm the eject was there to prevent, produced by the recovery. + """ + from meshbay_node.roster import Roster + state = _state(tmp_path) + usb = tmp_path / "USB" + usb.mkdir() + roots = RootSet.build([{"path": str(usb), "removable": True}]) + roots.roots[0].ejected = True + roots.roots[0].available = False + state["groups_ctx"]["g" * 32]["roots"] = roots + state["config"] = SimpleNamespace( + groups=[SimpleNamespace(id="g" * 32, roots=[])]) + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + state["roster"] = roster + usb.rmdir() + try: + with pytest.raises(ops.OpError, match="device connected"): + await ops.plug_root(state, "g" * 32, "USB") + assert roots.roots[0].ejected is True + finally: + await roster.close() # ── Reload ────────────────────────────────────────────────────────────────── @@ -279,13 +359,47 @@ def test_update_node_toml_forces_lf_and_keeps_standalone_comments(tmp_path): def test_a_backslash_path_written_into_node_toml_stays_parseable(): - # attach_group / add_root / init embed a directory into a TOML basic string. - # A raw Windows path there (drive + backslash + "Users" + ...) is a parse - # error since backslash sequences are escapes; the code writes as_posix(). + """ + attach_group and add_root embed a directory into a TOML basic string. A raw + Windows path there is a parse error, because backslash sequences are escapes + (`\\U`, `\\a`, ...); the code writes `as_posix()` and pathlib reads `/` back + on Windows. + + `PureWindowsPath`, not `Path`: on this suite's usual machine `Path` is a + `PosixPath`, where a backslash is an ordinary filename character and + `as_posix()` converts nothing — so the test modelled the wrong platform and + failed everywhere except the one it was written for. Naming the flavour + explicitly is what makes it the same assertion on all three. + """ import tomllib bs = chr(92) win_dir = f"C:{bs}Users{bs}alice{bs}Media" - assert tomllib.loads(f'path = "{Path(win_dir).as_posix()}"\n')["path"] == \ - "C:/Users/alice/Media" + + assert tomllib.loads( + f'path = "{PureWindowsPath(win_dir).as_posix()}"\n' + )["path"] == "C:/Users/alice/Media" + with pytest.raises(tomllib.TOMLDecodeError): tomllib.loads(f'path = "{win_dir}"\n') # the bug this guards against + + +def test_every_path_written_into_node_toml_goes_through_as_posix(): + """ + The half the round trip above cannot see. + + Proving `as_posix()` produces a parseable string says nothing about whether + the code calls it, and this is a defect no Linux machine can reproduce: the + config is written, parsed and served correctly here, and fails on the + operator's Windows box. So the source is read for the shape instead — + weak evidence, and the only kind available for a platform the suite does + not run on. + """ + import re + source = inspect.getsource(ops) + # Every f-string interpolation that lands on the right of a TOML `path =`. + writes = re.findall(r'path\s*=\s*\\?"\{([^}]+)\}', source) + assert writes, "no TOML path writer found — did the config writer move?" + for expr in writes: + assert "as_posix()" in expr, ( + f'node.toml path written as `{expr}` — a Windows path needs ' + f'as_posix(), or the file it lands in will not parse') |