""" 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()