mirror of
https://git.quad4.io/RNS-Things/MeshChatX.git
synced 2026-03-31 19:05:47 +00:00
824d84e3ac
- Updated the version in `__init__.py` and `version.py` to 4.3.0. - Introduced `PageNodeManager` to handle the lifecycle of page nodes, including creation, deletion, and management of their state. - Added API endpoints for listing, creating, retrieving, deleting, starting, stopping, and announcing page nodes. - Improved the frontend with a new `PageNodesPage` component for managing page nodes and integrated publishing functionality in the `MicronEditorPage`.
391 lines
14 KiB
Python
391 lines
14 KiB
Python
import os
|
|
import shutil
|
|
import tempfile
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def node_dir():
|
|
d = tempfile.mkdtemp()
|
|
yield d
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_rns():
|
|
with patch("meshchatx.src.backend.page_node.RNS") as mock:
|
|
mock_identity = MagicMock()
|
|
mock_identity.hash = b"\x01" * 16
|
|
mock_identity.get_public_key.return_value = b"\x02" * 64
|
|
|
|
mock_destination = MagicMock()
|
|
mock_destination.hash = b"\x03" * 16
|
|
|
|
mock.Identity.return_value = mock_identity
|
|
mock.Identity.from_file.return_value = mock_identity
|
|
mock.Destination.return_value = mock_destination
|
|
mock.Destination.IN = 1
|
|
mock.Destination.SINGLE = 0
|
|
mock.Destination.ALLOW_ALL = 0xFF
|
|
mock.Transport = MagicMock()
|
|
|
|
yield mock, mock_identity, mock_destination
|
|
|
|
|
|
def _make_node(node_dir, mock_rns):
|
|
from meshchatx.src.backend.page_node import PageNode
|
|
|
|
_, mock_identity, _ = mock_rns
|
|
return PageNode(
|
|
node_id="test-node-1",
|
|
name="Test Node",
|
|
base_dir=node_dir,
|
|
identity=mock_identity,
|
|
)
|
|
|
|
|
|
class TestPageNodeSetup:
|
|
def test_setup_creates_directories(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
assert os.path.isdir(node.pages_dir)
|
|
assert os.path.isdir(node.files_dir)
|
|
assert node.running is True
|
|
|
|
def test_setup_returns_destination_hash_hex(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
result = node.setup()
|
|
assert isinstance(result, str)
|
|
assert len(result) == 32
|
|
|
|
def test_setup_creates_identity_when_none(self, node_dir, mock_rns):
|
|
from meshchatx.src.backend.page_node import PageNode
|
|
|
|
rns_mock, _, _ = mock_rns
|
|
node = PageNode(
|
|
node_id="test-node-2",
|
|
name="No Identity Node",
|
|
base_dir=node_dir,
|
|
)
|
|
node.setup()
|
|
rns_mock.Identity.assert_called()
|
|
|
|
def test_setup_loads_identity_from_file(self, node_dir, mock_rns):
|
|
from meshchatx.src.backend.page_node import PageNode
|
|
|
|
rns_mock, mock_identity, _ = mock_rns
|
|
identity_path = os.path.join(node_dir, "identity")
|
|
with open(identity_path, "w") as f:
|
|
f.write("fake")
|
|
|
|
node = PageNode(
|
|
node_id="test-node-3",
|
|
name="File Identity Node",
|
|
base_dir=node_dir,
|
|
)
|
|
node.setup()
|
|
rns_mock.Identity.from_file.assert_called_with(identity_path)
|
|
|
|
def test_setup_registers_link_callback(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
_, _, mock_dest = mock_rns
|
|
mock_dest.set_link_established_callback.assert_called_once()
|
|
|
|
def test_setup_calls_ensure_local_path(self, node_dir, mock_rns):
|
|
rns_mock, _, _ = mock_rns
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
rns_mock.Identity.remember.assert_called_once()
|
|
|
|
|
|
class TestPageNodeTeardown:
|
|
def test_teardown_sets_running_false(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node.teardown()
|
|
assert node.running is False
|
|
assert node.destination is None
|
|
|
|
def test_teardown_deregisters_handlers(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node.add_page("test.mu", "content")
|
|
node.add_file("test.txt", b"data")
|
|
_, _, mock_dest = mock_rns
|
|
node.teardown()
|
|
assert mock_dest.deregister_request_handler.call_count >= 2
|
|
|
|
def test_teardown_clears_active_links(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
mock_link = MagicMock()
|
|
node.active_links.append(mock_link)
|
|
node.teardown()
|
|
assert len(node.active_links) == 0
|
|
mock_link.teardown.assert_called_once()
|
|
|
|
|
|
class TestPageNodeAnnounce:
|
|
def test_announce_calls_destination_announce(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
_, _, mock_dest = mock_rns
|
|
node.announce()
|
|
mock_dest.announce.assert_called_once()
|
|
|
|
def test_announce_skips_when_not_running(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.announce()
|
|
_, _, mock_dest = mock_rns
|
|
mock_dest.announce.assert_not_called()
|
|
|
|
def test_announce_passes_name_as_app_data(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
_, _, mock_dest = mock_rns
|
|
node.announce()
|
|
call_kwargs = mock_dest.announce.call_args
|
|
assert call_kwargs[1]["app_data"] == b"Test Node"
|
|
|
|
|
|
class TestPageNodePages:
|
|
def test_add_page_writes_file(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
name = node.add_page("hello", "Hello World")
|
|
assert name == "hello.mu"
|
|
path = os.path.join(node.pages_dir, "hello.mu")
|
|
assert os.path.isfile(path)
|
|
with open(path, "rb") as f:
|
|
assert f.read() == b"Hello World"
|
|
|
|
def test_add_page_appends_mu_extension(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
assert node.add_page("test", "x") == "test.mu"
|
|
|
|
def test_add_page_preserves_mu_extension(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
assert node.add_page("test.mu", "x") == "test.mu"
|
|
|
|
def test_add_page_registers_handler(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
_, _, mock_dest = mock_rns
|
|
node.add_page("index.mu", "content")
|
|
mock_dest.register_request_handler.assert_called()
|
|
assert "/page/index.mu" in node._registered_page_paths
|
|
|
|
def test_remove_page_deletes_file(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node.add_page("deleteme.mu", "gone")
|
|
assert node.remove_page("deleteme.mu") is True
|
|
assert not os.path.isfile(os.path.join(node.pages_dir, "deleteme.mu"))
|
|
|
|
def test_remove_nonexistent_page_returns_false(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
assert node.remove_page("nope.mu") is False
|
|
|
|
def test_list_pages(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node.add_page("b.mu", "b")
|
|
node.add_page("a.mu", "a")
|
|
pages = node.list_pages()
|
|
assert pages == ["a.mu", "b.mu"]
|
|
|
|
def test_get_page_content(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node.add_page("index.mu", "Hello Mesh")
|
|
content = node.get_page_content("index.mu")
|
|
assert content == "Hello Mesh"
|
|
|
|
def test_get_page_content_nonexistent(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
assert node.get_page_content("nope.mu") is None
|
|
|
|
|
|
class TestPageNodeFiles:
|
|
def test_add_file_writes_binary(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
name = node.add_file("doc.pdf", b"\x00\x01\x02")
|
|
assert name == "doc.pdf"
|
|
path = os.path.join(node.files_dir, "doc.pdf")
|
|
with open(path, "rb") as f:
|
|
assert f.read() == b"\x00\x01\x02"
|
|
|
|
def test_add_file_registers_handler(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node.add_file("info.txt", b"data")
|
|
assert "/file/info.txt" in node._registered_file_paths
|
|
|
|
def test_remove_file(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node.add_file("rm.bin", b"\xff")
|
|
assert node.remove_file("rm.bin") is True
|
|
assert not os.path.isfile(os.path.join(node.files_dir, "rm.bin"))
|
|
|
|
def test_remove_nonexistent_file(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
assert node.remove_file("nope.bin") is False
|
|
|
|
def test_list_files(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node.add_file("b.txt", b"bb")
|
|
node.add_file("a.txt", b"a")
|
|
files = node.list_files()
|
|
assert len(files) == 2
|
|
assert files[0]["name"] == "a.txt"
|
|
assert files[1]["name"] == "b.txt"
|
|
assert files[0]["size"] == 1
|
|
assert files[1]["size"] == 2
|
|
|
|
|
|
class TestPageNodeResponders:
|
|
def test_page_responder_returns_content(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node.add_page("index.mu", "page content")
|
|
responder = node._make_page_responder("index.mu")
|
|
result = responder("/page/index.mu", None, "req1", "link1", None, 0)
|
|
assert result == b"page content"
|
|
|
|
def test_page_responder_missing_returns_none(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
responder = node._make_page_responder("missing.mu")
|
|
result = responder("/page/missing.mu", None, "req1", "link1", None, 0)
|
|
assert result is None
|
|
|
|
def test_file_responder_returns_list(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node.add_file("data.bin", b"\x01\x02\x03")
|
|
responder = node._make_file_responder("data.bin")
|
|
result = responder("/file/data.bin", None, "req1", "link1", None, 0)
|
|
assert isinstance(result, list)
|
|
assert len(result) == 2
|
|
file_handle = result[0]
|
|
metadata = result[1]
|
|
assert file_handle.read() == b"\x01\x02\x03"
|
|
assert metadata["name"] == b"data.bin"
|
|
file_handle.close()
|
|
|
|
def test_file_responder_increments_stats(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node.add_file("f.txt", b"hello")
|
|
responder = node._make_file_responder("f.txt")
|
|
assert node._stats["files_served"] == 0
|
|
responder("/file/f.txt", None, "r", "l", None, 0)
|
|
assert node._stats["files_served"] == 1
|
|
result = responder("/file/f.txt", None, "r", "l", None, 0)
|
|
result[0].close()
|
|
assert node._stats["files_served"] == 2
|
|
|
|
|
|
class TestPageNodeConfig:
|
|
def test_save_and_load_config(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.save_config()
|
|
from meshchatx.src.backend.page_node import PageNode
|
|
|
|
config = PageNode.load_config(node_dir)
|
|
assert config is not None
|
|
assert config["node_id"] == "test-node-1"
|
|
assert config["name"] == "Test Node"
|
|
|
|
def test_load_config_missing(self, node_dir):
|
|
from meshchatx.src.backend.page_node import PageNode
|
|
|
|
assert PageNode.load_config(node_dir) is None
|
|
|
|
|
|
class TestPageNodeStatus:
|
|
def test_get_status(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node.add_page("index.mu", "hi")
|
|
status = node.get_status()
|
|
assert status["node_id"] == "test-node-1"
|
|
assert status["name"] == "Test Node"
|
|
assert status["running"] is True
|
|
assert status["destination_hash"] is not None
|
|
assert "index.mu" in status["pages"]
|
|
assert isinstance(status["stats"], dict)
|
|
|
|
def test_get_destination_hash_when_not_running(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
assert node.get_destination_hash() is None
|
|
|
|
|
|
class TestPageNodeLinkCallbacks:
|
|
def test_link_established_callback(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
mock_link = MagicMock()
|
|
node._link_established(mock_link)
|
|
assert mock_link in node.active_links
|
|
assert node._stats["links_established"] == 1
|
|
mock_link.set_link_closed_callback.assert_called_once()
|
|
|
|
def test_link_closed_callback(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
mock_link = MagicMock()
|
|
node.active_links.append(mock_link)
|
|
node._link_closed(mock_link)
|
|
assert mock_link not in node.active_links
|
|
|
|
def test_link_closed_callback_ignores_unknown(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
node._link_closed(MagicMock())
|
|
assert len(node.active_links) == 0
|
|
|
|
|
|
class TestPageNodeEdgeCases:
|
|
def test_add_page_before_running(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
os.makedirs(node.pages_dir, exist_ok=True)
|
|
name = node.add_page("offline.mu", "data")
|
|
assert name == "offline.mu"
|
|
assert os.path.isfile(os.path.join(node.pages_dir, "offline.mu"))
|
|
assert "/page/offline.mu" not in node._registered_page_paths
|
|
|
|
def test_setup_registers_existing_pages(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
os.makedirs(node.pages_dir, exist_ok=True)
|
|
with open(os.path.join(node.pages_dir, "pre.mu"), "w") as f:
|
|
f.write("existing")
|
|
node.setup()
|
|
assert "/page/pre.mu" in node._registered_page_paths
|
|
|
|
def test_setup_registers_existing_files(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
os.makedirs(node.files_dir, exist_ok=True)
|
|
with open(os.path.join(node.files_dir, "pre.txt"), "wb") as f:
|
|
f.write(b"existing")
|
|
node.setup()
|
|
assert "/file/pre.txt" in node._registered_file_paths
|
|
|
|
def test_path_traversal_blocked(self, node_dir, mock_rns):
|
|
node = _make_node(node_dir, mock_rns)
|
|
node.setup()
|
|
name = node.add_page("../../etc/passwd", "malicious")
|
|
assert name == "passwd.mu"
|
|
assert not os.path.exists(os.path.join(node_dir, "..", "etc"))
|