mirror of
https://git.quad4.io/RNS-Things/MeshChatX.git
synced 2026-04-03 16:26:12 +00:00
e2586e9052
- Introduced new test files for telemetry functionality, including integration, fuzzing, and extended tests to ensure robustness and performance. - Added tests for parsing LXMF display names and telemetry data, addressing potential bugs and ensuring correct handling of various input formats. - Implemented performance tests for the InterfacesPage component, validating rendering efficiency with a large number of discovered interfaces. - Enhanced existing tests for markdown rendering and link utilities to cover additional edge cases and improve stability.
114 lines
4.0 KiB
Python
114 lines
4.0 KiB
Python
from unittest.mock import MagicMock, patch, AsyncMock
|
|
|
|
import pytest
|
|
import RNS
|
|
from aiohttp import web
|
|
|
|
from meshchatx.meshchat import ReticulumMeshChat
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_rns_minimal():
|
|
with (
|
|
patch("RNS.Reticulum") as mock_rns,
|
|
patch("RNS.Transport"),
|
|
patch("LXMF.LXMRouter"),
|
|
patch("meshchatx.meshchat.get_file_path", return_value="/tmp/mock_path"),
|
|
):
|
|
mock_rns_instance = mock_rns.return_value
|
|
mock_rns_instance.configpath = "/tmp/mock_config"
|
|
mock_rns_instance.is_connected_to_shared_instance = False
|
|
mock_rns_instance.transport_enabled.return_value = True
|
|
|
|
mock_id = MagicMock(spec=RNS.Identity)
|
|
mock_id.hash = b"test_hash_32_bytes_long_01234567"
|
|
mock_id.hexhash = mock_id.hash.hex()
|
|
mock_id.get_private_key.return_value = b"test_private_key"
|
|
yield mock_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_csp_header_logic(mock_rns_minimal, tmp_path):
|
|
storage_dir = str(tmp_path / "storage")
|
|
config_dir = str(tmp_path / "config")
|
|
|
|
with patch("meshchatx.meshchat.generate_ssl_certificate"):
|
|
app_instance = ReticulumMeshChat(
|
|
identity=mock_rns_minimal,
|
|
storage_dir=storage_dir,
|
|
reticulum_config_dir=config_dir,
|
|
)
|
|
|
|
# Mock the config values
|
|
app_instance.config.csp_extra_connect_src.set("https://api.example.com")
|
|
app_instance.config.map_tile_server_url.set(
|
|
"https://tiles.example.com/{z}/{x}/{y}.png"
|
|
)
|
|
|
|
# Mock a request and handler
|
|
request = MagicMock(spec=web.Request)
|
|
request.path = "/"
|
|
request.app = {}
|
|
|
|
# We need to mock the handler to return a real response
|
|
async def mock_handler(req):
|
|
return web.Response(text="test")
|
|
|
|
# Call _define_routes to get the security_middleware
|
|
routes = web.RouteTableDef()
|
|
_, _, security_middleware = app_instance._define_routes(routes)
|
|
|
|
response = await security_middleware(request, mock_handler)
|
|
|
|
csp = response.headers.get("Content-Security-Policy", "")
|
|
assert "https://api.example.com" in csp
|
|
assert "https://tiles.example.com" in csp
|
|
assert "default-src 'self'" in csp
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_config_update_csp(mock_rns_minimal, tmp_path):
|
|
storage_dir = str(tmp_path / "storage")
|
|
config_dir = str(tmp_path / "config")
|
|
|
|
with patch("meshchatx.meshchat.generate_ssl_certificate"):
|
|
app_instance = ReticulumMeshChat(
|
|
identity=mock_rns_minimal,
|
|
storage_dir=storage_dir,
|
|
reticulum_config_dir=config_dir,
|
|
)
|
|
|
|
# Find the config update handler
|
|
config_update_handler = None
|
|
for route in app_instance.get_routes():
|
|
if route.path == "/api/v1/config" and route.method == "PATCH":
|
|
config_update_handler = route.handler
|
|
break
|
|
|
|
assert config_update_handler is not None
|
|
|
|
# Mock request with new CSP settings
|
|
request_data = {
|
|
"csp_extra_connect_src": "https://api1.com, https://api2.com",
|
|
"csp_extra_img_src": "https://img.com",
|
|
}
|
|
|
|
request = MagicMock(spec=web.Request)
|
|
# request.json() must be awaited, so it should return an awaitable
|
|
request.json = AsyncMock(return_value=request_data)
|
|
|
|
# To avoid the JSON serialization error of MagicMock in get_config_dict,
|
|
# we mock get_config_dict to return a serializable dict.
|
|
with patch.object(
|
|
app_instance, "get_config_dict", return_value={"status": "ok"}
|
|
):
|
|
with patch.object(app_instance, "send_config_to_websocket_clients"):
|
|
response = await config_update_handler(request)
|
|
assert response.status == 200
|
|
|
|
assert (
|
|
app_instance.config.csp_extra_connect_src.get()
|
|
== "https://api1.com, https://api2.com"
|
|
)
|
|
assert app_instance.config.csp_extra_img_src.get() == "https://img.com"
|