mirror of
https://github.com/Screenly/Anthias.git
synced 2026-06-11 01:28:23 -04:00
daa1d4bbb6
* fix(server): open SQLite with WAL journaling and a busy timeout - uvicorn, the celery worker, and the viewer share one SQLite file across containers; the stock rollback journal plus a 0s busy timeout raised OperationalError: database is locked fleet-wide (Sentry ANTHIAS-C, ANTHIAS-E, ANTHIAS-G) - timeout=20 waits for the lock instead of failing on the spot - journal_mode=WAL lets readers and the writer coexist; synchronous=NORMAL is the recommended WAL pairing - transaction_mode=IMMEDIATE queues concurrent writers on the busy handler instead of deadlocking on the read-to-write lock upgrade - Add regression tests covering the connection options Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(tests): annotate tmp_path for strict mypy Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
46 lines
1.9 KiB
Python
46 lines
1.9 KiB
Python
"""Tests for the shared SQLite connection options.
|
|
|
|
Regression coverage for the fleet-wide ``OperationalError: database
|
|
is locked`` crashes (Sentry ANTHIAS-C/E/G): uvicorn, the celery
|
|
worker, and the viewer all open the same SQLite file from separate
|
|
containers, so every consumer of the settings module must connect
|
|
with a busy timeout and WAL journaling instead of the stock
|
|
fail-immediately rollback-journal behaviour.
|
|
"""
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from django.conf import settings
|
|
|
|
|
|
class TestSqliteConnectionOptions:
|
|
def test_busy_timeout_is_set(self) -> None:
|
|
options = settings.DATABASES['default']['OPTIONS']
|
|
assert options['timeout'] == 20
|
|
|
|
def test_wal_and_synchronous_pragmas_in_init_command(self) -> None:
|
|
init_command = settings.DATABASES['default']['OPTIONS']['init_command']
|
|
assert 'PRAGMA journal_mode=WAL' in init_command
|
|
assert 'PRAGMA synchronous=NORMAL' in init_command
|
|
|
|
def test_transactions_start_immediate(self) -> None:
|
|
options = settings.DATABASES['default']['OPTIONS']
|
|
assert options['transaction_mode'] == 'IMMEDIATE'
|
|
|
|
def test_init_command_is_valid_sqlite(self, tmp_path: Path) -> None:
|
|
# Execute the exact configured init_command against a scratch
|
|
# database the way Django does on connect — a typo'd pragma
|
|
# would otherwise only surface at service startup on-device.
|
|
init_command = settings.DATABASES['default']['OPTIONS']['init_command']
|
|
conn = sqlite3.connect(tmp_path / 'scratch.db')
|
|
try:
|
|
conn.executescript(init_command)
|
|
journal_mode = conn.execute('PRAGMA journal_mode').fetchone()[0]
|
|
synchronous = conn.execute('PRAGMA synchronous').fetchone()[0]
|
|
finally:
|
|
conn.close()
|
|
assert journal_mode == 'wal'
|
|
# NORMAL reports as 1.
|
|
assert synchronous == 1
|