83 lines
3.6 KiB
Python
83 lines
3.6 KiB
Python
"""SQL dialect adapter for the SQLite/Postgres dual backend."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from sql_compat import adapt_sql, split_sql, sqlite_schema_to_postgres, use_postgres
|
|
|
|
|
|
def expect(ok: bool, message: str) -> None:
|
|
if not ok:
|
|
raise SystemExit(f"FAIL: {message}")
|
|
print(f"OK {message}")
|
|
|
|
|
|
def main() -> None:
|
|
os.environ.pop("KANSHO_DB_BACKEND", None)
|
|
os.environ.pop("KANSHO_DB_PATH", None)
|
|
os.environ.pop("DB_HOST", None)
|
|
os.environ.pop("DATABASE_URL", None)
|
|
expect(use_postgres() is False, "default engine is sqlite")
|
|
|
|
os.environ["KANSHO_DB_PATH"] = "/tmp/isolated.sqlite"
|
|
os.environ["DB_HOST"] = "postgres"
|
|
expect(use_postgres() is False, "KANSHO_DB_PATH without backend keeps Windows local on sqlite")
|
|
os.environ["KANSHO_DB_BACKEND"] = "postgres"
|
|
expect(use_postgres() is True, "explicit postgres backend wins")
|
|
os.environ["KANSHO_DB_BACKEND"] = "sqlite"
|
|
expect(use_postgres() is False, "explicit sqlite backend wins")
|
|
os.environ.pop("KANSHO_DB_BACKEND", None)
|
|
os.environ.pop("KANSHO_DB_PATH", None)
|
|
expect(use_postgres() is True, "DB_HOST alone selects postgres")
|
|
os.environ.pop("DB_HOST", None)
|
|
|
|
translated = sqlite_schema_to_postgres("created TEXT NOT NULL DEFAULT (datetime('now'))")
|
|
expect("CURRENT_TIMESTAMP::text" in translated, "schema datetime default")
|
|
expect("datetime('now')" not in translated, "no sqlite datetime left in schema")
|
|
|
|
ignore = adapt_sql("INSERT OR IGNORE INTO tiers (id) VALUES (?)")
|
|
expect("INSERT INTO tiers" in ignore, "insert or ignore becomes insert")
|
|
expect("ON CONFLICT DO NOTHING" in ignore, "conflict do nothing appended")
|
|
expect("%s" in ignore and "?" not in ignore, "placeholders converted")
|
|
|
|
upsert = adapt_sql(
|
|
"INSERT INTO app_settings (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value"
|
|
)
|
|
expect("ON CONFLICT (key)" in upsert, "on conflict parenthesis spacing")
|
|
expect(upsert.count("ON CONFLICT") == 1, "existing on conflict not duplicated")
|
|
expect("%s" in upsert, "upsert placeholders")
|
|
|
|
quoted = adapt_sql("SELECT '?' AS q, name FROM profiles WHERE id = ?")
|
|
expect("'?'" in quoted, "question mark inside quotes kept")
|
|
expect(quoted.endswith("%s") or quoted.rstrip().endswith("%s"), "trailing placeholder converted")
|
|
|
|
case_sql = adapt_sql("UPDATE t SET x = CASE WHEN ? = 1 THEN ? ELSE x END")
|
|
expect("CASE WHEN %s = 1 THEN %s ELSE x END" in case_sql, "boolean-safe case placeholder")
|
|
|
|
rowid_sql = adapt_sql("SELECT e.* FROM journal_entries e ORDER BY e.created, e.rowid")
|
|
expect("e.rowid" not in rowid_sql, "sqlite rowid removed for postgres")
|
|
expect("ORDER BY e.created, e.id" in rowid_sql, "rowid becomes primary id")
|
|
|
|
parts = split_sql("CREATE TABLE a (id TEXT); CREATE TABLE b (id TEXT);")
|
|
expect(parts == ["CREATE TABLE a (id TEXT)", "CREATE TABLE b (id TEXT)"], "split two statements")
|
|
schema = (ROOT / "schema.sql").read_text(encoding="utf-8")
|
|
pg_schema = sqlite_schema_to_postgres(schema)
|
|
expect("datetime('now')" not in pg_schema, "translated schema has no sqlite datetime")
|
|
expect("CURRENT_TIMESTAMP::text" in pg_schema, "translated schema uses timestamp text default")
|
|
expect(len(split_sql(pg_schema)) > 40, "schema splits into many statements")
|
|
from sqlite_to_postgres import TABLES
|
|
|
|
tables = re.findall(r"CREATE TABLE IF NOT EXISTS (\w+)", schema)
|
|
expect(tables == list(TABLES), "import table order matches schema.sql")
|
|
print("sql compat: OK")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|