diff options
Diffstat (limited to 'packages/meshbay-hub/tests')
| -rw-r--r-- | packages/meshbay-hub/tests/test_cast_discovery.py | 108 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_desktop_keyring.py | 22 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_desktop_shell.py | 12 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_upload_controls_hidden.py | 30 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_username_case.py | 24 |
5 files changed, 184 insertions, 12 deletions
diff --git a/packages/meshbay-hub/tests/test_cast_discovery.py b/packages/meshbay-hub/tests/test_cast_discovery.py new file mode 100644 index 0000000..9e9ce85 --- /dev/null +++ b/packages/meshbay-hub/tests/test_cast_discovery.py @@ -0,0 +1,108 @@ +""" +Cast receivers are listed as they answer, not at the end of the scan. + +The scan runs its full length because a receiver coming back from a reset can +take several seconds to answer, but most answer within two; a picker that showed +nothing until the end was slow every time it was opened. These tests run the real +`CastChromecast` with mDNS replaced by a stub that answers on cue, and the clock +shortened, so what they measure is what the page's poll would see. +""" + +import json +import shutil +import subprocess +from pathlib import Path + +import pytest + +CLIENT = Path(__file__).resolve().parents[2] / "meshbay-client" +CHROMECAST = CLIENT / "src" / "cast-chromecast.js" + +pytestmark = pytest.mark.skipif( + shutil.which("node") is None or not CHROMECAST.exists(), + reason="node or the desktop client sources are not available") + +# Stubs the two dependencies at `require` time, and makes six seconds of scan +# last sixty milliseconds by scaling every timer the module arms. +SCRIPT = r""" +const Module = require('node:module'); +const browsers = []; +const realLoad = Module._load; +Module._load = function (request, ...rest) { + if (request === 'bonjour-service') { + return { Bonjour: class { + find(_q, onUp) { + const b = { onUp, stopped: false, stop() { this.stopped = true; } }; + browsers.push(b); + return b; + } + } }; + } + if (request === 'castv2-client') return { Client: class {}, DefaultMediaReceiver: {} }; + return realLoad.call(this, request, ...rest); +}; +const realSetTimeout = global.setTimeout; +global.setTimeout = (fn, ms, ...a) => realSetTimeout(fn, ms / 100, ...a); +const wait = (ms) => new Promise((r) => realSetTimeout(r, ms)); + +const CastChromecast = require(process.argv[2]); +const tv = (id) => ({ name: id, txt: { id, fn: `TV ${id}` }, + addresses: ['10.0.0.9'], port: 8009 }); + +(async () => { + const cc = new CastChromecast(); + const out = {}; + + cc.startScan(); + out.atStart = cc.devices(); + browsers[0].onUp(tv('a')); + out.afterFirstAnswer = cc.devices(); + await wait(100); + out.afterScan = cc.devices(); + out.firstBrowserStopped = browsers[0].stopped; + + // A second scan started over a first: the first's timer must not end it. + cc.startScan(); + await wait(40); + cc.startScan(); + await wait(40); + out.restartedStillScanning = cc.devices().scanning; + out.restartClearedOldDevices = cc.devices().devices.length === 0; + await wait(60); + out.restartedEnds = !cc.devices().scanning; + console.log(JSON.stringify(out)); +})(); +""" + + +@pytest.fixture(scope="module") +def run(tmp_path_factory): + script = tmp_path_factory.mktemp("cd") / "discover.cjs" + script.write_text(SCRIPT, encoding="utf-8") + proc = subprocess.run( + ["node", str(script), str(CHROMECAST)], + capture_output=True, text=True, encoding="utf-8", timeout=60) + assert proc.returncode == 0, proc.stderr + return json.loads(proc.stdout) + + +def test_a_receiver_is_listed_before_the_scan_ends(run): + assert run["atStart"] == {"devices": [], "scanning": True} + assert run["afterFirstAnswer"]["scanning"] is True + assert [d["name"] for d in run["afterFirstAnswer"]["devices"]] == ["TV a"] + + +def test_the_scan_ends_on_its_own_and_keeps_what_it_found(run): + assert run["afterScan"]["scanning"] is False + assert [d["id"] for d in run["afterScan"]["devices"]] == ["a"] + assert run["firstBrowserStopped"] is True + + +def test_a_new_scan_is_not_cut_short_by_the_one_it_replaced(run): + """ + The old code left the first scan's timer armed, and it stopped whichever + browser was current when it fired — the new one, two thirds early. + """ + assert run["restartedStillScanning"] is True + assert run["restartClearedOldDevices"] is True + assert run["restartedEnds"] is True diff --git a/packages/meshbay-hub/tests/test_desktop_keyring.py b/packages/meshbay-hub/tests/test_desktop_keyring.py index e55e6d0..af7cc37 100644 --- a/packages/meshbay-hub/tests/test_desktop_keyring.py +++ b/packages/meshbay-hub/tests/test_desktop_keyring.py @@ -13,6 +13,7 @@ page, and a reference written from the specification in Python. import base64 import hashlib import json +import re import shutil import subprocess from pathlib import Path @@ -93,6 +94,9 @@ const v = JSON.parse(fs.readFileSync(input, 'utf8')); other_node: await refusal('join', { ...F.join, nodePk: 'T3RoZXJOb2Rl' }), other_account: await refusal('join', { ...F.join, userId: 'someone-else' }), stale: await refusal('join', { ...F.join, ts: ts - 3600 }), + root_add: await refusal('admin', { ...F.admin, op: 'root_add' }), + root_update: await refusal('admin', { ...F.admin, op: 'root_update' }), + group_attach: await refusal('admin', { ...F.admin, op: 'group_attach' }), }; const eph = require('crypto').generateKeyPairSync('x25519'); @@ -268,6 +272,24 @@ def test_the_page_names_a_kind_and_never_the_bytes(out): assert "not now" in r["stale"] +def test_what_widens_a_nodes_sharing_is_never_signed(out): + """Gone from MNP 6.0, and refused here as well: a node older than that + still accepts them, and a script in the page must not be able to get one + signed for it.""" + for op in ("root_add", "root_update", "group_attach"): + assert "not an operation" in out["refused"][op], op + + +def test_the_application_signs_exactly_the_nodes_operations(): + from meshbay_common import adminop + src = (Path(__file__).resolve().parents[2] / "meshbay-client" / "src" + / "transcripts.js").read_text(encoding="utf-8") + listed = set(re.findall(r"'([a-z_]+)'", + src.split("const ADMIN_OPS = new Set([")[1].split("]);")[0])) + catalogue = {v for k, v in vars(adminop).items() if k.startswith("OP_")} + assert listed == catalogue + + def test_nothing_is_sealed_for_a_browser_while_browser_access_is_off(out): for what, err in out["sealing_while_access_off"].items(): assert err and "browser access is off" in err, what diff --git a/packages/meshbay-hub/tests/test_desktop_shell.py b/packages/meshbay-hub/tests/test_desktop_shell.py index 60aa32f..c2ea4ca 100644 --- a/packages/meshbay-hub/tests/test_desktop_shell.py +++ b/packages/meshbay-hub/tests/test_desktop_shell.py @@ -425,14 +425,10 @@ def test_the_page_names_node_operations_not_routes(): assert defined <= used, f"defined but never called: {defined - used}" -@pytest.mark.parametrize("op", ["attachGroup", "addRoot", "initGek", "clearDenylist"]) -def test_what_widens_the_node_is_confirmed_natively(op): - """Sharing a folder, hosting a group, replacing the key, re-admitting a - revoked subject: asked by a dialog the main process draws, which a script in - the page cannot answer. A folder chosen in the native picker is its own - confirmation.""" - body = _node_ops()[op] - assert "confirmOrRefuse(" in body or "confirmFolder(" in body +def test_readmitting_a_revoked_subject_is_confirmed_natively(): + """Asked by a dialog the main process draws, which a script in the page + cannot answer.""" + assert "confirmOrRefuse(" in _node_ops()["clearDenylist"] def test_the_native_dialogs_are_worded_in_every_language(): diff --git a/packages/meshbay-hub/tests/test_upload_controls_hidden.py b/packages/meshbay-hub/tests/test_upload_controls_hidden.py index 6316e44..947ba75 100644 --- a/packages/meshbay-hub/tests/test_upload_controls_hidden.py +++ b/packages/meshbay-hub/tests/test_upload_controls_hidden.py @@ -231,7 +231,7 @@ def test_the_notice_also_answers_the_operators_own_request(): def test_changing_a_root_is_signed(): transport = transport_source() - for method in ("updateRoot", "ejectRoot", "plugRoot"): + for method in ("ejectRoot", "plugRoot", "removeRoot"): body = transport[transport.index(f"async {method}("):] body = body[:body.index("\n async ", 1)] assert "admin_challenge" in body and "_authorizeAdminOp" in body, ( @@ -261,12 +261,34 @@ def test_the_operator_is_offered_it_on_the_web_too(): "the shared directories section still requires a local node") table = _component(source, "SharedDirectoriesTable") - for call in ("transport.updateRoot", "transport.ejectRoot", - "transport.plugRoot", "transport.removeRoot", - "transport.addRoot"): + for call in ("transport.ejectRoot", "transport.plugRoot", "transport.removeRoot"): assert call in table, f"{call} has no MNP route from the table" +def test_what_widens_the_sharing_never_crosses_mnp(): + """ + Adding a directory and switching `writable` or `removable` are done on the + node's own machine (MNP 6.0). Over MNP they were signed ops, and a signature + proves that the operator's key signed: in a browser that key is driven by + code the hub serves. The list stays visible from anywhere; those controls + are read-only there. + """ + transport = transport_source() + for method in ("addRoot", "updateRoot", "attachGroup"): + assert f"async {method}(" not in transport, f"{method} is an MNP call again" + for mtype in ("'root_add'", "'root_update'", "'group_attach'"): + assert mtype not in transport, f"{mtype} is sent or expected again" + + table = _component(GROUP_SETTINGS.read_text(encoding="utf-8"), + "SharedDirectoriesTable") + assert "platform.node.op('addRoot'" in table + assert "platform.node.op('updateRoot'" in table + assert "const canWiden = isLocal || onNodeMachine;" in table + assert table.count("!canWiden") >= 3, ( + "a writable or removable switch is live where it cannot be honoured") + assert "settings_node.roots_local_only_hint" in table + + def test_the_roots_shown_come_from_the_live_connection_when_there_is_one(): """ The loopback list is a second source, and the two drift: it is read once on diff --git a/packages/meshbay-hub/tests/test_username_case.py b/packages/meshbay-hub/tests/test_username_case.py new file mode 100644 index 0000000..42f4815 --- /dev/null +++ b/packages/meshbay-hub/tests/test_username_case.py @@ -0,0 +1,24 @@ +""" +A username is unique whatever its case. + +Invitations and member management name people by username, so "Alice" beside +"alice" is one person to whoever reads the list — and a second account under +the other spelling is the way to be mistaken for them. +""" + +import pytest +from test_bundle_pepper import KEY + + +async def _register(client, username, email): + return await client.post("/v1/users/register", json={ + "username": username, "auth_key": KEY, "email": email}) + + +@pytest.mark.asyncio +async def test_a_name_differing_only_by_case_is_taken(client): + assert (await _register(client, "alice_case", "a1@example.invalid")).status_code == 201 + for other in ("Alice_case", "ALICE_CASE", "alice_CASE"): + r = await _register(client, other, "a2@example.invalid") + assert r.status_code == 409, (other, r.text) + |