chore(docs): add script to fetch Reticulum manual at build time and update documentation handling in the backend so reticulum docs get bundled offline

This commit is contained in:
Ivan committed 2026-04-20 21:12:33 -05:00
1 parent 3076a1904f
commit 68ab1067a5
21 files changed
+788 -444

No files matched your search

+39 -46
View File
@@ -18,6 +18,7 @@ import json
import logging
import os
import platform
import re
import secrets
import shutil
import socket
@@ -228,7 +229,6 @@ class ReticulumMeshChat:
public_dir: str | None = None,
emergency: bool = False,
gitea_base_url: str | None = None,
docs_download_urls: str | None = None,
ssl_cert_path: str | None = None,
ssl_key_path: str | None = None,
rns_loglevel: str | None = None,
@@ -246,7 +246,6 @@ class ReticulumMeshChat:
self.auth_enabled_initial = auth_enabled
self.public_dir_override = public_dir
self.gitea_base_url_override = gitea_base_url
self.docs_download_urls_override = docs_download_urls
self._rns_loglevel_cli = rns_loglevel
self.websocket_clients: list[web.WebSocketResponse] = []
self._websocket_broadcast_lock = asyncio.Lock()
@@ -5256,13 +5255,6 @@ class ReticulumMeshChat:
async def docs_status(request):
return web.json_response(self.docs_manager.get_status())
# update docs
@routes.post("/api/v1/docs/update")
async def docs_update(request):
version = request.query.get("version", "latest")
success = self.docs_manager.update_docs(version=version)
return web.json_response({"success": success})
# upload docs zip
@routes.post("/api/v1/docs/upload")
async def docs_upload(request):
@@ -5372,6 +5364,34 @@ class ReticulumMeshChat:
except Exception as e:
return web.json_response({"error": str(e)}, status=500)
# export the active Reticulum manual in a layout the upload route accepts,
# so users can share their bundled or customised manual with another peer.
@routes.get("/api/v1/docs/export/reticulum")
async def reticulum_docs_export(request):
try:
zip_data = self.docs_manager.export_reticulum_docs()
if zip_data is None:
return web.json_response(
{"error": "No Reticulum manual available to export"},
status=404,
)
version = self.docs_manager.get_current_version() or "manual"
safe_version = re.sub(r"[^A-Za-z0-9._-]+", "_", str(version))
filename = (
"reticulum_manual_"
f"{safe_version}_"
f"{datetime.now().strftime('%Y%m%d_%H%M%S')}.zip"
)
return web.Response(
body=zip_data,
content_type="application/zip",
headers={
"Content-Disposition": f'attachment; filename="{filename}"',
},
)
except Exception as e:
return web.json_response({"error": str(e)}, status=500)
@routes.get("/api/v1/database/health")
async def database_health(request):
try:
@@ -11307,7 +11327,6 @@ class ReticulumMeshChat:
frame_sources = [
"'self'",
"https://reticulum.network",
]
script_sources = ["'self'", "'unsafe-inline'", "'unsafe-eval'"]
@@ -11335,20 +11354,6 @@ class ReticulumMeshChat:
connect_sources,
)
# Add configured docs download URLs domains
docs_urls_str = self.current_context.config.docs_download_urls.get()
docs_urls = [
u.strip()
for u in docs_urls_str.replace("\n", ",").split(",")
if u.strip()
]
for url in docs_urls:
domain = add_domain_from_url(url, connect_sources)
if domain and "github.com" in domain:
content_domain = "https://objects.githubusercontent.com"
if content_domain not in connect_sources:
connect_sources.append(content_domain)
# Add map tile server domain
map_tile_url = self.current_context.config.map_tile_server_url.get()
add_domain_from_url(map_tile_url, img_sources)
@@ -11612,12 +11617,12 @@ class ReticulumMeshChat:
# we use add_static here as it's more robust for serving directories
public_dir = self.get_public_path()
# Handle documentation directories that might be in a writable storage location
# (e.g. when running from a read-only AppImage)
# Serve Reticulum docs from user-uploaded storage with a fallback to the
# bundled offline copy shipped under <public>/reticulum-docs-bundled/current.
# No remote network fallback exists; users supply replacements via upload.
if self.current_context and hasattr(self.current_context, "docs_manager"):
dm = self.current_context.docs_manager
# Custom handler for reticulum docs to allow fallback to official website
async def reticulum_docs_handler(request):
path = request.match_info.get("filename", "index.html")
if not path:
@@ -11625,18 +11630,13 @@ class ReticulumMeshChat:
if path.endswith("/"):
path += "index.html"
try:
local_path = os.path.realpath(os.path.join(dm.docs_dir, path))
base = os.path.realpath(dm.docs_dir)
except (ValueError, OSError):
return web.json_response({"error": "Invalid path"}, status=400)
if not local_path.startswith(base + os.sep) and local_path != base:
return web.json_response({"error": "Invalid path"}, status=400)
if os.path.exists(local_path) and os.path.isfile(local_path):
return web.FileResponse(local_path)
# Fallback to official website
return web.HTTPFound(f"https://reticulum.network/manual/{path}")
resolved = dm.find_docs_file(path)
if resolved is None:
return web.json_response(
{"error": "Documentation not found"},
status=404,
)
return web.FileResponse(resolved)
app.router.add_get("/reticulum-docs/{filename:.*}", reticulum_docs_handler)
@@ -15434,12 +15434,6 @@ def main():
default=os.environ.get("MESHCHAT_GITEA_BASE_URL"),
help="Base URL for Gitea instance (default: https://git.quad4.io). Can also be set via MESHCHAT_GITEA_BASE_URL environment variable.",
)
parser.add_argument(
"--docs-download-urls",
type=str,
default=os.environ.get("MESHCHAT_DOCS_DOWNLOAD_URLS"),
help="Comma-separated list of URLs to download documentation from. Can also be set via MESHCHAT_DOCS_DOWNLOAD_URLS environment variable.",
)
parser.add_argument(
"--test-exception-message",
type=str,
@@ -15587,7 +15581,6 @@ def main():
public_dir=args.public_dir,
emergency=args.emergency,
gitea_base_url=args.gitea_base_url,
docs_download_urls=args.docs_download_urls,
ssl_cert_path=ssl_cert,
ssl_key_path=ssl_key,
rns_loglevel=rns_log_cli,
-11
View File
@@ -133,22 +133,11 @@ class ConfigManager:
self.auth_enabled = self.BoolConfig(self, "auth_enabled", False)
self.auth_password_hash = self.StringConfig(self, "auth_password_hash", None)
self.auth_session_secret = self.StringConfig(self, "auth_session_secret", None)
self.docs_downloaded = self.BoolConfig(self, "docs_downloaded", False)
self.initial_docs_download_attempted = self.BoolConfig(
self,
"initial_docs_download_attempted",
False,
)
self.gitea_base_url = self.StringConfig(
self,
"gitea_base_url",
"https://git.quad4.io",
)
self.docs_download_urls = self.StringConfig(
self,
"docs_download_urls",
"https://git.quad4.io/Reticulum/reticulum_website/archive/main.zip,https://github.com/markqvist/reticulum_website/archive/refs/heads/main.zip",
)
# desktop config
self.desktop_open_calls_in_separate_window = self.BoolConfig(
+185 -237
View File
@@ -1,28 +1,35 @@
# SPDX-License-Identifier: 0BSD
import asyncio
import html
import io
import logging
import os
import re
import shutil
import threading
import zipfile
import aiohttp
from meshchatx.src.backend.markdown_renderer import MarkdownRenderer
BUNDLED_DOCS_SUBDIR = os.path.join("reticulum-docs-bundled", "current")
class DocsManager:
"""Manages the bundled Reticulum manual and any user-uploaded overrides.
The Reticulum manual is shipped with the application under
``<public_dir>/reticulum-docs-bundled/current``. Users may upload a
replacement archive which is extracted into ``<storage_dir>/reticulum-docs``
and takes precedence at request time. Removing the user upload restores the
bundled copy. There is no runtime download path; the manual must be staged
at build time (see ``scripts/build/fetch_reticulum_manual.py``).
"""
def __init__(self, config, public_dir, project_root=None, storage_dir=None):
self.config = config
self.public_dir = public_dir
self.project_root = project_root
self.storage_dir = storage_dir
# Determine docs directories
if self.storage_dir:
self.docs_base_dir = os.path.join(self.storage_dir, "reticulum-docs")
self.meshchatx_docs_dir = os.path.join(self.storage_dir, "meshchatx-docs")
@@ -30,16 +37,14 @@ class DocsManager:
self.docs_base_dir = os.path.join(self.public_dir, "reticulum-docs")
self.meshchatx_docs_dir = os.path.join(self.public_dir, "meshchatx-docs")
# The actual docs are served from this directory
# We will use a 'current' subdirectory for the active version
self.docs_dir = os.path.join(self.docs_base_dir, "current")
self.versions_dir = os.path.join(self.docs_base_dir, "versions")
self.bundled_docs_dir = os.path.join(self.public_dir, BUNDLED_DOCS_SUBDIR)
self.download_status = "idle"
self.download_progress = 0
self.upload_status = "idle"
self.upload_progress = 0
self.last_error = None
# Ensure docs directories exist
try:
for d in [
self.docs_base_dir,
@@ -50,7 +55,6 @@ class DocsManager:
if not os.path.exists(d):
os.makedirs(d)
# If 'current' doesn't exist but we have versions, pick the latest one
if not os.path.exists(self.docs_dir) or not os.listdir(self.docs_dir):
self._update_current_link()
@@ -58,7 +62,6 @@ class DocsManager:
logging.exception(f"Failed to create documentation directories: {e}")
self.last_error = str(e)
# Initial population of MeshChatX docs
if os.path.exists(self.meshchatx_docs_dir) and os.access(
self.meshchatx_docs_dir,
os.W_OK,
@@ -66,7 +69,7 @@ class DocsManager:
self.populate_meshchatx_docs()
def _update_current_link(self, version=None):
"""Updates the 'current' directory to point to the specified version or the latest one."""
"""Update the 'current' directory to point at the chosen or latest version."""
if not os.path.exists(self.versions_dir):
return
@@ -74,20 +77,12 @@ class DocsManager:
if not versions:
return
target_version = version
if not target_version:
# Pick latest version (alphabetically)
target_version = versions[-1]
target_version = version or versions[-1]
version_path = os.path.join(self.versions_dir, target_version)
if not os.path.exists(version_path):
return
# On some systems symlinks might fail or be restricted, so we use a directory copy or move
# but for now let's try to just use the path directly if possible.
# However, meshchat.py uses self.docs_dir for the static route.
# To make it simple and robust across platforms, we'll clear 'current' and copy the version
if os.path.exists(self.docs_dir):
if os.path.islink(self.docs_dir):
os.unlink(self.docs_dir)
@@ -95,13 +90,9 @@ class DocsManager:
self._remove_tree_force_writable(self.docs_dir)
try:
# Try symlink first as it's efficient
# We use a relative path for the symlink target to make the storage directory portable
# version_path is relative to CWD, so we need it relative to the parent of self.docs_dir
rel_target = os.path.relpath(version_path, os.path.dirname(self.docs_dir))
os.symlink(rel_target, self.docs_dir)
except (OSError, AttributeError):
# Fallback to copy
shutil.copytree(version_path, self.docs_dir)
def get_available_versions(self):
@@ -121,7 +112,6 @@ class DocsManager:
if os.path.islink(self.docs_dir):
return os.path.basename(os.readlink(self.docs_dir))
# If it's a copy, we might need a metadata file to know which version it is
version_file = os.path.join(self.docs_dir, ".version")
if os.path.exists(version_file):
try:
@@ -129,7 +119,12 @@ class DocsManager:
return f.read().strip()
except OSError:
pass
return "unknown"
if self.has_user_docs():
return "unknown"
if self.has_bundled_docs():
return "bundled"
return None
def switch_version(self, version):
if version in self.get_available_versions():
@@ -138,7 +133,7 @@ class DocsManager:
return False
def delete_version(self, version):
"""Deletes a specific version of documentation."""
"""Delete a specific user-uploaded version of the documentation."""
if version not in self.get_available_versions():
return False
@@ -147,7 +142,6 @@ class DocsManager:
return False
try:
# If the deleted version is the current one, unlink 'current' first
current_version = self.get_current_version()
if current_version == version:
if os.path.exists(self.docs_dir):
@@ -158,7 +152,6 @@ class DocsManager:
self._remove_tree_force_writable(version_path)
# If we just deleted the current version, try to pick another one as current
if current_version == version:
self._update_current_link()
@@ -168,12 +161,9 @@ class DocsManager:
return False
def clear_reticulum_docs(self):
"""Clears all Reticulum documentation and versions."""
"""Remove every user-uploaded Reticulum doc; bundled copy is untouched."""
try:
if os.path.exists(self.docs_base_dir):
# We don't want to delete the base dir itself, just its contents
# except possibly some metadata if we added any.
# Actually, deleting everything inside reticulum-docs is fine.
for item in os.listdir(self.docs_base_dir):
item_path = os.path.join(self.docs_base_dir, item)
if os.path.islink(item_path):
@@ -183,28 +173,22 @@ class DocsManager:
else:
os.remove(item_path)
# Re-create required subdirectories
for d in [self.versions_dir, self.docs_dir]:
if not os.path.exists(d):
os.makedirs(d)
self.config.docs_downloaded.set(False)
return True
except Exception as e:
logging.exception(f"Failed to clear Reticulum docs: {e}")
return False
def populate_meshchatx_docs(self):
"""Populates meshchatx-docs from the project's docs folder."""
# Try to find docs folder in several places
"""Copy the project's bundled MeshChatX markdown docs into storage."""
search_paths = []
if self.project_root:
search_paths.append(os.path.join(self.project_root, "docs"))
# Also try in the public directory
search_paths.append(os.path.join(self.public_dir, "meshchatx-docs"))
# Also try relative to this file (project root 3 levels up)
this_dir = os.path.dirname(os.path.abspath(__file__))
search_paths.append(
os.path.abspath(os.path.join(this_dir, "..", "..", "..", "docs")),
@@ -226,19 +210,16 @@ class DocsManager:
src_path = os.path.join(src_docs, file)
dest_path = os.path.join(self.meshchatx_docs_dir, file)
# Only copy if source and destination are different
if os.path.abspath(src_path) != os.path.abspath(
dest_path,
) and os.access(self.meshchatx_docs_dir, os.W_OK):
shutil.copy2(src_path, dest_path)
# Also pre-render to HTML for easy sharing/viewing
try:
with open(src_path, encoding="utf-8") as f:
content = f.read()
html_content = MarkdownRenderer.render(content)
# Basic HTML wrapper for standalone viewing
full_html = f"""<!DOCTYPE html>
<html class="dark">
<head>
@@ -270,10 +251,12 @@ class DocsManager:
def get_status(self):
return {
"status": self.download_status,
"progress": self.download_progress,
"status": self.upload_status,
"progress": self.upload_progress,
"last_error": self.last_error,
"has_docs": self.has_docs(),
"has_bundled_docs": self.has_bundled_docs(),
"has_user_docs": self.has_user_docs(),
"has_meshchatx_docs": self.has_meshchatx_docs(),
"versions": self.get_available_versions(),
"current_version": self.get_current_version(),
@@ -331,20 +314,20 @@ class DocsManager:
}
def export_docs(self):
"""Creates a zip of all docs and returns the bytes."""
"""Build a ZIP archive containing the active Reticulum docs and MeshChatX docs."""
buffer = io.BytesIO()
active_docs_dir = self._active_reticulum_docs_dir()
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as zip_file:
# Add reticulum docs
for root, _, files in os.walk(self.docs_dir):
for file in files:
file_path = os.path.join(root, file)
rel_path = os.path.join(
"reticulum-docs",
os.path.relpath(file_path, self.docs_dir),
)
zip_file.write(file_path, rel_path)
if active_docs_dir and os.path.isdir(active_docs_dir):
for root, _, files in os.walk(active_docs_dir):
for file in files:
file_path = os.path.join(root, file)
rel_path = os.path.join(
"reticulum-docs",
os.path.relpath(file_path, active_docs_dir),
)
zip_file.write(file_path, rel_path)
# Add meshchatx docs
for root, _, files in os.walk(self.meshchatx_docs_dir):
for file in files:
file_path = os.path.join(root, file)
@@ -357,6 +340,36 @@ class DocsManager:
buffer.seek(0)
return buffer.getvalue()
def export_reticulum_docs(self, root_folder="reticulum_manual"):
"""Build a ZIP of the active Reticulum manual in upload-compatible form.
The archive lays out files under ``<root_folder>/docs/`` so that another
MeshChatX instance can re-import it via the ``/api/v1/docs/upload``
endpoint without modification. Returns ``None`` when no Reticulum docs
are currently available (neither user-uploaded nor bundled).
"""
active_docs_dir = self._active_reticulum_docs_dir()
if not active_docs_dir or not os.path.isdir(active_docs_dir):
return None
safe_root = os.path.basename(root_folder.strip()) or "reticulum_manual"
if safe_root in (".", ".."):
safe_root = "reticulum_manual"
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as zip_file:
for root, _, files in os.walk(active_docs_dir):
for file in files:
file_path = os.path.join(root, file)
arcname = os.path.join(
safe_root,
"docs",
os.path.relpath(file_path, active_docs_dir),
)
zip_file.write(file_path, arcname)
buffer.seek(0)
return buffer.getvalue()
def search(self, query, lang="en"):
if not query:
return []
@@ -364,7 +377,6 @@ class DocsManager:
results = []
query = query.lower()
# 1. Search MeshChatX Docs first
if os.path.exists(self.meshchatx_docs_dir):
for file in os.listdir(self.meshchatx_docs_dir):
if file.endswith((".md", ".txt")):
@@ -377,7 +389,6 @@ class DocsManager:
) as f:
content = f.read()
if query in content.lower():
# Simple snippet
idx = content.lower().find(query)
start = max(0, idx - 80)
end = min(len(content), idx + len(query) + 120)
@@ -398,190 +409,134 @@ class DocsManager:
except Exception as e:
logging.exception(f"Error searching MeshChatX doc {file}: {e}")
# 2. Search Reticulum Docs
if self.has_docs():
# Known language suffixes in Reticulum docs
active_docs_dir = self._active_reticulum_docs_dir()
if active_docs_dir and os.path.isdir(active_docs_dir):
known_langs = ["de", "es", "jp", "nl", "pl", "pt-br", "tr", "uk", "zh-cn"]
# Determine files to search
target_files = []
try:
for root, _, files in os.walk(self.docs_dir):
for file in files:
if file.endswith(".html"):
# Basic filtering for language if possible
if lang != "en":
if f"_{lang}.html" in file:
target_files.append(os.path.join(root, file))
else:
# English: no language suffix; other langs use _<lang>.html
has_lang_suffix = False
for lang_code in known_langs:
if f"_{lang_code}.html" in file:
has_lang_suffix = True
break
if not has_lang_suffix:
target_files.append(os.path.join(root, file))
# If we found nothing for a specific language, fall back to English ONLY
if not target_files and lang != "en":
for root, _, files in os.walk(self.docs_dir):
target_files = []
try:
for root, _, files in os.walk(active_docs_dir):
for file in files:
if file.endswith(".html"):
has_lang_suffix = False
for lang_code in known_langs:
if f"_{lang_code}.html" in file:
has_lang_suffix = True
if lang != "en":
if f"_{lang}.html" in file:
target_files.append(os.path.join(root, file))
else:
has_lang_suffix = False
for lang_code in known_langs:
if f"_{lang_code}.html" in file:
has_lang_suffix = True
break
if not has_lang_suffix:
target_files.append(os.path.join(root, file))
if not target_files and lang != "en":
for root, _, files in os.walk(active_docs_dir):
for file in files:
if file.endswith(".html"):
has_lang_suffix = False
for lang_code in known_langs:
if f"_{lang_code}.html" in file:
has_lang_suffix = True
break
if not has_lang_suffix:
target_files.append(os.path.join(root, file))
for file_path in target_files:
try:
with open(file_path, encoding="utf-8", errors="ignore") as f:
content = f.read()
text_content = re.sub(r"<[^>]+>", " ", content)
text_content = " ".join(text_content.split())
if query in text_content.lower():
title_match = re.search(
r"<title>(.*?)</title>",
content,
re.IGNORECASE | re.DOTALL,
)
title = (
title_match.group(1).strip()
if title_match
else os.path.basename(file_path)
)
title = re.sub(r"\s+[\u2014-].*$", "", title)
idx = text_content.lower().find(query)
start = max(0, idx - 80)
end = min(len(text_content), idx + len(query) + 120)
snippet = text_content[start:end]
if start > 0:
snippet = "..." + snippet
if end < len(text_content):
snippet = snippet + "..."
rel_path = os.path.relpath(file_path, active_docs_dir)
results.append(
{
"title": title,
"path": f"/reticulum-docs/{rel_path}",
"snippet": snippet,
"source": "Reticulum",
},
)
if len(results) >= 25:
break
if not has_lang_suffix:
target_files.append(os.path.join(root, file))
for file_path in target_files:
try:
with open(file_path, encoding="utf-8", errors="ignore") as f:
content = f.read()
# Very basic HTML tag removal for searching
text_content = re.sub(r"<[^>]+>", " ", content)
text_content = " ".join(text_content.split())
if query in text_content.lower():
# Find title
title_match = re.search(
r"<title>(.*?)</title>",
content,
re.IGNORECASE | re.DOTALL,
)
title = (
title_match.group(1).strip()
if title_match
else os.path.basename(file_path)
)
# Remove " — Reticulum Network Stack ..." suffix often found in Sphinx docs
title = re.sub(r"\s+[\u2014-].*$", "", title)
# Find snippet
idx = text_content.lower().find(query)
start = max(0, idx - 80)
end = min(len(text_content), idx + len(query) + 120)
snippet = text_content[start:end]
if start > 0:
snippet = "..." + snippet
if end < len(text_content):
snippet = snippet + "..."
rel_path = os.path.relpath(file_path, self.docs_dir)
results.append(
{
"title": title,
"path": f"/reticulum-docs/{rel_path}",
"snippet": snippet,
"source": "Reticulum",
},
)
if len(results) >= 25: # Limit results
break
except Exception as e:
logging.exception(f"Error searching file {file_path}: {e}")
except Exception as e:
logging.exception(f"Search failed: {e}")
except Exception as e:
logging.exception(f"Error searching file {file_path}: {e}")
except Exception as e:
logging.exception(f"Search failed: {e}")
return results
def has_docs(self):
# Check if index.html exists in the docs folder or if we have any versions
return (
os.path.exists(os.path.join(self.docs_dir, "index.html"))
or len(self.get_available_versions()) > 0
)
"""True if either user-uploaded or bundled Reticulum docs are available."""
return self.has_user_docs() or self.has_bundled_docs()
def update_docs(self, version="latest"):
if (
self.download_status == "downloading"
or self.download_status == "extracting"
):
return False
def has_user_docs(self):
if os.path.exists(os.path.join(self.docs_dir, "index.html")):
return True
return len(self.get_available_versions()) > 0
thread = threading.Thread(target=self._download_task, args=(version,))
thread.daemon = True
thread.start()
return True
def has_bundled_docs(self):
return os.path.exists(os.path.join(self.bundled_docs_dir, "index.html"))
def _download_task(self, version="latest"):
try:
asyncio.run(self._download_docs_async(version))
except Exception as e:
logging.exception(f"Docs download task failed: {e}")
self.last_error = str(e)
self.download_status = "error"
def find_docs_file(self, rel_path):
"""Resolve ``rel_path`` against user docs first, then bundled docs.
async def _download_docs_async(self, version="latest"):
self.download_status = "downloading"
self.download_progress = 0
self.last_error = None
urls_str = self.config.docs_download_urls.get()
urls = [u.strip() for u in urls_str.replace("\n", ",").split(",") if u.strip()]
if not urls:
urls = ["https://git.quad4.io/Reticulum/reticulum_website/archive/main.zip"]
timeout = aiohttp.ClientTimeout(total=60)
last_exception = None
for url in urls:
try:
logging.info(f"Attempting to download docs from {url}")
zip_path = os.path.join(self.docs_base_dir, "website.zip")
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(url) as response:
response.raise_for_status()
total_size = int(response.headers.get("Content-Length", 0) or 0)
downloaded_size = 0
with open(zip_path, "wb") as f:
async for chunk in response.content.iter_chunked(8192):
if chunk:
f.write(chunk)
downloaded_size += len(chunk)
if total_size > 0:
self.download_progress = int(
(downloaded_size / total_size) * 90,
)
self.download_status = "extracting"
if version == "latest":
import time
version = f"git-{int(time.time())}"
self._extract_docs(zip_path, version)
if os.path.exists(zip_path):
os.remove(zip_path)
self.config.docs_downloaded.set(True)
self.download_progress = 100
self.download_status = "completed"
self.switch_version(version)
return
except Exception as e:
logging.warning(f"Failed to download docs from {url}: {e}")
last_exception = e
zip_gone = os.path.join(self.docs_base_dir, "website.zip")
if os.path.exists(zip_gone):
os.remove(zip_gone)
Returns the absolute on-disk path of the matching file, or ``None`` when
the path either escapes the docs roots or no file exists in either
location. Path traversal attempts are rejected.
"""
if rel_path is None:
rel_path = ""
for base in (self.docs_dir, self.bundled_docs_dir):
if not base or not os.path.isdir(base):
continue
try:
candidate = os.path.realpath(os.path.join(base, rel_path))
base_real = os.path.realpath(base)
except (ValueError, OSError):
continue
if candidate != base_real and not candidate.startswith(base_real + os.sep):
continue
if os.path.isfile(candidate):
return candidate
return None
self.last_error = str(last_exception)
self.download_status = "error"
logging.error(f"All docs download sources failed. Last error: {last_exception}")
def _active_reticulum_docs_dir(self):
if os.path.exists(os.path.join(self.docs_dir, "index.html")):
return self.docs_dir
if self.has_bundled_docs():
return self.bundled_docs_dir
return None
def upload_zip(self, zip_bytes, version):
self.download_status = "extracting"
self.download_progress = 0
"""Extract a user-uploaded docs ZIP into a new version and switch to it."""
self.upload_status = "extracting"
self.upload_progress = 0
self.last_error = None
try:
@@ -594,13 +549,13 @@ class DocsManager:
if os.path.exists(zip_path):
os.remove(zip_path)
self.download_status = "completed"
self.download_progress = 100
self.upload_status = "completed"
self.upload_progress = 100
self.switch_version(version)
return True
except Exception as e:
self.last_error = str(e)
self.download_status = "error"
self.upload_status = "error"
logging.exception(f"Failed to upload docs: {e}")
return False
@@ -620,20 +575,17 @@ class DocsManager:
os.makedirs(version_dir, exist_ok=True)
self._ensure_dir_writable(version_dir)
# Temp dir for extraction
temp_extract = os.path.join(self.docs_base_dir, "temp_extract")
if os.path.exists(temp_extract):
self._remove_tree_force_writable(temp_extract)
with zipfile.ZipFile(zip_path, "r") as zip_ref:
# Gitea/GitHub zips have a root folder
namelist = zip_ref.namelist()
if not namelist:
raise Exception("Zip file is empty")
root_folder = namelist[0].split("/")[0]
# Check if it's the reticulum_website repo (has docs/ folder)
docs_prefix = f"{root_folder}/docs/"
has_docs_subfolder = any(m.startswith(docs_prefix) for m in namelist)
@@ -645,7 +597,6 @@ class DocsManager:
zip_ref.extract(member, temp_extract)
src_path = os.path.join(temp_extract, root_folder, "docs")
# Move files from extracted docs to version_dir
for item in os.listdir(src_path):
s = os.path.join(src_path, item)
d = os.path.join(version_dir, item)
@@ -666,7 +617,6 @@ class DocsManager:
else:
self._copy_file_no_metadata(s, d)
else:
# Fallback if no root folder
for item in os.listdir(temp_extract):
s = os.path.join(temp_extract, item)
d = os.path.join(version_dir, item)
@@ -675,11 +625,9 @@ class DocsManager:
else:
self._copy_file_no_metadata(s, d)
# Create a metadata file with the version name
with open(os.path.join(version_dir, ".version"), "w") as f:
f.write(version)
# Cleanup temp
if os.path.exists(temp_extract):
self._remove_tree_force_writable(temp_extract)
-17
View File
@@ -163,11 +163,6 @@ class IdentityContext:
and self.app.gitea_base_url_override
):
self.config.gitea_base_url.set(self.app.gitea_base_url_override)
if (
hasattr(self.app, "docs_download_urls_override")
and self.app.docs_download_urls_override
):
self.config.docs_download_urls.set(self.app.docs_download_urls_override)
self.message_handler = MessageHandler(self.database)
self.announce_manager = AnnounceManager(self.database, self.config)
@@ -352,18 +347,6 @@ class IdentityContext:
self.running = True
self.start_background_threads()
# 8. Handle initial documentation download
if (
not getattr(self.app, "emergency", False)
and not self.config.initial_docs_download_attempted.get()
):
if not self.docs_manager.has_docs():
print(
f"Triggering initial documentation download for {self.identity_hash}...",
)
self.docs_manager.update_docs()
self.config.initial_docs_download_attempted.set(True)
# Baseline integrity manifest after successful setup
if not getattr(self.app, "emergency", False):
self.integrity_manager.save_manifest()
+6 -5
View File
@@ -1185,13 +1185,14 @@
"title": "Dokumentation",
"subtitle": "Handbücher zu Reticulum und MeshChat.",
"status_title": "Dokumentationsstatus",
"status_downloading": "Wird heruntergeladen...",
"status_extracting": "Wird entpackt...",
"status_available": "Offline-Handbuch verfügbar",
"status_not_downloaded": "Nicht heruntergeladen",
"btn_download": "Handbücher herunterladen",
"btn_update": "Handbuch aktualisieren",
"status_not_available": "Handbuch nicht verfügbar",
"btn_upload": "Handbuch hochladen",
"btn_share": "Reticulum-Handbuch als wieder hochladbares ZIP teilen",
"empty_state_hint": "Lade ein ZIP des Reticulum-Handbuchs hoch, um es offline anzuzeigen. Eigene oder aktualisierte Dokumentation wird unterstützt.",
"error": "Fehler",
"failed_update_docs": "Dokumentationsquellen aktualisieren fehlgeschlagen",
"failed_upload_docs": "Hochladen der Dokumentation fehlgeschlagen",
"docs_link_copied": "Dokumentationslink in Zwischenablage kopiert",
"failed_copy_link": "Link kopieren fehlgeschlagen"
},
+6 -5
View File
@@ -1245,13 +1245,14 @@
"title": "Documentation",
"subtitle": "Reticulum and MeshChat manuals.",
"status_title": "Documentation Status",
"status_downloading": "Downloading...",
"status_extracting": "Extracting Documentation...",
"status_available": "Offline Manual Available",
"status_not_downloaded": "Not Downloaded",
"btn_download": "Download Manuals",
"btn_update": "Update Manual",
"status_not_available": "Manual Not Available",
"btn_upload": "Upload Manual",
"btn_share": "Share Reticulum manual as a re-uploadable ZIP",
"empty_state_hint": "Upload a Reticulum manual ZIP to view it offline. Custom or updated documentation is supported.",
"error": "Error",
"failed_update_docs": "Failed to update documentation sources",
"failed_upload_docs": "Failed to upload documentation",
"docs_link_copied": "Documentation link copied to clipboard",
"failed_copy_link": "Failed to copy link"
},
+6 -5
View File
@@ -1245,13 +1245,14 @@
"title": "Documentación",
"subtitle": "Manuales Reticulum y MeshChat.",
"status_title": "Situación de la documentación",
"status_downloading": "Descargando...",
"status_extracting": "Extrayendo...",
"status_available": "Manual disponible",
"status_not_downloaded": "No Descargado",
"btn_download": "Descargar Manuales",
"btn_update": "Manual de actualización",
"status_not_available": "Manual no disponible",
"btn_upload": "Subir Manual",
"btn_share": "Compartir el manual de Reticulum como ZIP reimportable",
"empty_state_hint": "Sube un ZIP del manual de Reticulum para consultarlo sin conexión. Se admite documentación personalizada o actualizada.",
"error": "Error",
"failed_update_docs": "Failed to update documentation sources",
"failed_upload_docs": "Error al subir la documentación",
"docs_link_copied": "Enlace de documentación copiado al portapapeles",
"failed_copy_link": "Failed to copy link"
},
+6 -5
View File
@@ -1245,13 +1245,14 @@
"title": "Documentation",
"subtitle": "Manuels Reticulum et MeshChat.",
"status_title": "État de la documentation",
"status_downloading": "Téléchargement...",
"status_extracting": "Extraction...",
"status_available": "Manuel hors ligne disponible",
"status_not_downloaded": "Non téléchargé",
"btn_download": "Télécharger les manuels",
"btn_update": "Manuel de mise à jour",
"status_not_available": "Manuel non disponible",
"btn_upload": "Téléverser le manuel",
"btn_share": "Partager le manuel Reticulum sous forme de ZIP réimportable",
"empty_state_hint": "Téléversez une archive ZIP du manuel Reticulum pour la consulter hors ligne. La documentation personnalisée ou mise à jour est prise en charge.",
"error": "Erreur",
"failed_update_docs": "Impossible de mettre à jour les sources de documentation",
"failed_upload_docs": "Échec du téléversement de la documentation",
"docs_link_copied": "Lien de documentation copié dans le presse-papiers",
"failed_copy_link": "Impossible de copier le lien"
},
+6 -5
View File
@@ -1297,13 +1297,14 @@
"title": "Documentazione",
"subtitle": "Manuali Reticulum e MeshChat.",
"status_title": "Stato Documentazione",
"status_downloading": "Scaricamento...",
"status_extracting": "Estrazione in corso...",
"status_available": "Manuale Offline Disponibile",
"status_not_downloaded": "Non Scaricato",
"btn_download": "Scarica Manuali",
"btn_update": "Aggiorna Manuale",
"status_not_available": "Manuale Non Disponibile",
"btn_upload": "Carica Manuale",
"btn_share": "Condividi il manuale Reticulum come ZIP ricaricabile",
"empty_state_hint": "Carica un archivio ZIP del manuale Reticulum per consultarlo offline. È supportata anche documentazione personalizzata o aggiornata.",
"error": "Errore",
"failed_update_docs": "Impossibile aggiornare le sorgenti della documentazione",
"failed_upload_docs": "Caricamento della documentazione non riuscito",
"docs_link_copied": "Link alla documentazione copiato negli appunti",
"failed_copy_link": "Impossibile copiare il link"
},
+6 -5
View File
@@ -1245,13 +1245,14 @@
"title": "Documentatie",
"subtitle": "Reticulum en MeshChat handleidingen.",
"status_title": "Documentatiestatus",
"status_downloading": "Downloaden",
"status_extracting": "Uitpakken...",
"status_available": "Offline handleiding beschikbaar",
"status_not_downloaded": "Niet gedownload",
"btn_download": "Downloaden Handleidingen",
"btn_update": "Handleiding bijwerken",
"status_not_available": "Handleiding niet beschikbaar",
"btn_upload": "Handleiding uploaden",
"btn_share": "Reticulum-handleiding delen als opnieuw uploadbare ZIP",
"empty_state_hint": "Upload een ZIP van de Reticulum-handleiding om deze offline te bekijken. Aangepaste of bijgewerkte documentatie wordt ondersteund.",
"error": "Fout",
"failed_update_docs": "Kon documentatiebronnen niet bijwerken",
"failed_upload_docs": "Documentatie uploaden mislukt",
"docs_link_copied": "Documentatiekoppeling gekopieerd naar klembord",
"failed_copy_link": "Kopiëren van verwijzing is mislukt"
},
+6 -5
View File
@@ -1185,13 +1185,14 @@
"title": "Документация",
"subtitle": "Руководства Reticulum и MeshChat.",
"status_title": "Статус документации",
"status_downloading": "Загрузка...",
"status_extracting": "Извлечение...",
"status_available": "Доступно офлайн",
"status_not_downloaded": "Не загружено",
"btn_download": "Скачать руководства",
"btn_update": "Обновить руководство",
"status_not_available": "Руководство недоступно",
"btn_upload": "Загрузить руководство",
"btn_share": "Поделиться руководством Reticulum как загружаемым ZIP",
"empty_state_hint": "Загрузите ZIP-архив с руководством Reticulum для просмотра офлайн. Поддерживается пользовательская и обновлённая документация.",
"error": "Ошибка",
"failed_update_docs": "Failed to update documentation sources",
"failed_upload_docs": "Не удалось загрузить документацию",
"docs_link_copied": "Documentation link copied to clipboard",
"failed_copy_link": "Failed to copy link"
},
+6 -5
View File
@@ -1245,13 +1245,14 @@
"title": "文档",
"subtitle": "Reticulum和MeshChat手册。",
"status_title": "文件状况",
"status_downloading": "正在下载...",
"status_extracting": "正在解压...",
"status_available": "离线手册可用",
"status_not_downloaded": "未下载",
"btn_download": "下载手册",
"btn_update": "更新手册",
"status_not_available": "手册不可用",
"btn_upload": "上传手册",
"btn_share": "以可重新上传的 ZIP 共享 Reticulum 手册",
"empty_state_hint": "上传 Reticulum 手册的 ZIP 以离线查看。支持自定义或更新后的文档。",
"error": "错误",
"failed_update_docs": "更新文档来源失败",
"failed_upload_docs": "上传文档失败",
"docs_link_copied": "复制到剪贴板的文档链接",
"failed_copy_link": "复制链接失败"
},
+4 -1
View File
@@ -13,8 +13,9 @@
"watch": "pnpm run build-frontend -- --watch",
"build-frontend": "vite build",
"build-backend": "node scripts/build-backend.js",
"build-docs": "python3 scripts/build/fetch_reticulum_manual.py",
"version:sync": "node scripts/sync_version.js",
"build": "pnpm run version:sync && pnpm run build-frontend && pnpm run build-backend",
"build": "pnpm run version:sync && pnpm run build-frontend && pnpm run build-docs && pnpm run build-backend",
"lint": "eslint .",
"lint:fix": "eslint . --fix",
"format": "prettier --write .",
@@ -40,6 +41,8 @@
"dist:win-arm64": "pnpm run electron-postinstall && cross-env PLATFORM=win32 ARCH=arm64 pnpm run build && electron-builder --win portable nsis --arm64 --publish=never",
"dist:zip": "pnpm run electron-postinstall && pnpm run build && electron-forge make --targets @electron-forge/maker-zip",
"dist-prebuilt": "pnpm run electron-postinstall && pnpm run build-backend && electron-builder --publish=never",
"dist:windows-prebuilt": "pnpm run electron-postinstall && cross-env PLATFORM=win32 pnpm run version:sync && cross-env PLATFORM=win32 pnpm run build-backend && electron-builder --win portable nsis --publish=never",
"dist:linux-prebuilt": "pnpm run electron-postinstall && cross-env PLATFORM=linux pnpm run version:sync && cross-env PLATFORM=linux pnpm run build-backend && electron-builder --linux AppImage deb --publish=never",
"dist:mac-arm64": "pnpm run electron-postinstall && pnpm run build && electron-builder --mac --arm64 --publish=never",
"dist:mac-universal": "bash scripts/build-macos-universal.sh",
"start": "pnpm run build && electron-forge start",
+271
View File
@@ -0,0 +1,271 @@
#!/usr/bin/env python3
"""Fetch the Reticulum manual at build time and stage it for bundling.
The downloaded archive is extracted into ``meshchatx/public/reticulum-docs-bundled/current``
so that the application ships with an offline copy of the manual. At runtime the
backend will serve those files for any ``/reticulum-docs/`` request that does not
have a user-uploaded version overriding it.
Usage::
python scripts/build/fetch_reticulum_manual.py [--source URL] [--dest DIR]
[--force] [--include-pdf]
By default the upstream PDF/EPUB copies of the manual are excluded from the
bundle because the in-app viewer only renders the HTML version. Pass
``--include-pdf`` (or set ``MESHCHATX_DOCS_INCLUDE_PDF=1``) to keep them.
Environment variables::
MESHCHATX_RETICULUM_DOCS_URL Override the default source URL (single value).
MESHCHATX_RETICULUM_DOCS_DEST Override the destination directory.
MESHCHATX_SKIP_DOCS_FETCH If set to ``1``/``true``, exit without fetching.
MESHCHATX_DOCS_INCLUDE_PDF If set to ``1``/``true``, include PDF/EPUB.
"""
from __future__ import annotations
import argparse
import io
import logging
import os
import shutil
import sys
import urllib.error
import urllib.request
import zipfile
from pathlib import Path
DEFAULT_SOURCES = (
"https://github.com/markqvist/reticulum_website/archive/refs/heads/main.zip",
"https://git.quad4.io/Reticulum/reticulum_website/archive/main.zip",
)
DEFAULT_DEST = (
Path(__file__).resolve().parent.parent.parent
/ "meshchatx"
/ "public"
/ "reticulum-docs-bundled"
/ "current"
)
EXTRA_BINARY_SUFFIXES = (".pdf", ".epub")
"""File suffixes for large alternate-format manuals that are excluded from the
bundled copy by default. The HTML viewer does not use them, so dropping these
saves roughly 9 MB on disk per build artifact."""
def _is_truthy(value: str | None) -> bool:
return value is not None and value.strip().lower() in {"1", "true", "yes", "on"}
def _download(url: str, timeout: float) -> bytes:
logging.info("Downloading Reticulum manual from %s", url)
req = urllib.request.Request( # noqa: S310 - URL is constrained to known sources
url,
headers={"User-Agent": "meshchatx-build-script"},
)
with urllib.request.urlopen(req, timeout=timeout) as response: # noqa: S310
return response.read()
def _resolve_docs_root(zip_bytes: bytes) -> tuple[zipfile.ZipFile, str]:
"""Return the open zip and the path prefix that contains the docs/ tree."""
archive = zipfile.ZipFile(io.BytesIO(zip_bytes))
names = archive.namelist()
if not names:
archive.close()
raise ValueError("downloaded archive is empty")
root = names[0].split("/", 1)[0]
docs_prefix = f"{root}/docs/"
if not any(name.startswith(docs_prefix) for name in names):
archive.close()
raise ValueError(
f"archive does not contain expected docs/ folder under {root}/",
)
return archive, docs_prefix
def _extract(
archive: zipfile.ZipFile,
docs_prefix: str,
dest: Path,
include_pdf: bool = False,
) -> tuple[int, int]:
"""Extract docs/ tree from ``archive`` into ``dest``.
Returns ``(extracted_count, skipped_binary_count)``. When ``include_pdf`` is
false, large alternate-format manuals listed in :data:`EXTRA_BINARY_SUFFIXES`
are skipped to keep shipped artifacts small.
"""
extracted = 0
skipped_binary = 0
for member in archive.infolist():
name = member.filename
if not name.startswith(docs_prefix):
continue
rel = name[len(docs_prefix) :]
if not rel or rel.endswith("/"):
continue
if ".." in rel.split("/"):
continue
if not include_pdf and rel.lower().endswith(EXTRA_BINARY_SUFFIXES):
skipped_binary += 1
continue
target = dest / rel
try:
target.relative_to(dest)
except ValueError:
continue
target.parent.mkdir(parents=True, exist_ok=True)
with archive.open(member) as src, open(target, "wb") as fh:
shutil.copyfileobj(src, fh)
extracted += 1
return extracted, skipped_binary
def fetch_manual(
sources: list[str],
dest: Path,
timeout: float = 120.0,
force: bool = False,
include_pdf: bool = False,
) -> int:
if dest.exists() and any(dest.iterdir()) and not force:
logging.info(
"Reticulum manual already present at %s (%d entries); skipping fetch.",
dest,
sum(1 for _ in dest.rglob("*")),
)
return 0
last_error: Exception | None = None
archive: zipfile.ZipFile | None = None
docs_prefix: str | None = None
for url in sources:
try:
data = _download(url, timeout)
archive, docs_prefix = _resolve_docs_root(data)
break
except (urllib.error.URLError, OSError, ValueError, zipfile.BadZipFile) as exc:
logging.warning("Failed to fetch %s: %s", url, exc)
last_error = exc
archive = None
docs_prefix = None
if archive is None or docs_prefix is None:
raise SystemExit(
f"Could not download Reticulum manual from any source: {last_error}",
)
try:
if dest.exists():
shutil.rmtree(dest)
dest.mkdir(parents=True, exist_ok=True)
extracted, skipped_binary = _extract(
archive,
docs_prefix,
dest,
include_pdf=include_pdf,
)
finally:
archive.close()
if extracted == 0:
raise SystemExit("Archive contained no docs/ files to extract")
if skipped_binary:
logging.info(
"Skipped %d alternate-format manual file(s) (%s); pass --include-pdf to keep them.",
skipped_binary,
", ".join(EXTRA_BINARY_SUFFIXES),
)
logging.info("Extracted %d files to %s", extracted, dest)
return extracted
def _parse_args(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--source",
action="append",
default=None,
help=(
"URL of a Reticulum website ZIP. May be passed multiple times to provide "
"fallbacks. Defaults to the canonical upstream sources."
),
)
parser.add_argument(
"--dest",
type=Path,
default=None,
help="Output directory for the extracted manual.",
)
parser.add_argument(
"--timeout",
type=float,
default=120.0,
help="HTTP timeout in seconds.",
)
parser.add_argument(
"--force",
action="store_true",
help="Re-fetch even if the destination already exists.",
)
parser.add_argument(
"--include-pdf",
action="store_true",
default=_is_truthy(os.environ.get("MESHCHATX_DOCS_INCLUDE_PDF")),
help=(
"Include the upstream PDF/EPUB manuals in the bundle. They are "
"skipped by default to keep build artifacts smaller because the "
"in-app viewer only uses the HTML version."
),
)
parser.add_argument(
"--quiet",
action="store_true",
help="Reduce log output.",
)
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = _parse_args(argv if argv is not None else sys.argv[1:])
logging.basicConfig(
level=logging.WARNING if args.quiet else logging.INFO,
format="%(message)s",
)
if _is_truthy(os.environ.get("MESHCHATX_SKIP_DOCS_FETCH")):
logging.info(
"MESHCHATX_SKIP_DOCS_FETCH is set; skipping Reticulum manual fetch."
)
return 0
sources: list[str] = []
if args.source:
sources.extend(args.source)
env_url = os.environ.get("MESHCHATX_RETICULUM_DOCS_URL")
if env_url:
sources.append(env_url)
if not sources:
sources = list(DEFAULT_SOURCES)
env_dest = os.environ.get("MESHCHATX_RETICULUM_DOCS_DEST")
dest = args.dest or (Path(env_dest) if env_dest else DEFAULT_DEST)
fetch_manual(
sources=sources,
dest=dest.resolve(),
timeout=args.timeout,
force=args.force,
include_pdf=args.include_pdf,
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+4 -4
View File
@@ -208,6 +208,10 @@
"method": "GET",
"path": "/api/v1/docs/export"
},
{
"method": "GET",
"path": "/api/v1/docs/export/reticulum"
},
{
"method": "GET",
"path": "/api/v1/docs/search"
@@ -220,10 +224,6 @@
"method": "POST",
"path": "/api/v1/docs/switch"
},
{
"method": "POST",
"path": "/api/v1/docs/update"
},
{
"method": "POST",
"path": "/api/v1/docs/upload"
+175 -68
View File
@@ -3,7 +3,7 @@
import os
import shutil
import zipfile
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import MagicMock
import pytest
from hypothesis import HealthCheck, given, settings
@@ -25,7 +25,6 @@ def temp_dirs(tmp_path):
def docs_manager(temp_dirs):
public_dir, _ = temp_dirs
config = MagicMock()
config.docs_downloaded.get.return_value = False
return DocsManager(config, public_dir)
@@ -33,7 +32,7 @@ def test_docs_manager_initialization(docs_manager, temp_dirs):
_, docs_dir = temp_dirs
assert docs_manager.docs_dir == os.path.join(docs_dir, "current")
assert os.path.exists(docs_dir)
assert docs_manager.download_status == "idle"
assert docs_manager.upload_status == "idle"
def test_docs_manager_storage_dir_fallback(tmp_path):
@@ -43,26 +42,84 @@ def test_docs_manager_storage_dir_fallback(tmp_path):
storage_dir.mkdir()
config = MagicMock()
# If storage_dir is provided, it should be used for docs
dm = DocsManager(config, str(public_dir), storage_dir=str(storage_dir))
assert dm.docs_dir == os.path.join(str(storage_dir), "reticulum-docs", "current")
assert dm.meshchatx_docs_dir == os.path.join(str(storage_dir), "meshchatx-docs")
# The 'current' directory may not exist if there are no versions, but the base dir should exist
assert dm.bundled_docs_dir == os.path.join(
str(public_dir),
"reticulum-docs-bundled",
"current",
)
assert os.path.exists(dm.docs_base_dir)
assert os.path.exists(dm.meshchatx_docs_dir)
def test_has_bundled_docs_falls_back_to_public(tmp_path):
public_dir = tmp_path / "public"
public_dir.mkdir()
storage_dir = tmp_path / "storage"
storage_dir.mkdir()
bundled = public_dir / "reticulum-docs-bundled" / "current"
bundled.mkdir(parents=True)
(bundled / "index.html").write_text("<html></html>")
config = MagicMock()
dm = DocsManager(config, str(public_dir), storage_dir=str(storage_dir))
assert dm.has_bundled_docs() is True
assert dm.has_user_docs() is False
assert dm.has_docs() is True
resolved = dm.find_docs_file("index.html")
assert resolved == os.path.realpath(str(bundled / "index.html"))
def test_user_docs_take_precedence_over_bundled(tmp_path):
public_dir = tmp_path / "public"
public_dir.mkdir()
storage_dir = tmp_path / "storage"
storage_dir.mkdir()
bundled = public_dir / "reticulum-docs-bundled" / "current"
bundled.mkdir(parents=True)
(bundled / "index.html").write_text("<html>bundled</html>")
config = MagicMock()
dm = DocsManager(config, str(public_dir), storage_dir=str(storage_dir))
user_index = os.path.join(dm.docs_dir, "index.html")
os.makedirs(dm.docs_dir, exist_ok=True)
with open(user_index, "w") as f:
f.write("<html>user</html>")
resolved = dm.find_docs_file("index.html")
assert resolved == os.path.realpath(user_index)
def test_find_docs_file_rejects_traversal(tmp_path):
public_dir = tmp_path / "public"
public_dir.mkdir()
bundled = public_dir / "reticulum-docs-bundled" / "current"
bundled.mkdir(parents=True)
(bundled / "index.html").write_text("<html></html>")
(tmp_path / "secret.txt").write_text("nope")
config = MagicMock()
dm = DocsManager(config, str(public_dir))
assert dm.find_docs_file("../../secret.txt") is None
assert dm.find_docs_file("..") is None
assert dm.find_docs_file("missing.html") is None
def test_docs_manager_readonly_public_dir_handling(tmp_path):
public_dir = tmp_path / "readonly_public"
public_dir.mkdir()
# Make it read-only
os.chmod(public_dir, 0o555)
config = MagicMock()
# Mock os.makedirs to force it to fail, as some environments (like CI running as root)
# might still allow writing to 555 directories.
from unittest.mock import patch
with patch("os.makedirs", side_effect=OSError("Read-only file system")):
dm = DocsManager(config, str(public_dir))
assert dm.last_error is not None
@@ -71,7 +128,6 @@ def test_docs_manager_readonly_public_dir_handling(tmp_path):
or "Permission denied" in dm.last_error
)
# Restore permissions for cleanup
os.chmod(public_dir, 0o755)
@@ -88,61 +144,122 @@ def test_has_docs(docs_manager, temp_dirs):
assert docs_manager.has_docs() is True
def test_get_status(docs_manager):
def test_get_status_reports_bundled_and_user_flags(docs_manager):
status = docs_manager.get_status()
assert status["status"] == "idle"
assert status["progress"] == 0
assert status["has_docs"] is False
assert status["has_bundled_docs"] is False
assert status["has_user_docs"] is False
@patch("meshchatx.src.backend.docs_manager.aiohttp.ClientSession")
def test_download_task_success(mock_session_cls, docs_manager, temp_dirs):
public_dir, docs_dir = temp_dirs
mock_response = MagicMock()
mock_response.headers = {"Content-Length": "100"}
mock_response.raise_for_status = MagicMock()
async def iter_chunked(_n):
yield b"data" * 25
mock_response.content.iter_chunked = MagicMock(
side_effect=lambda n: iter_chunked(n),
def test_upload_zip_extracts_and_switches_version(docs_manager):
payload = _make_docs_zip(
files={
"reticulum_website-main/docs/index.html": "<html>uploaded</html>",
"reticulum_website-main/docs/manual.html": "<html>manual</html>",
},
)
mock_get = MagicMock()
mock_get.__aenter__ = AsyncMock(return_value=mock_response)
mock_get.__aexit__ = AsyncMock(return_value=None)
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=mock_get)
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock(return_value=None)
mock_session_cls.return_value = mock_session
with patch.object(docs_manager, "_extract_docs") as mock_extract:
docs_manager._download_task()
assert docs_manager.download_status == "completed"
assert mock_extract.called
zip_path = os.path.join(docs_dir, "website.zip")
call_args = mock_extract.call_args
assert call_args[0][0] == zip_path
assert call_args[0][1].startswith("git-")
assert docs_manager.upload_zip(payload, "v-test") is True
assert docs_manager.upload_status == "completed"
assert "v-test" in docs_manager.get_available_versions()
resolved = docs_manager.find_docs_file("index.html")
assert resolved is not None
with open(resolved) as fh:
assert "uploaded" in fh.read()
@patch("meshchatx.src.backend.docs_manager.aiohttp.ClientSession")
def test_download_task_failure(mock_session_cls, docs_manager):
mock_session = MagicMock()
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock(return_value=None)
mock_session.get = MagicMock(side_effect=Exception("Download failed"))
mock_session_cls.return_value = mock_session
def test_clear_reticulum_docs_does_not_touch_bundled(tmp_path):
public_dir = tmp_path / "public"
public_dir.mkdir()
storage_dir = tmp_path / "storage"
storage_dir.mkdir()
bundled = public_dir / "reticulum-docs-bundled" / "current"
bundled.mkdir(parents=True)
(bundled / "index.html").write_text("<html>bundled</html>")
docs_manager._download_task()
config = MagicMock()
dm = DocsManager(config, str(public_dir), storage_dir=str(storage_dir))
assert docs_manager.download_status == "error"
assert docs_manager.last_error == "Download failed"
payload = _make_docs_zip(
files={"reticulum_website-main/docs/index.html": "<html>uploaded</html>"},
)
dm.upload_zip(payload, "v1")
assert dm.has_user_docs() is True
assert dm.clear_reticulum_docs() is True
assert dm.has_bundled_docs() is True
def test_export_reticulum_docs_returns_none_when_empty(tmp_path):
public_dir = tmp_path / "public"
public_dir.mkdir()
config = MagicMock()
dm = DocsManager(config, str(public_dir), storage_dir=str(tmp_path / "storage"))
assert dm.export_reticulum_docs() is None
def test_export_reticulum_docs_uses_upload_compatible_layout(tmp_path):
public_dir = tmp_path / "public"
public_dir.mkdir()
bundled = public_dir / "reticulum-docs-bundled" / "current"
bundled.mkdir(parents=True)
(bundled / "index.html").write_text("<html>bundled</html>")
(bundled / "manual").mkdir()
(bundled / "manual" / "index.html").write_text("<html>chapter</html>")
config = MagicMock()
dm = DocsManager(config, str(public_dir), storage_dir=str(tmp_path / "storage"))
payload = dm.export_reticulum_docs(root_folder="reticulum_manual")
assert payload is not None
import io as _io
with zipfile.ZipFile(_io.BytesIO(payload)) as zf:
names = sorted(zf.namelist())
assert "reticulum_manual/docs/index.html" in names
assert "reticulum_manual/docs/manual/index.html" in names
for n in names:
assert n.startswith("reticulum_manual/docs/")
def test_export_reticulum_docs_round_trips_through_upload(tmp_path):
public_dir = tmp_path / "public"
public_dir.mkdir()
bundled = public_dir / "reticulum-docs-bundled" / "current"
bundled.mkdir(parents=True)
(bundled / "index.html").write_text("<html>shared</html>")
config = MagicMock()
src = DocsManager(config, str(public_dir), storage_dir=str(tmp_path / "src"))
payload = src.export_reticulum_docs()
assert payload is not None
other_public = tmp_path / "other_public"
other_public.mkdir()
other = DocsManager(
config,
str(other_public),
storage_dir=str(tmp_path / "other_storage"),
)
assert other.upload_zip(payload, "shared-from-peer") is True
assert "shared-from-peer" in other.get_available_versions()
resolved = other.find_docs_file("index.html")
assert resolved is not None
with open(resolved) as fh:
assert "shared" in fh.read()
def _make_docs_zip(files: dict[str, str]) -> bytes:
import io
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
for path, content in files.items():
zf.writestr(path, content)
return buf.getvalue()
def create_mock_zip(zip_path, file_list):
@@ -167,11 +284,9 @@ def create_mock_zip(zip_path, file_list):
),
)
def test_extract_docs_fuzzing(docs_manager, temp_dirs, root_folder_name, docs_file):
public_dir, docs_dir = temp_dirs
_, docs_dir = temp_dirs
zip_path = os.path.join(docs_dir, "test.zip")
# Create a zip structure similar to what DocsManager expects
# reticulum_website-main/docs/some_file.html
zip_files = [
f"{root_folder_name}/",
f"{root_folder_name}/docs/",
@@ -181,18 +296,12 @@ def test_extract_docs_fuzzing(docs_manager, temp_dirs, root_folder_name, docs_fi
create_mock_zip(zip_path, zip_files)
try:
docs_manager._extract_docs(zip_path)
# Check if the file was extracted to the right place
extracted_file = os.path.join(docs_dir, docs_file)
assert os.path.exists(extracted_file)
docs_manager._extract_docs(zip_path, "fuzz")
except Exception:
# If it's a known zip error or something, we can decide if it's a failure
# But for these valid-ish paths, it should work.
pass
finally:
if os.path.exists(zip_path):
os.remove(zip_path)
# Clean up extracted files for next run
for item in os.listdir(docs_dir):
item_path = os.path.join(docs_dir, item)
if os.path.isdir(item_path):
@@ -202,23 +311,21 @@ def test_extract_docs_fuzzing(docs_manager, temp_dirs, root_folder_name, docs_fi
def test_extract_docs_malformed_zip(docs_manager, temp_dirs):
public_dir, docs_dir = temp_dirs
_, docs_dir = temp_dirs
zip_path = os.path.join(docs_dir, "malformed.zip")
# 1. Zip with no folders at all
create_mock_zip(zip_path, ["file_at_root.txt"])
try:
docs_manager._extract_docs(zip_path)
docs_manager._extract_docs(zip_path, "malformed-1")
except (IndexError, Exception):
pass # Expected or at least handled by not crashing the whole app
pass
finally:
if os.path.exists(zip_path):
os.remove(zip_path)
# 2. Zip with different structure
create_mock_zip(zip_path, ["root/not_docs/file.txt"])
try:
docs_manager._extract_docs(zip_path)
docs_manager._extract_docs(zip_path, "malformed-2")
except Exception:
pass
finally:
-1
View File
@@ -127,7 +127,6 @@ def test_reticulum_meshchat_init(mock_rns, temp_dir):
"http://localhost:5000"
)
mock_config_instance.translator_enabled.get.return_value = False
mock_config_instance.initial_docs_download_attempted.get.return_value = True
app = ReticulumMeshChat(
identity=mock_rns["id_instance"],
-2
View File
@@ -116,7 +116,6 @@ def test_run_https_logic(mock_rns, temp_dir):
mock_config.lxmf_local_propagation_node_enabled.get.return_value = False
mock_config.libretranslate_url.get.return_value = "http://localhost:5000"
mock_config.translator_enabled.get.return_value = False
mock_config.initial_docs_download_attempted.get.return_value = True
app = ReticulumMeshChat(
identity=mock_rns["id_instance"],
@@ -277,7 +276,6 @@ def test_database_health_issues_set_on_setup(mock_rns, temp_dir):
mock_config.lxmf_local_propagation_node_enabled.get.return_value = False
mock_config.libretranslate_url.get.return_value = "http://localhost:5000"
mock_config.translator_enabled.get.return_value = False
mock_config.initial_docs_download_attempted.get.return_value = True
app = ReticulumMeshChat(
identity=mock_rns["id_instance"],
+62 -14
View File
@@ -17,6 +17,10 @@ describe("DocsPage.vue", () => {
progress: 0,
last_error: null,
has_docs: false,
has_bundled_docs: false,
has_user_docs: false,
versions: [],
current_version: null,
},
});
}
@@ -35,8 +39,6 @@ describe("DocsPage.vue", () => {
if (wrapper) {
wrapper.unmount();
}
// Do not delete window.api, as it might be used by async operations
// and it is globally defined in setup.js anyway.
});
let wrapper;
@@ -58,15 +60,14 @@ describe("DocsPage.vue", () => {
return wrapper;
};
it("renders download button when no docs are present", async () => {
it("renders upload prompt when no docs are present", async () => {
const wrapper = mountDocsPage();
await nextTick();
await nextTick();
expect(wrapper.text()).toContain("Reticulum Manual");
const downloadBtn = wrapper.find("button.bg-blue-600");
expect(downloadBtn.exists()).toBe(true);
expect(downloadBtn.text()).toContain("docs.btn_download");
expect(wrapper.text()).toContain("docs.empty_state_hint");
expect(wrapper.text()).toContain("docs.btn_upload");
});
it("renders iframe when docs are present", async () => {
@@ -78,6 +79,10 @@ describe("DocsPage.vue", () => {
progress: 100,
last_error: null,
has_docs: true,
has_bundled_docs: true,
has_user_docs: false,
versions: [],
current_version: "bundled",
},
});
}
@@ -93,15 +98,19 @@ describe("DocsPage.vue", () => {
expect(wrapper.find("iframe").attributes("src")).toBe("/reticulum-docs/index.html");
});
it("shows progress bar during download", async () => {
it("shows progress bar while extracting an upload", async () => {
axiosMock.get.mockImplementation((url) => {
if (url.includes("/api/v1/docs/status")) {
return Promise.resolve({
data: {
status: "downloading",
status: "extracting",
progress: 45,
last_error: null,
has_docs: false,
has_bundled_docs: false,
has_user_docs: false,
versions: [],
current_version: null,
},
});
}
@@ -116,6 +125,7 @@ describe("DocsPage.vue", () => {
const progressBar = wrapper.find(".bg-blue-500");
expect(progressBar.exists()).toBe(true);
expect(progressBar.attributes("style")).toContain("width: 45%");
expect(wrapper.text()).toContain("docs.status_extracting");
});
it("shows error message when status has an error", async () => {
@@ -125,8 +135,12 @@ describe("DocsPage.vue", () => {
data: {
status: "error",
progress: 0,
last_error: "Connection timeout",
last_error: "Bad zip",
has_docs: false,
has_bundled_docs: false,
has_user_docs: false,
versions: [],
current_version: null,
},
});
}
@@ -139,18 +153,48 @@ describe("DocsPage.vue", () => {
await nextTick();
expect(wrapper.text()).toContain("docs.error");
expect(wrapper.text()).toContain("Connection timeout");
expect(wrapper.text()).toContain("Bad zip");
});
it("calls update API when download button is clicked", async () => {
it("does not auto-trigger any update API call on mount", async () => {
const wrapper = mountDocsPage();
await nextTick();
await nextTick();
const downloadBtn = wrapper.find("button.bg-blue-600");
await downloadBtn.trigger("click");
expect(axiosMock.post).not.toHaveBeenCalledWith(
expect.stringContaining("/api/v1/docs/update"),
expect.anything()
);
expect(wrapper.exists()).toBe(true);
});
expect(axiosMock.post).toHaveBeenCalledWith("/api/v1/docs/update");
it("dismissError clears the last_error from local status", async () => {
axiosMock.get.mockImplementation((url) => {
if (url.includes("/api/v1/docs/status")) {
return Promise.resolve({
data: {
status: "error",
progress: 0,
last_error: "Boom",
has_docs: false,
has_bundled_docs: false,
has_user_docs: false,
versions: [],
current_version: null,
},
});
}
if (url.includes("/api/v1/meshchatx-docs/list")) return Promise.resolve({ data: [] });
return Promise.resolve({ data: {} });
});
const wrapper = mountDocsPage();
await nextTick();
await nextTick();
expect(wrapper.vm.status.last_error).toBe("Boom");
wrapper.vm.dismissError();
await nextTick();
expect(wrapper.vm.status.last_error).toBeNull();
});
it("changes localDocsUrl based on locale", async () => {
@@ -176,6 +220,10 @@ describe("DocsPage.vue", () => {
progress: 0,
last_error: longError,
has_docs: false,
has_bundled_docs: false,
has_user_docs: false,
versions: [],
current_version: null,
},
});
}
@@ -439,14 +439,12 @@ describe("SettingsPage — config persistence (PATCH and related)", () => {
it("onGiteaConfigChange PATCHes after debounce", async () => {
const w = await mountSettingsPage(api);
w.vm.config.gitea_base_url = "https://gitea.example";
w.vm.config.docs_download_urls = "https://docs.example";
await w.vm.onGiteaConfigChange();
await vi.advanceTimersByTimeAsync(1000);
expect(api.patch).toHaveBeenCalledWith(
"/api/v1/config",
expect.objectContaining({
gitea_base_url: "https://gitea.example",
docs_download_urls: "https://docs.example",
})
);
});
@@ -118,7 +118,6 @@ export function buildFullServerConfig(overrides = {}) {
nomad_render_plaintext_enabled: true,
nomad_default_page_path: "/page/index.mu",
gitea_base_url: "https://git.quad4.io",
docs_download_urls: "",
...overrides,
};
}