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

"""
One shape for every application's directories.

`video_root` (a string), `audio_root` (a string) and `photo_roots` (a list)
said the same thing three ways, and each needed its own op, its own MNP message
and its own settings widget. They are one function keyed by the app's own name
now, which is what lets an application be added without touching this layer at
all — the whole claim of the plugin architecture.

Two properties are new rather than moved, and both matter more than the tidying:

* **the paths are validated.** The setters this replaces accepted anything. A
  typo, or a path left behind when a root was removed, was stored happily and
  then matched no entry — an app showing an empty tab, with nothing to
  distinguish "misconfigured" from "no files yet". The moment of setting is the
  only one where the operator is present to be told;
* **the legacy scalar is derived, never stored.** `video_root` still rides on
  the handshake ack for MNP 1.0 clients. Kept as a second stored value it would
  drift from the list within one run — the shape of bug that reads as "it works
  after a restart".
"""

from pathlib import Path
from types import SimpleNamespace

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from meshbay_common.adminop import OP_APP_DIRECTORIES
from meshbay_node import ops
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roots import RootSet
from meshbay_node.roster import Roster
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

from conftest import one_root

pytestmark = pytest.mark.asyncio

GROUP = "g" * 32


async def _state(tmp_path: Path, *, writable: bool = True) -> tuple[dict, Roster]:
    media = tmp_path / "Media"
    (media / "Films").mkdir(parents=True)
    (media / "Albums").mkdir()
    published = tmp_path / "Published"
    published.mkdir()

    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    roots = RootSet.build([
        {"path": str(media), "writable": writable},
        {"path": str(published)},
    ])
    state = {
        "roster": roster,
        "node_user_id": "operator",
        "groups_ctx": {GROUP: {"index": index, "roots": roots}},
        "config": SimpleNamespace(groups=[SimpleNamespace(id=GROUP, roots=[])]),
    }
    return state, roster


# ── One function, any app ────────────────────────────────────────────────────

async def test_an_app_nobody_wrote_code_for_stores_its_directories(tmp_path):
    """
    The point of the generic pair. Nothing in ops.py, roster.py or the daemon
    names this app, and it round-trips anyway — which is the difference between
    a plugin architecture and a list of special cases.
    """
    state, roster = await _state(tmp_path)
    try:
        await ops.set_app_directories(state, GROUP, "helloworld", ["Media/Films"])
        assert await roster.app_directories(GROUP, "helloworld") == ["Media/Films"]
    finally:
        await roster.close()


async def test_directories_are_deduplicated_and_ordered(tmp_path):
    """
    The stored form is what the operator signs a subject built from, on both
    sides. Two clients sending the same set in different orders must produce
    the same bytes, or one of them refuses to sign its own request.
    """
    state, roster = await _state(tmp_path)
    try:
        out = await ops.set_app_directories(
            state, GROUP, "video", ["Media/Films", "Media", "Media/Films"])
        assert out["directories"] == ["Media", "Media/Films"]
    finally:
        await roster.close()


async def test_a_single_directory_app_stores_a_one_element_list(tmp_path):
    state, roster = await _state(tmp_path)
    try:
        out = await ops.set_app_directory(state, GROUP, "chat", "Media/Films",
                                          require_writable=True)
        assert out["path"] == "Media/Films"
        assert await roster.app_directories(GROUP, "chat") == ["Media/Films"]

        cleared = await ops.set_app_directory(state, GROUP, "chat", "")
        assert cleared["path"] == ""
        assert await roster.app_directories(GROUP, "chat") == []
    finally:
        await roster.close()


# ── Validation ───────────────────────────────────────────────────────────────

@pytest.mark.parametrize("bad", [
    "Nowhere", "Nowhere/Deeper", "/etc", "Media/../../etc", "..",
])
async def test_a_directory_outside_every_root_is_refused(tmp_path, bad):
    state, roster = await _state(tmp_path)
    try:
        with pytest.raises(ops.OpError):
            await ops.set_app_directories(state, GROUP, "video", [bad])
        assert await roster.app_directories(GROUP, "video") == []
    finally:
        await roster.close()


async def test_a_read_only_root_is_refused_where_writability_is_required(tmp_path):
    """
    Chat's directory is a destination, not a view. Storing one on a read-only
    root would produce a paperclip that fails at the moment somebody uses it,
    which is the failure the RO/RW model exists to move earlier.
    """
    state, roster = await _state(tmp_path)
    try:
        with pytest.raises(ops.OpError, match="read-only"):
            await ops.set_app_directory(state, GROUP, "chat", "Published",
                                        require_writable=True)
        # The same path is fine for an app that only reads it.
        await ops.set_app_directories(state, GROUP, "video", ["Published"])
        assert await roster.app_directories(GROUP, "video") == ["Published"]
    finally:
        await roster.close()


async def test_a_directory_on_an_unplugged_drive_can_still_be_configured(tmp_path):
    """
    Deliberately *not* `RootSet.resolve()`, which also refuses a root that is
    currently unavailable. An operator must be able to point an app at a
    library on a drive they have ejected — what is checked is the shape, which
    does not change with what happens to be mounted.
    """
    state, roster = await _state(tmp_path)
    roots = state["groups_ctx"][GROUP]["roots"]
    roots.roots[0].available = False
    try:
        out = await ops.set_app_directories(state, GROUP, "video", ["Media/Films"])
        assert out["directories"] == ["Media/Films"]
    finally:
        await roster.close()


# ── The derived scalar ───────────────────────────────────────────────────────

async def test_the_list_is_what_the_live_context_carries(tmp_path):
    """
    One name per app in the group context, and it is the list. A second name
    for the same idea is a second thing a save has to update, and the one it
    forgets disagrees until the next restart.
    """
    state, roster = await _state(tmp_path)
    ctx = state["groups_ctx"][GROUP]
    try:
        await ops.set_app_directories(state, GROUP, "video",
                                      ["Media/Films", "Media/Albums"])
        assert ctx["video_directories"] == ["Media/Albums", "Media/Films"]
        assert "video_root" not in ctx, (
            "a scalar nothing reads is a scalar that can go stale unnoticed")

        await ops.set_app_directories(state, GROUP, "video", [])
        assert ctx["video_directories"] == []
    finally:
        await roster.close()


async def test_chat_is_also_published_under_the_name_the_ack_uses(tmp_path):
    """
    Chat has one destination and the handshake ack publishes it as
    `chat_directory`, which the paperclip reads. That second name is derived
    from the list on every save, never stored beside it.
    """
    state, roster = await _state(tmp_path)
    ctx = state["groups_ctx"][GROUP]
    try:
        await ops.set_app_directory(state, GROUP, "chat", "Media",
                                    require_writable=True)
        assert ctx["chat_directory"] == "Media"
        assert ctx["chat_directories"] == ["Media"]

        await ops.set_app_directories(state, GROUP, "photo",
                                      ["Media/Films", "Media/Albums"])
        assert ctx["photo_directories"] == ["Media/Albums", "Media/Films"]
        assert "photo_roots" not in ctx
    finally:
        await roster.close()


# ── Reading what an older node stored ────────────────────────────────────────

async def test_an_existing_video_root_is_read_without_a_migration(tmp_path):
    """
    A node upgraded into this reads its old key until the first save through
    the new path. Requiring a migration script to run before the Videos tab
    works again would be a step nobody performs on the machine where it
    matters.
    """
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    try:
        await roster.set_setting(GROUP, "video_root", "Media/Films")
        assert await roster.app_directories(GROUP, "video") == ["Media/Films"]

        await roster.set_setting(GROUP, "audio_root", "Media/Albums")
        assert await roster.app_directories(GROUP, "music") == ["Media/Albums"]

        await roster.set_setting(GROUP, "photo_roots", '["A", "B"]')
        assert await roster.app_directories(GROUP, "photo") == ["A", "B"]
    finally:
        await roster.close()


async def test_an_empty_legacy_value_means_nothing_configured(tmp_path):
    """`video_root = ""` was how "unset" was spelled; it must not become `[""]`."""
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    try:
        await roster.set_setting(GROUP, "video_root", "")
        assert await roster.app_directories(GROUP, "video") == []
    finally:
        await roster.close()


async def test_the_new_key_wins_over_the_legacy_one(tmp_path):
    """
    Both present is a group saved once through the new path. Reading the old
    key there would undo that save on every load.
    """
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    try:
        await roster.set_setting(GROUP, "video_root", "Old/Place")
        await roster.set_app_directories(GROUP, "video", ["New/Place"])
        assert await roster.app_directories(GROUP, "video") == ["New/Place"]
    finally:
        await roster.close()


async def test_an_app_with_no_legacy_name_simply_has_none(tmp_path):
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    try:
        assert await roster.app_directories(GROUP, "helloworld") == []
    finally:
        await roster.close()


# ── Chat's own settings ──────────────────────────────────────────────────────

async def test_link_previews_default_on_and_survive_a_restart(tmp_path):
    """
    Absent means on, because that is what the node did before the switch
    existed — an upgrade must not silently change what a group's chat does.
    """
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    try:
        assert await roster.chat_link_preview(GROUP) is True
        await roster.set_chat_link_preview(GROUP, False, set_by="op")
        assert await roster.chat_link_preview(GROUP) is False
    finally:
        await roster.close()

    reopened = Roster(db_path=tmp_path / "roster.db")
    await reopened.open()
    try:
        assert await reopened.chat_link_preview(GROUP) is False
        assert await reopened.chat_link_preview("other") is True, (
            "one group's setting must not answer for another")
    finally:
        await reopened.close()


async def test_setting_link_previews_updates_the_live_context(tmp_path):
    """
    The unfurl handler reads the context, not the database — it runs per
    message and a round trip there would be absurd. So the two are kept in
    step by the op that changes it.
    """
    state, roster = await _state(tmp_path)
    try:
        await ops.set_chat_link_preview(state, GROUP, False)
        assert state["groups_ctx"][GROUP]["chat_link_preview"] is False
    finally:
        await roster.close()


# ── The handler, before any signature ────────────────────────────────────────
#
# `_do_app_directories` refuses three things up front and asks for a signature
# for everything else. The paths are not among the three: `_validate_app_dirs`
# checks those after the signature, deliberately (a settings change is not a
# capability), and `test_app_directories_signed.py` drives that whole path.

def _handler_session(tmp_path, *, authorized: bool) -> WebRTCPeerSession:
    shared = tmp_path / "shared"
    (shared / "Films").mkdir(parents=True, exist_ok=True)
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = {"roots": one_root(shared), "index": index,
                    "sk_node": index.sk_node}
    session._group_id = GROUP
    session._user_id = "op"
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    session._has_admin_authority = lambda: authorized
    session.issued = []
    session._issue_admin_challenge = lambda op, subject: session.issued.append(
        (op, subject))
    return session


async def test_an_app_this_node_does_not_know_is_refused(tmp_path):
    """A client-supplied key is otherwise a way to write arbitrary rows into
    `group_settings`."""
    session = _handler_session(tmp_path, authorized=True)
    session._do_app_directories({"app": "../etc", "directories": ["shared"]})
    assert session.issued == []
    assert [m for m in session.sent if m.get("type") == "error"]


@pytest.mark.parametrize("dirs", [None, "shared/Films", [1], {"a": 1}])
async def test_directories_that_are_not_a_list_of_strings_are_refused(tmp_path, dirs):
    session = _handler_session(tmp_path, authorized=True)
    session._do_app_directories({"app": "video", "directories": dirs})
    assert session.issued == []
    assert [m for m in session.sent if m.get("type") == "error"]


async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
    """No operator key on this node's roster means no signature can ever
    arrive, so asking for one would be a dialog that cannot be completed."""
    session = _handler_session(tmp_path, authorized=False)
    session._do_app_directories({"app": "video", "directories": ["shared/Films"]})
    assert session.issued == []
    assert [m for m in session.sent if m.get("type") == "error"]


async def test_the_subject_names_the_app_and_the_cleaned_paths(tmp_path):
    """What the operator is shown before signing has to say which application
    is about to be pointed where — two apps' challenges are otherwise
    indistinguishable — and it must match what the node will store."""
    session = _handler_session(tmp_path, authorized=True)
    session._do_app_directories(
        {"app": "video", "directories": ["/b/", "a", "a", ""]})
    assert session.issued == [(OP_APP_DIRECTORIES, "video:a,b")]


async def test_an_empty_set_is_signable(tmp_path):
    """Clearing an app's folders is an instruction like any other."""
    session = _handler_session(tmp_path, authorized=True)
    session._do_app_directories({"app": "video", "directories": []})
    assert session.issued == [(OP_APP_DIRECTORIES, "video:")]