host
dc3f572b2b
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.
173 lines
7.3 KiB
Python
173 lines
7.3 KiB
Python
"""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()
|