Harden runtime and add backup foundation

This commit is contained in:
2026-04-06 10:15:00 -04:00
parent 9715fe3143
commit c9757e313a
11 changed files with 331 additions and 10 deletions

71
tests/test_backup.py Normal file
View File

@@ -0,0 +1,71 @@
"""Tests for runtime backup creation."""
import json
import sqlite3
from datetime import UTC, datetime
import atocore.config as config
from atocore.models.database import init_db
from atocore.ops.backup import create_runtime_backup
def test_create_runtime_backup_copies_db_and_registry(tmp_path, monkeypatch):
monkeypatch.setenv("ATOCORE_DATA_DIR", str(tmp_path / "data"))
monkeypatch.setenv("ATOCORE_BACKUP_DIR", str(tmp_path / "backups"))
monkeypatch.setenv(
"ATOCORE_PROJECT_REGISTRY_PATH", str(tmp_path / "config" / "project-registry.json")
)
registry_path = tmp_path / "config" / "project-registry.json"
registry_path.parent.mkdir(parents=True)
registry_path.write_text('{"projects":[{"id":"p01-example","aliases":[],"ingest_roots":[{"source":"vault","subpath":"incoming/projects/p01-example"}]}]}\n', encoding="utf-8")
original_settings = config.settings
try:
config.settings = config.Settings()
init_db()
with sqlite3.connect(str(config.settings.db_path)) as conn:
conn.execute("INSERT INTO projects (id, name) VALUES (?, ?)", ("p01", "P01 Example"))
conn.commit()
result = create_runtime_backup(datetime(2026, 4, 6, 18, 0, 0, tzinfo=UTC))
finally:
config.settings = original_settings
db_snapshot = tmp_path / "backups" / "snapshots" / "20260406T180000Z" / "db" / "atocore.db"
registry_snapshot = (
tmp_path / "backups" / "snapshots" / "20260406T180000Z" / "config" / "project-registry.json"
)
metadata_path = (
tmp_path / "backups" / "snapshots" / "20260406T180000Z" / "backup-metadata.json"
)
assert result["db_snapshot_path"] == str(db_snapshot)
assert db_snapshot.exists()
assert registry_snapshot.exists()
assert metadata_path.exists()
with sqlite3.connect(str(db_snapshot)) as conn:
row = conn.execute("SELECT name FROM projects WHERE id = ?", ("p01",)).fetchone()
assert row[0] == "P01 Example"
metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
assert metadata["registry_snapshot_path"] == str(registry_snapshot)
def test_create_runtime_backup_handles_missing_registry(tmp_path, monkeypatch):
monkeypatch.setenv("ATOCORE_DATA_DIR", str(tmp_path / "data"))
monkeypatch.setenv("ATOCORE_BACKUP_DIR", str(tmp_path / "backups"))
monkeypatch.setenv(
"ATOCORE_PROJECT_REGISTRY_PATH", str(tmp_path / "config" / "project-registry.json")
)
original_settings = config.settings
try:
config.settings = config.Settings()
init_db()
result = create_runtime_backup(datetime(2026, 4, 6, 19, 0, 0, tzinfo=UTC))
finally:
config.settings = original_settings
assert result["registry_snapshot_path"] == ""

49
tests/test_database.py Normal file
View File

@@ -0,0 +1,49 @@
"""Tests for SQLite connection pragmas and runtime behavior."""
import sqlite3
import atocore.config as config
from atocore.models.database import get_connection, init_db
def test_get_connection_applies_busy_timeout_and_wal(tmp_path, monkeypatch):
monkeypatch.setenv("ATOCORE_DATA_DIR", str(tmp_path / "data"))
monkeypatch.setenv("ATOCORE_DB_BUSY_TIMEOUT_MS", "7000")
original_settings = config.settings
try:
config.settings = config.Settings()
init_db()
with get_connection() as conn:
busy_timeout = conn.execute("PRAGMA busy_timeout").fetchone()[0]
journal_mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
foreign_keys = conn.execute("PRAGMA foreign_keys").fetchone()[0]
finally:
config.settings = original_settings
assert busy_timeout == 7000
assert str(journal_mode).lower() == "wal"
assert foreign_keys == 1
def test_get_connection_uses_configured_timeout_value(tmp_path, monkeypatch):
monkeypatch.setenv("ATOCORE_DATA_DIR", str(tmp_path / "data"))
monkeypatch.setenv("ATOCORE_DB_BUSY_TIMEOUT_MS", "2500")
original_settings = config.settings
original_connect = sqlite3.connect
calls = []
def fake_connect(*args, **kwargs):
calls.append(kwargs.get("timeout"))
return original_connect(*args, **kwargs)
try:
config.settings = config.Settings()
monkeypatch.setattr("atocore.models.database.sqlite3.connect", fake_connect)
init_db()
finally:
config.settings = original_settings
assert calls
assert calls[0] == 2.5