path: root/packages/meshbay-node/tests/test_roots.py

"""
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,
    SAFE_UPLOAD_NAME, safe_subdir, _free_name,
)
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"]


# ── Writable roots ───────────────────────────────────────────────────────────

def test_a_root_is_read_only_unless_it_says_otherwise(tmp_path):
    """
    The default is the safe one. An operator who shares a directory has not
    thereby agreed to let anyone write into it, and the version of this that
    guessed — one root, so it must be the upload target — meant adding a
    second directory silently changed what the first one was.
    """
    (tmp_path / "Media").mkdir()
    roots = RootSet.build([_spec(tmp_path / "Media")])
    assert roots.roots[0].writable is False
    assert roots.writable_roots == []


def test_several_roots_can_be_writable_at_once(tmp_path):
    (tmp_path / "A").mkdir()
    (tmp_path / "B").mkdir()
    (tmp_path / "C").mkdir()
    roots = RootSet.build([_spec(tmp_path / "A", writable=True),
                           _spec(tmp_path / "B"),
                           _spec(tmp_path / "C", writable=True)])
    assert [r.name for r in roots.writable_roots] == ["A", "C"]


def test_a_fully_read_only_group_is_valid(tmp_path):
    """
    A group that only publishes is the point of the read-only model, not a
    misconfiguration — build must not refuse it, and nothing downstream may
    promote a root to writable to have somewhere to put an upload.
    """
    (tmp_path / "A").mkdir()
    (tmp_path / "B").mkdir()
    roots = RootSet.build([_spec(tmp_path / "A"), _spec(tmp_path / "B")])
    assert roots.writable_roots == []
    assert len(roots) == 2


def test_the_old_upload_flag_still_reads_as_writable(tmp_path):
    """A node.toml written before this refactor must not change meaning."""
    (tmp_path / "Media").mkdir()
    roots = RootSet.build([_spec(tmp_path / "Media", upload=True)])
    assert roots.roots[0].writable is True
    assert roots.describe()[0]["writable"] is True


def test_writable_wins_over_a_leftover_upload_flag(tmp_path):
    """
    A config carrying both is one a migration touched. `writable` is the field
    the operator's tooling writes now, so it is the one that decides — reading
    the legacy field there would undo the migration on the next load.
    """
    (tmp_path / "Media").mkdir()
    roots = RootSet.build([_spec(tmp_path / "Media", upload=True, writable=False)])
    assert roots.roots[0].writable is False


# ── 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()
    try:
        (tmp_path / "Media" / "escape").symlink_to(outside)
    except OSError as e:   # Windows without Developer Mode / SeCreateSymbolicLink
        pytest.skip(f"cannot create a symlink here: {e}")
    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", writable=True),
                           _spec(tmp_path / "Music", kind="audio",
                                 removable=True)])
    described = roots.describe()
    assert described == [
        {"name": "Media", "kind": "generic", "available": True,
         "writable": True, "removable": False, "ejected": False,
         "upload": True},
        {"name": "Music", "kind": "audio", "available": True,
         "writable": False, "removable": True, "ejected": False,
         "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)


def test_describe_still_carries_upload_for_mnp_1_0_clients(tmp_path):
    """
    `upload` is `writable` under its old name, kept because an MNP 1.0 client
    reads no other field and would otherwise decide the group takes no uploads
    at all. It is derived, never stored — the two can never disagree.
    """
    (tmp_path / "Media").mkdir()
    roots = RootSet.build([_spec(tmp_path / "Media", writable=True)])
    described = roots.describe()[0]
    assert described["upload"] == described["writable"] is True


# ── SAFE_UPLOAD_NAME ────────────────────────────────────────────────────────

def test_safe_name_accepts_unicode_letters():
    assert SAFE_UPLOAD_NAME.match("rapport (1).pdf")
    assert SAFE_UPLOAD_NAME.match("hello.txt")


def test_safe_name_rejects_dotfiles():
    assert not SAFE_UPLOAD_NAME.match(".hidden")
    assert not SAFE_UPLOAD_NAME.match("..secret")


def test_safe_name_rejects_trailing_dot_or_space():
    assert not SAFE_UPLOAD_NAME.match("file.")
    assert not SAFE_UPLOAD_NAME.match("file ")


# ── _free_name ──────────────────────────────────────────────────────────────

def test_free_name_returns_original_when_not_taken(tmp_path):
    assert _free_name(tmp_path, "photo.jpg") == "photo.jpg"


def test_free_name_appends_counter_on_collision(tmp_path):
    (tmp_path / "photo.jpg").write_text("x")
    assert _free_name(tmp_path, "photo.jpg") == "photo (2).jpg"


def test_free_name_increments_past_multiple_collisions(tmp_path):
    (tmp_path / "photo.jpg").write_text("x")
    (tmp_path / "photo (2).jpg").write_text("x")
    assert _free_name(tmp_path, "photo.jpg") == "photo (3).jpg"


# ── safe_subdir ─────────────────────────────────────────────────────────────

def test_safe_subdir_resolves_valid_path(tmp_path):
    (tmp_path / "Films").mkdir()
    (tmp_path / "Films" / "2024").mkdir()
    roots = RootSet.build([_spec(tmp_path / "Films")])
    assert safe_subdir(roots, "Films/2024") == (tmp_path / "Films" / "2024").resolve()


def test_safe_subdir_refuses_traversal(tmp_path):
    (tmp_path / "Films").mkdir()
    roots = RootSet.build([_spec(tmp_path / "Films")])
    assert safe_subdir(roots, "Films/../../etc") is None


def test_safe_subdir_refuses_empty_virtual_root(tmp_path):
    (tmp_path / "Films").mkdir()
    roots = RootSet.build([_spec(tmp_path / "Films")])
    assert safe_subdir(roots, "") is None


def test_safe_subdir_refuses_unavailable_root(tmp_path):
    (tmp_path / "Films").mkdir()
    roots = RootSet.build([_spec(tmp_path / "Films")])
    roots.roots[0].available = False
    assert safe_subdir(roots, "Films/2024") is None