fix: ruff and mypy compliance across utility scripts and tests

Fix ruff violations (import sort, unused imports, whitespace) in
backup_database.py, generate_website.py, meshcore_bot.py,
migrate_webviewer_db.py, packet_capture_service.py, version_info.py,
scripts/config_tui.py, scripts/update_todos.py, and validate_config.py.

Fix mypy stale type-ignore and remove unused imports in test files.
Add is_radio_zombie/is_radio_offline mock attributes to command_manager
test fixtures to prevent truthy Mock early returns.
This commit is contained in:
Stacy Olivas authored and agessaman committed 2026-04-14 10:02:36 -07:00
1 parent ce7adc55f8
commit 7679eb7e2c
9 files changed
+367 -365

No files matched your search

+36 -37
View File
@@ -4,9 +4,8 @@ Database Backup Script for MeshCore Bot
Creates timestamped backups of all SQLite database files
"""
import sqlite3
import shutil
import os
import sqlite3
import sys
from contextlib import closing
from datetime import datetime
@@ -25,36 +24,36 @@ def backup_database(db_path, backup_dir="backups"):
Path to the backup file if successful, None otherwise
"""
db_path = Path(db_path)
if not db_path.exists():
print(f"Warning: Database file {db_path} does not exist, skipping...")
return None
# Create backup directory if it doesn't exist
backup_path = Path(backup_dir)
backup_path.mkdir(exist_ok=True)
# Generate timestamp string
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
# Create backup filename with timestamp
db_name = db_path.stem
backup_filename = f"{db_name}_{timestamp}.db"
backup_file = backup_path / backup_filename
try:
# Use SQLite backup API for proper backup (handles WAL files correctly)
with closing(sqlite3.connect(str(db_path))) as source_conn, closing(sqlite3.connect(str(backup_file))) as backup_conn:
# Perform the backup
source_conn.backup(backup_conn)
# Get file size for reporting
file_size = backup_file.stat().st_size
file_size_mb = file_size / (1024 * 1024)
print(f"✓ Backed up {db_path.name} -> {backup_file.name} ({file_size_mb:.2f} MB)")
return str(backup_file)
except sqlite3.Error as e:
print(f"✗ Error backing up {db_path.name}: {e}")
# Clean up failed backup file
@@ -79,25 +78,25 @@ def check_database_status(db_path):
Tuple of (is_used, table_count, file_size_mb)
"""
db_path = Path(db_path)
if not db_path.exists():
return (False, 0, 0.0)
file_size_mb = db_path.stat().st_size / (1024 * 1024)
try:
conn = sqlite3.connect(str(db_path))
with closing(conn):
cursor = conn.cursor()
# Get list of tables
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
table_count = len(tables)
# Consider database used if it has tables or is larger than 1KB
is_used = table_count > 0 or file_size_mb > 0.001
return (is_used, table_count, file_size_mb)
except sqlite3.Error:
return (False, 0, file_size_mb)
@@ -116,13 +115,13 @@ def find_database_files(root_dir=".", check_usage=True):
"""
root = Path(root_dir)
db_files = []
# Find all .db files in root directory (not in subdirectories)
for db_file in root.glob("*.db"):
# Skip WAL and SHM files (they're not standalone databases)
if db_file.suffix == ".db":
db_files.append(db_file)
# Sort by priority: meshcore_bot.db first (main database), then others
def sort_key(db_file):
name = db_file.name.lower()
@@ -132,7 +131,7 @@ def find_database_files(root_dir=".", check_usage=True):
return (1, name)
else:
return (2, name)
return sorted(db_files, key=sort_key)
@@ -141,75 +140,75 @@ def main():
# Get script directory (project root)
script_dir = Path(__file__).parent.absolute()
os.chdir(script_dir)
print("MeshCore Bot Database Backup")
print("=" * 50)
print(f"Working directory: {script_dir}")
print()
# Find all database files
db_files = find_database_files(script_dir)
if not db_files:
print("No database files found in the project root.")
return 1
print(f"Found {len(db_files)} database file(s):")
# Check which databases are actually used
active_databases = []
inactive_databases = []
for db_file in db_files:
is_used, table_count, file_size = check_database_status(db_file)
status = "✓ ACTIVE" if is_used else "○ EMPTY/UNUSED"
print(f" {status}: {db_file.name} ({table_count} tables, {file_size:.2f} MB)")
if is_used:
active_databases.append(db_file)
else:
inactive_databases.append(db_file)
print()
if not active_databases:
print("No active databases found. Nothing to backup.")
return 0
if inactive_databases:
print(f"Note: {len(inactive_databases)} database(s) appear to be empty/unused and will be skipped.")
print()
# Create backups (only for active databases)
backup_dir = script_dir / "backups"
successful_backups = []
failed_backups = []
for db_file in active_databases:
result = backup_database(db_file, backup_dir)
if result:
successful_backups.append(result)
else:
failed_backups.append(db_file.name)
print()
print("=" * 50)
print(f"Backup Summary:")
print("Backup Summary:")
print(f" Successful: {len(successful_backups)}")
print(f" Failed: {len(failed_backups)}")
if successful_backups:
print(f"\nBackups saved to: {backup_dir}")
print("\nRecent backups:")
for backup in successful_backups:
print(f" - {Path(backup).name}")
if failed_backups:
print(f"\nFailed to backup:")
print("\nFailed to backup:")
for db in failed_backups:
print(f" - {db}")
return 1
return 0
+286 -288
View File
File diff suppressed because it is too large. Load diff
+2 -3
View File
@@ -56,7 +56,6 @@ def main():
if args.validate_config:
from modules.config_validation import (
SEVERITY_ERROR,
SEVERITY_INFO,
SEVERITY_WARNING,
validate_config,
)
@@ -74,7 +73,7 @@ def main():
from modules.core import MeshCoreBot
bot = MeshCoreBot(config_file=args.config)
# Use asyncio.run() which handles KeyboardInterrupt properly
# For SIGTERM, we'll handle it in the async context
async def run_bot():
@@ -149,7 +148,7 @@ def main():
await bot.start()
finally:
await bot.stop()
try:
asyncio.run(run_bot())
except KeyboardInterrupt:
+1 -1
View File
@@ -57,7 +57,7 @@ def main() -> int:
)
if cur.fetchone() is None:
conn.execute("DETACH DATABASE src")
print(f"Source database has no packet_stream table; nothing to migrate.", file=sys.stderr)
print("Source database has no packet_stream table; nothing to migrate.", file=sys.stderr)
return 0
# Copy rows from source; skip ids that already exist in target (INSERT OR IGNORE)
+8
View File
@@ -18,6 +18,14 @@ if [[ -z "${VERSION}" ]]; then
grep -Po '(?<=^version = ")([^"]+)' "${PROJECT_ROOT}/pyproject.toml" || echo "0.9.0")"
fi
# Validate VERSION: must be semver-like (digits and dots only) to prevent
# shell injection via the unquoted <<EOF heredoc below.
if [[ ! "${VERSION}" =~ ^[0-9]+\.[0-9]+(\.[0-9]+)?([~+][A-Za-z0-9._-]+)?$ ]]; then
echo "ERROR: Invalid version string: '${VERSION}'" >&2
echo " Must match semver format, e.g. 1.2.3 or 1.2.3~rc1" >&2
exit 1
fi
PACKAGE_NAME="meshcore-bot"
ARCH="all"
INSTALL_ROOT="/opt/meshcore-bot"
+31 -34
View File
@@ -29,13 +29,10 @@ Navigation:
import configparser
import curses
import os
import re
import sys
import textwrap
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from typing import Optional
# ---------------------------------------------------------------------------
# Helpers
@@ -58,20 +55,20 @@ def load_config(path: Path) -> configparser.ConfigParser:
if path.exists():
try:
cfg.read(str(path), encoding="utf-8")
except (configparser.Error, IOError, OSError, UnicodeDecodeError) as e:
except (configparser.Error, OSError, UnicodeDecodeError) as e:
# Return empty config rather than crash; caller may show a warning
cfg._load_error = str(e) # type: ignore[attr-defined]
return cfg
def load_example_comments(example_path: Path) -> Dict[str, Dict[str, str]]:
def load_example_comments(example_path: Path) -> dict[str, dict[str, str]]:
"""Parse config.ini.example — return {section: {key: comment_text}}."""
comments: Dict[str, Dict[str, str]] = {}
comments: dict[str, dict[str, str]] = {}
if not example_path.exists():
return comments
try:
current_section = ""
pending: List[str] = []
pending: list[str] = []
with open(example_path, encoding="utf-8", errors="replace") as fh:
for line in fh:
stripped = line.strip()
@@ -88,14 +85,14 @@ def load_example_comments(example_path: Path) -> Dict[str, Dict[str, str]]:
pending = []
else:
pending = []
except (IOError, OSError, UnicodeDecodeError):
except (OSError, UnicodeDecodeError):
pass
return comments
def load_example_lines(example_path: Path) -> Dict[str, Dict[str, str]]:
def load_example_lines(example_path: Path) -> dict[str, dict[str, str]]:
"""Return {section: {key: 'full raw line from example'}}."""
lines: Dict[str, Dict[str, str]] = {}
lines: dict[str, dict[str, str]] = {}
if not example_path.exists():
return lines
try:
@@ -110,14 +107,14 @@ def load_example_lines(example_path: Path) -> Dict[str, Dict[str, str]]:
key = stripped.split("=", 1)[0].strip()
if current_section:
lines.setdefault(current_section, {})[key] = stripped
except (IOError, OSError, UnicodeDecodeError):
except (OSError, UnicodeDecodeError):
pass
return lines
def load_example_keys(example_path: Path) -> Dict[str, Dict[str, str]]:
def load_example_keys(example_path: Path) -> dict[str, dict[str, str]]:
"""Return {section: {key: default_value}} from config.ini.example."""
keys: Dict[str, Dict[str, str]] = {}
keys: dict[str, dict[str, str]] = {}
if not example_path.exists():
return keys
try:
@@ -129,17 +126,17 @@ def load_example_keys(example_path: Path) -> Dict[str, Dict[str, str]]:
keys[section] = dict(cfg.items(section))
except configparser.Error:
keys[section] = {}
except (configparser.Error, IOError, OSError, UnicodeDecodeError):
except (configparser.Error, OSError, UnicodeDecodeError):
pass
return keys
def validate_config(
cfg: configparser.ConfigParser,
example_keys: Dict[str, Dict[str, str]],
) -> List[Tuple[str, str, str]]:
example_keys: dict[str, dict[str, str]],
) -> list[tuple[str, str, str]]:
"""Return list of (severity, section, message) issues."""
issues: List[Tuple[str, str, str]] = []
issues: list[tuple[str, str, str]] = []
try:
required = {
"Connection": {"connection_type"},
@@ -174,10 +171,10 @@ def validate_config(
def migrate_config(
cfg: configparser.ConfigParser,
example_keys: Dict[str, Dict[str, str]],
example_comments: Dict[str, Dict[str, str]],
) -> List[str]:
changes: List[str] = []
example_keys: dict[str, dict[str, str]],
example_comments: dict[str, dict[str, str]],
) -> list[str]:
changes: list[str] = []
for section, keys in example_keys.items():
if not cfg.has_section(section):
try:
@@ -201,13 +198,13 @@ def migrate_config(
def save_config(cfg: configparser.ConfigParser, path: Path) -> None:
try:
path.parent.mkdir(parents=True, exist_ok=True)
except (IOError, OSError, PermissionError) as e:
raise IOError(f"Cannot create directory {path.parent}: {e}") from e
except (OSError, PermissionError) as e:
raise OSError(f"Cannot create directory {path.parent}: {e}") from e
try:
with open(path, "w", encoding="utf-8") as fh:
cfg.write(fh)
except (IOError, OSError, PermissionError) as e:
raise IOError(f"Cannot write {path}: {e}") from e
except (OSError, PermissionError) as e:
raise OSError(f"Cannot write {path}: {e}") from e
# ---------------------------------------------------------------------------
@@ -219,9 +216,9 @@ class TUIState:
self,
cfg: configparser.ConfigParser,
path: Path,
example_keys: Dict[str, Dict[str, str]],
example_comments: Dict[str, Dict[str, str]],
example_lines: Dict[str, Dict[str, str]],
example_keys: dict[str, dict[str, str]],
example_comments: dict[str, dict[str, str]],
example_lines: dict[str, dict[str, str]],
) -> None:
self.cfg = cfg
self.path = path
@@ -229,7 +226,7 @@ class TUIState:
self.example_comments = example_comments
self.example_lines = example_lines
self.dirty = False
self.sections: List[str] = []
self.sections: list[str] = []
try:
self.sections = cfg.sections()
except Exception:
@@ -245,7 +242,7 @@ class TUIState:
return ""
return self.sections[self.section_idx]
def current_keys(self) -> List[str]:
def current_keys(self) -> list[str]:
s = self.current_section()
if not s:
return []
@@ -442,7 +439,7 @@ def draw_hint_line(win, state: TUIState, top: int, left: int, width: int) -> Non
# Overlays
# ---------------------------------------------------------------------------
def show_overlay(stdscr, title: str, items: List[Tuple[str, str, str]]) -> None:
def show_overlay(stdscr, title: str, items: list[tuple[str, str, str]]) -> None:
"""Generic scrollable list overlay. Items: (tag, section, message)."""
try:
curses.curs_set(0)
@@ -561,7 +558,7 @@ def show_key_help(stdscr, state: TUIState) -> None:
current_val = "(error reading value)"
default_val = state.example_keys.get(section, {}).get(key, "")
lines: List[Tuple[str, str, str]] = [
lines: list[tuple[str, str, str]] = [
("INFO", "Section", section),
("INFO", "Key", key),
("INFO", "Current value", repr(current_val)),
@@ -1189,7 +1186,7 @@ def main() -> None:
print(f"config.ini not found — creating from {example_path.name}…")
shutil.copy(str(example_path), str(config_path))
print(f"Created {config_path}")
except (IOError, OSError, shutil.Error) as e:
except (OSError, shutil.Error) as e:
print(f"ERROR: Could not create config from example: {e}", file=sys.stderr)
sys.exit(1)
+1 -1
View File
@@ -92,7 +92,7 @@ def build_section(todos, today: str) -> str:
continue
lines.append(f"### {emoji.get(marker, '')} {marker}\n")
for rel, lineno, text in sorted(by_marker[marker]):
label = text if text else f"(no description — see file)"
label = text if text else "(no description — see file)"
lines.append(f"- [ ] **`{rel}:{lineno}`** — {label}")
lines.append("")
+2
View File
@@ -41,6 +41,7 @@ def cm_bot(mock_logger):
bot.bot_tx_rate_limiter = Mock()
bot.bot_tx_rate_limiter.wait_for_tx = Mock()
bot.tx_delay_ms = 0
bot.is_radio_zombie = False
return bot
@@ -553,6 +554,7 @@ class TestSendChannelMessageRetry:
def _setup_bot(self, cm_bot):
cm_bot.connected = True
cm_bot.is_radio_zombie = False
cm_bot.channel_manager = Mock()
cm_bot.channel_manager.get_channel_number = Mock(return_value=2)
cm_bot.meshcore = Mock()
-1
View File
@@ -11,7 +11,6 @@ import sys
from modules.config_validation import (
SEVERITY_ERROR,
SEVERITY_INFO,
SEVERITY_WARNING,
validate_config,
)