mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 10:51:36 -04:00
The database now lives at <media>/config/wrolpi.db so a library (files, configs, AND database) travels with the drive between WROLPi systems. Core: - wrolpi/db.py: lazy SQLite engine factory (WAL, busy_timeout=30s, foreign_keys=ON); get_db_curs yields a plain sqlite3.Cursor; native sqlite3 paramstyle everywhere (no compatibility shims); global sqlite3 adapters keep raw-SQL datetime formats identical to the ORM's. - wrolpi/fts.py: FTS5 external-content search replacing tsvector/GIN. Injection-proof websearch->FTS5 translator, bm25(10,4,2,1) reproducing A/B/C/D weighting, fast(abc)/deep column filtering, snippet headlines. Stemming contract pinned by dedicated tests (watch/watches/watched/ watching are interchangeable). - wrolpi/schema_ddl.py: SQLite triggers shared verbatim by the alembic baseline and test DBs; fixes the channel video_count DELETE/UPDATE bugs and a latent total_size bug (it used the video id as channel id). - wrolpi/db_bootstrap.py: startup auto-create/migrate; refuses databases written by a newer WROLPi (db_version_mismatch); SQLite 3.40 floor; network-filesystem refusal; never creates/migrates/trusts a database beneath an unmounted production media directory (shadow-DB guard). - Alembic: 85 Postgres-era migrations squashed into one baseline; revision ids are dates (2026_07_10_1350) so filenames self-sort and alembic_version says when a schema was created; render_as_batch=True. - All raw Postgres SQL rewritten (ANY->json_each IN, array_agg @> -> HAVING COUNT(DISTINCT), unnest->executemany/UPDATE..FROM, generate_series histograms->strftime GROUP BY, DISTINCT ON-> row_number(), writable CTEs->two statements, jsonb_*->json_*, explicit NULLS FIRST/LAST, LIMIT-before-OFFSET). Simple statements converted to SQLAlchemy instead (flag upsert, download retry/summary/tag checks, channel-by-tag estimate, Directory delete). - Config imports count skipped entries; background dumps are blocked after a lossy import so partial imports can never overwrite YAML configs (they are the sole source of truth for rebuilds now). - Search tolerates explicit JSON nulls for limit/offset (Postgres treated OFFSET NULL as 0; SQLite errors). - file_group.id/doc_section.id render as INTEGER PRIMARY KEY on SQLite (BIGINT is not a rowid alias; FTS5 content_rowid requires it). Ops teardown: docker-compose db service, CI postgres sidecars, initialize_api_db.sh, controller postgresql service, psycopg2-binary all removed. upgrade.sh stops/disables postgresql (data is NOT deleted; `systemctl enable --now postgresql` restores it); systemd units no longer Want postgresql.service. Existing-user upgrade: first boot creates+migrates the DB, configs import (tags recreate their FileGroups from disk), a refresh rebuilds the library index. QA'd on the 10.0.0.8 RPi: rebuild matched the Postgres baseline exactly; survived an uncontrolled hard reboot with clean integrity checks. Accepted behavior changes: FTS5 has no stopwords; bm25 length normalization shifts some rank orders; fast-path d_headline is un-highlighted; all-negative queries match nothing. Deferred: Debian/trixie package lists (incl. sqlite3 CLI), Cypress re-baseline, VACUUM INTO snapshots, exFAT/NTFS warning, auto-kicked rebuild refresh with progress UX. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
130 lines
5 KiB
Python
130 lines
5 KiB
Python
"""
|
|
Unit tests for controller.defaults module.
|
|
"""
|
|
import pytest
|
|
|
|
from controller.defaults import DEFAULT_CONFIG
|
|
|
|
|
|
class TestDefaultConfig:
|
|
"""Tests for DEFAULT_CONFIG structure."""
|
|
|
|
def test_default_config_is_dict(self):
|
|
"""DEFAULT_CONFIG should be a dictionary."""
|
|
assert isinstance(DEFAULT_CONFIG, dict)
|
|
|
|
def test_default_port(self):
|
|
"""Default port should be 80."""
|
|
assert DEFAULT_CONFIG["port"] == 80
|
|
|
|
def test_default_media_directory(self):
|
|
"""Default media directory should be /media/wrolpi."""
|
|
assert DEFAULT_CONFIG["media_directory"] == "/media/wrolpi"
|
|
|
|
|
|
class TestDrivesConfig:
|
|
"""Tests for drives configuration."""
|
|
|
|
def test_drives_config_exists(self):
|
|
"""Drives config should exist."""
|
|
assert "drives" in DEFAULT_CONFIG
|
|
assert isinstance(DEFAULT_CONFIG["drives"], dict)
|
|
|
|
@pytest.mark.parametrize("filesystem", ["ext4", "btrfs", "vfat", "exfat"])
|
|
def test_supported_filesystem(self, filesystem):
|
|
"""Should have expected filesystems in supported list."""
|
|
filesystems = DEFAULT_CONFIG["drives"]["supported_filesystems"]
|
|
assert isinstance(filesystems, list)
|
|
assert filesystem in filesystems
|
|
|
|
def test_auto_mount_default(self):
|
|
"""Auto mount should be True by default."""
|
|
assert DEFAULT_CONFIG["drives"]["auto_mount"] is True
|
|
|
|
def test_mounts_empty_by_default(self):
|
|
"""Mounts list should be empty by default."""
|
|
assert DEFAULT_CONFIG["drives"]["mounts"] == []
|
|
|
|
|
|
class TestManagedServicesConfig:
|
|
"""Tests for managed services configuration."""
|
|
|
|
def test_managed_services_exists(self):
|
|
"""Managed services config should exist."""
|
|
assert "managed_services" in DEFAULT_CONFIG
|
|
assert isinstance(DEFAULT_CONFIG["managed_services"], list)
|
|
|
|
def test_managed_services_count(self):
|
|
"""Should have expected number of managed services."""
|
|
services = DEFAULT_CONFIG["managed_services"]
|
|
assert len(services) == 11
|
|
|
|
@pytest.mark.parametrize("name,expected", [
|
|
("wrolpi-api",
|
|
{"systemd_name": "wrolpi-api", "port": 8081, "viewable": True, "description": "Python API (Sanic)"}),
|
|
("wrolpi-api-dev",
|
|
{"systemd_name": "wrolpi-api-dev", "port": 8081, "viewable": True, "show_only_when_running": True}),
|
|
("wrolpi-app", {"port": 3000, "viewable": False}),
|
|
("wrolpi-app-dev", {"port": 3000, "viewable": False, "show_only_when_running": True}),
|
|
("caddy", {"port": 443}),
|
|
("wrolpi-upgrade", {"port": None, "viewable": False, "show_only_when_running": True}),
|
|
])
|
|
def test_service_config(self, name, expected):
|
|
"""Should have service configured with expected values."""
|
|
services = DEFAULT_CONFIG["managed_services"]
|
|
service = next((s for s in services if s["name"] == name), None)
|
|
assert service is not None, f"Service {name} not found"
|
|
for key, value in expected.items():
|
|
assert service[key] == value, f"Service {name} has {key}={service.get(key)}, expected {value}"
|
|
|
|
@pytest.mark.parametrize("name,use_https", [
|
|
("wrolpi-help", True),
|
|
("wrolpi-kiwix", True),
|
|
("wrolpi-api", False),
|
|
("wrolpi-app", False),
|
|
])
|
|
def test_service_use_https(self, name, use_https):
|
|
"""HTTPS services should have use_https=True, others should have False."""
|
|
services = DEFAULT_CONFIG["managed_services"]
|
|
service = next((s for s in services if s["name"] == name), None)
|
|
assert service is not None, f"Service {name} not found"
|
|
assert service.get("use_https", False) == use_https, \
|
|
f"Service {name} should have use_https={use_https}"
|
|
|
|
def test_all_services_have_required_fields(self):
|
|
"""All services should have required fields."""
|
|
required_fields = ["name", "systemd_name", "port", "viewable", "use_https", "description"]
|
|
for service in DEFAULT_CONFIG["managed_services"]:
|
|
for field in required_fields:
|
|
assert field in service, f"Service {service.get('name')} missing {field}"
|
|
|
|
|
|
class TestHotspotConfig:
|
|
"""Tests for hotspot configuration."""
|
|
|
|
def test_hotspot_config_exists(self):
|
|
"""Hotspot config should exist."""
|
|
assert "hotspot" in DEFAULT_CONFIG
|
|
assert isinstance(DEFAULT_CONFIG["hotspot"], dict)
|
|
|
|
@pytest.mark.parametrize("key,expected", [
|
|
("device", "wlan0"),
|
|
("ssid", "WROLPi"),
|
|
("password", "wrolpi hotspot"),
|
|
])
|
|
def test_hotspot_defaults(self, key, expected):
|
|
"""Hotspot should have expected default values."""
|
|
assert DEFAULT_CONFIG["hotspot"][key] == expected
|
|
|
|
|
|
class TestThrottleConfig:
|
|
"""Tests for throttle configuration."""
|
|
|
|
def test_throttle_config_exists(self):
|
|
"""Throttle config should exist."""
|
|
assert "throttle" in DEFAULT_CONFIG
|
|
assert isinstance(DEFAULT_CONFIG["throttle"], dict)
|
|
|
|
def test_default_governor(self):
|
|
"""Default governor should be ondemand."""
|
|
assert DEFAULT_CONFIG["throttle"]["default_governor"] == "ondemand"
|