Test, Benchmark, Back Up, Restore, and Secure the Database

import sqlite3import tempfilefrom pathlib import Pathwith tempfile.TemporaryDirectory() as td:    db = Path(td) / "fieldnotes.sqlite"    con = open_db(db)    migrate(con)    con.execute("INSERT INTO site(site_code,site_name) VALUES('PLANT-A','Plant A')")    con.execute("INSERT INTO device(site_id,device_code,device_name) VALUES(1,'P-007','Pump 007')")    # Invalid JSON must be rejected by the database, not only the UI.    try:        con.execute("UPDATE device SET metadata_json=? WHERE device_id=1", ("{bad",))        raise AssertionError("invalid JSON was accepted")    except sqlite3.IntegrityError:        pass    before = con.execute("SELECT count(*) FROM inspection").fetchone()[0]    con.execute("BEGIN IMMEDIATE")    try:        con.execute("""          INSERT INTO inspection(device_id,request_id,technician_name,                                 started_at,outcome,summary)          VALUES(1,'inject-1','Tech A','2026-08-12T08:00:00Z','pass','demo')        """)        raise RuntimeError("injected")    except RuntimeError:        con.execute("ROLLBACK")    after = con.execute("SELECT count(*) FROM inspection").fetchone()[0]    assert after == before    con.close()

kwargs = dict(    request_id="req-0001",    device_id=1,    technician_name="Tech A",    started_at="2026-08-12T08:00:00Z",    finished_at="2026-08-12T08:10:00Z",    outcome="follow_up",    summary="seal temperature elevated",    measurements={"temperature_c": 83.2},    severity="warning",)first, created1 = record_inspection(con, **kwargs)second, created2 = record_inspection(con, **kwargs)assert created1 is Trueassert created2 is Falseassert first["inspection_id"] == second["inspection_id"]assert con.execute(    "SELECT count(*) FROM inspection WHERE request_id='req-0001'").fetchone()[0] == 1assert con.execute(    "SELECT count(*) FROM sync_outbox WHERE event_id='inspection:req-0001'").fetchone()[0] == 1

SELECT sqlite_version();SELECT sqlite_source_id();PRAGMA journal_mode;PRAGMA synchronous;PRAGMA page_size;PRAGMA cache_size;PRAGMA mmap_size;PRAGMA compile_options;SELECT count(*) AS devices FROM device;SELECT count(*) AS inspections FROM inspection;SELECT count(*) AS notes FROM maintenance_note;EXPLAIN QUERY PLANSELECT inspection_id, started_at, outcome, summaryFROM inspectionWHERE device_id=?ORDER BY started_at DESCLIMIT 20;

from statistics import medianfrom time import perf_counterdef time_recent_history(con, device_id, trials=25):    samples = []    for _ in range(trials):        t0 = perf_counter()        rows = con.execute(            """SELECT inspection_id, started_at, outcome, summary               FROM inspection               WHERE device_id=?               ORDER BY started_at DESC LIMIT 20""",            (device_id,),        ).fetchall()        samples.append(perf_counter() - t0)    return {"rows": len(rows), "median_s": median(samples),            "min_s": min(samples), "max_s": max(samples)}

DROP INDEX idx_inspection_device_started;EXPLAIN QUERY PLANSELECT inspection_id, started_at, outcomeFROM inspectionWHERE device_id=17ORDER BY started_at DESC LIMIT 20;CREATE INDEX idx_inspection_device_startedON inspection(device_id, started_at DESC);PRAGMA optimize;EXPLAIN QUERY PLANSELECT inspection_id, started_at, outcomeFROM inspectionWHERE device_id=17ORDER BY started_at DESC LIMIT 20;

from datetime import datetime, timezonefrom pathlib import Pathimport sqlite3stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")backup_dir = Path("backups")backup_dir.mkdir(exist_ok=True)candidate = backup_dir / f"fieldnotes-{stamp}.sqlite.partial"final = backup_dir / f"fieldnotes-{stamp}.sqlite"source = open_db("fieldnotes.sqlite")dest = sqlite3.connect(candidate, autocommit=True)source.backup(dest)dest.close(); source.close()check = sqlite3.connect(candidate, autocommit=True)assert check.execute("PRAGMA quick_check").fetchone()[0] == "ok"assert check.execute("PRAGMA integrity_check").fetchone()[0] == "ok"assert check.execute("PRAGMA foreign_key_check").fetchall() == []expected = check.execute("SELECT count(*) FROM inspection").fetchone()[0]check.close()candidate.replace(final)print(final, "verified inspections=", expected)

PRAGMA user_version;PRAGMA quick_check;PRAGMA integrity_check;PRAGMA foreign_key_check;SELECT count(*) FROM site;SELECT count(*) FROM device;SELECT count(*) FROM inspection;SELECT count(*) FROM maintenance_note;SELECT count(*) FROM sync_outbox WHERE delivered_at IS NULL;-- Domain consistency: every inspection device must be queryable through FK.SELECT count(*) AS orphan_inspectionsFROM inspection AS iLEFT JOIN device AS d ON d.device_id=i.device_idWHERE d.device_id IS NULL;-- Idempotency should remain unique after restore.SELECT request_id, count(*)FROM inspectionGROUP BY request_idHAVING count(*) > 1;

FieldNotes Release Acceptance=============================Application version: __________________SQLite version/source id: _____________Compile/capability snapshot attached: [ ]DB user_version: ______________________Correctness[ ] schema manifest matches release[ ] constraint-negative tests pass[ ] idempotency replay test passes[ ] transaction failure injection leaves no partial state[ ] concurrency test reproduces + recovers from BUSYHealth[ ] quick_check = ok[ ] integrity_check = ok[ ] foreign_key_check = no rows[ ] domain consistency queries = expectedPerformance (target device)[ ] hot-query plans recorded[ ] read/write benchmark dataset recorded[ ] product latency/throughput SLOs met[ ] no unexplained busy/retry exhaustionRecovery[ ] online backup candidate verified[ ] restore drill completed[ ] restored counts/domain checks match policy[ ] off-device copy/retention verifiedSecurity[ ] parameter binding review pass[ ] file/backup permission review pass[ ] trusted_schema/extension policy pass[ ] logs/secrets review passRelease decision: PASS / FAILOwner: __________________ Date: _______