diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_roots.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_roots.py | 242 |
1 files changed, 242 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_roots.py b/packages/meshbay-node/tests/test_roots.py new file mode 100644 index 0000000..ea4ba6a --- /dev/null +++ b/packages/meshbay-node/tests/test_roots.py @@ -0,0 +1,242 @@ +""" +Several named roots per group. + +Most of these are negative assertions — a root set that would be ambiguous is +refused rather than resolved, because every ambiguity here ends as either "my +file went to the wrong disk" or "the same film is listed twice and deleting one +copy breaks the other". +""" + +from pathlib import Path + +import pytest + +from meshbay_node.roots import Root, RootError, RootSet, entry_abs_path +from meshbay_common.protocol import IndexEntry + + +def _spec(path, **kw): + return {"path": str(path), **kw} + + +def _entry(path: str, name: str) -> IndexEntry: + return IndexEntry(id="f" * 64, name=name, path=path, size=1, + type="other", added_at=0) + + +# ── Naming ─────────────────────────────────────────────────────────────────── + +def test_the_name_is_the_directory_basename(tmp_path): + (tmp_path / "Films").mkdir() + roots = RootSet.build([_spec(tmp_path / "Films")]) + assert roots.names == ["Films"] + + +def test_an_explicit_name_wins_over_the_basename(tmp_path): + (tmp_path / "Films").mkdir() + roots = RootSet.build([_spec(tmp_path / "Films", name="Cinema")]) + assert roots.names == ["Cinema"] + + +def test_two_roots_cannot_share_a_name(tmp_path): + for parent in ("a", "b"): + (tmp_path / parent / "Films").mkdir(parents=True) + with pytest.raises(RootError, match="both be called"): + RootSet.build([_spec(tmp_path / "a" / "Films"), + _spec(tmp_path / "b" / "Films")]) + + +def test_names_clash_without_regard_to_case(tmp_path): + """ + `Films` and `films` are one directory on NTFS and exFAT, which is where most + of these live. A comparison that respected case would let the pair through + and produce two roots a Windows member cannot tell apart. + """ + (tmp_path / "a" / "Films").mkdir(parents=True) + (tmp_path / "b" / "films").mkdir(parents=True) + with pytest.raises(RootError, match="both be called"): + RootSet.build([_spec(tmp_path / "a" / "Films"), + _spec(tmp_path / "b" / "films")]) + + +def test_a_name_windows_cannot_write_is_refused(tmp_path): + """ + The root name is a folder every member sees, including on Windows, where + `AUX` cannot be created at all. + """ + (tmp_path / "AUX").mkdir() + with pytest.raises(RootError, match="reserved on Windows"): + RootSet.build([_spec(tmp_path / "AUX")]) + + +# ── Nesting ────────────────────────────────────────────────────────────────── + +def test_a_root_inside_another_is_refused(tmp_path): + """ + Both roots would index the same bytes under two identities, and deleting + through one would leave the other pointing at nothing. + """ + (tmp_path / "Media" / "Films").mkdir(parents=True) + with pytest.raises(RootError, match="is inside root"): + RootSet.build([_spec(tmp_path / "Media"), + _spec(tmp_path / "Media" / "Films")]) + + +def test_nesting_is_refused_in_either_order(tmp_path): + (tmp_path / "Media" / "Films").mkdir(parents=True) + with pytest.raises(RootError, match="is inside root"): + RootSet.build([_spec(tmp_path / "Media" / "Films"), + _spec(tmp_path / "Media")]) + + +def test_the_same_directory_twice_is_refused(tmp_path): + (tmp_path / "Media").mkdir() + with pytest.raises(RootError, match="same directory"): + RootSet.build([_spec(tmp_path / "Media"), + _spec(tmp_path / "Media", name="Other")]) + + +def test_a_sibling_with_a_shared_prefix_is_fine(tmp_path): + """`/data/Media` and `/data/Media2` are unrelated — a string prefix test + would wrongly call the second nested inside the first.""" + (tmp_path / "Media").mkdir() + (tmp_path / "Media2").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media"), _spec(tmp_path / "Media2")]) + assert roots.names == ["Media", "Media2"] + + +# ── Uploads ────────────────────────────────────────────────────────────────── + +def test_a_single_root_receives_uploads_without_being_asked(tmp_path): + (tmp_path / "Media").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media")]) + assert roots.upload_root is roots.roots[0] + + +def test_several_roots_and_no_designation_means_no_uploads(tmp_path): + """ + Refused, never guessed: picking one would send a member's file to a disk the + operator did not intend, and that is discovered weeks later. + """ + (tmp_path / "A").mkdir() + (tmp_path / "B").mkdir() + roots = RootSet.build([_spec(tmp_path / "A"), _spec(tmp_path / "B")]) + assert roots.upload_root is None + + +def test_two_upload_roots_are_refused(tmp_path): + (tmp_path / "A").mkdir() + (tmp_path / "B").mkdir() + with pytest.raises(RootError, match="exactly one"): + RootSet.build([_spec(tmp_path / "A", upload=True), + _spec(tmp_path / "B", upload=True)]) + + +# ── Resolution ─────────────────────────────────────────────────────────────── + +def test_resolution_finds_a_path_inside_its_root(tmp_path): + (tmp_path / "Media" / "2024").mkdir(parents=True) + roots = RootSet.build([_spec(tmp_path / "Media")]) + assert roots.resolve("Media/2024") == (tmp_path / "Media" / "2024").resolve() + + +def test_the_virtual_root_resolves_to_nothing(tmp_path): + """ + It is not a directory on anyone's disk — it belongs to no volume — so a file + cannot be written there and a directory cannot be created there. + """ + (tmp_path / "Media").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media")]) + for attempt in ("", "/", ".", " "): + assert roots.resolve(attempt) is None, f"{attempt!r} resolved" + + +@pytest.mark.parametrize("attempt", [ + "Media/../..", "Media/../../etc", "Media/sub/../../../etc", + "../Media", "..", "Unknown/x", +]) +def test_escaping_a_root_is_refused(tmp_path, attempt): + (tmp_path / "Media" / "sub").mkdir(parents=True) + roots = RootSet.build([_spec(tmp_path / "Media")]) + assert roots.resolve(attempt) is None, f"{attempt!r} escaped its root" + + +def test_a_symlink_out_of_the_root_is_refused(tmp_path): + """Resolved before comparing, so a link is followed and then rejected — + checking the string would have accepted it.""" + (tmp_path / "Media").mkdir() + outside = tmp_path / "outside" + outside.mkdir() + (tmp_path / "Media" / "escape").symlink_to(outside) + roots = RootSet.build([_spec(tmp_path / "Media")]) + assert roots.resolve("Media/escape") is None + + +def test_resolution_is_case_insensitive_on_the_root_name(tmp_path): + (tmp_path / "Media").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media")]) + assert roots.resolve("media") == (tmp_path / "Media").resolve() + + +def test_an_unavailable_root_resolves_to_nothing(tmp_path): + (tmp_path / "Media").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media")]) + roots.roots[0].available = False + assert roots.resolve("Media") is None + # …but the mapping is still known, so entries can still be listed as frozen + # rather than vanishing from the index. + assert roots.split("Media")[0].name == "Media" + + +def test_an_entry_under_a_missing_root_has_no_path(tmp_path): + """ + An unplugged drive must answer "nowhere", not open a file that happens to + share a relative path with another root. + """ + (tmp_path / "Media").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media")]) + entry = _entry("Media", "film.mkv") + assert entry_abs_path(roots, entry) == (tmp_path / "Media" / "film.mkv").resolve() + roots.roots[0].available = False + assert entry_abs_path(roots, entry) is None + + +def test_virtual_path_round_trips(tmp_path): + (tmp_path / "Media" / "2024").mkdir(parents=True) + roots = RootSet.build([_spec(tmp_path / "Media")]) + real = roots.resolve("Media/2024") + assert roots.virtual_of(real) == "Media/2024" + assert roots.virtual_of(roots.resolve("Media")) == "Media" + + +# ── Availability ───────────────────────────────────────────────────────────── + +def test_availability_follows_the_directory(tmp_path): + target = tmp_path / "Media" + target.mkdir() + roots = RootSet.build([_spec(target)]) + assert roots.refresh_availability() == [] + + target.rmdir() # stands in for an unmounted volume + changed = roots.refresh_availability() + assert [(r.name, live) for r, live in changed] == [("Media", False)] + assert roots.roots[0].available is False + + target.mkdir() + changed = roots.refresh_availability() + assert [(r.name, live) for r, live in changed] == [("Media", True)] + + +def test_describe_reports_what_a_member_needs(tmp_path): + (tmp_path / "Media").mkdir() + (tmp_path / "Music").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media", upload=True), + _spec(tmp_path / "Music", kind="audio")]) + described = roots.describe() + assert described == [ + {"name": "Media", "kind": "generic", "available": True, "upload": True}, + {"name": "Music", "kind": "audio", "available": True, "upload": False}, + ] + # Deliberately no paths: a member is told what exists and whether it is + # readable, not where on the operator's disk it lives. + assert not any("path" in d for d in described) |