46 lines
1.8 KiB
Python
46 lines
1.8 KiB
Python
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 (")
|