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_node import ops
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roots import RootSet
from meshbay_node.roster import Roster

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_legacy_scalar_follows_the_list_in_the_live_context(tmp_path):
    """
    `video_root` rides on the handshake ack for MNP 1.0 clients and is read
    from the group context. Left behind by a save, it would disagree with the
    list 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 ctx["video_root"] == "Media/Albums", (
            "the scalar must be the first of the list, not a stale value")

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


async def test_the_photo_alias_stays_a_list_and_chat_stays_a_string(tmp_path):
    """The alias table has to carry the shape, not just the name."""
    state, roster = await _state(tmp_path)
    ctx = state["groups_ctx"][GROUP]
    try:
        await ops.set_app_directories(state, GROUP, "photo",
                                      ["Media/Films", "Media/Albums"])
        assert ctx["photo_roots"] == ["Media/Albums", "Media/Films"]
        await ops.set_app_directory(state, GROUP, "chat", "Media",
                                    require_writable=True)
        assert ctx["chat_directory"] == "Media"
    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()