feat(db_manager, web_viewer): enhance database path diagnostics and validation

- Added detailed diagnostics for database initialization failures in `DBManager`, improving error logging with filesystem checks.
- Implemented a new method `_resolve_viewer_db_path` in `WebViewerIntegration` to determine the database path for the viewer subprocess.
- Introduced `_preflight_database_path` to validate the viewer database path, ensuring the parent directory exists, is writable, and is a directory.
- Added unit tests to verify logging behavior for missing or invalid database paths in both `DBManager` and `WebViewerIntegration`.
This commit is contained in:
agessaman
2026-06-27 16:56:13 -07:00
parent 986fe67c01
commit 4da77b7527
4 changed files with 194 additions and 1 deletions
+29 -1
View File
@@ -5,11 +5,13 @@ Provides common database operations and table management for the MeshCore Bot
"""
import json
import os
import re
import sqlite3
from collections.abc import AsyncGenerator, Generator
from contextlib import asynccontextmanager, contextmanager
from datetime import date, datetime
from pathlib import Path
from typing import Any, Optional
from .db_migrations import MigrationRunner
@@ -73,9 +75,35 @@ class DBManager:
self.logger.info("Database manager initialized successfully")
except Exception as e:
self.logger.error(f"Failed to initialize database: {e}")
self.logger.error(
"Failed to initialize database at %s: %s. %s",
self.db_path,
e,
self._database_path_diagnostics(),
)
raise
def _database_path_diagnostics(self) -> str:
"""Return filesystem details that explain common SQLite open failures."""
try:
if str(self.db_path) == ":memory:":
return "Using in-memory SQLite database."
db_file = Path(self.db_path)
parent = db_file.parent if db_file.parent != Path("") else Path(".")
parent_exists = parent.exists()
parent_is_dir = parent.is_dir() if parent_exists else False
parent_writable = os.access(str(parent), os.W_OK) if parent_exists else False
file_exists = db_file.exists()
file_readable = os.access(str(db_file), os.R_OK) if file_exists else False
file_writable = os.access(str(db_file), os.W_OK) if file_exists else False
return (
f"parent={parent} "
f"(exists={parent_exists}, is_dir={parent_is_dir}, writable={parent_writable}); "
f"file_exists={file_exists}, readable={file_readable}, writable={file_writable}"
)
except Exception as diag_error:
return f"could not inspect database path: {diag_error}"
# Geocoding cache methods
def get_cached_geocoding(self, query: str) -> tuple[Optional[float], Optional[float]]:
"""Get cached geocoding result for a query.
+74
View File
@@ -695,12 +695,86 @@ class WebViewerIntegration:
"host = 127.0.0.1 for local-only access."
)
def _resolve_viewer_db_path(self) -> Path:
"""Return the database path the viewer subprocess will use."""
config_path = getattr(self.bot, 'config_file', 'config.ini')
config_base = Path(config_path).resolve().parent if config_path else Path(".").resolve()
if self.bot.config.has_section('Web_Viewer') and self.bot.config.has_option('Web_Viewer', 'db_path'):
raw = self.bot.config.get('Web_Viewer', 'db_path', fallback='').strip()
if raw:
return Path(resolve_path(raw, config_base))
return Path(self.bot.db_manager.db_path)
def _preflight_database_path(self) -> bool:
"""Validate the web viewer DB path before spawning the subprocess."""
try:
db_path = self._resolve_viewer_db_path()
except Exception as e:
self.logger.error("Web viewer database path could not be resolved: %s", e)
return False
parent = db_path.parent if db_path.parent != Path("") else Path(".")
parent_exists = parent.exists()
parent_is_dir = parent.is_dir() if parent_exists else False
parent_writable = os.access(str(parent), os.W_OK) if parent_exists else False
file_exists = db_path.exists()
file_readable = os.access(str(db_path), os.R_OK) if file_exists else False
file_writable = os.access(str(db_path), os.W_OK) if file_exists else False
self.logger.info("Web viewer database path: %s", db_path)
if not parent_exists:
self.logger.error(
"Web viewer database parent directory does not exist: %s. "
"Create it, fix [Web_Viewer] db_path, or remove [Web_Viewer] db_path to use [Bot] db_path.",
parent,
)
return False
if not parent_is_dir:
self.logger.error(
"Web viewer database parent path is not a directory: %s. "
"Fix [Web_Viewer] db_path or remove it to use [Bot] db_path.",
parent,
)
return False
if not parent_writable:
self.logger.error(
"Web viewer database parent directory is not writable: %s. "
"Fix permissions or choose a writable db_path.",
parent,
)
return False
if file_exists and (not file_readable or not file_writable):
self.logger.error(
"Web viewer database file is not readable and writable: %s "
"(readable=%s, writable=%s). Fix file permissions.",
db_path,
file_readable,
file_writable,
)
return False
if self.bot.config.has_section('Web_Viewer') and self.bot.config.get('Web_Viewer', 'db_path', fallback='').strip():
bot_db_path = Path(self.bot.db_manager.db_path)
if db_path.resolve() != bot_db_path.resolve():
self.logger.warning(
"Web viewer database path differs from bot database: viewer=%s, bot=%s. "
"Remove [Web_Viewer] db_path unless you intentionally use a separate viewer database.",
db_path,
bot_db_path,
)
return True
def start_viewer(self):
"""Start the web viewer in a separate thread"""
if self.running:
self.logger.warning("Web viewer is already running")
return
if not self._preflight_database_path():
self.running = False
return
# Intentional (re)start after stop_viewer / restart_viewer — allow monitor thread to run
self.shutting_down = False
+19
View File
@@ -17,6 +17,25 @@ def db(mock_logger, tmp_path):
return DBManager(bot, str(tmp_path / "test.db"))
class TestDatabaseInitialization:
def test_missing_parent_logs_path_diagnostics(self, mock_logger, tmp_path):
bot = Mock()
bot.logger = mock_logger
db_path = tmp_path / "missing" / "test.db"
with pytest.raises(sqlite3.OperationalError, match="unable to open database file"):
DBManager(bot, str(db_path))
logged = " ".join(
str(arg)
for call in mock_logger.error.call_args_list
for arg in call.args
)
assert str(db_path) in logged
assert "parent=" in logged
assert "exists=False" in logged
class TestGeocoding:
"""Tests for geocoding cache."""
+72
View File
@@ -4,6 +4,7 @@ import json
import queue
import time
from configparser import ConfigParser
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
import pytest
@@ -442,6 +443,77 @@ class TestWebViewerIntegrationValidation:
WebViewerIntegration(bot)
bot.logger.error.assert_not_called()
def test_start_viewer_refuses_missing_db_parent(self, tmp_path):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
bot.config_file = str(tmp_path / "config.ini")
Path(bot.config_file).write_text("[Bot]\n", encoding="utf-8")
bot.config.set("Web_Viewer", "db_path", "missing/viewer.db")
bot.db_manager.db_path = str(tmp_path / "bot.db")
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
wvi = WebViewerIntegration(bot)
with patch.object(wvi, "_run_viewer") as run_viewer:
wvi.start_viewer()
run_viewer.assert_not_called()
assert wvi.running is False
assert any(
"parent directory does not exist" in str(call.args[0])
for call in bot.logger.error.call_args_list
)
def test_start_viewer_refuses_db_parent_that_is_file(self, tmp_path):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
bot.config_file = str(tmp_path / "config.ini")
Path(bot.config_file).write_text("[Bot]\n", encoding="utf-8")
not_dir = tmp_path / "not_a_directory"
not_dir.write_text("not a directory", encoding="utf-8")
bot.config.set("Web_Viewer", "db_path", "not_a_directory/viewer.db")
bot.db_manager.db_path = str(tmp_path / "bot.db")
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
wvi = WebViewerIntegration(bot)
with patch.object(wvi, "_run_viewer") as run_viewer:
wvi.start_viewer()
run_viewer.assert_not_called()
assert wvi.running is False
assert any(
"parent path is not a directory" in str(call.args[0])
for call in bot.logger.error.call_args_list
)
def test_start_viewer_allows_valid_db_parent(self, tmp_path):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
bot.config_file = str(tmp_path / "config.ini")
Path(bot.config_file).write_text("[Bot]\n", encoding="utf-8")
bot.config.set("Web_Viewer", "db_path", "viewer.db")
bot.db_manager.db_path = str(tmp_path / "bot.db")
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
wvi = WebViewerIntegration(bot)
with patch("modules.web_viewer.integration.threading.Thread") as thread_cls:
thread = thread_cls.return_value
wvi.start_viewer()
thread.start.assert_called_once()
assert wvi.running is True
class TestNormalizedWebViewerPassword:
def test_blank_and_null_placeholders(self):