Postgres real backend: psycopg adapter, schema, migration

Replace the PostgresDatabase skeleton with a real implementation:
psycopg (v3) first, psycopg2 second, lazy load, missing driver
still raises PostgresDriverMissingError with no silent SQLite
fallback. A sqlite3-style adapter (PgConnection/PgCursor/PgRow)
translates ? placeholders to %s outside string literals, maps
UNIQUE/FK violations, lock errors and connection failures to
explicit Postgres error types, and handles INSERT OR REPLACE,
PRAGMA table_info, AUTOINCREMENT and FTS5 refusal. POSTGRES_SCHEMA
covers all business tables with tasks result columns; server
init_db uses it under CLOUD_DB_BACKEND=postgres, SQLite path
unchanged. migrate_sqlite_to_postgres.py copies rows with
ON CONFLICT DO NOTHING and reseeds identity sequences.
Tests: cloud 239 (235 pass, 4 skipped without a real server),
host 27 pass; real-server tests skip unless CLOUD_TEST_PG_DSN
is set.
This commit is contained in:
host committed 2026-10-26 03:23:06 +08:00
1 parent a19e77ee44
commit dc3f572b2b
10 files changed
+1542 -66

No files matched your search

+88 -16
View File
@@ -207,22 +207,94 @@ class PostgresSkeletonTestCase(_EnvCleanMixin, unittest.TestCase):
self.assertEqual(db2.host, "explicit-host")
self.assertEqual(db2.dbname, "env-db")
def test_postgres_methods_are_skeleton_not_fake_success(self):
# 绕过 __init__(驱动探测)构造骨架实例,验证方法体本身
# 就是明确的 NotImplementedError,而不是假装成功。
skeleton = object.__new__(PostgresDatabase)
for name in ("connect", "read_all", "transaction",
"is_locked_error"):
method = getattr(skeleton, name)
with self.assertRaises(NotImplementedError):
if name in ("connect", "transaction"):
method(None) if name == "transaction" else method()
elif name == "is_locked_error":
method(ValueError("x"))
else:
method(None, "SELECT 1")
with self.assertRaises(NotImplementedError):
skeleton.execute_with_retry(None, "SELECT 1", ())
def test_postgres_methods_are_implemented_via_fake_driver(self):
# 骨架已由真实实现取代:用 fake 驱动(不连真实 Postgres)
# 验证 connect / read_all / transaction / is_locked_error
# 均可用,且驱动唯一冲突被映射成模块内完整性错误。
from cloud.db import (
PgConnection,
PostgresIntegrityError,
PostgresOperationalError,
)
class _FakeUnique(Exception):
pass
class _FakeCursor:
def __init__(self):
self.executed = []
self.description = [("id",), ("name",)]
def execute(self, sql, params=()):
self.executed.append((sql, params))
if "dup" in sql:
raise _FakeUnique("duplicate key")
return self
def fetchall(self):
return [("t1", "hello")]
def fetchone(self):
return ("t1", "hello")
@property
def rowcount(self):
return 1
def close(self):
pass
class _FakeConn:
def __init__(self):
self.cursor_obj = _FakeCursor()
self.committed = False
self.rolled_back = False
def cursor(self):
return self.cursor_obj
def commit(self):
self.committed = True
def rollback(self):
self.rolled_back = True
def close(self):
pass
class _FakeDriver:
UniqueViolation = _FakeUnique
OperationalError = Exception
Error = Exception
def __init__(self):
self.conn = _FakeConn()
def connect(self, **kwargs):
return self.conn
fake = _FakeDriver()
with mock.patch("cloud.db._load_postgres_driver",
return_value=("psycopg", fake)):
db = PostgresDatabase(host="test-host", dbname="test-db")
conn = db.connect()
self.assertIsInstance(conn, PgConnection)
rows = db.read_all(conn, "SELECT * FROM t WHERE id = ?",
("t1",))
self.assertEqual(rows[0]["name"], "hello")
self.assertEqual(rows[0][0], "t1")
# 唯一冲突 -> 模块内错误,不是驱动原始异常、不是 sqlite 异常。
with self.assertRaises(PostgresIntegrityError):
db.execute_with_retry(conn, "SELECT dup", ())
# 事务回滚:第二条失败后 rollback 被调用且异常原样抛出。
with self.assertRaises(PostgresIntegrityError):
with db.transaction(conn):
conn.execute("SELECT dup", ())
self.assertTrue(fake.conn.rolled_back)
# 锁判定:本模块锁错误 True,其它错误 False。
self.assertTrue(db.is_locked_error(
PostgresOperationalError("deadlock", locked=True)))
self.assertFalse(db.is_locked_error(ValueError("x")))
def test_postgres_repr_never_contains_password(self):
skeleton = object.__new__(PostgresDatabase)
+336
View File
@@ -0,0 +1,336 @@
"""Postgres 后端单测:适配层(fake 驱动,常跑)+ 真实 PG(skip)。
分两类,泾渭分明:
- FakeDriver 类:不连任何真实 Postgres,验证占位符转换、
行按名访问、唯一冲突映射、事务回滚、锁重试、schema 语句
覆盖——这些在无 Postgres 的机器(含 CI runner)上也必须通过。
- RealPostgres 类:需要真实 Postgres 时才跑,判定条件见
_real_pg_params():须显式设置 CLOUD_TEST_PG_DSN(形如
postgres://user:pass@host:port/dbname)且主机端口 2 秒内
可达,否则 skipTest,绝不 fail。DSN 必须指向一次性测试库,
本文件不在其中建生产表之外的任何东西(schema 即业务表)。
测试里出现的密码一律是 fake/test 占位("test-placeholder"),
绝不写真实凭证;DSN 只从环境变量读,测试不打印它。
"""
import os
import socket
import sys
import unittest
from pathlib import Path
from unittest import mock
from urllib.parse import unquote, urlparse
try:
from cloud.db import (
POSTGRES_SCHEMA,
PgConnection,
PostgresDatabase,
PostgresDriverMissingError,
PostgresIntegrityError,
PostgresOperationalError,
_convert_placeholders,
_translate_sql,
init_postgres_schema,
)
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.db import (
POSTGRES_SCHEMA,
PgConnection,
PostgresDatabase,
PostgresDriverMissingError,
PostgresIntegrityError,
PostgresOperationalError,
_convert_placeholders,
_translate_sql,
init_postgres_schema,
)
def _has_real_driver():
for name in ("psycopg", "psycopg2"):
try:
__import__(name)
return True
except ImportError:
continue
return False
def _real_pg_params():
"""返回 (host, port, dbname, user, password) 或 None。
仅当 CLOUD_TEST_PG_DSN 已设且主机端口 2 秒内可达时返回;
其余情况返回 None,调用方 skipTest。绝不打印 DSN。
"""
dsn = os.environ.get("CLOUD_TEST_PG_DSN")
if not dsn:
return None
parsed = urlparse(dsn)
if parsed.scheme not in ("postgres", "postgresql") or not parsed.hostname:
return None
host = parsed.hostname
port = parsed.port or 5432
try:
with socket.create_connection((host, port), timeout=2):
pass
except OSError:
return None
return (
host,
port,
parsed.path.lstrip("/") or "postgres",
unquote(parsed.username) if parsed.username else None,
unquote(parsed.password) if parsed.password else None,
)
class _FakeUnique(Exception):
pass
class _FakeDeadlock(Exception):
pass
class _FakeCursor:
"""记录 SQL、按配置失败或返回预置行;description 为列名元组。"""
def __init__(self, rows=None, fail_sequence=None):
self.executed = []
self.rows = list(rows or [])
self.fail_sequence = list(fail_sequence or [])
self.description = [("id",), ("name",)]
def execute(self, sql, params=()):
self.executed.append((sql, params))
if self.fail_sequence:
exc = self.fail_sequence.pop(0)
if exc is not None:
raise exc
return self
def fetchall(self):
return list(self.rows)
def fetchone(self):
return self.rows[0] if self.rows else None
@property
def rowcount(self):
return len(self.rows) if self.rows else 1
def close(self):
pass
class _FakeConn:
def __init__(self, cursor=None):
self.cursor_obj = cursor or _FakeCursor()
self.committed = False
self.rolled_back = False
self.closed = False
def cursor(self):
return self.cursor_obj
def commit(self):
self.committed = True
def rollback(self):
self.rolled_back = True
def close(self):
self.closed = True
class _FakeDriver:
UniqueViolation = _FakeUnique
DeadlockDetected = _FakeDeadlock
OperationalError = Exception
Error = Exception
def __init__(self, cursor=None):
self.conn = _FakeConn(cursor)
def connect(self, **kwargs):
self.connect_kwargs = kwargs
return self.conn
def _make_db(fake):
with mock.patch("cloud.db._load_postgres_driver",
return_value=("psycopg", fake)):
return PostgresDatabase(host="test-host", dbname="test-db",
user="test-user",
password="test-placeholder")
class DriverMissingTestCase(unittest.TestCase):
def test_missing_driver_raises_psycopg_error(self):
if _has_real_driver():
self.skipTest("环境已装 psycopg 驱动,无驱动报错路径不适用")
with self.assertRaises(PostgresDriverMissingError) as ctx:
PostgresDatabase(host="test-host", dbname="test-db")
self.assertIn("psycopg", str(ctx.exception))
class PlaceholderTranslationTestCase(unittest.TestCase):
def test_question_marks_outside_strings_become_percent_s(self):
self.assertEqual(
_convert_placeholders(
"SELECT * FROM t WHERE a = ? AND b = 'x?y' AND c = ?"),
"SELECT * FROM t WHERE a = %s AND b = 'x?y' AND c = %s")
self.assertEqual(
_convert_placeholders("SELECT 'it''s ?' , ?"),
"SELECT 'it''s ?' , %s")
def test_insert_or_replace_becomes_on_conflict_update(self):
translated, kind = _translate_sql(
"INSERT OR REPLACE INTO project_members"
"(project_id, user_id, role) VALUES(?, ?, ?)")
self.assertEqual(kind, "normal")
self.assertIn("ON CONFLICT (project_id, user_id) DO UPDATE SET",
translated)
self.assertIn("role = EXCLUDED.role", translated)
self.assertIn("%s", translated)
self.assertNotIn("?", translated)
def test_autoincrement_becomes_identity(self):
translated, _ = _translate_sql(
"CREATE TABLE t(id INTEGER PRIMARY KEY AUTOINCREMENT, x TEXT)")
self.assertIn("GENERATED BY DEFAULT AS IDENTITY", translated)
self.assertNotIn("AUTOINCREMENT", translated)
def test_fts5_is_rejected_not_silently_created(self):
with self.assertRaises(PostgresOperationalError):
_translate_sql(
"CREATE VIRTUAL TABLE kb_fts USING fts5(content)")
class AdapterFakeDriverTestCase(unittest.TestCase):
def test_row_name_and_index_access(self):
fake = _FakeDriver(_FakeCursor(rows=[("t1", "hello")]))
db = _make_db(fake)
conn = db.connect()
self.assertIsInstance(conn, PgConnection)
rows = db.read_all(conn, "SELECT * FROM t WHERE id = ?", ("t1",))
self.assertEqual(rows[0]["name"], "hello")
self.assertEqual(rows[0][0], "t1")
# 占位符已转换:fake 游标收到的 SQL 应含 %s 不含 ?。
self.assertIn("%s", fake.conn.cursor_obj.executed[0][0])
self.assertNotIn("?", fake.conn.cursor_obj.executed[0][0])
def test_unique_violation_maps_to_integrity_error(self):
import sqlite3
fake = _FakeDriver(_FakeCursor(fail_sequence=[_FakeUnique("dup")]))
db = _make_db(fake)
conn = db.connect()
with self.assertRaises(PostgresIntegrityError) as ctx:
conn.execute("INSERT INTO t VALUES(?)", ("x",))
# 不是 sqlite3 异常:调用方按 sqlite 异常捕获时会漏掉,
# 因此 server.py 已同步捕获本模块错误(见该文件)。
self.assertNotIsInstance(ctx.exception, sqlite3.IntegrityError)
def test_transaction_rolls_back_on_error(self):
fake = _FakeDriver(_FakeCursor(fail_sequence=[None, _FakeUnique("dup")]))
db = _make_db(fake)
conn = db.connect()
with self.assertRaises(PostgresIntegrityError):
with db.transaction(conn):
conn.execute("INSERT INTO t VALUES(?)", ("a",))
conn.execute("INSERT INTO t VALUES(?)", ("b",))
self.assertTrue(fake.conn.rolled_back)
def test_execute_with_retry_retries_only_locked_errors(self):
fake = _FakeDriver(_FakeCursor(
fail_sequence=[_FakeDeadlock("deadlock"), None]))
db = _make_db(fake)
conn = db.connect()
# DeadlockDetected 在适配层映射为 locked=True,应重试一次后成功。
cur = db.execute_with_retry(conn, "UPDATE t SET x = ?", ("y",),
backoff_seconds=(0,))
self.assertEqual(len(fake.conn.cursor_obj.executed), 2)
self.assertIsNotNone(cur)
def test_repr_never_contains_password(self):
fake = _FakeDriver()
db = _make_db(fake)
self.assertNotIn("test-placeholder", repr(db))
self.assertNotIn("test-placeholder", repr(db.connect()))
class SchemaStatementTestCase(unittest.TestCase):
EXPECTED_TABLES = (
"users", "project_members", "audit_log", "projects", "tasks",
"summaries", "kb_docs", "cost_records", "oauth_states",
"slack_users", "push_subscriptions",
)
def test_schema_covers_all_business_tables(self):
joined = "\n".join(POSTGRES_SCHEMA)
for table in self.EXPECTED_TABLES:
self.assertIn("CREATE TABLE IF NOT EXISTS %s(" % table, joined)
# tasks 含迁移补的 6 列;绝无 FTS5 / AUTOINCREMENT 残留。
tasks_stmt = [s for s in POSTGRES_SCHEMA
if "CREATE TABLE IF NOT EXISTS tasks(" in s][0]
for col in ("result_text", "error_text", "actual_model",
"started_at", "finished_at", "child_pid"):
self.assertIn(col, tasks_stmt)
self.assertNotIn("fts5", joined.lower())
self.assertNotIn("AUTOINCREMENT", joined)
def test_schema_builds_via_fake_driver(self):
fake = _FakeDriver()
db = _make_db(fake)
conn = db.connect()
init_postgres_schema(conn)
executed = [sql for sql, _ in fake.conn.cursor_obj.executed]
self.assertEqual(len(executed), len(POSTGRES_SCHEMA))
self.assertTrue(fake.conn.committed)
class RealPostgresSchemaTestCase(unittest.TestCase):
"""真实 PG schema:无 DSN 或不可达时 skip。"""
def test_schema_creates_tables_and_roundtrips(self):
params = _real_pg_params()
if params is None:
self.skipTest("未设置可达的 CLOUD_TEST_PG_DSN,跳过真实 PG 测试")
if not _has_real_driver():
self.skipTest("未装 psycopg 驱动,跳过真实 PG 测试")
host, port, dbname, user, password = params
db = PostgresDatabase(host=host, port=port, dbname=dbname,
user=user, password=password)
conn = db.connect()
try:
init_postgres_schema(conn)
rows = db.read_all(
conn,
"SELECT table_name FROM information_schema.tables"
" WHERE table_schema = 'public'",
())
names = {r["table_name"] for r in rows}
for table in SchemaStatementTestCase.EXPECTED_TABLES:
self.assertIn(table, names)
# 最小往返:插一条项目再按名读回,随后删掉测试行。
with db.transaction(conn):
conn.execute(
"INSERT INTO projects(id, name, owner_id, created_at)"
" VALUES(?, ?, ?, ?)",
("pg-test-proj", "pg-test", "pg-test-user", 0.0))
rows = db.read_all(
conn, "SELECT * FROM projects WHERE id = ?",
("pg-test-proj",))
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["name"], "pg-test")
finally:
try:
with db.transaction(conn):
conn.execute("DELETE FROM projects WHERE id = ?",
("pg-test-proj",))
except Exception:
pass
conn.close()
+172
View File
@@ -0,0 +1,172 @@
"""Postgres 集成测试:真实服务器上的端到端路径。
全部测试在无真实 Postgres 时 skip,绝不 fail。判定条件与
test_postgres_backend.py 的 _real_pg_params() 一致:须显式
设置 CLOUD_TEST_PG_DSN(postgres://user:pass@host:port/dbname)
且主机端口 2 秒内可达,且装有 psycopg 驱动。DSN 必须指向
一次性测试库;本文件只写业务表与测试行,跑完删掉测试行,
不删库、不动其它库。DSN 只从环境变量读,测试不打印它。
"""
import os
import sys
import tempfile
import unittest
from pathlib import Path
try:
from cloud.tests.test_postgres_backend import (
_has_real_driver,
_real_pg_params,
)
from cloud.db import PostgresDatabase, init_postgres_schema
from cloud.migrations import migrate_sqlite_to_postgres
from cloud.server import init_db
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.tests.test_postgres_backend import (
_has_real_driver,
_real_pg_params,
)
from cloud.db import PostgresDatabase, init_postgres_schema
from cloud.migrations import migrate_sqlite_to_postgres
from cloud.server import init_db
def _pg_db_or_skip(testcase):
params = _real_pg_params()
if params is None:
testcase.skipTest("未设置可达的 CLOUD_TEST_PG_DSN,跳过真实 PG 测试")
if not _has_real_driver():
testcase.skipTest("未装 psycopg 驱动,跳过真实 PG 测试")
host, port, dbname, user, password = params
return PostgresDatabase(host=host, port=port, dbname=dbname,
user=user, password=password)
class PostgresIntegrationTestCase(unittest.TestCase):
def test_server_init_db_and_adapter_crud(self):
db = _pg_db_or_skip(self)
conn = db.connect()
try:
init_postgres_schema(conn)
# 适配层 CRUD:projects 插、读、改、删均走 ? 占位符。
with db.transaction(conn):
conn.execute(
"INSERT INTO projects(id, name, owner_id, created_at)"
" VALUES(?, ?, ?, ?)",
("pg-int-proj", "pg-int", "pg-int-user", 0.0))
rows = db.read_all(
conn, "SELECT * FROM projects WHERE id = ?",
("pg-int-proj",))
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["name"], "pg-int")
with db.transaction(conn):
conn.execute(
"UPDATE projects SET name = ? WHERE id = ?",
("pg-int-2", "pg-int-proj"))
rows = db.read_all(
conn, "SELECT name FROM projects WHERE id = ?",
("pg-int-proj",))
self.assertEqual(rows[0]["name"], "pg-int-2")
finally:
try:
with db.transaction(conn):
conn.execute("DELETE FROM projects WHERE id = ?",
("pg-int-proj",))
except Exception:
pass
conn.close()
def test_sqlite_to_postgres_migration_is_idempotent(self):
import sqlite3
db = _pg_db_or_skip(self)
# 先建一个最小 SQLite 库:含 projects/tasks 两行。
with tempfile.TemporaryDirectory() as tmp:
src = os.path.join(tmp, "src.db")
sconn = sqlite3.connect(src)
sconn.execute(
"CREATE TABLE projects(id TEXT PRIMARY KEY, name TEXT,"
" owner_id TEXT, created_at REAL)")
sconn.execute(
"CREATE TABLE tasks(id TEXT PRIMARY KEY, project_id TEXT,"
" prompt TEXT, status TEXT)")
sconn.execute(
"INSERT INTO projects VALUES(?, ?, ?, ?)",
("mig-proj", "mig", "mig-user", 1.5))
sconn.execute(
"INSERT INTO tasks VALUES(?, ?, ?, ?)",
("mig-task", "mig-proj", "hello", "queued"))
sconn.commit()
sconn.close()
conn = db.connect()
try:
first = migrate_sqlite_to_postgres(src, pg_conn=conn)
self.assertGreaterEqual(first.get("projects", 0), 1)
self.assertGreaterEqual(first.get("tasks", 0), 1)
# 第二次跑:ON CONFLICT DO NOTHING,不重复插。
second = migrate_sqlite_to_postgres(src, pg_conn=conn)
self.assertEqual(second.get("projects", -1), 0)
self.assertEqual(second.get("tasks", -1), 0)
rows = db.read_all(
conn, "SELECT * FROM projects WHERE id = ?",
("mig-proj",))
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["name"], "mig")
finally:
try:
with db.transaction(conn):
conn.execute("DELETE FROM tasks WHERE id = ?",
("mig-task",))
conn.execute("DELETE FROM projects WHERE id = ?",
("mig-proj",))
except Exception:
pass
conn.close()
def test_server_init_db_on_postgres(self):
# 真走 server.init_db 的 PG 分支:环境变量切到 postgres
# 后,init_db 不应建任何 SQLite 文件,而应在 PG 里建表
# (含 tasks 的 6 个结果列);本测试不启动 HTTP server。
import os
from unittest import mock
params = _real_pg_params()
if params is None:
self.skipTest("未设置可达的 CLOUD_TEST_PG_DSN,跳过真实 PG 测试")
if not _has_real_driver():
self.skipTest("未装 psycopg 驱动,跳过真实 PG 测试")
host, port, dbname, user, password = params
env = {
"CLOUD_DB_BACKEND": "postgres",
"CLOUD_DB_HOST": host,
"CLOUD_DB_PORT": str(port),
"CLOUD_DB_NAME": dbname,
"CLOUD_DB_USER": user or "",
"CLOUD_DB_PASSWORD": password or "",
}
with mock.patch.dict(os.environ, env):
init_db("ignored-in-postgres-mode.db")
db = PostgresDatabase(host=host, port=port, dbname=dbname,
user=user, password=password)
conn = db.connect()
try:
rows = db.read_all(
conn,
"SELECT table_name FROM information_schema.tables"
" WHERE table_schema = 'public'",
())
names = {r["table_name"] for r in rows}
self.assertIn("projects", names)
self.assertIn("tasks", names)
self.assertIn("audit_log", names)
cols = db.read_all(
conn,
"SELECT column_name FROM information_schema.columns"
" WHERE table_name = 'tasks'",
())
colnames = {r["column_name"] for r in cols}
for col in ("result_text", "error_text", "actual_model",
"started_at", "finished_at", "child_pid"):
self.assertIn(col, colnames)
finally:
conn.close()