From 2b4423e3d158f1c4baf9077ac4f4d47d4306b5d8 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Tue, 7 Jul 2026 12:21:54 +0200 Subject: [PATCH] Add SQLite to Postgres restore canary (#43) --- docs/gcp-infra-hardening-20260707.md | 9 + docs/gcp-kb-cloudsql-restore-runbook.md | 37 ++- ops/check_gcp_infra_readiness.py | 45 +++- ops/run_sqlite_postgres_restore_canary.sh | 139 +++++++++++ ops/sqlite_to_postgres_dump.py | 272 ++++++++++++++++++++++ tests/test_sqlite_to_postgres_dump.py | 46 ++++ 6 files changed, 542 insertions(+), 6 deletions(-) create mode 100755 ops/run_sqlite_postgres_restore_canary.sh create mode 100755 ops/sqlite_to_postgres_dump.py create mode 100644 tests/test_sqlite_to_postgres_dump.py diff --git a/docs/gcp-infra-hardening-20260707.md b/docs/gcp-infra-hardening-20260707.md index 767db51..471840e 100644 --- a/docs/gcp-infra-hardening-20260707.md +++ b/docs/gcp-infra-hardening-20260707.md @@ -39,6 +39,10 @@ This records the current GCP hardening contract for `teleo-501523`. - source DB: `/opt/teleo-eval/pipeline/pipeline.db` - source Leo/KB files: `workspaces/*/agents/leo` plus `agent-state` - excludes secrets and logs +- Local SQLite-to-Postgres restore canary exists: + - scripts: `ops/sqlite_to_postgres_dump.py` and `ops/run_sqlite_postgres_restore_canary.sh` + - target: disposable local `postgres:16-alpine` shadow schema + - verifies source SQLite integrity and per-table source/target row-count parity ## How To Build @@ -106,6 +110,11 @@ Database redundancy is not complete. The current project now has a GCP Cloud SQL Do not call the empty `teleo-pgvector-standby` instance redundancy by itself. It only counts after source data, restore/replication, access controls, and query readback are proven. +The local restore canary narrows the remaining gap: the source SQLite backup can +be converted and restored into PostgreSQL with table/row-count parity, but the +same import still needs to run against the GCP Cloud SQL standby through an +approved GCP auth and network path. + ## Communication Posture Current GCP communication hardening is limited to the surfaces already read back: diff --git a/docs/gcp-kb-cloudsql-restore-runbook.md b/docs/gcp-kb-cloudsql-restore-runbook.md index ad4335b..a5d3a49 100644 --- a/docs/gcp-kb-cloudsql-restore-runbook.md +++ b/docs/gcp-kb-cloudsql-restore-runbook.md @@ -43,6 +43,29 @@ The script: This proves source exportability and local restore integrity. It does not prove GCP DB redundancy until a GCP restore/import/query canary also passes. +## Local SQLite-To-Postgres Restore Canary + +Before importing into Cloud SQL, prove that the current SQLite backup can be +converted and restored into PostgreSQL without row loss: + +```bash +SQLITE_BACKUP=./outputs/gcp-infra-hardening-20260707/private-backups/teleo-pipeline-sqlite-.db.gz \ + ops/run_sqlite_postgres_restore_canary.sh +``` + +The canary: + +- generates a PostgreSQL import script with `ops/sqlite_to_postgres_dump.py`; +- recreates a shadow schema in a disposable `postgres:16-alpine` container; +- imports all user tables from the SQLite backup; +- compares source and target row counts for every table; +- writes a proof JSON under `outputs/gcp-infra-hardening-20260707/proofs/`; +- removes only its temporary canary container. + +This is a local restore/parity proof, not GCP redundancy by itself. It is the +preflight that should pass before the same generated import is applied through +the approved Cloud SQL connector/VPC path. + ## Required Proof A successful restore or replication canary must retain: @@ -68,14 +91,15 @@ Use this while the canonical DB remains SQLite on the VPS and we need a GCP rest 1. Run `ops/backup_vps_sqlite_kb.sh`. 2. Upload the resulting SQLite backup and Leo/KB tarball to a versioned GCS bucket such as `gs://teleo-501523-prod-backups/kb-dumps/`. -3. Convert or replay the SQLite data into Cloud SQL with an explicit migration script. Do not run blind string rewrites against the SQLite dump. -4. Install required extensions on Cloud SQL: +3. Run the local SQLite-to-Postgres restore canary above and retain its proof. +4. Convert or replay the SQLite data into Cloud SQL with the explicit generated migration/import script. Do not run blind string rewrites against the SQLite dump. +5. Install required extensions on Cloud SQL: ```sql create extension if not exists vector; ``` -5. From a trusted VPC runtime or Cloud SQL connector path, run readbacks: +6. From a trusted VPC runtime or Cloud SQL connector path, run readbacks: ```sql select current_database(); @@ -83,7 +107,7 @@ select extname, extversion from pg_extension where extname = 'vector'; select schemaname, tablename from pg_tables where schemaname not in ('pg_catalog', 'information_schema') order by 1, 2 limit 50; ``` -6. Retain the SQLite backup hash, GCS object generation, import/conversion operation, query output, and row-count sample. +7. Retain the SQLite backup hash, GCS object generation, import/conversion operation, query output, and row-count sample. ## Logical Replication Path @@ -108,7 +132,10 @@ Retain only redacted connection metadata. Do not commit or paste credentials. ## Current Blocker -As of this run, GCP has the standby target and a repeatable SQLite/KB export script, but it does not have an approved GCP upload/restore credential in the current local session, nor a retained Cloud SQL import/conversion/query proof. That is why the readiness checker still reports: +As of this run, GCP has the standby target, a repeatable SQLite/KB export script, +and a local SQLite-to-Postgres restore canary. It does not have an approved GCP +upload/restore credential in the current local session, nor a retained Cloud SQL +import/query proof. That is why the readiness checker still reports: - `kb_source_restore_access = blocked` - `kb_restore_or_replication = blocked` diff --git a/ops/check_gcp_infra_readiness.py b/ops/check_gcp_infra_readiness.py index 005c863..b02167a 100755 --- a/ops/check_gcp_infra_readiness.py +++ b/ops/check_gcp_infra_readiness.py @@ -9,11 +9,11 @@ accidentally round up to "ready" without live readbacks. from __future__ import annotations import json +import os import subprocess from dataclasses import dataclass from pathlib import Path - PROJECT = "teleo-501523" REGION = "europe-west6" ZONE = "europe-west6-a" @@ -43,6 +43,7 @@ GITHUB_REPOSITORY = "living-ip/teleo-infrastructure" CLOUDSQL_STANDBY_INSTANCE = "teleo-pgvector-standby" CLOUDSQL_STANDBY_DATABASE = "teleo_kb" CLOUDSQL_STANDBY_PASSWORD_SECRET = "gcp-teleo-pgvector-standby-postgres-password" +PROOF_ROOT = Path(os.environ.get("TELEO_GCP_PROOF_ROOT", "outputs/gcp-infra-hardening-20260707/proofs")) @dataclass @@ -516,6 +517,47 @@ def check_kb_source_restore_access() -> Check: return Check("kb_source_restore_access", "pass", "candidate_secret_names=" + ",".join(source_candidates)) +def check_local_sqlite_postgres_restore_canary() -> Check: + candidates = sorted(PROOF_ROOT.glob("sqlite-postgres-restore-canary-*.json")) + if not candidates: + return Check( + "local_sqlite_postgres_restore_canary", + "blocked", + f"no sqlite-postgres restore canary proof found under {PROOF_ROOT}", + required_tier="T2_local_restore_canary", + current_tier="T1_foundation", + ) + + latest = candidates[-1] + proof = json.loads(latest.read_text()) + mismatches = proof.get("mismatched_tables") or [] + source_tables = proof.get("source_table_count") + target_tables = proof.get("target_table_count") + source_rows = proof.get("source_total_rows") + target_rows = proof.get("target_total_rows") + if ( + proof.get("status") == "pass" + and proof.get("source_integrity_check") == "ok" + and not mismatches + and source_tables == target_tables + and source_rows == target_rows + ): + return Check( + "local_sqlite_postgres_restore_canary", + "pass", + f"latest={latest} tables={source_tables} rows={source_rows} postgres_image={proof.get('postgres_image')}", + required_tier="T2_local_restore_canary", + current_tier="T2_local_restore_canary", + ) + return Check( + "local_sqlite_postgres_restore_canary", + "fail", + f"latest={latest} status={proof.get('status')} mismatches={len(mismatches)} source_tables={source_tables} target_tables={target_tables} source_rows={source_rows} target_rows={target_rows}", + required_tier="T2_local_restore_canary", + current_tier="T1_foundation", + ) + + def check_kb_restore_or_replication() -> Check: return Check( "kb_restore_or_replication", @@ -538,6 +580,7 @@ def main() -> int: safe_check("backup_buckets", check_backup_buckets), safe_check("cloud_sql_standby_target", check_cloud_sql_standby), safe_check("kb_source_restore_access", check_kb_source_restore_access), + safe_check("local_sqlite_postgres_restore_canary", check_local_sqlite_postgres_restore_canary), safe_check("kb_restore_or_replication", check_kb_restore_or_replication), ] payload = { diff --git a/ops/run_sqlite_postgres_restore_canary.sh b/ops/run_sqlite_postgres_restore_canary.sh new file mode 100755 index 0000000..195fda1 --- /dev/null +++ b/ops/run_sqlite_postgres_restore_canary.sh @@ -0,0 +1,139 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Restores a Teleo SQLite backup into a disposable local Postgres container and +# compares source/target table row counts. It does not touch GCP or production. + +SQLITE_BACKUP="${SQLITE_BACKUP:-}" +POSTGRES_IMAGE="${POSTGRES_IMAGE:-postgres:16-alpine}" +OUTPUT_ROOT="${OUTPUT_ROOT:-./outputs/gcp-infra-hardening-20260707/private-restore-canaries}" +PROOF_ROOT="${PROOF_ROOT:-./outputs/gcp-infra-hardening-20260707/proofs}" +TS="${TS:-$(date -u +%Y%m%dT%H%M%SZ)}" +SCHEMA="${SCHEMA:-teleo_restore}" +CONTAINER_NAME="${CONTAINER_NAME:-teleo-pg-restore-canary-${TS}}" + +if [[ -z "$SQLITE_BACKUP" ]]; then + echo "SQLITE_BACKUP is required, for example SQLITE_BACKUP=outputs/.../teleo-pipeline-sqlite-...db.gz" >&2 + exit 2 +fi + +if [[ ! -f "$SQLITE_BACKUP" ]]; then + echo "SQLite backup not found: $SQLITE_BACKUP" >&2 + exit 2 +fi + +mkdir -p "$OUTPUT_ROOT" "$PROOF_ROOT" +chmod 700 "$OUTPUT_ROOT" + +WORK_DIR="$OUTPUT_ROOT/sqlite-postgres-restore-canary-${TS}" +mkdir -p "$WORK_DIR" +chmod 700 "$WORK_DIR" + +IMPORT_SQL="$WORK_DIR/teleo-pipeline-postgres-import.sql" +SOURCE_SUMMARY="$WORK_DIR/source-summary.json" +TARGET_COUNTS_SQL="$WORK_DIR/target-counts.sql" +TARGET_COUNTS_CSV="$WORK_DIR/target-counts.csv" +IMPORT_LOG="$WORK_DIR/postgres-import.log" +PROOF_JSON="$PROOF_ROOT/sqlite-postgres-restore-canary-${TS}.json" + +cleanup() { + if [[ "${KEEP_CONTAINER:-0}" != "1" ]]; then + docker rm -f "$CONTAINER_NAME" >/dev/null 2>&1 || true + fi +} +trap cleanup EXIT + +python3 ops/sqlite_to_postgres_dump.py \ + --sqlite-backup "$SQLITE_BACKUP" \ + --output-sql "$IMPORT_SQL" \ + --summary-json "$SOURCE_SUMMARY" \ + --schema "$SCHEMA" \ + --target-counts-sql "$TARGET_COUNTS_SQL" + +docker run --rm -d \ + --name "$CONTAINER_NAME" \ + -e POSTGRES_PASSWORD=postgres \ + -e POSTGRES_DB=teleo_restore_canary \ + "$POSTGRES_IMAGE" >/dev/null + +for _ in $(seq 1 60); do + if docker exec "$CONTAINER_NAME" pg_isready -U postgres -d teleo_restore_canary >/dev/null 2>&1; then + break + fi + sleep 1 +done + +docker exec "$CONTAINER_NAME" pg_isready -U postgres -d teleo_restore_canary >/dev/null + +docker exec -i "$CONTAINER_NAME" \ + psql -v ON_ERROR_STOP=1 -U postgres -d teleo_restore_canary \ + < "$IMPORT_SQL" > "$IMPORT_LOG" + +docker exec -i "$CONTAINER_NAME" \ + psql -v ON_ERROR_STOP=1 -U postgres -d teleo_restore_canary \ + < "$TARGET_COUNTS_SQL" > "$TARGET_COUNTS_CSV" + +python3 - "$SOURCE_SUMMARY" "$TARGET_COUNTS_CSV" "$PROOF_JSON" "$SQLITE_BACKUP" "$IMPORT_SQL" "$POSTGRES_IMAGE" <<'PY' +import csv +import hashlib +import json +import sys +from datetime import UTC, datetime +from pathlib import Path + +source_summary_path = Path(sys.argv[1]) +target_counts_path = Path(sys.argv[2]) +proof_path = Path(sys.argv[3]) +sqlite_backup = Path(sys.argv[4]) +import_sql = Path(sys.argv[5]) +postgres_image = sys.argv[6] + +source_summary = json.loads(source_summary_path.read_text()) +source_counts = {table["name"]: int(table["row_count"]) for table in source_summary["tables"]} +with target_counts_path.open(newline="") as handle: + target_counts = {row["table_name"]: int(row["row_count"]) for row in csv.DictReader(handle)} + +all_tables = sorted(set(source_counts) | set(target_counts)) +mismatches = [ + { + "table": table, + "source_rows": source_counts.get(table), + "target_rows": target_counts.get(table), + } + for table in all_tables + if source_counts.get(table) != target_counts.get(table) +] + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + +proof = { + "artifact": "sqlite_postgres_restore_canary", + "generated_at_utc": datetime.now(UTC).isoformat(), + "status": "pass" if not mismatches and source_summary["source_integrity_check"] == "ok" else "fail", + "source_sqlite_backup": str(sqlite_backup), + "source_sqlite_backup_sha256": sha256(sqlite_backup), + "postgres_image": postgres_image, + "source_integrity_check": source_summary["source_integrity_check"], + "postgres_schema": source_summary["schema"], + "source_table_count": len(source_counts), + "target_table_count": len(target_counts), + "source_total_rows": sum(source_counts.values()), + "target_total_rows": sum(target_counts.values()), + "mismatched_tables": mismatches, + "conversion_notes": source_summary["conversion_notes"], + "conversion_stats": source_summary["conversion_stats"], + "private_artifacts": { + "source_summary_json": str(source_summary_path), + "postgres_import_sql": str(import_sql), + "target_counts_csv": str(target_counts_path), + }, +} +proof_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n") +print(json.dumps(proof, indent=2, sort_keys=True)) +sys.exit(0 if proof["status"] == "pass" else 1) +PY diff --git a/ops/sqlite_to_postgres_dump.py b/ops/sqlite_to_postgres_dump.py new file mode 100755 index 0000000..d097026 --- /dev/null +++ b/ops/sqlite_to_postgres_dump.py @@ -0,0 +1,272 @@ +#!/usr/bin/env python3 +"""Generate a PostgreSQL restore script from a Teleo SQLite backup. + +This is a shadow-restore converter for restore drills while the canonical +runtime database is still SQLite. It preserves tables and rows, infers safe +Postgres column types from actual SQLite storage classes, and avoids blind +string rewrites of `.dump` output. +""" + +from __future__ import annotations + +import argparse +import gzip +import json +import shutil +import sqlite3 +import tempfile +from contextlib import contextmanager +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +PG_NUMERIC_TYPES = {"bigint", "double precision"} + + +def pg_quote_identifier(value: str) -> str: + return '"' + value.replace('"', '""') + '"' + + +def sqlite_quote_identifier(value: str) -> str: + return '"' + value.replace('"', '""') + '"' + + +def copy_escape(value: Any, pg_type: str, stats: dict[str, int]) -> str: + if value is None: + return r"\N" + if isinstance(value, bytes): + text = value.hex() + stats["blob_values_hex_encoded"] += 1 + elif pg_type in PG_NUMERIC_TYPES: + text = str(value) + else: + text = str(value) + + if "\x00" in text: + stats["nul_chars_replaced"] += text.count("\x00") + text = text.replace("\x00", r"\u0000") + + return ( + text.replace("\\", "\\\\") + .replace("\t", r"\t") + .replace("\n", r"\n") + .replace("\r", r"\r") + ) + + +def infer_pg_type(declared_type: str | None, storage_counts: dict[str, int]) -> tuple[str, str]: + storage_types = {name for name, count in storage_counts.items() if count and name != "null"} + declared = (declared_type or "").upper() + + if not storage_types: + if "INT" in declared: + return "bigint", "all_null_declared_integer" + if any(token in declared for token in ("REAL", "FLOA", "DOUB")): + return "double precision", "all_null_declared_real" + return "text", "all_null_text_default" + + if storage_types == {"integer"}: + return "bigint", "sqlite_integer_storage" + if storage_types <= {"integer", "real"}: + return "double precision", "sqlite_numeric_storage" + return "text", "sqlite_text_or_mixed_storage" + + +def read_storage_counts(conn: sqlite3.Connection, table: str, column: str) -> dict[str, int]: + sql = ( + f"select typeof({sqlite_quote_identifier(column)}) as storage_type, count(*) " + f"from {sqlite_quote_identifier(table)} group by storage_type" + ) + return {str(row[0]): int(row[1]) for row in conn.execute(sql)} + + +def user_tables(conn: sqlite3.Connection) -> list[dict[str, str]]: + rows = conn.execute( + """ + select name, sql + from sqlite_master + where type = 'table' + and name not like 'sqlite_%' + order by name + """ + ).fetchall() + return [{"name": str(row[0]), "sqlite_sql": row[1] or ""} for row in rows] + + +@contextmanager +def sqlite_database_path(path: Path): + if path.suffix != ".gz": + yield path + return + + tmpdir = Path(tempfile.mkdtemp(prefix="teleo-sqlite-restore-")) + try: + db_path = tmpdir / "pipeline.db" + with gzip.open(path, "rb") as source, db_path.open("wb") as target: + shutil.copyfileobj(source, target) + yield db_path + finally: + shutil.rmtree(tmpdir, ignore_errors=True) + + +def build_table_metadata(conn: sqlite3.Connection, table: str) -> dict[str, Any]: + columns: list[dict[str, Any]] = [] + for cid, name, declared_type, notnull, default_value, primary_key in conn.execute( + f"pragma table_info({sqlite_quote_identifier(table)})" + ): + storage_counts = read_storage_counts(conn, table, name) + pg_type, type_reason = infer_pg_type(declared_type, storage_counts) + columns.append( + { + "cid": cid, + "name": name, + "sqlite_declared_type": declared_type or "", + "sqlite_notnull": bool(notnull), + "sqlite_default": default_value, + "sqlite_primary_key_position": primary_key, + "postgres_type": pg_type, + "postgres_type_reason": type_reason, + "sqlite_storage_counts": storage_counts, + } + ) + row_count = conn.execute(f"select count(*) from {sqlite_quote_identifier(table)}").fetchone()[0] + return {"name": table, "row_count": int(row_count), "columns": columns} + + +def write_target_counts_sql(path: Path, schema: str, tables: list[dict[str, Any]]) -> None: + selects = [] + for table in tables: + table_name = table["name"] + table_literal = table_name.replace("'", "''") + table_ref = f"{pg_quote_identifier(schema)}.{pg_quote_identifier(table_name)}" + selects.append(f"select '{table_literal}' as table_name, count(*)::bigint as row_count from {table_ref}") + body = "\nunion all\n".join(selects) or "select '' as table_name, 0::bigint as row_count where false" + path.write_text( + "copy (\n" + f"{body}\n" + "order by table_name\n" + ") to stdout with (format csv, header true);\n" + ) + + +def convert_sqlite_to_postgres( + sqlite_path: Path, + output_sql: Path, + summary_json: Path, + schema: str, + target_counts_sql: Path | None = None, +) -> dict[str, Any]: + output_sql.parent.mkdir(parents=True, exist_ok=True) + summary_json.parent.mkdir(parents=True, exist_ok=True) + if target_counts_sql: + target_counts_sql.parent.mkdir(parents=True, exist_ok=True) + + with sqlite_database_path(sqlite_path) as db_path: + conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) + conn.row_factory = sqlite3.Row + integrity = conn.execute("pragma integrity_check").fetchone()[0] + tables = [] + table_names = [table["name"] for table in user_tables(conn)] + for table_name in table_names: + tables.append(build_table_metadata(conn, table_name)) + + stats = {"blob_values_hex_encoded": 0, "nul_chars_replaced": 0} + with output_sql.open("w", encoding="utf-8", newline="\n") as out: + out.write("\\set ON_ERROR_STOP on\n") + out.write("begin;\n") + out.write(f"drop schema if exists {pg_quote_identifier(schema)} cascade;\n") + out.write(f"create schema {pg_quote_identifier(schema)};\n") + out.write(f"set search_path to {pg_quote_identifier(schema)};\n") + + for table in tables: + table_ref = f"{pg_quote_identifier(schema)}.{pg_quote_identifier(table['name'])}" + column_defs = [ + f"{pg_quote_identifier(column['name'])} {column['postgres_type']}" + for column in table["columns"] + ] + out.write(f"create table {table_ref} (\n " + ",\n ".join(column_defs) + "\n);\n") + + if not table["columns"]: + continue + + column_list = ", ".join(pg_quote_identifier(column["name"]) for column in table["columns"]) + out.write(f"copy {table_ref} ({column_list}) from stdin with (format text, delimiter E'\\t', null '\\N');\n") + + select_columns = ", ".join(sqlite_quote_identifier(column["name"]) for column in table["columns"]) + for row in conn.execute(f"select {select_columns} from {sqlite_quote_identifier(table['name'])}"): + fields = [ + copy_escape(row[column["name"]], column["postgres_type"], stats) + for column in table["columns"] + ] + out.write("\t".join(fields)) + out.write("\n") + out.write("\\.\n") + + out.write("commit;\n") + out.write("analyze;\n") + + if target_counts_sql: + write_target_counts_sql(target_counts_sql, schema, tables) + + summary = { + "artifact": "teleo_sqlite_to_postgres_dump", + "generated_at_utc": datetime.now(UTC).isoformat(), + "source_sqlite_path": str(sqlite_path), + "source_integrity_check": integrity, + "schema": schema, + "table_count": len(tables), + "source_total_rows": sum(table["row_count"] for table in tables), + "tables": tables, + "conversion_notes": { + "mode": "shadow_restore", + "constraints_and_indexes": "not_recreated", + "blob_storage": "hex_encoded_text_when_needed", + "postgres_types": "inferred_from_sqlite_storage_classes", + }, + "conversion_stats": stats, + "output_sql": str(output_sql), + "target_counts_sql": str(target_counts_sql) if target_counts_sql else None, + } + summary_json.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n") + return summary + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--sqlite-backup", required=True, type=Path, help="SQLite .db or .db.gz backup") + parser.add_argument("--output-sql", required=True, type=Path, help="Postgres import SQL path") + parser.add_argument("--summary-json", required=True, type=Path, help="Non-secret conversion summary JSON") + parser.add_argument("--schema", default="teleo_restore", help="Postgres schema to recreate") + parser.add_argument("--target-counts-sql", type=Path, help="Optional psql query file for target row counts") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + summary = convert_sqlite_to_postgres( + sqlite_path=args.sqlite_backup, + output_sql=args.output_sql, + summary_json=args.summary_json, + schema=args.schema, + target_counts_sql=args.target_counts_sql, + ) + print( + json.dumps( + { + "artifact": summary["artifact"], + "source_integrity_check": summary["source_integrity_check"], + "schema": summary["schema"], + "table_count": summary["table_count"], + "source_total_rows": summary["source_total_rows"], + "output_sql": summary["output_sql"], + "target_counts_sql": summary["target_counts_sql"], + }, + indent=2, + sort_keys=True, + ) + ) + return 0 if summary["source_integrity_check"] == "ok" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_sqlite_to_postgres_dump.py b/tests/test_sqlite_to_postgres_dump.py new file mode 100644 index 0000000..1fdfc18 --- /dev/null +++ b/tests/test_sqlite_to_postgres_dump.py @@ -0,0 +1,46 @@ +import json +import sqlite3 + +from ops.sqlite_to_postgres_dump import convert_sqlite_to_postgres + + +def test_sqlite_to_postgres_dump_escapes_and_counts(tmp_path): + db_path = tmp_path / "source.db" + conn = sqlite3.connect(db_path) + conn.execute("create table audit_log (id integer primary key, detail text, score real, payload blob)") + conn.execute( + "insert into audit_log (id, detail, score, payload) values (?, ?, ?, ?)", + (1, "line one\nline two\tbackslash\\literal \\N", 1.25, b"\x00\x01abc"), + ) + conn.execute("insert into audit_log (id, detail, score, payload) values (?, ?, ?, ?)", (2, None, None, None)) + conn.execute("create table mixed (value integer)") + conn.execute("insert into mixed (value) values (?)", ("not actually integer",)) + conn.commit() + conn.close() + + sql_path = tmp_path / "restore.sql" + summary_path = tmp_path / "summary.json" + target_counts_path = tmp_path / "counts.sql" + + summary = convert_sqlite_to_postgres( + sqlite_path=db_path, + output_sql=sql_path, + summary_json=summary_path, + schema="teleo_restore", + target_counts_sql=target_counts_path, + ) + + sql = sql_path.read_text() + assert 'create schema "teleo_restore";' in sql + assert 'copy "teleo_restore"."audit_log"' in sql + assert r"line one\nline two\tbackslash\\literal \\N" in sql + assert "0001616263" in sql + assert summary["source_integrity_check"] == "ok" + assert summary["table_count"] == 2 + assert summary["source_total_rows"] == 3 + + stored_summary = json.loads(summary_path.read_text()) + mixed_column = stored_summary["tables"][1]["columns"][0] + assert mixed_column["name"] == "value" + assert mixed_column["postgres_type"] == "text" + assert target_counts_path.read_text().startswith("copy (")