1 files changed, 99 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_attach_from_the_hub.py b/packages/meshbay-node/tests/test_attach_from_the_hub.py
new file mode 100644
index 0000000..f4aaa25
--- /dev/null
+++ b/packages/meshbay-node/tests/test_attach_from_the_hub.py
@@ -0,0 +1,99 @@
+"""
+Hosting a group writes what the operator said, never what the hub says.
+
+`attach_group` looks the group up on the hub, because the node is the process
+signed in there. What it must not take from that answer is how people join: a
+hub able to declare a group open would be handed its key by anyone it sent
+(admission in `transport/webrtc/admission.py` admits a stranger to an open
+group). And every string it writes into node.toml is someone else's text — a
+group name chosen on the hub, a folder name — so none of it may end a TOML
+string and write lines of its own.
+"""
+
+import tomllib
+from pathlib import Path
+
+import pytest
+from meshbay_node import ops
+from meshbay_node.config import load_config
+from meshbay_node.ops.node_toml import toml_string
+
+GID = "0f8fad5b-d9cb-469f-a165-70867728950e"
+HOSTILE = 'Films"\n[node]\nui_port = 1\n# '
+
+
+class _Hub:
+    _session = object()   # signed in
+
+    def __init__(self, group):
+        self._group = group
+
+    async def list_my_groups(self):
+        return [self._group]
+
+
+def _state(tmp_path: Path, group: dict) -> dict:
+    conf = tmp_path / "node.toml"
+    conf.write_text('[hub]\nurl = "https://hub.invalid"\nusername = "op"\n\n'
+                    '[node]\nui_port = 18000\n', encoding="utf-8")
+    return {"config": load_config(conf), "config_path": str(conf), "hub": _Hub(group)}
+
+
+def _hosted(tmp_path: Path) -> dict:
+    parsed = tomllib.loads((tmp_path / "node.toml").read_text(encoding="utf-8"))
+    return parsed["groups"][0] | {"node": parsed["node"]}
+
+
+async def test_a_group_the_hub_calls_open_is_hosted_by_invitation(tmp_path):
+    state = _state(tmp_path, {"id": GID, "name": "Films", "visibility": "public",
+                              "join_policy": "open"})
+    out = await ops.attach_group(state, "Films", str(tmp_path / "share"))
+
+    hosted = _hosted(tmp_path)
+    assert hosted["join_policy"] == "invite"
+    assert hosted["visibility"] == "private"
+    assert out["hub_join_policy"] == "open", "the caller is not told the two differ"
+
+
+async def test_the_operator_opens_it(tmp_path):
+    state = _state(tmp_path, {"id": GID, "name": "Films", "join_policy": "invite"})
+    await ops.attach_group(state, "Films", str(tmp_path / "share"), join_policy="open")
+
+    hosted = _hosted(tmp_path)
+    assert (hosted["join_policy"], hosted["visibility"]) == ("open", "public")
+
+
+async def test_an_unknown_policy_is_refused(tmp_path):
+    state = _state(tmp_path, {"id": GID, "name": "Films"})
+    with pytest.raises(ops.OpError):
+        await ops.attach_group(state, "Films", str(tmp_path / "share"),
+                               join_policy="anyone")
+
+
+async def test_a_group_name_cannot_write_lines_into_node_toml(tmp_path):
+    state = _state(tmp_path, {"id": GID, "name": HOSTILE})
+    await ops.attach_group(state, HOSTILE, str(tmp_path / "share"))
+
+    hosted = _hosted(tmp_path)
+    assert hosted["name"] == HOSTILE
+    assert hosted["node"]["ui_port"] == 18000
+
+
+async def test_a_folder_name_cannot_either(tmp_path):
+    state = _state(tmp_path, {"id": GID, "name": "Films"})
+    await ops.attach_group(state, "Films", str(tmp_path / "share"))
+    state["config"] = load_config(tmp_path / "node.toml")
+    # Root names are refused with such characters already (roots.py); the
+    # path is not, and is the operator's own folder or a member's request.
+    weird = tmp_path / 'a "quoted"\\ folder\n[node]'
+    await ops.add_root(state, GID, str(weird), name="extra")
+
+    hosted = _hosted(tmp_path)
+    assert Path(hosted["roots"][-1]["path"]) == weird
+    assert hosted["node"]["ui_port"] == 18000
+
+
+@pytest.mark.parametrize("value", ["plain", 'q"uote', "back\\slash", "line\nbreak",
+                                   "tab\there", "del\x7f", "café 日本"])
+def test_every_string_reads_back_as_written(value):
+    assert tomllib.loads(f"v = {toml_string(value)}")["v"] == value