host
64995fafab
- migrations: MIGRATION_TABLES now equals POSTGRES_SCHEMA tables
(uploads, task_attachments, task_followups, environments added);
SEQUENCE_TABLES matches IDENTITY tables; task_followups
sender_task_id column ensured before insert; UNMIGRATED_TABLES
and migration_coverage() report tables with no PG schema instead
of silently skipping them; CLI prints the coverage report.
- deploy: DEFAULT_TEST_SUITES gains deploy/tests (discover -s only,
never package path); regression tests pin the default list.
- monitor: MONITOR_SINGLE_HOST=1 makes the server check single-host;
default stays dual to avoid silent downgrade; server_mode is in
the JSON. Optional fail-only alert outlet (webhook or mail,
default off, alert body is pass/fail JSON only, no secrets).
- backup/rollback: README and restore-drill document what is
recoverable (DB, code) vs not (worktrees, traces, uploads, host
configs); rollback.sh shares the server-unit list and dual-port
health success standard with host_deploy.sh (sync test guards it).
- public entry: Caddyfile uses native {$DOMAIN}/{$ACME_EMAIL}
variables plus a caddy.service template; README documents the
steps and the current BT-nginx production status.
204 lines
8.6 KiB
Python
204 lines
8.6 KiB
Python
"""P2-ops 迁移对齐：迁移清单必须等于 Postgres schema 表集合，
|
||
未迁移项必须被显式报告，绝不能静默跳过。
|
||
|
||
不用真实 Postgres：迁移本体用一个假 PG 连接跑（只记录
|
||
INSERT / ALTER，不连库）；schema 对齐直接读 POSTGRES_SCHEMA。
|
||
真实 PG 上的幂等性仍由 test_postgres_integration.py 覆盖
|
||
（有 CLOUD_TEST_PG_DSN 时才跑）。
|
||
"""
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from cloud.db import POSTGRES_SCHEMA
|
||
from cloud.migrations import (
|
||
MIGRATION_TABLES,
|
||
SEQUENCE_TABLES,
|
||
UNMIGRATED_TABLES,
|
||
migrate_sqlite_to_postgres,
|
||
migration_coverage,
|
||
schema_tables,
|
||
)
|
||
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.db import POSTGRES_SCHEMA
|
||
from cloud.migrations import (
|
||
MIGRATION_TABLES,
|
||
SEQUENCE_TABLES,
|
||
UNMIGRATED_TABLES,
|
||
migrate_sqlite_to_postgres,
|
||
migration_coverage,
|
||
schema_tables,
|
||
)
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
|
||
|
||
class FakeCursor:
|
||
def __init__(self, rows=None, rowcount=1):
|
||
self._rows = rows or []
|
||
self.rowcount = rowcount
|
||
|
||
def fetchall(self):
|
||
return self._rows
|
||
|
||
|
||
class FakePgConn:
|
||
"""只记录 SQL 的假 PG 连接：init_postgres_schema 的 DDL
|
||
直接忽略；PRAGMA 按构造时给的列集合回答（row[1] 为列名，
|
||
与 init_postgres_schema / 迁移补列代码的取法一致）。"""
|
||
|
||
def __init__(self, pragma_columns=None):
|
||
# pragma_columns: {表名: {列名}}；缺省表示“列齐全”。
|
||
self.pragma_columns = pragma_columns or {}
|
||
self.statements = []
|
||
self.inserts = [] # (表名, 列名元组)
|
||
self.alters = []
|
||
self.commits = 0
|
||
|
||
def execute(self, sql, params=()):
|
||
self.statements.append(sql)
|
||
if sql.startswith("PRAGMA table_info("):
|
||
table = sql[len("PRAGMA table_info("):].rstrip(")")
|
||
cols = self.pragma_columns.get(table)
|
||
if cols is None:
|
||
# 未指定时假装列齐全：projects 含 repo_url，
|
||
# task_followups 含 sender_task_id。
|
||
cols = {"repo_url"} if table == "projects" \
|
||
else {"sender_task_id"}
|
||
rows = [(i, c, None) for i, c in enumerate(sorted(cols))]
|
||
return FakeCursor(rows=rows)
|
||
m = re.search(r'INSERT INTO "(\w+)" \(([^)]*)\)', sql)
|
||
if m:
|
||
cols = tuple(c.strip().strip('"')
|
||
for c in m.group(2).split(","))
|
||
self.inserts.append((m.group(1), cols))
|
||
return FakeCursor(rowcount=1)
|
||
if sql.startswith("ALTER TABLE"):
|
||
self.alters.append(sql)
|
||
return FakeCursor(rowcount=0)
|
||
return FakeCursor(rowcount=0)
|
||
|
||
def commit(self):
|
||
self.commits += 1
|
||
|
||
|
||
def _make_sqlite(path):
|
||
conn = sqlite3.connect(path)
|
||
conn.execute("CREATE TABLE uploads(id TEXT PRIMARY KEY, created_by TEXT,"
|
||
" created_at REAL, status TEXT)")
|
||
conn.execute("CREATE TABLE task_attachments(task_id TEXT, bundle_id TEXT,"
|
||
" rel_path TEXT, size INTEGER,"
|
||
" PRIMARY KEY(task_id, rel_path))")
|
||
conn.execute("CREATE TABLE task_followups(id TEXT PRIMARY KEY,"
|
||
" task_id TEXT, project_id TEXT, body TEXT, mode TEXT,"
|
||
" status TEXT, created_by TEXT, sender_task_id TEXT,"
|
||
" created_at REAL, updated_at REAL)")
|
||
conn.execute("CREATE TABLE environments(id TEXT PRIMARY KEY,"
|
||
" project_id TEXT, name TEXT, branch TEXT, executor TEXT,"
|
||
" status TEXT, error_text TEXT, repo_url TEXT,"
|
||
" created_by TEXT, created_at REAL, updated_at REAL)")
|
||
conn.execute("INSERT INTO uploads VALUES(?, ?, ?, ?)",
|
||
("up1", "u1", 1.0, "done"))
|
||
conn.execute("INSERT INTO task_attachments VALUES(?, ?, ?, ?)",
|
||
("t1", "b1", "a.txt", 3))
|
||
conn.execute("INSERT INTO task_followups VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
("f1", "t1", "p1", "hi", "queued", "pending", "u1",
|
||
"sender-task", 1.0, 1.0))
|
||
conn.execute("INSERT INTO environments VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
("e1", "p1", "dev", "main", "docker", "ready", "",
|
||
"https://example.invalid/r", "u1", 1.0, 1.0))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
class TestMigrationAlignment(unittest.TestCase):
|
||
def test_migration_tables_equal_schema_tables(self):
|
||
# 核心对齐断言：以前漏掉的 4 张表补进后，两边集合必须一致。
|
||
self.assertEqual(set(MIGRATION_TABLES), schema_tables())
|
||
for table in ("uploads", "task_attachments",
|
||
"task_followups", "environments", "user_quotas"):
|
||
self.assertIn(table, MIGRATION_TABLES)
|
||
|
||
def test_sequence_tables_equal_identity_tables(self):
|
||
# 序列拨动清单必须等于 schema 里 IDENTITY 列的表集合：
|
||
# TEXT 主键的新表不需要拨序列，漏拨或多拨都会在这里变红。
|
||
identity = set()
|
||
for statement in POSTGRES_SCHEMA:
|
||
m = re.search(r"CREATE TABLE IF NOT EXISTS (\w+)", statement)
|
||
if m and "GENERATED BY DEFAULT AS IDENTITY" in statement:
|
||
identity.add(m.group(1))
|
||
self.assertEqual(set(SEQUENCE_TABLES), identity)
|
||
|
||
def test_unmigrated_are_real_sqlite_tables_not_in_schema(self):
|
||
# 未迁移清单不得是凭空编的：每项都必须在 cloud/ 源码里
|
||
# 真有 CREATE TABLE，且确实不在 PG schema 里；同时不得
|
||
# 与迁移清单重叠（重叠即“既迁又报未迁”，自相矛盾）。
|
||
sources = ""
|
||
for name in os.listdir(REPO_ROOT / "cloud"):
|
||
if name.endswith(".py"):
|
||
with open(REPO_ROOT / "cloud" / name) as f:
|
||
sources += f.read()
|
||
schema = schema_tables()
|
||
for table in UNMIGRATED_TABLES:
|
||
self.assertIn("CREATE TABLE IF NOT EXISTS %s" % table, sources,
|
||
"%s 不在任何 SQLite DDL 里，清单已过期" % table)
|
||
self.assertNotIn(table, schema)
|
||
self.assertNotIn(table, MIGRATION_TABLES)
|
||
|
||
def test_coverage_reports_no_missing(self):
|
||
coverage = migration_coverage()
|
||
self.assertEqual(coverage["missing_from_migration"], ())
|
||
self.assertEqual(coverage["unmigrated_no_pg_schema"],
|
||
tuple(UNMIGRATED_TABLES))
|
||
self.assertEqual(set(coverage["skipped_by_design"]),
|
||
{"kb_fts", "sqlite_sequence"})
|
||
|
||
|
||
class TestMigrateNewTables(unittest.TestCase):
|
||
def test_new_tables_migrated_and_sender_column_added(self):
|
||
# PG 侧 task_followups 尚无 sender_task_id：迁移必须先
|
||
# ALTER 补列，再把含该列的行插过去，而不是撞未知列失败。
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
src = os.path.join(tmp, "src.db")
|
||
_make_sqlite(src)
|
||
fake = FakePgConn(pragma_columns={
|
||
"projects": {"repo_url"},
|
||
"task_followups": {"id", "task_id", "project_id", "body",
|
||
"mode", "status", "created_by",
|
||
"created_at", "updated_at"},
|
||
})
|
||
result = migrate_sqlite_to_postgres(src, pg_conn=fake)
|
||
for table in ("uploads", "task_attachments",
|
||
"task_followups", "environments"):
|
||
self.assertEqual(result.get(table), 1, table)
|
||
self.assertTrue(
|
||
any("task_followups" in a and "sender_task_id" in a
|
||
for a in fake.alters),
|
||
"未发出 sender_task_id 补列 ALTER：%s" % fake.alters)
|
||
fu_inserts = [cols for t, cols in fake.inserts
|
||
if t == "task_followups"]
|
||
self.assertEqual(len(fu_inserts), 1)
|
||
self.assertIn("sender_task_id", fu_inserts[0])
|
||
|
||
def test_sender_column_present_skips_alter(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
src = os.path.join(tmp, "src.db")
|
||
_make_sqlite(src)
|
||
fake = FakePgConn() # 缺省：列齐全
|
||
migrate_sqlite_to_postgres(src, pg_conn=fake,
|
||
tables=("task_followups",))
|
||
# init_postgres_schema 自身会给 projects/tasks 补列，
|
||
# 这里只断言 task_followups 没有重复补列。
|
||
self.assertEqual(
|
||
[a for a in fake.alters if "task_followups" in a], [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|