diff --git a/lambda/backup-verification/app.py b/lambda/backup-verification/app.py index 2dec9aa..4935e03 100644 --- a/lambda/backup-verification/app.py +++ b/lambda/backup-verification/app.py @@ -124,11 +124,12 @@ def _restore_test(): names = tf.getnames() results.append({"pass": True, "msg": f"Restore test: Archive OK — {len(names)} entries"}) - db_entries = [n for n in names if n.endswith(".sqlite3") or n.endswith(".sql")] - if db_entries: + sqlite_entries = [n for n in names if n.endswith(".sqlite3")] + sql_entries = [n for n in names if n.endswith(".sql")] + if sqlite_entries: import sqlite3 as sqlite_mod - tf.extract(db_entries[0], path=tmpdir, filter="data") - db_path = os.path.join(tmpdir, db_entries[0]) + tf.extract(sqlite_entries[0], path=tmpdir, filter="data") + db_path = os.path.join(tmpdir, sqlite_entries[0]) conn = sqlite_mod.connect(db_path) result = conn.execute("PRAGMA integrity_check").fetchone() conn.close() @@ -136,8 +137,26 @@ def _restore_test(): results.append({"pass": True, "msg": "Restore test: SQLite integrity OK"}) else: results.append({"pass": False, "msg": f"Restore test: SQLite integrity FAILED — {result[0]}"}) + elif sql_entries: + import sqlite3 as sqlite_mod + tf.extract(sql_entries[0], path=tmpdir, filter="data") + sql_path = os.path.join(tmpdir, sql_entries[0]) + with open(sql_path, "r") as f: + sql_text = f.read() + if len(sql_text) < 100: + results.append({"pass": False, "msg": f"Restore test: SQL dump suspiciously small ({len(sql_text)} bytes)"}) + else: + db_path = os.path.join(tmpdir, "restore-test.db") + conn = sqlite_mod.connect(db_path) + conn.executescript(sql_text) + result = conn.execute("PRAGMA integrity_check").fetchone() + conn.close() + if result[0] == "ok": + results.append({"pass": True, "msg": "Restore test: SQL dump import + integrity OK"}) + else: + results.append({"pass": False, "msg": f"Restore test: Integrity FAILED after SQL import — {result[0]}"}) else: - results.append({"pass": False, "msg": "Restore test: No SQLite DB file found in archive"}) + results.append({"pass": False, "msg": "Restore test: No database file found in archive"}) except tarfile.TarError as e: results.append({"pass": False, "msg": f"Restore test: Archive extraction FAILED — {e}"}) except Exception as e: