Improve RNS management and settings interface in Ren Browser

- Introduced a new rns.py module to encapsulate Reticulum lifecycle management.
- Simplified RNS initialization and error handling in app.py.
- Enhanced settings.py to improve configuration management and user feedback.
- Updated UI components for better interaction and status display.
- Added tests for settings functionality and RNS integration.
This commit is contained in:
2025-11-30 15:21:18 -06:00
parent d1536aa05a
commit d8de2b1150
7 changed files with 732 additions and 450 deletions

View File

@@ -5,11 +5,14 @@ Ren Browser, a browser for the Reticulum Network built with Flet.
""" """
import argparse import argparse
import os
from pathlib import Path
import flet as ft import flet as ft
import RNS import RNS
from flet import AppView, Page from flet import AppView, Page
from ren_browser import rns
from ren_browser.storage.storage import initialize_storage from ren_browser.storage.storage import initialize_storage
from ren_browser.ui.ui import build_ui from ren_browser.ui.ui import build_ui
@@ -50,44 +53,36 @@ async def main(page: Page):
page.add(loader) page.add(loader)
page.update() page.update()
# Initialize storage system initialize_storage(page)
storage = initialize_storage(page)
# Get Reticulum config directory from storage manager config_override = RNS_CONFIG_DIR
config_dir = storage.get_reticulum_config_path()
# Update the global RNS_CONFIG_DIR so RNS uses the right path print("Initializing Reticulum Network...")
global RNS_CONFIG_DIR
RNS_CONFIG_DIR = str(config_dir)
# Ensure any saved config is written to filesystem before RNS init
try: try:
saved_config = storage.load_config()
if saved_config and saved_config.strip():
config_file_path = config_dir / "config"
config_file_path.parent.mkdir(parents=True, exist_ok=True)
config_file_path.write_text(saved_config, encoding="utf-8")
except Exception as e:
print(f"Warning: Failed to write config file: {e}")
print(f"Initializing RNS with config directory: {config_dir}")
print(f"Config directory exists: {config_dir.exists()}")
print(f"Config directory is writable: {config_dir.is_dir() if config_dir.exists() else 'N/A'}")
try:
# Set up logging capture first, before RNS init
import ren_browser.logs import ren_browser.logs
ren_browser.logs.setup_rns_logging() ren_browser.logs.setup_rns_logging()
except Exception:
pass
success = rns.initialize_reticulum(config_override)
if not success:
error_text = rns.get_last_error() or "Unknown error"
print(f"Error initializing Reticulum: {error_text}")
else:
global RNS_INSTANCE global RNS_INSTANCE
RNS_INSTANCE = RNS.Reticulum(str(config_dir)) RNS_INSTANCE = rns.get_reticulum_instance()
config_dir = rns.get_config_path()
if config_dir:
config_path = Path(config_dir)
print(f"RNS config directory: {config_path}")
print(f"Config directory exists: {config_path.exists()}")
print(
"Config directory is writable: "
f"{config_path.is_dir() and os.access(config_path, os.W_OK)}",
)
print("RNS initialized successfully") print("RNS initialized successfully")
except Exception as e:
print(f"Error initializing Reticulum: {e}")
print(f"Config directory: {config_dir}")
import traceback
traceback.print_exc()
page.controls.clear() page.controls.clear()
build_ui(page) build_ui(page)
page.update() page.update()
@@ -113,46 +108,24 @@ async def reload_reticulum(page: Page, on_complete=None):
except Exception as e: except Exception as e:
print(f"Warning during RNS shutdown: {e}") print(f"Warning during RNS shutdown: {e}")
rns.shutdown_reticulum()
RNS.Reticulum._Reticulum__instance = None RNS.Reticulum._Reticulum__instance = None
RNS.Transport.destinations = [] RNS.Transport.destinations = []
RNS_INSTANCE = None RNS_INSTANCE = None
print("RNS instance cleared") print("RNS instance cleared")
await asyncio.sleep(0.5) await asyncio.sleep(0.5)
# Initialize storage system success = rns.initialize_reticulum(RNS_CONFIG_DIR)
storage = initialize_storage(page) if success:
RNS_INSTANCE = rns.get_reticulum_instance()
# Get Reticulum config directory from storage manager
config_dir = storage.get_reticulum_config_path()
# Ensure any saved config is written to filesystem before RNS init
try:
saved_config = storage.load_config()
if saved_config and saved_config.strip():
config_file_path = config_dir / "config"
config_file_path.parent.mkdir(parents=True, exist_ok=True)
config_file_path.write_text(saved_config, encoding="utf-8")
except Exception as e:
print(f"Warning: Failed to write config file: {e}")
try:
# Re-initialize Reticulum
import ren_browser.logs
ren_browser.logs.setup_rns_logging()
RNS_INSTANCE = RNS.Reticulum(str(config_dir))
# Success
if on_complete: if on_complete:
on_complete(True, None) on_complete(True, None)
else:
except Exception as e: error_text = rns.get_last_error() or "Unknown error"
print(f"Error reinitializing Reticulum: {e}") print(f"Error reinitializing Reticulum: {error_text}")
if on_complete: if on_complete:
on_complete(False, str(e)) on_complete(False, error_text)
except Exception as e: except Exception as e:
print(f"Error during reload: {e}") print(f"Error during reload: {e}")
@@ -198,9 +171,7 @@ def run():
if args.config_dir: if args.config_dir:
RNS_CONFIG_DIR = args.config_dir RNS_CONFIG_DIR = args.config_dir
else: else:
import pathlib RNS_CONFIG_DIR = None
RNS_CONFIG_DIR = str(pathlib.Path.home() / ".reticulum")
if args.web: if args.web:
if args.port is not None: if args.port is not None:

284
ren_browser/rns.py Normal file
View File

@@ -0,0 +1,284 @@
"""Reticulum helper utilities for Ren Browser."""
from __future__ import annotations
import os
import tempfile
from pathlib import Path
from typing import Optional
import RNS
class RNSManager:
"""Manage Reticulum lifecycle and configuration."""
def __init__(self):
self.reticulum = None
self.config_path: Optional[str] = None
self.last_error: Optional[str] = None
def _is_android(self) -> bool:
vendor = getattr(RNS, "vendor", None)
platformutils = getattr(vendor, "platformutils", None)
if platformutils and hasattr(platformutils, "is_android"):
try:
return bool(platformutils.is_android())
except Exception:
return False
return "ANDROID_ROOT" in os.environ
def _android_storage_root(self) -> Path:
candidates = [
os.environ.get("ANDROID_APP_PATH"),
os.environ.get("ANDROID_PRIVATE"),
os.environ.get("ANDROID_ARGUMENT"),
]
for raw_path in candidates:
if not raw_path:
continue
path = Path(raw_path).expanduser()
if path.name == "app":
path = path.parent
if path.is_file():
path = path.parent
if path.is_dir():
return path
return Path(tempfile.gettempdir())
def _default_config_root(self) -> Path:
override = (
os.environ.get("REN_BROWSER_RNS_DIR")
or os.environ.get("REN_RETICULUM_CONFIG_DIR")
)
if override:
return Path(override).expanduser()
if self._is_android():
return self._android_storage_root() / "ren_browser" / "reticulum"
return Path.home() / ".reticulum"
def _resolve_config_dir(self, preferred: Optional[str | Path]) -> Path:
target = Path(preferred).expanduser() if preferred else self._default_config_root()
allow_fallback = preferred is None
try:
target.mkdir(parents=True, exist_ok=True)
except Exception:
if not allow_fallback:
raise
fallback = Path(tempfile.gettempdir()) / "ren_browser" / "reticulum"
fallback.mkdir(parents=True, exist_ok=True)
target = fallback
self._seed_config_if_missing(target)
return target
def _default_tcp_interfaces_snippet(self) -> str:
return """
[[Quad4 Node 1]]
type = TCPClientInterface
interface_enabled = true
target_host = rns.quad4.io
target_port = 4242
name = Quad4 Node 1
selected_interface_mode = 1
[[Quad4 Node 2]]
type = TCPClientInterface
interface_enabled = true
target_host = rns2.quad4.io
target_port = 4242
name = Quad4 Node 2
selected_interface_mode = 1
""".strip(
"\n",
)
def _seed_config_if_missing(self, target: Path) -> None:
config_file = target / "config"
if config_file.exists():
return
base_content = None
try:
default_lines = getattr(RNS.Reticulum, "__default_rns_config__", None)
if default_lines:
if isinstance(default_lines, list):
base_content = "\n".join(default_lines)
else:
base_content = str(default_lines)
except Exception:
base_content = None
if not base_content:
base_content = (
"[reticulum]\n"
"share_instance = Yes\n\n"
"[interfaces]\n\n"
" [[Default Interface]]\n"
" type = AutoInterface\n"
" enabled = Yes\n"
)
snippet = self._default_tcp_interfaces_snippet()
if snippet and snippet not in base_content:
base_content = base_content.rstrip() + "\n\n" + snippet + "\n"
try:
config_file.write_text(base_content, encoding="utf-8")
os.chmod(config_file, 0o600)
except Exception:
pass
def _ensure_default_tcp_interfaces(self) -> None:
if not self.config_path:
return
config_file = Path(self.config_path) / "config"
if not config_file.exists():
return
try:
content = config_file.read_text(encoding="utf-8")
except Exception:
return
snippet = self._default_tcp_interfaces_snippet()
if "target_host = rns.quad4.io" in content or "Quad4 Node 1" in content:
return
try:
with open(config_file, "a", encoding="utf-8") as cfg:
if not content.endswith("\n"):
cfg.write("\n")
cfg.write("\n" + snippet + "\n")
except Exception:
pass
def _get_or_create_config_dir(self) -> Path:
if self.config_path:
return Path(self.config_path)
resolved = self._resolve_config_dir(None)
self.config_path = str(resolved)
return resolved
def initialize(self, config_dir: Optional[str] = None) -> bool:
"""Initialize the Reticulum instance."""
self.last_error = None
try:
use_custom_dir = bool(config_dir or self._is_android())
if use_custom_dir:
resolved = self._resolve_config_dir(config_dir)
self.config_path = str(resolved)
self.reticulum = RNS.Reticulum(configdir=self.config_path)
else:
self.reticulum = RNS.Reticulum()
self.config_path = getattr(
RNS.Reticulum,
"configdir",
str(Path.home() / ".reticulum"),
)
self._ensure_default_tcp_interfaces()
return True
except Exception as exc:
self.last_error = str(exc)
return False
def shutdown(self) -> bool:
"""Shut down the active Reticulum instance."""
try:
if self.reticulum and hasattr(self.reticulum, "exit_handler"):
self.reticulum.exit_handler()
except Exception:
return False
finally:
self.reticulum = None
return True
def read_config_file(self) -> str:
"""Return the current configuration file contents."""
config_dir = self._get_or_create_config_dir()
config_file = config_dir / "config"
try:
return config_file.read_text(encoding="utf-8")
except FileNotFoundError:
self._seed_config_if_missing(config_dir)
try:
return config_file.read_text(encoding="utf-8")
except Exception:
return ""
except Exception:
return ""
def write_config_file(self, content: str) -> bool:
"""Persist configuration text to disk."""
config_dir = self._get_or_create_config_dir()
config_file = config_dir / "config"
try:
config_dir.mkdir(parents=True, exist_ok=True)
config_file.write_text(content, encoding="utf-8")
os.chmod(config_file, 0o600)
return True
except Exception as exc:
self.last_error = str(exc)
return False
def get_config_path(self) -> Optional[str]:
"""Return the directory holding the active Reticulum config."""
if self.config_path:
return self.config_path
try:
default_path = self._resolve_config_dir(None)
self.config_path = str(default_path)
return self.config_path
except Exception:
return None
def get_reticulum_instance(self):
"""Return the current Reticulum instance, if any."""
return self.reticulum
def get_last_error(self) -> Optional[str]:
"""Return the last recorded error string."""
return self.last_error
rns_manager = RNSManager()
def initialize_reticulum(config_dir: Optional[str] = None) -> bool:
"""Initialize Reticulum using the shared manager."""
return rns_manager.initialize(config_dir)
def shutdown_reticulum() -> bool:
"""Shut down the shared Reticulum instance."""
return rns_manager.shutdown()
def get_reticulum_instance():
"""Expose the active Reticulum instance."""
return rns_manager.get_reticulum_instance()
def get_config_path() -> Optional[str]:
"""Expose the active configuration directory."""
return rns_manager.get_config_path()
def read_config_file() -> str:
"""Read the Reticulum configuration file."""
return rns_manager.read_config_file()
def write_config_file(content: str) -> bool:
"""Write the Reticulum configuration file."""
return rns_manager.write_config_file(content)
def get_last_error() -> Optional[str]:
"""Return the last recorded Reticulum error."""
return rns_manager.get_last_error()

View File

@@ -1,30 +1,210 @@
"""Settings interface for Ren Browser. """Settings interface for Ren Browser."""
Provides configuration management, log viewing, and storage from __future__ import annotations
information display.
""" from datetime import datetime
from pathlib import Path
import flet as ft import flet as ft
import RNS
from ren_browser.logs import ERROR_LOGS, RET_LOGS from ren_browser import rns
from ren_browser.storage.storage import get_storage_manager from ren_browser.storage.storage import get_storage_manager
BUTTON_BG = "#0B3D91"
BUTTON_BG_HOVER = "#082C6C"
def _blue_button_style() -> ft.ButtonStyle:
return ft.ButtonStyle(
bgcolor=BUTTON_BG,
color=ft.Colors.WHITE,
overlay_color=BUTTON_BG_HOVER,
)
def _get_config_file_path() -> Path:
config_dir = rns.get_config_path()
if config_dir:
return Path(config_dir) / "config"
return Path.home() / ".reticulum" / "config"
def _read_config_text(config_path: Path) -> str:
try:
return config_path.read_text(encoding="utf-8")
except FileNotFoundError:
config_path.parent.mkdir(parents=True, exist_ok=True)
config_path.write_text("", encoding="utf-8")
return ""
except Exception as exc: # noqa: BLE001
return f"# Error loading config: {exc}"
def _write_config_text(config_path: Path, content: str) -> None:
config_path.parent.mkdir(parents=True, exist_ok=True)
config_path.write_text(content, encoding="utf-8")
def _get_interface_statuses():
statuses = []
interfaces = getattr(RNS.Transport, "interfaces", []) or []
for interface in interfaces:
if interface is None:
continue
if (
interface.__class__.__name__ == "LocalClientInterface"
and getattr(interface, "is_connected_to_shared_instance", False)
):
continue
statuses.append(
{
"name": getattr(interface, "name", None) or interface.__class__.__name__,
"online": bool(getattr(interface, "online", False)),
"type": interface.__class__.__name__,
"bitrate": getattr(interface, "bitrate", None),
},
)
return statuses
def _format_bitrate(bitrate: int | None) -> str | None:
if not bitrate:
return None
if bitrate >= 1_000_000:
return f"{bitrate / 1_000_000:.1f} Mbps"
if bitrate >= 1_000:
return f"{bitrate / 1_000:.0f} kbps"
return f"{bitrate} bps"
def _build_interface_chip_controls(statuses):
if not statuses:
return [
ft.Text(
"No interfaces detected",
size=11,
color=ft.Colors.ON_SURFACE_VARIANT,
),
]
chips = []
for status in statuses:
indicator_color = ft.Colors.GREEN if status["online"] else ft.Colors.ERROR
tooltip = status["type"]
bitrate_label = _format_bitrate(status.get("bitrate"))
if bitrate_label:
tooltip = f"{tooltip}{bitrate_label}"
chips.append(
ft.Container(
content=ft.Row(
[
ft.Icon(ft.Icons.CIRCLE, size=10, color=indicator_color),
ft.Text(status["name"], size=11),
],
spacing=4,
vertical_alignment=ft.CrossAxisAlignment.CENTER,
),
bgcolor="#1C1F2B",
border_radius=999,
padding=ft.padding.symmetric(horizontal=10, vertical=4),
tooltip=tooltip,
),
)
return chips
def _refresh_interface_status(summary_text, chip_wrap, updated_text):
statuses = _get_interface_statuses()
total = len(statuses)
online = sum(1 for entry in statuses if entry["online"])
if total == 0:
summary_text.value = "No active interfaces"
summary_text.color = ft.Colors.ERROR
else:
summary_text.value = f"{online}/{total} interfaces online"
summary_text.color = ft.Colors.GREEN if online else ft.Colors.ERROR
chip_wrap.controls = _build_interface_chip_controls(statuses)
updated_text.value = f"Updated {datetime.now().strftime('%H:%M:%S')}"
def _build_status_section(page: ft.Page):
summary_text = ft.Text("", size=16, weight=ft.FontWeight.BOLD)
updated_text = ft.Text("", size=12, color=ft.Colors.ON_SURFACE_VARIANT)
chip_wrap = ft.Row(
spacing=6,
run_spacing=6,
vertical_alignment=ft.CrossAxisAlignment.CENTER,
)
def refresh(_=None):
_refresh_interface_status(summary_text, chip_wrap, updated_text)
page.update()
refresh()
refresh_button = ft.IconButton(
icon=ft.Icons.REFRESH,
tooltip="Refresh status",
on_click=refresh,
icon_color=ft.Colors.BLUE_200,
)
section = ft.Column(
spacing=12,
controls=[
ft.Row(
controls=[
ft.Row(
controls=[
ft.Icon(ft.Icons.LAN, size=18, color=ft.Colors.BLUE_200),
summary_text,
],
spacing=6,
vertical_alignment=ft.CrossAxisAlignment.CENTER,
),
refresh_button,
],
alignment=ft.MainAxisAlignment.SPACE_BETWEEN,
vertical_alignment=ft.CrossAxisAlignment.CENTER,
),
chip_wrap,
updated_text,
],
)
return section, refresh
def _build_storage_field(storage):
storage_field = ft.TextField(
label="Storage Information",
value="",
expand=True,
multiline=True,
read_only=True,
min_lines=10,
max_lines=15,
border_color=ft.Colors.GREY_700,
text_style=ft.TextStyle(font_family="monospace", size=12),
)
def refresh():
info = storage.get_storage_info()
storage_field.value = "\n".join(f"{key}: {value}" for key, value in info.items())
refresh()
return storage_field, refresh
def open_settings_tab(page: ft.Page, tab_manager): def open_settings_tab(page: ft.Page, tab_manager):
"""Open a settings tab with configuration and debugging options. """Open a settings tab with configuration, status, and storage info."""
Args:
page: Flet page instance for UI updates.
tab_manager: Tab manager to add the settings tab to.
"""
storage = get_storage_manager(page) storage = get_storage_manager(page)
config_path = _get_config_file_path()
try: config_text = _read_config_text(config_path)
config_text = storage.load_config()
except Exception as ex:
config_text = f"Error reading config: {ex}"
app_settings = storage.load_app_settings() app_settings = storage.load_app_settings()
config_field = ft.TextField( config_field = ft.TextField(
@@ -61,7 +241,7 @@ def open_settings_tab(page: ft.Page, tab_manager):
border=ft.border.all(1, ft.Colors.GREY_700), border=ft.border.all(1, ft.Colors.GREY_700),
) )
def on_bgcolor_change(e): def on_bgcolor_change(_):
try: try:
color_preview.bgcolor = page_bgcolor_field.value color_preview.bgcolor = page_bgcolor_field.value
page.update() page.update()
@@ -70,196 +250,81 @@ def open_settings_tab(page: ft.Page, tab_manager):
page_bgcolor_field.on_change = on_bgcolor_change page_bgcolor_field.on_change = on_bgcolor_change
def on_save_config(ev): def show_snack(message, *, success=True):
snack = ft.SnackBar(
content=ft.Row(
controls=[
ft.Icon(
ft.Icons.CHECK_CIRCLE if success else ft.Icons.ERROR,
color=ft.Colors.GREEN_400 if success else ft.Colors.RED_400,
size=20,
),
ft.Text(message, color=ft.Colors.WHITE),
],
tight=True,
),
bgcolor=ft.Colors.GREEN_900 if success else ft.Colors.RED_900,
duration=3000 if success else 4000,
)
page.overlay.append(snack)
snack.open = True
page.update()
def on_save_config(_):
try: try:
success = storage.save_config(config_field.value) _write_config_text(config_path, config_field.value)
if success: show_snack(f"Configuration saved to {config_path}")
snack = ft.SnackBar( except Exception as exc: # noqa: BLE001
content=ft.Row( show_snack(f"Failed to save configuration: {exc}", success=False)
controls=[
ft.Icon(
ft.Icons.CHECK_CIRCLE,
color=ft.Colors.GREEN_400,
size=20,
),
ft.Text(
"Configuration saved! Restart app to apply changes.",
color=ft.Colors.WHITE,
),
],
tight=True,
),
bgcolor=ft.Colors.GREEN_900,
duration=3000,
)
page.overlay.append(snack)
snack.open = True
page.update()
else:
snack = ft.SnackBar(
content=ft.Row(
controls=[
ft.Icon(ft.Icons.ERROR, color=ft.Colors.RED_400, size=20),
ft.Text(
"Failed to save configuration", color=ft.Colors.WHITE
),
],
tight=True,
),
bgcolor=ft.Colors.RED_900,
duration=3000,
)
page.overlay.append(snack)
snack.open = True
page.update()
except Exception as ex:
snack = ft.SnackBar(
content=ft.Row(
controls=[
ft.Icon(ft.Icons.ERROR, color=ft.Colors.RED_400, size=20),
ft.Text(f"Error: {ex}", color=ft.Colors.WHITE),
],
tight=True,
),
bgcolor=ft.Colors.RED_900,
duration=4000,
)
page.overlay.append(snack)
snack.open = True
page.update()
def on_save_and_reload_config(ev): def on_save_and_reload_config(_):
try: try:
success = storage.save_config(config_field.value) _write_config_text(config_path, config_field.value)
if not success: except Exception as exc: # noqa: BLE001
snack = ft.SnackBar( show_snack(f"Failed to save configuration: {exc}", success=False)
content=ft.Row( return
controls=[
ft.Icon(ft.Icons.ERROR, color=ft.Colors.RED_400, size=20), loading_snack = ft.SnackBar(
ft.Text( content=ft.Row(
"Failed to save configuration", color=ft.Colors.WHITE controls=[
), ft.ProgressRing(
], width=16,
tight=True, height=16,
stroke_width=2,
color=ft.Colors.BLUE_400,
), ),
bgcolor=ft.Colors.RED_900, ft.Text("Reloading Reticulum...", color=ft.Colors.WHITE),
duration=3000, ],
) tight=True,
page.overlay.append(snack) ),
snack.open = True bgcolor=ft.Colors.BLUE_900,
page.update() duration=10000,
return )
page.overlay.append(loading_snack)
loading_snack.open = True
page.update()
loading_snack = ft.SnackBar( async def do_reload():
content=ft.Row( import ren_browser.app as app_module
controls=[
ft.ProgressRing(
width=16,
height=16,
stroke_width=2,
color=ft.Colors.BLUE_400,
),
ft.Text("Reloading Reticulum...", color=ft.Colors.WHITE),
],
tight=True,
),
bgcolor=ft.Colors.BLUE_900,
duration=10000,
)
page.overlay.append(loading_snack)
loading_snack.open = True
page.update()
async def do_reload(): try:
import ren_browser.app as app_module await app_module.reload_reticulum(page, on_reload_complete)
except Exception as exc: # noqa: BLE001
try:
await app_module.reload_reticulum(page, on_reload_complete)
except Exception as e:
loading_snack.open = False
page.update()
snack = ft.SnackBar(
content=ft.Row(
controls=[
ft.Icon(
ft.Icons.ERROR, color=ft.Colors.RED_400, size=20
),
ft.Text(
f"Reload failed: {str(e)}", color=ft.Colors.WHITE
),
],
tight=True,
),
bgcolor=ft.Colors.RED_900,
duration=4000,
)
page.overlay.append(snack)
snack.open = True
page.update()
def on_reload_complete(success, error):
loading_snack.open = False loading_snack.open = False
page.update() page.update()
show_snack(f"Reload failed: {exc}", success=False)
if success: def on_reload_complete(success, error):
snack = ft.SnackBar( loading_snack.open = False
content=ft.Row(
controls=[
ft.Icon(
ft.Icons.CHECK_CIRCLE,
color=ft.Colors.GREEN_400,
size=20,
),
ft.Text(
"Reticulum reloaded successfully!",
color=ft.Colors.WHITE,
),
],
tight=True,
),
bgcolor=ft.Colors.GREEN_900,
duration=3000,
)
else:
snack = ft.SnackBar(
content=ft.Row(
controls=[
ft.Icon(
ft.Icons.ERROR, color=ft.Colors.RED_400, size=20
),
ft.Text(
f"Reload failed: {error}", color=ft.Colors.WHITE
),
],
tight=True,
),
bgcolor=ft.Colors.RED_900,
duration=4000,
)
page.overlay.append(snack)
snack.open = True
page.update()
page.run_task(do_reload)
except Exception as ex:
snack = ft.SnackBar(
content=ft.Row(
controls=[
ft.Icon(ft.Icons.ERROR, color=ft.Colors.RED_400, size=20),
ft.Text(f"Error: {ex}", color=ft.Colors.WHITE),
],
tight=True,
),
bgcolor=ft.Colors.RED_900,
duration=4000,
)
page.overlay.append(snack)
snack.open = True
page.update() page.update()
if success:
show_snack("Reticulum reloaded successfully!")
else:
show_snack(f"Reload failed: {error}", success=False)
def on_save_app_settings(ev): page.run_task(do_reload)
def on_save_app_settings(_):
try: try:
new_settings = { new_settings = {
"horizontal_scroll": horizontal_scroll_switch.value, "horizontal_scroll": horizontal_scroll_switch.value,
@@ -267,123 +332,35 @@ def open_settings_tab(page: ft.Page, tab_manager):
} }
success = storage.save_app_settings(new_settings) success = storage.save_app_settings(new_settings)
if success: if success:
tab_manager.apply_settings(new_settings) if hasattr(tab_manager, "apply_settings"):
snack = ft.SnackBar( tab_manager.apply_settings(new_settings)
content=ft.Row( show_snack("Appearance settings saved and applied!")
controls=[
ft.Icon(
ft.Icons.CHECK_CIRCLE,
color=ft.Colors.GREEN_400,
size=20,
),
ft.Text(
"Appearance settings saved and applied!",
color=ft.Colors.WHITE,
),
],
tight=True,
),
bgcolor=ft.Colors.GREEN_900,
duration=2000,
)
page.overlay.append(snack)
snack.open = True
page.update()
else: else:
snack = ft.SnackBar( show_snack("Failed to save appearance settings", success=False)
content=ft.Row( except Exception as exc: # noqa: BLE001
controls=[ show_snack(f"Error saving appearance: {exc}", success=False)
ft.Icon(ft.Icons.ERROR, color=ft.Colors.RED_400, size=20),
ft.Text(
"Failed to save appearance settings",
color=ft.Colors.WHITE,
),
],
tight=True,
),
bgcolor=ft.Colors.RED_900,
duration=3000,
)
page.overlay.append(snack)
snack.open = True
page.update()
except Exception as ex:
snack = ft.SnackBar(
content=ft.Row(
controls=[
ft.Icon(ft.Icons.ERROR, color=ft.Colors.RED_400, size=20),
ft.Text(f"Error: {ex}", color=ft.Colors.WHITE),
],
tight=True,
),
bgcolor=ft.Colors.RED_900,
duration=4000,
)
page.overlay.append(snack)
snack.open = True
page.update()
save_btn = ft.ElevatedButton( save_btn = ft.ElevatedButton(
"Save Configuration", "Save Configuration",
icon=ft.Icons.SAVE, icon=ft.Icons.SAVE,
on_click=on_save_config, on_click=on_save_config,
bgcolor=ft.Colors.BLUE_700, style=_blue_button_style(),
color=ft.Colors.WHITE,
) )
save_reload_btn = ft.ElevatedButton( save_reload_btn = ft.ElevatedButton(
"Save & Hot Reload", "Save & Hot Reload",
icon=ft.Icons.REFRESH, icon=ft.Icons.REFRESH,
on_click=on_save_and_reload_config, on_click=on_save_and_reload_config,
bgcolor=ft.Colors.GREEN_700, style=_blue_button_style(),
color=ft.Colors.WHITE,
) )
save_appearance_btn = ft.ElevatedButton( save_appearance_btn = ft.ElevatedButton(
"Save Appearance", "Save Appearance",
icon=ft.Icons.PALETTE, icon=ft.Icons.PALETTE,
on_click=on_save_app_settings, on_click=on_save_app_settings,
bgcolor=ft.Colors.BLUE_700, style=_blue_button_style(),
color=ft.Colors.WHITE,
)
error_field = ft.TextField(
label="Error Logs",
value="",
expand=True,
multiline=True,
read_only=True,
min_lines=15,
max_lines=20,
border_color=ft.Colors.GREY_700,
text_style=ft.TextStyle(font_family="monospace", size=12),
)
ret_field = ft.TextField(
label="Reticulum Logs",
value="",
expand=True,
multiline=True,
read_only=True,
min_lines=15,
max_lines=20,
border_color=ft.Colors.GREY_700,
text_style=ft.TextStyle(font_family="monospace", size=12),
) )
storage_info = storage.get_storage_info() status_content, refresh_status_section = _build_status_section(page)
storage_text = "\n".join([f"{key}: {value}" for key, value in storage_info.items()]) storage_field, refresh_storage_info = _build_storage_field(storage)
storage_field = ft.TextField(
label="Storage Information",
value=storage_text,
expand=True,
multiline=True,
read_only=True,
min_lines=10,
max_lines=15,
border_color=ft.Colors.GREY_700,
text_style=ft.TextStyle(font_family="monospace", size=12),
)
content_placeholder = ft.Container(expand=True)
appearance_content = ft.Column( appearance_content = ft.Column(
spacing=16, spacing=16,
@@ -391,10 +368,7 @@ def open_settings_tab(page: ft.Page, tab_manager):
ft.Text("Appearance Settings", size=18, weight=ft.FontWeight.BOLD), ft.Text("Appearance Settings", size=18, weight=ft.FontWeight.BOLD),
horizontal_scroll_switch, horizontal_scroll_switch,
ft.Row( ft.Row(
controls=[ controls=[page_bgcolor_field, color_preview],
page_bgcolor_field,
color_preview,
],
alignment=ft.MainAxisAlignment.START, alignment=ft.MainAxisAlignment.START,
spacing=16, spacing=16,
), ),
@@ -402,68 +376,55 @@ def open_settings_tab(page: ft.Page, tab_manager):
], ],
) )
def show_config(ev): content_placeholder = ft.Container(expand=True, content=config_field)
def show_config(_):
content_placeholder.content = config_field content_placeholder.content = config_field
page.update() page.update()
def show_appearance(ev): def show_appearance(_):
content_placeholder.content = appearance_content content_placeholder.content = appearance_content
page.update() page.update()
def show_errors(ev): def show_status(_):
error_field.value = "\n".join(ERROR_LOGS) or "No errors logged." content_placeholder.content = status_content
content_placeholder.content = error_field refresh_status_section()
page.update()
def show_ret_logs(ev): def show_storage_info(_):
ret_field.value = "\n".join(RET_LOGS) or "No Reticulum logs." refresh_storage_info()
content_placeholder.content = ret_field
page.update()
def show_storage_info(ev):
storage_info = storage.get_storage_info()
storage_field.value = "\n".join(
[f"{key}: {value}" for key, value in storage_info.items()],
)
content_placeholder.content = storage_field content_placeholder.content = storage_field
page.update() page.update()
def refresh_current_view(ev): def refresh_current_view(_):
if content_placeholder.content == error_field: if content_placeholder.content == status_content:
show_errors(ev) refresh_status_section()
elif content_placeholder.content == ret_field:
show_ret_logs(ev)
elif content_placeholder.content == storage_field: elif content_placeholder.content == storage_field:
show_storage_info(ev) refresh_storage_info()
elif content_placeholder.content == appearance_content: page.update()
show_appearance(ev)
elif content_placeholder.content == config_field:
show_config(ev)
btn_config = ft.FilledButton( btn_config = ft.FilledButton(
"Configuration", "Configuration",
icon=ft.Icons.SETTINGS, icon=ft.Icons.SETTINGS,
on_click=show_config, on_click=show_config,
style=_blue_button_style(),
) )
btn_appearance = ft.FilledButton( btn_appearance = ft.FilledButton(
"Appearance", "Appearance",
icon=ft.Icons.PALETTE, icon=ft.Icons.PALETTE,
on_click=show_appearance, on_click=show_appearance,
style=_blue_button_style(),
) )
btn_errors = ft.FilledButton( btn_status = ft.FilledButton(
"Errors", "Status",
icon=ft.Icons.ERROR_OUTLINE, icon=ft.Icons.LAN,
on_click=show_errors, on_click=show_status,
) style=_blue_button_style(),
btn_ret = ft.FilledButton(
"Reticulum Logs",
icon=ft.Icons.TERMINAL,
on_click=show_ret_logs,
) )
btn_storage = ft.FilledButton( btn_storage = ft.FilledButton(
"Storage", "Storage",
icon=ft.Icons.STORAGE, icon=ft.Icons.STORAGE,
on_click=show_storage_info, on_click=show_storage_info,
style=_blue_button_style(),
) )
btn_refresh = ft.IconButton( btn_refresh = ft.IconButton(
icon=ft.Icons.REFRESH, icon=ft.Icons.REFRESH,
@@ -474,14 +435,7 @@ def open_settings_tab(page: ft.Page, tab_manager):
nav_card = ft.Container( nav_card = ft.Container(
content=ft.Row( content=ft.Row(
controls=[ controls=[btn_config, btn_appearance, btn_status, btn_storage, btn_refresh],
btn_config,
btn_appearance,
btn_errors,
btn_ret,
btn_storage,
btn_refresh,
],
spacing=8, spacing=8,
wrap=True, wrap=True,
), ),
@@ -507,7 +461,6 @@ def open_settings_tab(page: ft.Page, tab_manager):
padding=ft.padding.symmetric(horizontal=16, vertical=8), padding=ft.padding.symmetric(horizontal=16, vertical=8),
) )
content_placeholder.content = config_field
settings_content = ft.Column( settings_content = ft.Column(
expand=True, expand=True,
spacing=16, spacing=16,
@@ -526,7 +479,9 @@ def open_settings_tab(page: ft.Page, tab_manager):
action_row, action_row,
], ],
) )
tab_manager._add_tab_internal("Settings", settings_content) tab_manager._add_tab_internal("Settings", settings_content)
idx = len(tab_manager.manager.tabs) - 1 idx = len(tab_manager.manager.tabs) - 1
tab_manager.select_tab(idx) tab_manager.select_tab(idx)
page.update() page.update()

View File

@@ -76,6 +76,11 @@ def mock_storage_manager():
mock_storage.save_config.return_value = True mock_storage.save_config.return_value = True
mock_storage.get_config_path.return_value = Mock() mock_storage.get_config_path.return_value = Mock()
mock_storage.get_reticulum_config_path.return_value = Mock() mock_storage.get_reticulum_config_path.return_value = Mock()
mock_storage.load_app_settings.return_value = {
"horizontal_scroll": False,
"page_bgcolor": "#000000",
}
mock_storage.save_app_settings.return_value = True
mock_storage.get_storage_info.return_value = { mock_storage.get_storage_info.return_value = {
"storage_dir": "/mock/storage", "storage_dir": "/mock/storage",
"config_path": "/mock/storage/config.txt", "config_path": "/mock/storage/config.txt",

View File

@@ -1,5 +1,6 @@
from unittest.mock import Mock from unittest.mock import Mock
import flet as ft
import pytest import pytest
from ren_browser import app from ren_browser import app
@@ -14,16 +15,21 @@ class TestAppIntegration:
mock_page = Mock() mock_page = Mock()
mock_page.add = Mock() mock_page.add = Mock()
mock_page.update = Mock() mock_page.update = Mock()
mock_page.run_thread = Mock()
mock_page.controls = Mock() mock_page.controls = Mock()
mock_page.controls.clear = Mock() mock_page.controls.clear = Mock()
mock_page.width = 1024
mock_page.window = Mock()
mock_page.window.maximized = False
mock_page.appbar = Mock()
mock_page.drawer = Mock()
mock_page.theme_mode = ft.ThemeMode.DARK
await app.main(mock_page) await app.main(mock_page)
# Verify that the main function sets up the loading screen assert mock_page.add.call_count >= 1
mock_page.add.assert_called_once() loader_call = mock_page.add.call_args_list[0][0][0]
assert isinstance(loader_call, ft.Container)
mock_page.update.assert_called() mock_page.update.assert_called()
mock_page.run_thread.assert_called_once()
def test_entry_points_exist(self): def test_entry_points_exist(self):
"""Test that all expected entry points exist and are callable.""" """Test that all expected entry points exist and are callable."""

View File

@@ -12,26 +12,34 @@ class TestApp:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_main_initializes_loader(self, mock_page, mock_rns): async def test_main_initializes_loader(self, mock_page, mock_rns):
"""Test that main function initializes with loading screen.""" """Test that main function initializes with loading screen."""
with patch("ren_browser.ui.ui.build_ui"): with (
patch("ren_browser.rns.initialize_reticulum", return_value=True),
patch("ren_browser.rns.get_reticulum_instance"),
patch("ren_browser.rns.get_config_path", return_value="/tmp/.reticulum"),
patch("ren_browser.app.build_ui"),
):
await app.main(mock_page) await app.main(mock_page)
mock_page.add.assert_called_once() assert mock_page.add.call_count >= 1
loader_call = mock_page.add.call_args_list[0][0][0]
assert isinstance(loader_call, ft.Container)
mock_page.update.assert_called() mock_page.update.assert_called()
mock_page.run_thread.assert_called_once()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_main_function_structure(self, mock_page, mock_rns): async def test_main_function_structure(self, mock_page, mock_rns):
"""Test that main function sets up the expected structure.""" """Test that main function sets up the expected structure."""
await app.main(mock_page) with (
patch("ren_browser.rns.initialize_reticulum", return_value=True),
patch("ren_browser.rns.get_reticulum_instance"),
patch("ren_browser.rns.get_config_path"),
patch("ren_browser.app.build_ui"),
):
await app.main(mock_page)
# Verify that main function adds content and sets up threading assert mock_page.add.call_count >= 1
mock_page.add.assert_called_once() loader_call = mock_page.add.call_args_list[0][0][0]
assert isinstance(loader_call, ft.Container)
mock_page.update.assert_called() mock_page.update.assert_called()
mock_page.run_thread.assert_called_once()
# Verify that a function was passed to run_thread
init_function = mock_page.run_thread.call_args[0][0]
assert callable(init_function)
def test_run_with_default_args(self, mock_rns): def test_run_with_default_args(self, mock_rns):
"""Test run function with default arguments.""" """Test run function with default arguments."""

View File

@@ -93,28 +93,46 @@ class TestBuildUI:
class TestOpenSettingsTab: class TestOpenSettingsTab:
"""Test cases for the open_settings_tab function.""" """Test cases for the open_settings_tab function."""
def test_open_settings_tab_basic(self, mock_page): def test_open_settings_tab_basic(self, mock_page, mock_storage_manager):
"""Test opening settings tab with basic functionality.""" """Test opening settings tab with basic functionality."""
mock_tab_manager = Mock() mock_tab_manager = Mock()
mock_tab_manager.manager.tabs = [] mock_tab_manager.manager.tabs = []
mock_tab_manager._add_tab_internal = Mock() mock_tab_manager._add_tab_internal = Mock()
mock_tab_manager.select_tab = Mock() mock_tab_manager.select_tab = Mock()
with patch("pathlib.Path.read_text", return_value="config content"): mock_page.overlay = []
with (
patch(
"ren_browser.ui.settings.get_storage_manager",
return_value=mock_storage_manager,
),
patch("ren_browser.ui.settings.rns.get_config_path", return_value="/tmp/rns"),
patch("pathlib.Path.read_text", return_value="config content"),
):
open_settings_tab(mock_page, mock_tab_manager) open_settings_tab(mock_page, mock_tab_manager)
mock_tab_manager._add_tab_internal.assert_called_once() mock_tab_manager._add_tab_internal.assert_called_once()
mock_tab_manager.select_tab.assert_called_once() mock_tab_manager.select_tab.assert_called_once()
mock_page.update.assert_called() mock_page.update.assert_called()
def test_open_settings_tab_config_error(self, mock_page): def test_open_settings_tab_config_error(self, mock_page, mock_storage_manager):
"""Test opening settings tab when config file cannot be read.""" """Test opening settings tab when config file cannot be read."""
mock_tab_manager = Mock() mock_tab_manager = Mock()
mock_tab_manager.manager.tabs = [] mock_tab_manager.manager.tabs = []
mock_tab_manager._add_tab_internal = Mock() mock_tab_manager._add_tab_internal = Mock()
mock_tab_manager.select_tab = Mock() mock_tab_manager.select_tab = Mock()
with patch("pathlib.Path.read_text", side_effect=Exception("File not found")): mock_page.overlay = []
with (
patch(
"ren_browser.ui.settings.get_storage_manager",
return_value=mock_storage_manager,
),
patch("ren_browser.ui.settings.rns.get_config_path", return_value="/tmp/rns"),
patch("pathlib.Path.read_text", side_effect=Exception("File not found")),
):
open_settings_tab(mock_page, mock_tab_manager) open_settings_tab(mock_page, mock_tab_manager)
mock_tab_manager._add_tab_internal.assert_called_once() mock_tab_manager._add_tab_internal.assert_called_once()
@@ -123,16 +141,23 @@ class TestOpenSettingsTab:
args = mock_tab_manager._add_tab_internal.call_args args = mock_tab_manager._add_tab_internal.call_args
assert args[0][0] == "Settings" assert args[0][0] == "Settings"
def test_settings_save_config_success(self, mock_page): def test_settings_save_config_success(self, mock_page, mock_storage_manager):
"""Test saving config successfully in settings.""" """Test saving config successfully in settings."""
mock_tab_manager = Mock() mock_tab_manager = Mock()
mock_tab_manager.manager.tabs = [] mock_tab_manager.manager.tabs = []
mock_tab_manager._add_tab_internal = Mock() mock_tab_manager._add_tab_internal = Mock()
mock_tab_manager.select_tab = Mock() mock_tab_manager.select_tab = Mock()
mock_page.overlay = []
with ( with (
patch(
"ren_browser.ui.settings.get_storage_manager",
return_value=mock_storage_manager,
),
patch("ren_browser.ui.settings.rns.get_config_path", return_value="/tmp/rns"),
patch("pathlib.Path.read_text", return_value="config"), patch("pathlib.Path.read_text", return_value="config"),
patch("pathlib.Path.write_text"), patch("pathlib.Path.write_text") as mock_write,
): ):
open_settings_tab(mock_page, mock_tab_manager) open_settings_tab(mock_page, mock_tab_manager)
@@ -152,40 +177,68 @@ class TestOpenSettingsTab:
break break
assert save_btn is not None assert save_btn is not None
save_btn.on_click(None)
assert mock_write.called
def test_settings_save_config_error(self, mock_page, mock_storage_manager): def test_settings_save_config_error(self, mock_page, mock_storage_manager):
"""Test saving config with error in settings.""" """Test saving config error path does not crash."""
mock_tab_manager = Mock() mock_tab_manager = Mock()
mock_tab_manager.manager.tabs = [] mock_tab_manager.manager.tabs = []
mock_tab_manager._add_tab_internal = Mock() mock_tab_manager._add_tab_internal = Mock()
mock_tab_manager.select_tab = Mock() mock_tab_manager.select_tab = Mock()
with patch( mock_page.overlay = []
"ren_browser.ui.settings.get_storage_manager",
return_value=mock_storage_manager,
):
open_settings_tab(mock_page, mock_tab_manager)
settings_content = mock_tab_manager._add_tab_internal.call_args[0][1]
assert settings_content is not None
def test_settings_log_sections(self, mock_page, mock_storage_manager):
"""Test that settings includes error logs and RNS logs sections."""
mock_tab_manager = Mock()
mock_tab_manager.manager.tabs = []
mock_tab_manager._add_tab_internal = Mock()
mock_tab_manager.select_tab = Mock()
with ( with (
patch( patch(
"ren_browser.ui.settings.get_storage_manager", "ren_browser.ui.settings.get_storage_manager",
return_value=mock_storage_manager, return_value=mock_storage_manager,
), ),
patch("ren_browser.logs.ERROR_LOGS", ["Error 1", "Error 2"]), patch("ren_browser.ui.settings.rns.get_config_path", return_value="/tmp/rns"),
patch("ren_browser.logs.RET_LOGS", ["RNS log 1", "RNS log 2"]), patch("pathlib.Path.read_text", return_value="config"),
patch("pathlib.Path.write_text", side_effect=Exception("disk full")),
): ):
open_settings_tab(mock_page, mock_tab_manager) open_settings_tab(mock_page, mock_tab_manager)
mock_tab_manager._add_tab_internal.assert_called_once() settings_content = mock_tab_manager._add_tab_internal.call_args[0][1]
args = mock_tab_manager._add_tab_internal.call_args save_btn = None
assert args[0][0] == "Settings" for control in settings_content.controls:
if hasattr(control, "content") and hasattr(control.content, "controls"):
for sub_control in control.content.controls:
if (
hasattr(sub_control, "text")
and sub_control.text == "Save Configuration"
):
save_btn = sub_control
break
assert save_btn is not None
# Should not raise despite write failure
save_btn.on_click(None)
def test_settings_status_section_present(self, mock_page, mock_storage_manager):
"""Ensure the status navigation button is present."""
mock_tab_manager = Mock()
mock_tab_manager.manager.tabs = []
mock_tab_manager._add_tab_internal = Mock()
mock_tab_manager.select_tab = Mock()
mock_page.overlay = []
with (
patch(
"ren_browser.ui.settings.get_storage_manager",
return_value=mock_storage_manager,
),
patch("ren_browser.ui.settings.rns.get_config_path", return_value="/tmp/rns"),
patch("pathlib.Path.read_text", return_value="config"),
):
open_settings_tab(mock_page, mock_tab_manager)
settings_content = mock_tab_manager._add_tab_internal.call_args[0][1]
nav_container = settings_content.controls[1]
button_labels = [
ctrl.text
for ctrl in nav_container.content.controls
if hasattr(ctrl, "text")
]
assert "Status" in button_labels