diff options
Diffstat (limited to 'packages/meshbay-node/tests')
7 files changed, 947 insertions, 16 deletions
diff --git a/packages/meshbay-node/tests/test_disk_io_off_loop.py b/packages/meshbay-node/tests/test_disk_io_off_loop.py index c0ab052..cacf936 100644 --- a/packages/meshbay-node/tests/test_disk_io_off_loop.py +++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py @@ -260,7 +260,7 @@ def test_no_handler_touches_the_disk_on_the_loop(): # Written to block, and reached only through `off_disk`. on_the_disk_thread = {"_locate", "_append_chunk", "_read_and_encrypt", "_mkdir_if_absent", "_is_empty_dir", "_rmdir_if_empty", - "safe_subdir"} + "_unlink", "_exists", "safe_subdir"} # ffmpeg's own output goes through `_read_scratch_capped` and # `_discard_scratch` on a worker thread — `asyncio.to_thread` and not # `off_disk`, because a temp file is not a group root and has no platter to diff --git a/packages/meshbay-node/tests/test_index_jobs_are_described.py b/packages/meshbay-node/tests/test_index_jobs_are_described.py index dee6e39..1eb203e 100644 --- a/packages/meshbay-node/tests/test_index_jobs_are_described.py +++ b/packages/meshbay-node/tests/test_index_jobs_are_described.py @@ -55,11 +55,11 @@ class _Held(DirectoryIndexer): self.walked.append(root.name) return await super()._scan_root(root, *args, **kwargs) - async def _hash_or_cached(self, root, file_path): + async def _hash_or_cached(self, root, file_path, **kw): if self.hold_under is not None and file_path.is_relative_to(self.hold_under): self.at_gate.set() await self.gate.wait() - return await super()._hash_or_cached(root, file_path) + return await super()._hash_or_cached(root, file_path, **kw) def _dirs(tmp_path: Path) -> tuple[Path, Path, Path]: @@ -149,7 +149,7 @@ async def test_a_failed_initial_scan_leaves_nothing_announced(tmp_path): _, r1, r2 = _dirs(tmp_path) class _Broken(DirectoryIndexer): - async def _hash_or_cached(self, root, file_path): + async def _hash_or_cached(self, root, file_path, **kw): raise RuntimeError("simulated failure mid-scan") idx = _Broken(roots=_set(r1, r2), group_id=GROUP, diff --git a/packages/meshbay-node/tests/test_index_progress.py b/packages/meshbay-node/tests/test_index_progress.py index ecdc223..2fa8a5c 100644 --- a/packages/meshbay-node/tests/test_index_progress.py +++ b/packages/meshbay-node/tests/test_index_progress.py @@ -14,7 +14,7 @@ from fastapi.testclient import TestClient from meshbay_node.config import Config, HubConfig, KeystoreConfig, NodeConfig from meshbay_node.daemon import NodeDaemon from meshbay_node.indexer.group_index import GroupIndex -from meshbay_node.indexer.indexer import IndexProgress +from meshbay_node.indexer.indexer import NO_ANSWER, IndexProgress, UnreadFile from meshbay_node.transport.webrtc_server import WebRTCPeerSession from meshbay_node.ui.app import create_ui_app @@ -64,19 +64,30 @@ def _ui_client(state: dict) -> TestClient: "indexes": {}, **state})) +def _indexer(progress: IndexProgress, *, unread=(), stalled=None, slow=None) -> MagicMock: + """What the routes read off an indexer, with nothing wrong unless said.""" + indexer = MagicMock() + indexer.progress = progress + indexer.unread = {u.path: u for u in unread} + indexer.unread_count.return_value = len(indexer.unread) + indexer.stalled = dict(stalled or {}) + indexer.disk_wait.return_value = slow + return indexer + + def test_index_status_route_idle_for_unknown_group(): client = _ui_client({"indexers": {}}) resp = client.get("/api/groups/unknown-group/index-status") assert resp.status_code == 200 assert resp.json() == {"scanning": False, "scanned_bytes": 0, - "total_bytes": 0, "current_dir": ""} + "total_bytes": 0, "current_dir": "", + "unread": [], "unread_total": 0, "stalled": []} def test_index_status_route_reflects_indexer_progress(): - fake_indexer = MagicMock() - fake_indexer.progress = IndexProgress( + fake_indexer = _indexer(IndexProgress( scanning=True, scanned_bytes=1_000_000, total_bytes=4_000_000_000, - current_dir="2024") + current_dir="2024")) client = _ui_client({"indexers": {"g" * 32: fake_indexer}}) resp = client.get(f"/api/groups/{'g' * 32}/index-status") @@ -84,20 +95,47 @@ def test_index_status_route_reflects_indexer_progress(): assert resp.json() == { "scanning": True, "scanned_bytes": 1_000_000, "total_bytes": 4_000_000_000, "current_dir": "2024", + "unread": [], "unread_total": 0, "stalled": [], } +def test_the_operator_is_shown_what_a_scan_could_not_read_and_where_it_stopped(): + """On the node's own machine only: these are paths. A scan that sat still + used to say nothing about the file it sat on, and a file it gave up on was + one line in a log the operator does not read.""" + silent = UnreadFile(path="/srv/music/a/one.flac", root="music", reason=NO_ANSWER, + size=1, mtime=1.0) + refused = UnreadFile(path="/srv/music/b/two.flac", root="music", + reason="Permission denied", size=1, mtime=1.0) + indexer = _indexer(IndexProgress(), unread=[silent, refused], + stalled={"music": "/srv/music/a/three.flac"}, + slow=("/srv/music/a/four.flac", 23.7)) + config = MagicMock() + config.groups = [] + client = _ui_client({"config": config, "indexers": {"g" * 32: indexer}}) + + one = client.get(f"/api/groups/{'g' * 32}/index-status").json() + assert one["unread"] == [ + {"root": "music", "path": "/srv/music/a/one.flac", "reason": "no_answer"}, + {"root": "music", "path": "/srv/music/b/two.flac", "reason": "Permission denied"}] + assert one["unread_total"] == 2 + assert one["stalled"] == [{"root": "music", "path": "/srv/music/a/three.flac"}] + + (band,) = client.get("/api/index-status").json()["groups"] + assert (band["slow_path"], band["slow_secs"]) == ("/srv/music/a/four.flac", 23) + assert (band["unread"], band["stalled"]) == (2, ["music"]) + + # ── /api/index-status (loopback, every group) ─────────────────────────────── def test_every_group_is_described_including_one_still_being_attached(): """The band reads one route for the whole node. A group in its initial scan is in state["indexers"] and not yet in groups_ctx, and must be there.""" - busy, idle = MagicMock(), MagicMock() - busy.progress = IndexProgress( + busy = _indexer(IndexProgress( scanning=True, scanned_bytes=10, total_bytes=40, current_dir="2024", root="results", root_pos=1, kind="scan", files_done=1, files_total=4, - queued=["archive"]) - idle.progress = IndexProgress() + queued=["archive"])) + idle = _indexer(IndexProgress()) config = MagicMock() named = MagicMock() named.id, named.name = "a" * 32, "outputs" @@ -112,6 +150,7 @@ def test_every_group_is_described_including_one_still_being_attached(): "kind": "scan", "root": "results", "current_dir": "2024", "scanned_bytes": 10, "total_bytes": 40, "files_done": 1, "files_total": 4, "queued": ["archive"], + "slow_path": "", "slow_secs": 0, "unread": 0, "stalled": [], } assert groups["b" * 32]["group_name"] == "b" * 8 assert groups["b" * 32]["scanning"] is False diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py index fda40c8..5540be8 100644 --- a/packages/meshbay-node/tests/test_indexer.py +++ b/packages/meshbay-node/tests/test_indexer.py @@ -418,7 +418,14 @@ async def test_realtime_watchdog_add_reports_progress_like_a_bulk_scan(tmp_path, assert indexer.progress.total_bytes == total_size assert indexer.progress.scanned_bytes == 0 - await asyncio.sleep(0.05) # past debounce_secs; lets all three fire and finish + # Past debounce_secs, until all three have fired and finished. Not a fixed + # 50 ms: a Windows timer ticks every 15.6 ms, the debounce alone can take + # two of them, and the test failed one run in five there on a count of + # milliseconds that says nothing about the indexer. + for _ in range(200): + if not indexer.progress.scanning: + break + await asyncio.sleep(0.01) assert indexer.progress.scanning is False, "must end idle, not stuck scanning" assert indexer.progress.scanned_bytes == total_size diff --git a/packages/meshbay-node/tests/test_indexer_silent_disk.py b/packages/meshbay-node/tests/test_indexer_silent_disk.py new file mode 100644 index 0000000..4073ac5 --- /dev/null +++ b/packages/meshbay-node/tests/test_indexer_silent_disk.py @@ -0,0 +1,485 @@ +""" +A disk that stops answering, and the scan that used to wait on it for ever. + +A read on a failing drive, a share that dropped or a file a sync client has not +fetched does not fail, it does not return. The hash ran on the indexer's one +thread with nothing to bound it, so one such file stopped the scan of its group +where it stood: every root behind it unread, the enrichment that follows a scan +never started, a pause unable to take effect (it is looked at between two +files), and nothing in the log to say which file, since a file is named there +only when it fails. + +What is timed is silence, not work: a listing of a million entries and a file +on a slow link both take as long as they take. The files below "never answer" +by waiting on an event the test sets when it ends. +""" + +import asyncio +import os +import threading +from pathlib import Path +from types import SimpleNamespace + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_node import ops +from meshbay_node.indexer import indexer as indexer_mod +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.indexer.indexer import NO_ANSWER, DirectoryIndexer +from meshbay_node.roots import RootSet +from meshbay_node.roster import Roster + +pytestmark = pytest.mark.asyncio + +GID = "g" * 32 + + +@pytest.fixture(autouse=True) +def _short_patience(monkeypatch): + """Tenths of a second where the node waits a minute.""" + monkeypatch.setattr(DirectoryIndexer, "STALL_SECS", 0.3) + monkeypatch.setattr(DirectoryIndexer, "SLOW_SECS", 0.1) + monkeypatch.setattr(DirectoryIndexer, "DISK_POLL_SECS", 0.02) + + +@pytest.fixture +def never(): + """What a read that never returns waits on. Set at the end, so the threads + the tests leave behind do end.""" + event = threading.Event() + yield event + event.set() + + +def _folder(tmp_path: Path, name: str, files: list[str]) -> Path: + d = tmp_path / name + d.mkdir() + for f in files: + (d / f).write_bytes(f"{name}/{f}".encode()) + return d + + +def _indexer(roots: RootSet, **kw) -> DirectoryIndexer: + return DirectoryIndexer(roots=roots, group_id=GID, + sk_node=Ed25519PrivateKey.generate(), gek=None, **kw) + + +def _names(idx: DirectoryIndexer) -> set[str]: + return {e.name for e in idx.index.entries} + + +def _silent_files(monkeypatch, never: threading.Event, silent=lambda name: False) -> list[str]: + """Reads of the files `silent` names never return. Every read is recorded.""" + read: list[str] = [] + real = indexer_mod._scan_file + + def scan(root, path): + read.append(path.name) + if silent(path.name): + never.wait() + return None + return real(root, path) + + monkeypatch.setattr(indexer_mod, "_scan_file", scan) + return read + + +# ── One file ───────────────────────────────────────────────────────────────── + +async def test_a_read_that_never_returns_costs_its_file_not_the_scan( + tmp_path, monkeypatch, never): + films = _folder(tmp_path, "Films", ["a.mkv", "b.mkv", "stuck.mkv", "c.mkv", "d.mkv"]) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + _silent_files(monkeypatch, never, lambda name: name == "stuck.mkv") + + await asyncio.wait_for(idx.initial_scan(), timeout=10) + + assert _names(idx) == {"a.mkv", "b.mkv", "c.mkv", "d.mkv"} + (unread,) = idx.unread.values() + assert (Path(unread.path).name, unread.root, unread.reason) == ( + "stuck.mkv", "Films", NO_ANSWER) + assert roots.roots[0].paused is False, "one file is not the root's fault" + assert idx.progress.scanning is False + assert idx.progress.files_done == 5, "the file given up on is done with, too" + + +async def test_reconcile_does_not_wait_on_that_file_again_until_it_changes( + tmp_path, monkeypatch, never): + """Reconcile indexes every file the index lacks, at every pass. Without a + memory of this one it would wait out the limit on it each time.""" + films = _folder(tmp_path, "Films", ["a.mkv", "stuck.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + answering = indexer_mod._scan_file + read = _silent_files(monkeypatch, never, lambda name: name == "stuck.mkv") + await idx.initial_scan() + assert read.count("stuck.mkv") == 1 + + await idx.reconcile() + await idx.reconcile() + assert read.count("stuck.mkv") == 1, "asked the disk again for a file it left unanswered" + + monkeypatch.setattr(indexer_mod, "_scan_file", answering) # the disk is back... + await idx.reconcile() + assert "stuck.mkv" not in _names(idx), "...which nothing has told the node" + + (films / "stuck.mkv").write_bytes(b"written again, and longer") + await idx.reconcile() + assert "stuck.mkv" in _names(idx) + assert idx.unread == {} + + +async def test_a_file_the_system_refuses_is_listed_with_what_it_said( + tmp_path, monkeypatch): + films = _folder(tmp_path, "Films", ["a.mkv", "locked.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + + def refuse(path, *args, **kw): + # What a locked file does: it lists and stats, and refuses to open. + if Path(path).name == "locked.mkv": + raise PermissionError(13, "Permission denied") + return open(path, *args, **kw) + + monkeypatch.setattr(indexer_mod, "open", refuse, raising=False) + await idx.initial_scan() + + assert _names(idx) == {"a.mkv"} + (unread,) = idx.unread.values() + assert (Path(unread.path).name, unread.reason) == ("locked.mkv", "Permission denied") + + # A refusal costs nothing to ask again, and permissions change without the + # file changing: it is not remembered the way a silence is. + monkeypatch.delattr(indexer_mod, "open") + await idx.reconcile() + assert "locked.mkv" in _names(idx) + assert idx.unread == {} + + +async def test_a_file_that_has_gone_is_no_longer_unread(tmp_path, monkeypatch, never): + films = _folder(tmp_path, "Films", ["a.mkv", "stuck.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: name == "stuck.mkv") + await idx.initial_scan() + assert len(idx.unread) == 1 + + (films / "stuck.mkv").unlink() + await idx.reconcile() + + assert idx.unread == {} + + +# ── What counts as silence ─────────────────────────────────────────────────── + +async def test_a_slow_file_that_keeps_arriving_is_not_given_up_on(tmp_path, monkeypatch): + """Twice the limit in all, never the limit without a megabyte.""" + films = _folder(tmp_path, "Films", ["slow.mkv"]) + root = RootSet.build([{"path": str(films)}]) + idx = _indexer(root) + real = indexer_mod._scan_file + + def trickle(root, path): + for _ in range(6): + threading.Event().wait(0.1) + indexer_mod._beat() + return real(root, path) + + monkeypatch.setattr(indexer_mod, "_scan_file", trickle) + await idx.initial_scan() + + assert _names(idx) == {"slow.mkv"} + assert idx._executor.left_behind == 0 + + +async def test_files_waiting_behind_a_slow_one_are_not_timed_for_the_wait( + tmp_path, monkeypatch): + """A season dropped into a watched folder is hashed one file at a time on + the one thread, each event its own task. Timed from when it was asked, the + last of them would be given up on for having queued.""" + names = [f"e{i}.mkv" for i in range(5)] + films = _folder(tmp_path, "Films", names) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + real = indexer_mod._scan_file + + def unhurried(root, path): + threading.Event().wait(0.15) + return real(root, path) + + monkeypatch.setattr(indexer_mod, "_scan_file", unhurried) + entries = await asyncio.gather(*( + idx._hash_or_cached(roots.roots[0], films / name) for name in names)) + + assert all(entries), "0.75 s of queue against a limit of 0.3 s" + assert idx.unread == {} and idx._executor.left_behind == 0 + + +# ── Three in a row ─────────────────────────────────────────────────────────── + +async def test_three_in_a_row_pause_the_root_and_the_next_root_is_indexed( + tmp_path, monkeypatch, never): + """A minute a file is weeks on a music library, with a thread left behind + each time. After three with none read in between it is the disk.""" + sick = _folder(tmp_path, "Sick", [f"s{i}.mkv" for i in range(6)]) + well = _folder(tmp_path, "Well", ["w0.mkv", "w1.mkv"]) + roots = RootSet.build([{"path": str(sick)}, {"path": str(well)}]) + told = [] + + async def on_change(indexer): + told.append([r["paused"] for r in indexer.index.roots]) + + idx = _indexer(roots, on_change=on_change) + read = _silent_files(monkeypatch, never, lambda name: name.startswith("s")) + + await asyncio.wait_for(idx.initial_scan(), timeout=10) + + assert sum(name.startswith("s") for name in read) == 3, read + assert roots.by_name("Sick").paused is True + assert Path(idx.stalled["Sick"]).parent == sick + assert _names(idx) == {"w0.mkv", "w1.mkv"}, "the root behind it waited for ever" + assert told and told[0] == [True, False], "members are told the root is paused" + assert idx.progress.scanning is False and idx.progress.queued == [] + + +async def test_a_file_read_in_between_starts_the_count_again(tmp_path, monkeypatch, never): + names = ["s0", "s1", "good", "s2", "s3", "s4"] + films = _folder(tmp_path, "Films", [n + ".mkv" for n in names]) + roots = RootSet.build([{"path": str(films)}]) + root = roots.roots[0] + idx = _indexer(roots) + _silent_files(monkeypatch, never, lambda name: name.startswith("s")) + + for name in names[:5]: + await idx._hash_or_cached(root, films / f"{name}.mkv") + assert root.paused is False, "two, one read, two: never three in a row" + + await idx._hash_or_cached(root, films / "s4.mkv") + assert root.paused is True + + +async def test_resume_gets_past_the_files_that_gave_no_answer(tmp_path, monkeypatch, never): + films = _folder(tmp_path, "Films", ["s0.mkv", "s1.mkv", "s2.mkv"]) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + read = _silent_files(monkeypatch, never, lambda name: name.startswith("s")) + await idx.initial_scan() + assert roots.roots[0].paused is True + (films / "new.mkv").write_bytes(b"dropped in while it was paused") + + await idx.resume_root("Films") + await asyncio.gather(*idx._scan_tasks) + + assert _names(idx) == {"new.mkv"} + assert len(read) == 4, f"read again what gave no answer: {read}" + assert roots.roots[0].paused is False and idx.stalled == {} + assert len(idx.unread) == 3, "still listed for the operator" + + +async def test_a_reload_does_not_start_a_stalled_root_again(tmp_path, monkeypatch, never): + """A reload builds the set from the roster, which holds the operator's + pauses. This one is the node's own.""" + films = _folder(tmp_path, "Films", ["s0.mkv", "s1.mkv", "s2.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: True) + await idx.initial_scan() + + again = RootSet.build([{"path": str(films)}]) + await idx.retarget(again) + + assert again.roots[0].paused is True + + +def _state(roots: RootSet, roster: Roster, idx: DirectoryIndexer) -> dict: + index = GroupIndex(group_id=GID, sk_node=Ed25519PrivateKey.generate()) + return { + "groups_ctx": {GID: {"index": index, "roots": roots, "gek": None}}, + "indexers": {GID: idx}, + "config": SimpleNamespace(groups=[SimpleNamespace(id=GID, roots=[])]), + "roster": roster, + "node_user_id": "operator", + } + + +async def test_the_operators_resume_reaches_a_root_the_node_paused( + tmp_path, monkeypatch, never): + """Paused on the indexer's set and nowhere in the roster; and after a + reload the set the operations look at may be another object.""" + films = _folder(tmp_path, "Films", ["s0.mkv", "s1.mkv", "s2.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: True) + await idx.initial_scan() + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + seen_by_ops = RootSet.build([{"path": str(films)}]) + out = await ops.resume_root(_state(seen_by_ops, roster, idx), GID, "Films") + await asyncio.gather(*idx._scan_tasks) + + assert out["status"] == "resumed" + assert idx.roots.roots[0].paused is False and idx.stalled == {} + finally: + await roster.close() + + +# ── The listing ────────────────────────────────────────────────────────────── + +async def test_a_listing_that_stops_answering_pauses_the_root(tmp_path, monkeypatch, never): + sick = _folder(tmp_path, "Sick", ["a.mkv"]) + well = _folder(tmp_path, "Well", ["w.mkv"]) + roots = RootSet.build([{"path": str(sick)}, {"path": str(well)}]) + idx = _indexer(roots) + real = indexer_mod._walk_root + + def walk(root): + if root.name == "Sick": + indexer_mod._beat(sick / "deep" / "down") + never.wait() + return real(root) + + monkeypatch.setattr(indexer_mod, "_walk_root", walk) + await asyncio.wait_for(idx.initial_scan(), timeout=10) + + assert roots.by_name("Sick").paused is True + assert idx.stalled == {"Sick": str(sick / "deep" / "down")}, "where it went silent" + assert _names(idx) == {"w.mkv"} + + +async def test_a_rescan_that_gets_no_listing_keeps_what_the_root_held( + tmp_path, monkeypatch, never): + """A rescan drops the root's entries before it walks. A walk that learns + nothing must not leave the root empty for everyone.""" + films = _folder(tmp_path, "Films", ["a.mkv", "b.mkv"]) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + await idx.initial_scan() + for entry in idx.index.entries: + entry.display_title = "Kept" + + monkeypatch.setattr(indexer_mod, "_walk_root", lambda root: never.wait()) + await asyncio.wait_for(idx._rescan_root(roots.roots[0]), timeout=10) + + assert _names(idx) == {"a.mkv", "b.mkv"} + assert all(e.display_title == "Kept" for e in idx.index.entries) + assert idx.drain_rescanned_ids() == set(), "nothing was rebuilt, nothing to enrich again" + + +async def test_a_reconcile_that_gets_no_listing_pauses_and_removes_nothing( + tmp_path, monkeypatch, never): + films = _folder(tmp_path, "Films", ["a.mkv"]) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + await idx.initial_scan() + + monkeypatch.setattr(indexer_mod, "_walk_root", lambda root: never.wait()) + await asyncio.wait_for(idx.reconcile(), timeout=10) + + assert roots.roots[0].paused is True + assert _names(idx) == {"a.mkv"} + + +def _listed_before(root_path: Path) -> set[Path]: + """The walk this replaced.""" + return {p for p in root_path.rglob("*") if p.is_file()} + + +async def test_the_walk_lists_what_it_always_listed(tmp_path): + root = _folder(tmp_path, "Root", ["top.mkv", ".hidden", "x.part"]) + (root / "a" / "b" / "c").mkdir(parents=True) + (root / "a" / "one.mkv").write_bytes(b"1") + (root / "a" / "b" / "c" / "deep.mkv").write_bytes(b"2") + (root / "empty").mkdir() + outside = _folder(tmp_path, "Outside", ["o.mkv"]) + try: + os.symlink(outside / "o.mkv", root / "link_to_file.mkv") + os.symlink(outside, root / "link_to_dir", target_is_directory=True) + except OSError: + pass # no right to make links here: the rest still holds + listed = indexer_mod._walk_root(RootSet.build([{"path": str(root)}]).roots[0]) + + assert set(listed) == _listed_before(root) + assert len(listed) == len(set(listed)) + + +async def test_a_directory_that_cannot_be_listed_costs_only_itself(tmp_path, monkeypatch): + """One refusal used to be silent, and any other error ended the walk of the + whole root: "Cannot scan root", and nothing indexed on it at all.""" + root = _folder(tmp_path, "Root", ["top.mkv"]) + for sub in ("ok", "broken"): + (root / sub).mkdir() + (root / sub / f"{sub}.mkv").write_bytes(sub.encode()) + real = os.scandir + + def scandir(path): + if Path(path).name == "broken": + raise OSError(5, "Input/output error") + return real(path) + + monkeypatch.setattr(indexer_mod.os, "scandir", scandir) + idx = _indexer(RootSet.build([{"path": str(root)}])) + await idx.initial_scan() + + assert _names(idx) == {"top.mkv", "ok.mkv"} + + +# ── Said, and survivable ───────────────────────────────────────────────────── + +async def test_the_operator_is_told_which_file_the_disk_is_silent_on( + tmp_path, monkeypatch, never): + monkeypatch.setattr(DirectoryIndexer, "STALL_SECS", 30) + films = _folder(tmp_path, "Films", ["stuck.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: True) + assert idx.disk_wait() is None + + scan = asyncio.create_task(idx.initial_scan()) + try: + for _ in range(200): + if idx.disk_wait(): + break + await asyncio.sleep(0.02) + path, seconds = idx.disk_wait() + assert Path(path) == films / "stuck.mkv" and seconds >= 0.1 + finally: + never.set() + await asyncio.wait_for(scan, timeout=10) + assert idx.disk_wait() is None + + +async def test_the_node_can_stop_with_a_read_still_stuck(tmp_path, monkeypatch, never): + """The interpreter joins the threads of a pool on its way out. One read + that never returned kept the node from stopping at all.""" + films = _folder(tmp_path, "Films", ["stuck.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: True) + await idx.initial_scan() + assert idx._executor.left_behind == 1 + + await asyncio.wait_for(idx.stop(), timeout=5) + + stuck = [t for t in threading.enumerate() if t.name.startswith("indexer_")] + assert stuck and all(t.daemon for t in stuck) + + +async def test_a_scan_asks_the_disk_nothing_from_the_event_loop(tmp_path, monkeypatch): + """The hash was off the loop and the stat before it was not. On a disk that + stops answering that stat is where the loop would wait, and with it every + connection of the node: no limit on the indexer's thread helps a call made + somewhere else.""" + films = _folder(tmp_path, "Films", ["a.mkv", "b.mkv"]) + (films / "sub").mkdir() + (films / "sub" / "c.mkv").write_bytes(b"c") + idx = _indexer(RootSet.build([{"path": str(films)}])) + asked: set[str] = set() + real = Path.stat + + def stat(self, **kw): + if films in self.parents: + asked.add(threading.current_thread().name.split("_")[0]) + return real(self, **kw) + + monkeypatch.setattr(Path, "stat", stat) + await idx.initial_scan() + (films / "new.mkv").write_bytes(b"missed by the watcher") + await idx.reconcile() + + assert _names(idx) == {"a.mkv", "b.mkv", "c.mkv", "new.mkv"} + assert asked == {"indexer"}, f"a stat ran on: {asked}" diff --git a/packages/meshbay-node/tests/test_long_paths.py b/packages/meshbay-node/tests/test_long_paths.py new file mode 100644 index 0000000..9f7876a --- /dev/null +++ b/packages/meshbay-node/tests/test_long_paths.py @@ -0,0 +1,400 @@ +""" +Files whose full path is longer than Windows' MAX_PATH (260 characters). + +A deep library reaches it without trying: "Artist / Album (year, edition) / +01 - a long track title.flac" below a folder that is already deep. With the +system's default setting (`LongPathsEnabled` off) every call given a plain path +that long refuses it, whatever the program declares, and such a file was +invisible to the members: the walk could not list its directory, and said so +only in the log. The node now hands the system the extended form (`\\\\?\\`) at +each call that touches a group's disk, which works whatever the setting, and +keeps the plain form everywhere else: in the index, the hash cache, the log. + +ffmpeg and ffprobe are given the plain form on purpose: they extend a long path +themselves, which the probe and thumbnail tests below hold. + +These run only where the limit exists: with the setting on, a plain path works +too and they would prove nothing. +""" + +import asyncio +import os +import shutil +import sys +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.protocol import file_chunk_plaintext +from meshbay_node import ops +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roots import RootSet, publish_upload, safe_subdir +from meshbay_node.transport.webrtc.disk import ( + _append_chunk, + _free_bytes, + _is_empty_dir, + _locate, + _mkdir_if_absent, + _read_and_encrypt, + _rmdir_if_empty, +) + + +def _limit_in_force() -> bool: + if sys.platform != "win32": + return False + import winreg + try: + with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, + r"SYSTEM\CurrentControlSet\Control\FileSystem") as key: + return winreg.QueryValueEx(key, "LongPathsEnabled")[0] == 0 + except OSError: + return True + + +pytestmark = pytest.mark.skipif( + not _limit_in_force(), reason="MAX_PATH binds only on Windows with LongPathsEnabled off") + +GID = "g" * 32 +SEGMENT = "A folder name as long as the ones a real library has" + + +def _x(path: Path) -> str: + """The extended form, for the test's own setup and checks.""" + return "\\\\?\\" + str(path) + + +def _write(path: Path, data: bytes) -> Path: + with open(_x(path), "wb") as f: + f.write(data) + return path + + +def _read(path: Path) -> bytes: + with open(_x(path), "rb") as f: + return f.read() + + +@pytest.fixture +def library(tmp_path): + """A root with one short file at the top and a directory past the limit. + Removed here: the plain-path cleanup pytest does would stop at it.""" + root = tmp_path / "Library" + deep = root + while len(str(deep)) < 280: + deep = deep / SEGMENT + os.makedirs(_x(deep)) + _write(root / "short.txt", b"near the top") + yield root, deep + shutil.rmtree(_x(root), ignore_errors=True) + + +def _indexer(roots: RootSet, **kw) -> DirectoryIndexer: + return DirectoryIndexer(roots=roots, group_id=GID, + sk_node=Ed25519PrivateKey.generate(), gek=None, **kw) + + +def _names(idx: DirectoryIndexer) -> set[str]: + return {e.name for e in idx.index.entries} + + +def _entry(idx: DirectoryIndexer, name: str): + (entry,) = [e for e in idx.index.entries if e.name == name] + return entry + + +def _ffmpeg() -> str: + exe = shutil.which("ffmpeg") + if exe is None: + pytest.skip("ffmpeg is not on PATH") + return exe + + +async def _make_media(args: list[str], out: Path) -> None: + proc = await asyncio.create_subprocess_exec( + _ffmpeg(), "-v", "error", "-y", *args, str(out), + stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE) + _, err = await proc.communicate() + assert proc.returncode == 0, err.decode(errors="replace") + + +def test_the_fixture_is_past_the_limit(library): + """The premise: a plain path this long is refused on this machine.""" + _, deep = library + track = _write(deep / "track.mkv", b"deep down") + assert len(str(track)) > 260 + assert not os.path.exists(str(track)) + with pytest.raises(OSError): + os.scandir(str(deep)) + + +# ── Indexing ──────────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_a_scan_indexes_a_file_past_the_limit(library): + root, deep = library + _write(deep / "track.mkv", b"deep down" * 100) + idx = _indexer(RootSet.build([{"path": str(root)}])) + + await idx.initial_scan() + + assert _names(idx) == {"short.txt", "track.mkv"} + assert _entry(idx, "track.mkv").size == 900 + assert idx.unread == {} + + +@pytest.mark.asyncio +async def test_reconcile_finds_a_new_one_and_leaves_a_known_one_alone(library): + """Reconcile compares the disk with the index by path. Were the two sides + to spell a long path differently, it would rewrite the entry at every + pass and push the change to every member.""" + root, deep = library + _write(deep / "track.mkv", b"deep down") + idx = _indexer(RootSet.build([{"path": str(root)}])) + await idx.initial_scan() + + _write(deep / "later.mkv", b"added while the node ran") + await idx.reconcile() + assert "later.mkv" in _names(idx) + + version = idx.index.version + await idx.reconcile() + assert idx.index.version == version, "a pass with nothing new changed the index" + + +@pytest.mark.asyncio +async def test_a_watcher_event_indexes_it(library): + root, deep = library + idx = _indexer(RootSet.build([{"path": str(root)}]), debounce_secs=0.01) + await idx.initial_scan() + idx._loop = asyncio.get_running_loop() + + _write(deep / "arrived.mkv", b"one more") + idx._schedule_update(deep / "arrived.mkv") + for _ in range(200): + if "arrived.mkv" in _names(idx): + break + await asyncio.sleep(0.01) + + assert "arrived.mkv" in _names(idx) + + +@pytest.mark.asyncio +async def test_the_real_watcher_sees_it(library): + root, deep = library + idx = _indexer(RootSet.build([{"path": str(root)}]), debounce_secs=0.05) + await idx.start() + try: + _write(deep / "watched.mkv", b"seen by the system's own notification") + for _ in range(100): + if "watched.mkv" in _names(idx): + break + await asyncio.sleep(0.05) + finally: + await idx.stop() + + assert "watched.mkv" in _names(idx) + + +@pytest.mark.asyncio +async def test_a_root_that_is_itself_past_the_limit(library): + _, deep = library + _write(deep / "track.mkv", b"deep down") + roots = RootSet.build([{"path": str(deep)}]) + assert roots.roots[0].is_live() + idx = _indexer(roots) + + await idx.initial_scan() + + assert _names(idx) == {"track.mkv"} + + +# ── Serving and deleting ──────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_it_is_served_to_a_member(library): + root, deep = library + data = os.urandom(5000) + _write(deep / "track.mkv", data) + roots = RootSet.build([{"path": str(root)}]) + idx = _indexer(roots) + await idx.initial_scan() + entry = _entry(idx, "track.mkv") + gek = os.urandom(32) + + path, refusal = _locate(roots, entry) + assert refusal is None + msg = _read_and_encrypt(gek, path, 0, bytes.fromhex(entry.id), entry.id) + + assert file_chunk_plaintext(gek, msg) == data + + +@pytest.mark.asyncio +async def test_it_is_served_over_quic_too(library): + from meshbay_node.transport import quic_server + + root, deep = library + _write(deep / "track.mkv", b"over the other transport") + roots = RootSet.build([{"path": str(root)}]) + idx = _indexer(roots) + await idx.initial_scan() + entry = _entry(idx, "track.mkv") + gek = os.urandom(32) + + path, refusal = _locate(roots, entry) + msg = quic_server._read_and_encrypt(gek, path, 0, bytes.fromhex(entry.id), entry.id) + + assert file_chunk_plaintext(gek, msg) == b"over the other transport" + + +@pytest.mark.asyncio +async def test_the_operator_can_delete_it(library): + root, deep = library + track = _write(deep / "track.mkv", b"to be deleted") + roots = RootSet.build([{"path": str(root)}]) + idx = _indexer(roots) + await idx.initial_scan() + state = {"groups_ctx": {GID: {"index": idx.index, "roots": roots}}} + + await ops.delete_file(state, GID, _entry(idx, "track.mkv").id) + + assert not os.path.exists(_x(track)) + assert "track.mkv" not in _names(idx) + + +# ── Uploads and folders ───────────────────────────────────────────────────── + +def test_an_upload_lands_in_a_directory_past_the_limit(library): + root, deep = library + roots = RootSet.build([{"path": str(root), "writable": True}]) + rel = "/".join(["Library", *deep.relative_to(root).parts]) + target = safe_subdir(roots, rel) + assert target == deep + + assert _free_bytes(target) > 0 + part = target / "up.bin.0123abcd.part" + _append_chunk(part, b"first ", True) + _append_chunk(part, b"second", False) + assert publish_upload(part, target, "up.bin", "up.bin") == "up.bin" + assert _read(target / "up.bin") == b"first second" + + again = target / "up.bin.4567cdef.part" + _append_chunk(again, b"same name", True) + assert publish_upload(again, target, "up.bin", "up.bin") == "up (2).bin", \ + "a long path made the existing file look absent, and it was replaced" + assert _read(target / "up.bin") == b"first second" + + +def test_an_abandoned_upload_is_found(library): + from meshbay_node import uploads + + root, deep = library + roots = RootSet.build([{"path": str(root), "writable": True}]) + part = _write(deep / "film.mkv.89abcdef.part", b"half a film") + + found = [p for p, _ in uploads.find_parts(roots.roots)] + + assert found == [part] + + +def test_members_see_the_folders_past_the_limit(library): + """The folder tree a member browses and uploads into.""" + from meshbay_node.transport.wire import list_dirs + + root, deep = library + os.makedirs(_x(deep / "Deeper still")) + roots = RootSet.build([{"path": str(root)}]) + + dirs = list_dirs(roots) + + rel = "/".join(["Library", *deep.relative_to(root).parts]) + assert rel in dirs + assert f"{rel}/Deeper still" in dirs + + +def test_a_folder_is_created_and_removed_past_the_limit(library): + _, deep = library + folder = deep / "New folder" + assert _mkdir_if_absent(folder) is None + assert _mkdir_if_absent(folder) == "Already exists" + assert _is_empty_dir(folder) + assert _rmdir_if_empty(folder) + assert not os.path.exists(_x(folder)) + + +# ── Enrichment ────────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +@pytest.mark.daemon_loop # a subprocess needs the proactor loop +async def test_audio_tags_and_the_cover_beside_it_are_read(library): + from meshbay_node.indexer.enrich_audio import _read_tags_and_cover + from PIL import Image + + _, deep = library + track = deep / "01 - A Long Track Title.flac" + await _make_media(["-f", "lavfi", "-i", "sine=duration=1", + "-metadata", "title=Some Song", "-metadata", "artist=Some Artist", + "-metadata", "album=Some Album"], track) + with open(_x(deep / "cover.jpg"), "wb") as f: + Image.new("RGB", (8, 8), "red").save(f, "JPEG") + + tags, duration, cover = await asyncio.to_thread(_read_tags_and_cover, track) + + assert tags["title"] == "Some Song" and tags["album"] == "Some Album" + assert duration == pytest.approx(1.0, abs=0.1) + assert cover == _read(deep / "cover.jpg") + + +@pytest.mark.asyncio +async def test_a_photo_is_read(library): + from meshbay_node.indexer.enrich_photo import _read_image + from PIL import Image + + _, deep = library + with open(_x(deep / "photo.jpg"), "wb") as f: + Image.new("RGB", (64, 48), "blue").save(f, "JPEG") + + thumb, width, height, _, _ = await asyncio.to_thread(_read_image, deep / "photo.jpg") + + assert (width, height) == (64, 48) and thumb + + +@pytest.mark.asyncio +@pytest.mark.daemon_loop # a subprocess needs the proactor loop +async def test_ffmpeg_probes_and_thumbnails_it_from_the_plain_path(library): + from meshbay_node.indexer.enrich import _make_thumbnail + from meshbay_node.media_probe import probe_video + + _, deep = library + clip = deep / "clip.mp4" + await _make_media(["-f", "lavfi", "-i", "testsrc=size=64x48:rate=5:duration=2", + "-pix_fmt", "yuv420p"], clip) + + probe = await probe_video(str(clip)) + thumb = await _make_thumbnail(clip, probe.duration) + + assert probe.duration == pytest.approx(2.0, abs=0.3) + assert thumb and thumb[:2] == b"\xff\xd8" + + +@pytest.mark.asyncio +async def test_a_show_folder_past_the_limit_numbers_its_episodes(library): + from meshbay_node.indexer.enrich import ( + _season_and_show_from_ancestors, + _synthetic_episode_number, + _title_from_siblings, + ) + + _, deep = library + season = deep / "Some Show" / "Season 1" + os.makedirs(_x(season)) + _write(season / "Some Show S01E01.mkv", b"") + _write(season / "Pilot without a number.mkv", b"") + _write(season / "Second without a number.mkv", b"") + later = season / "Second without a number.mkv" + + found = _season_and_show_from_ancestors(later) + assert found is not None + number, show = found + assert await asyncio.to_thread(_synthetic_episode_number, later, show, number) == 2 + assert await asyncio.to_thread(_title_from_siblings, later) == "Some Show" diff --git a/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py b/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py index 1f93ffa..056c0b9 100644 --- a/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py +++ b/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py @@ -56,13 +56,13 @@ class _Held(DirectoryIndexer): self.rescanned: list[str] = [] self._seen = 0 - async def _hash_or_cached(self, root, file_path): + async def _hash_or_cached(self, root, file_path, **kw): if self.hold_under is not None and file_path.is_relative_to(self.hold_under): if self._seen == self.hold_after: self.at_gate.set() await self.gate.wait() self._seen += 1 - return await super()._hash_or_cached(root, file_path) + return await super()._hash_or_cached(root, file_path, **kw) async def _rescan_root(self, root): self.rescanned.append(root.name) |