浅念
552 lines
22 KiB
Python
552 lines
22 KiB
Python
"""scripts/import_legacy.py(Phase 4 Task 2): 老单租户库 → 多租户租户库。
|
|
|
|
契约(计划原文):
|
|
1. 五表对账: 行数源/目标一致 + 最近 10 条会话逐字段抽查
|
|
(id/session_id/status/page_token)+ conversation_id 原样(max 相等);
|
|
2. pending 推送日志 → failed, error_message 追加"[迁移] 切换前未完成态";
|
|
3. 幂等: 跑两遍第二遍零变化(对账报告 skipped==全部);
|
|
4. 老库结构假设防御: 关键列缺失 → 显性报错列名, 不静默丢数据;
|
|
新库多列 → 默认值填充并在报告标注; 老库多列 → 不迁移并标注;
|
|
5. 对账不一致(源≠目标合计) → 非零退出码。
|
|
|
|
测试方式(照 test_bootstrap_script.py): import + sys.argv 注入, 非 subprocess。
|
|
真实老库用例经 env 指路(ALL_KEFU_TEST_LEGACY_JC / _AUTO), 缺失 skip(机器
|
|
相关); 用前 VACUUM INTO 拷到 tmp, 绝不直连原库——原库 sha256 前后比对
|
|
钉死"零接触"。合成库用例(结构防御/幂等/pending 转换)不依赖 env, 常跑。
|
|
"""
|
|
import hashlib
|
|
import importlib.util
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.engine.models import (
|
|
Base,
|
|
Conversation,
|
|
ConversationSource,
|
|
ConversationStatus,
|
|
CustomerProfile,
|
|
Evaluation,
|
|
IntentLevel,
|
|
LeadPushLog,
|
|
Message,
|
|
MessageRole,
|
|
)
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
SEED_DIR = REPO_ROOT / "seed_profiles"
|
|
LEGACY_TABLES = (
|
|
"conversations", "messages", "customer_profiles", "evaluations",
|
|
"lead_push_logs",
|
|
)
|
|
# 行级比对列(名取交集两边都有——jc/auto 老库实测均有这些列)
|
|
COMPARE_COLUMNS = {
|
|
"conversations": ("id", "session_id", "page_token", "status", "source",
|
|
"client_info", "created_at", "updated_at"),
|
|
"messages": ("id", "conversation_id", "role", "content", "created_at"),
|
|
"customer_profiles": ("id", "conversation_id", "raw_info", "summary",
|
|
"updated_at"),
|
|
"evaluations": ("id", "conversation_id", "intent_level", "reasoning",
|
|
"contact_info", "email_sent", "correction_count",
|
|
"evaluated_at"),
|
|
"lead_push_logs": ("id", "evaluation_id", "conversation_id", "push_type",
|
|
"status", "attempts", "request_payload",
|
|
"response_status", "outcome", "error_message",
|
|
"created_at", "updated_at"),
|
|
}
|
|
PENDING_MARKER = "[迁移] 切换前未完成态"
|
|
|
|
|
|
def _load(name: str):
|
|
spec = importlib.util.spec_from_file_location(
|
|
name, REPO_ROOT / "scripts" / f"{name}.py")
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
# 模块级加载一次(脚本 import 期只做 sys.path 锚定, 无副作用)
|
|
import_legacy = _load("import_legacy")
|
|
bootstrap_platform = _load("bootstrap_platform")
|
|
|
|
|
|
def run_import(monkeypatch, *argv):
|
|
monkeypatch.setattr(sys, "argv", ["import_legacy.py", *argv])
|
|
import_legacy.main()
|
|
|
|
|
|
def run_bootstrap(monkeypatch, db: Path, *profiles: str):
|
|
monkeypatch.setattr(
|
|
sys, "argv",
|
|
["bootstrap_platform.py", "--db", str(db), "--seed", str(SEED_DIR),
|
|
"--profiles", *profiles])
|
|
bootstrap_platform.main()
|
|
|
|
|
|
# ---- 夹具/辅助 ---------------------------------------------------------------
|
|
|
|
def _ro(db) -> sqlite3.Connection:
|
|
return sqlite3.connect(f"file:{db}?mode=ro", uri=True)
|
|
|
|
|
|
def _count(db, table) -> int:
|
|
con = _ro(db)
|
|
try:
|
|
return con.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def _rows(db, table, columns, order="id"):
|
|
con = _ro(db)
|
|
con.row_factory = sqlite3.Row
|
|
try:
|
|
return [dict(r) for r in con.execute(
|
|
f"SELECT {', '.join(columns)} FROM {table} ORDER BY {order}")]
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def _spot_rows(db):
|
|
"""最近 10 条会话(id/session_id/status/page_token), 源目标同一排序。"""
|
|
con = _ro(db)
|
|
try:
|
|
return con.execute(
|
|
"SELECT id, session_id, status, page_token FROM conversations "
|
|
"ORDER BY created_at DESC, id DESC LIMIT 10").fetchall()
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def _max_conv_id(db, table):
|
|
con = _ro(db)
|
|
try:
|
|
return con.execute(
|
|
f"SELECT MAX(conversation_id) FROM {table}").fetchone()[0]
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def _sha256(path) -> str:
|
|
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
|
|
|
|
|
def _expected_lpl(rows):
|
|
"""源 lead_push_logs 行 → 导入后期望形态(pending→failed+追加备注)。"""
|
|
out = []
|
|
for r in rows:
|
|
r = dict(r)
|
|
if r["status"] == "pending":
|
|
r["status"] = "failed"
|
|
old = r["error_message"]
|
|
r["error_message"] = (old + " " + PENDING_MARKER) if old else \
|
|
PENDING_MARKER
|
|
out.append(r)
|
|
return out
|
|
|
|
|
|
def _parse_report_line(out: str, table: str):
|
|
"""' conversations: 源=3 导入=3 跳过=0 目标=3 一致' → (3,3,0,3)。"""
|
|
import re
|
|
m = re.search(rf"{table}: 源=(\d+) 导入=(\d+) 跳过=(\d+) 目标=(\d+)", out)
|
|
assert m, f"报告缺 {table} 对账行:\n{out}"
|
|
return tuple(int(g) for g in m.groups())
|
|
|
|
|
|
def _assert_tables_identical(source_db, target_db):
|
|
"""五表行级比对: 四表逐字相等, lead_push_logs 按转换后期望相等;
|
|
附带计划点名的 conversation_id 原样性(max 相等)。"""
|
|
for table in LEGACY_TABLES:
|
|
src = _rows(source_db, table, COMPARE_COLUMNS[table])
|
|
dst = _rows(target_db, table, COMPARE_COLUMNS[table])
|
|
expected = _expected_lpl(src) if table == "lead_push_logs" else src
|
|
assert dst == expected, f"{table} 行级不一致"
|
|
for table in ("messages", "customer_profiles", "evaluations",
|
|
"lead_push_logs"):
|
|
assert _max_conv_id(source_db, table) == _max_conv_id(target_db, table)
|
|
|
|
|
|
# ---- 合成老库(确定性, 不依赖 env)----------------------------------------------
|
|
|
|
C1 = "11111111-1111-1111-1111-111111111111"
|
|
C2 = "22222222-2222-2222-2222-222222222222"
|
|
C3 = "33333333-3333-3333-3333-333333333333"
|
|
E2 = "eeeeeeee-2222-2222-2222-222222222222"
|
|
E3 = "eeeeeeee-3333-3333-3333-333333333333"
|
|
|
|
|
|
def _make_legacy_db(path: Path) -> None:
|
|
"""3 会话小库: 状态三态齐备 + 2 条 pending 推送(一条带旧错误信息,
|
|
验证追加语义)+ success/failed 各一(验证非 pending 不动)。"""
|
|
engine = create_engine(f"sqlite:///{path}")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as s:
|
|
def ts(day, minute=0):
|
|
return datetime(2026, 9, day, 10, minute, tzinfo=timezone.utc)
|
|
|
|
s.add_all([
|
|
Conversation(id=C1, session_id="sess-c1", page_token="tok-c1",
|
|
status=ConversationStatus.completed,
|
|
source=ConversationSource.widget,
|
|
client_info={"ip": "1.2.3.4"},
|
|
created_at=ts(1), updated_at=ts(1, 5)),
|
|
Conversation(id=C2, session_id="sess-c2", page_token=None,
|
|
status=ConversationStatus.active,
|
|
source=ConversationSource.standalone,
|
|
created_at=ts(2), updated_at=ts(2, 5)),
|
|
Conversation(id=C3, session_id="sess-c3", page_token="tok-c3",
|
|
status=ConversationStatus.abandoned,
|
|
source=ConversationSource.embedded,
|
|
created_at=ts(3), updated_at=ts(3, 5)),
|
|
])
|
|
msgs = []
|
|
for i, (mid, cid, role, day) in enumerate([
|
|
("aaaa0000-0000-0000-0000-000000000001", C1,
|
|
MessageRole.user, 1),
|
|
("aaaa0000-0000-0000-0000-000000000002", C1,
|
|
MessageRole.assistant, 1),
|
|
("aaaa0000-0000-0000-0000-000000000003", C2,
|
|
MessageRole.user, 2),
|
|
("aaaa0000-0000-0000-0000-000000000004", C2,
|
|
MessageRole.assistant, 2),
|
|
("aaaa0000-0000-0000-0000-000000000005", C3,
|
|
MessageRole.user, 3)]):
|
|
msgs.append(Message(id=mid, conversation_id=cid, role=role,
|
|
content=f"msg-{i}", created_at=ts(day, i)))
|
|
s.add_all(msgs)
|
|
s.add_all([
|
|
CustomerProfile(id="bbbb0000-0000-0000-0000-000000000001",
|
|
conversation_id=C1,
|
|
raw_info={"name": "张三"}, summary="老客",
|
|
updated_at=ts(1, 6)),
|
|
CustomerProfile(id="bbbb0000-0000-0000-0000-000000000002",
|
|
conversation_id=C2, raw_info={}, summary=None,
|
|
updated_at=ts(2, 6)),
|
|
CustomerProfile(id="bbbb0000-0000-0000-0000-000000000003",
|
|
conversation_id=C3, raw_info=None,
|
|
summary="仅询价", updated_at=ts(3, 6)),
|
|
])
|
|
s.add_all([
|
|
Evaluation(id=E2, conversation_id=C2,
|
|
intent_level=IntentLevel.interested, reasoning="明确",
|
|
contact_info={"phone": "13800000000"}, email_sent=True,
|
|
correction_count=1, evaluated_at=ts(2, 7)),
|
|
Evaluation(id=E3, conversation_id=C3,
|
|
intent_level=IntentLevel.pending, reasoning="",
|
|
contact_info={}, email_sent=False, correction_count=0,
|
|
evaluated_at=ts(3, 7)),
|
|
])
|
|
s.add_all([
|
|
LeadPushLog(id="dddd0000-0000-0000-0000-000000000001",
|
|
evaluation_id=E2, conversation_id=C2,
|
|
push_type="lead", status="success", attempts=1,
|
|
response_status=200, outcome="accepted",
|
|
created_at=ts(2, 8), updated_at=ts(2, 8)),
|
|
LeadPushLog(id="dddd0000-0000-0000-0000-000000000002",
|
|
evaluation_id=E3, conversation_id=C3,
|
|
push_type="lead", status="pending", attempts=2,
|
|
error_message=None,
|
|
created_at=ts(3, 8), updated_at=ts(3, 8)),
|
|
LeadPushLog(id="dddd0000-0000-0000-0000-000000000003",
|
|
evaluation_id=E3, conversation_id=C3,
|
|
push_type="correction", status="pending", attempts=1,
|
|
error_message="旧错误: 推送超时",
|
|
created_at=ts(3, 9), updated_at=ts(3, 9)),
|
|
LeadPushLog(id="dddd0000-0000-0000-0000-000000000004",
|
|
evaluation_id=E2, conversation_id=C2,
|
|
push_type="returning", status="failed", attempts=3,
|
|
error_message="HTTP 500",
|
|
created_at=ts(2, 9), updated_at=ts(2, 9)),
|
|
])
|
|
s.commit()
|
|
engine.dispose()
|
|
|
|
|
|
LEGACY_COUNTS = {"conversations": 3, "messages": 5, "customer_profiles": 3,
|
|
"evaluations": 2, "lead_push_logs": 4}
|
|
|
|
|
|
class Env:
|
|
"""tmp 演练环境: platform.db(bootstrap 过 jiancai)+ tenants 目录。"""
|
|
|
|
def __init__(self, tmp_path: Path, monkeypatch):
|
|
self.platform_db = tmp_path / "platform.db"
|
|
self.tenants_dir = tmp_path / "tenants"
|
|
run_bootstrap(monkeypatch, self.platform_db, "jiancai")
|
|
|
|
def import_args(self, source: Path, tenant: str = "jiancai"):
|
|
return ("--source", str(source), "--tenant", tenant,
|
|
"--platform-db", str(self.platform_db),
|
|
"--tenants-dir", str(self.tenants_dir))
|
|
|
|
@property
|
|
def tenant_db(self) -> Path:
|
|
return self.tenants_dir / "jiancai.db"
|
|
|
|
|
|
# ---- 合成库: 导入正确性 -------------------------------------------------------
|
|
|
|
def test_import_preserves_all_five_tables(tmp_path, monkeypatch, capsys):
|
|
"""五表对账全绿: 行数一致 + 行级逐字(四表)+ conversation_id 原样 +
|
|
最近会话抽查逐字段 + 报告每表 一致。"""
|
|
env = Env(tmp_path, monkeypatch)
|
|
legacy = tmp_path / "legacy.db"
|
|
_make_legacy_db(legacy)
|
|
|
|
run_import(monkeypatch, *env.import_args(legacy))
|
|
out = capsys.readouterr().out
|
|
for table, n in LEGACY_COUNTS.items():
|
|
src, imp, skip, dst = _parse_report_line(out, table)
|
|
assert (src, imp, skip, dst) == (n, n, 0, n)
|
|
assert _count(env.tenant_db, table) == n
|
|
assert "对账结论: 全部一致" in out
|
|
assert _spot_rows(legacy) == _spot_rows(env.tenant_db)
|
|
_assert_tables_identical(legacy, env.tenant_db)
|
|
|
|
|
|
def test_pending_push_logs_converted_to_failed(tmp_path, monkeypatch, capsys):
|
|
"""pending → failed + 追加备注: 空 error 只剩备注; 旧错误为前缀追加;
|
|
非 pending 行(status/error_message)逐字不动。"""
|
|
env = Env(tmp_path, monkeypatch)
|
|
legacy = tmp_path / "legacy.db"
|
|
_make_legacy_db(legacy)
|
|
|
|
run_import(monkeypatch, *env.import_args(legacy))
|
|
capsys.readouterr()
|
|
lpl = _rows(env.tenant_db, "lead_push_logs", COMPARE_COLUMNS["lead_push_logs"])
|
|
by_id = {r["id"]: r for r in lpl}
|
|
p1 = by_id["dddd0000-0000-0000-0000-000000000002"]
|
|
assert p1["status"] == "failed"
|
|
assert p1["error_message"] == PENDING_MARKER
|
|
p2 = by_id["dddd0000-0000-0000-0000-000000000003"]
|
|
assert p2["status"] == "failed"
|
|
assert p2["error_message"] == f"旧错误: 推送超时 {PENDING_MARKER}"
|
|
assert by_id["dddd0000-0000-0000-0000-000000000001"]["status"] == "success"
|
|
assert by_id["dddd0000-0000-0000-0000-000000000001"]["error_message"] is None
|
|
assert by_id["dddd0000-0000-0000-0000-000000000004"]["status"] == "failed"
|
|
assert by_id["dddd0000-0000-0000-0000-000000000004"]["error_message"] == "HTTP 500"
|
|
|
|
|
|
def test_idempotent_second_run_zero_change(tmp_path, monkeypatch, capsys):
|
|
"""跑两遍: 第二遍每表 导入=0 跳过=源, DB 行级快照逐字不变。"""
|
|
env = Env(tmp_path, monkeypatch)
|
|
legacy = tmp_path / "legacy.db"
|
|
_make_legacy_db(legacy)
|
|
|
|
run_import(monkeypatch, *env.import_args(legacy))
|
|
capsys.readouterr()
|
|
snapshot = {t: _rows(env.tenant_db, t, COMPARE_COLUMNS[t])
|
|
for t in LEGACY_TABLES}
|
|
|
|
run_import(monkeypatch, *env.import_args(legacy))
|
|
out = capsys.readouterr().out
|
|
for table, n in LEGACY_COUNTS.items():
|
|
assert _parse_report_line(out, table) == (n, 0, n, n)
|
|
assert {t: _rows(env.tenant_db, t, COMPARE_COLUMNS[t])
|
|
for t in LEGACY_TABLES} == snapshot
|
|
|
|
|
|
# ---- 合成库: 结构防御 ---------------------------------------------------------
|
|
|
|
def test_missing_critical_column_fails_naming_column(tmp_path, monkeypatch):
|
|
"""保真关键列(conversations.status)缺失 → 显性报列名, 不静默丢数据。"""
|
|
env = Env(tmp_path, monkeypatch)
|
|
legacy = tmp_path / "legacy.db"
|
|
_make_legacy_db(legacy)
|
|
con = sqlite3.connect(legacy)
|
|
con.execute("ALTER TABLE conversations DROP COLUMN status")
|
|
con.commit()
|
|
con.close()
|
|
|
|
with pytest.raises(SystemExit, match="status"):
|
|
run_import(monkeypatch, *env.import_args(legacy))
|
|
|
|
|
|
def test_missing_optional_column_defaulted_and_noted(
|
|
tmp_path, monkeypatch, capsys):
|
|
"""非关键列(page_token 可空)缺失 → 默认值 NULL 填充 + 报告标注,
|
|
导入照常成功(对账仍绿)。"""
|
|
env = Env(tmp_path, monkeypatch)
|
|
legacy = tmp_path / "legacy.db"
|
|
_make_legacy_db(legacy)
|
|
con = sqlite3.connect(legacy)
|
|
con.execute("DROP INDEX ix_conversations_page_token")
|
|
con.execute("ALTER TABLE conversations DROP COLUMN page_token")
|
|
con.commit()
|
|
con.close()
|
|
|
|
run_import(monkeypatch, *env.import_args(legacy))
|
|
out = capsys.readouterr().out
|
|
assert "老库缺列 page_token" in out
|
|
assert _count(env.tenant_db, "conversations") == 3
|
|
con = _ro(env.tenant_db)
|
|
vals = [r[0] for r in con.execute("SELECT page_token FROM conversations")]
|
|
con.close()
|
|
assert vals == [None, None, None]
|
|
|
|
|
|
def test_extra_source_column_disclosed_not_fatal(tmp_path, monkeypatch, capsys):
|
|
"""老库多列(新库已废弃, 如 auto 真库的 correction_sent)→ 不迁移 +
|
|
报告标注, 导入照常成功。"""
|
|
env = Env(tmp_path, monkeypatch)
|
|
legacy = tmp_path / "legacy.db"
|
|
_make_legacy_db(legacy)
|
|
con = sqlite3.connect(legacy)
|
|
con.execute("ALTER TABLE evaluations ADD COLUMN legacy_note TEXT")
|
|
con.execute("UPDATE evaluations SET legacy_note = '老数据'")
|
|
con.commit()
|
|
con.close()
|
|
|
|
run_import(monkeypatch, *env.import_args(legacy))
|
|
out = capsys.readouterr().out
|
|
assert "老库多列 legacy_note" in out
|
|
assert _count(env.tenant_db, "evaluations") == 2
|
|
|
|
|
|
def test_orphan_rows_refused(tmp_path, monkeypatch):
|
|
"""老库孤儿行(子表 conversation_id 无父)→ 显性报错, 不静默导入悬挂数据。"""
|
|
env = Env(tmp_path, monkeypatch)
|
|
legacy = tmp_path / "legacy.db"
|
|
_make_legacy_db(legacy)
|
|
con = sqlite3.connect(legacy)
|
|
con.execute(
|
|
"INSERT INTO messages VALUES "
|
|
"('ffff0000-0000-0000-0000-000000000009', 'no-such-conv', "
|
|
"'user', '孤儿消息', '2026-09-04 10:00:00+00:00')")
|
|
con.commit()
|
|
con.close()
|
|
|
|
with pytest.raises(SystemExit, match="孤儿"):
|
|
run_import(monkeypatch, *env.import_args(legacy))
|
|
|
|
|
|
# ---- 合成库: 门禁与退出码 -----------------------------------------------------
|
|
|
|
def test_unbootstrapped_tenant_refused(tmp_path, monkeypatch):
|
|
"""租户未 bootstrap: platform.db 缺失 / slug 未注册 → 显性提示先跑
|
|
bootstrap, 绝不摸租户库。"""
|
|
legacy = tmp_path / "legacy.db"
|
|
_make_legacy_db(legacy)
|
|
no_dir = tmp_path / "tenants"
|
|
|
|
# platform.db 不存在
|
|
with pytest.raises(SystemExit, match="bootstrap"):
|
|
run_import(monkeypatch, "--source", str(legacy), "--tenant", "jiancai",
|
|
"--platform-db", str(tmp_path / "nope" / "platform.db"),
|
|
"--tenants-dir", str(no_dir))
|
|
|
|
# platform.db 存在但只注册了 auto(目标 jiancai 未注册)
|
|
platform_db = tmp_path / "platform.db"
|
|
run_bootstrap(monkeypatch, platform_db, "auto")
|
|
with pytest.raises(SystemExit, match="bootstrap"):
|
|
run_import(monkeypatch, "--source", str(legacy), "--tenant", "jiancai",
|
|
"--platform-db", str(platform_db),
|
|
"--tenants-dir", str(no_dir))
|
|
assert not no_dir.exists() # 门禁未过绝不建租户库
|
|
|
|
|
|
def test_reconciliation_mismatch_nonzero_exit(tmp_path, monkeypatch, capsys):
|
|
"""目标已有源外数据(源≠目标合计)→ 打印对账报告后非零退出。
|
|
形态: 先正常导入(目标库随之建立)→ 目标混入一条"原生"会话 → 第二遍
|
|
导入对账炸(模拟切换前误跑过流量再补导的分歧场景)。"""
|
|
env = Env(tmp_path, monkeypatch)
|
|
legacy = tmp_path / "legacy.db"
|
|
_make_legacy_db(legacy)
|
|
run_import(monkeypatch, *env.import_args(legacy))
|
|
capsys.readouterr()
|
|
|
|
pre = sqlite3.connect(env.tenant_db)
|
|
pre.execute(
|
|
"INSERT INTO conversations VALUES "
|
|
"('99999999-9999-9999-9999-999999999999', 'sess-native', NULL, "
|
|
"'active', 'standalone', NULL, '2026-09-05 10:00:00+00:00', "
|
|
"'2026-09-05 10:00:00+00:00')")
|
|
pre.commit()
|
|
pre.close()
|
|
|
|
with pytest.raises(SystemExit, match="对账不一致"):
|
|
run_import(monkeypatch, *env.import_args(legacy))
|
|
out = capsys.readouterr().out
|
|
assert "conversations" in out # 报告先于退出码打印
|
|
assert _parse_report_line(out, "conversations") == (3, 0, 3, 4)
|
|
|
|
|
|
def test_source_file_missing_refused(tmp_path, monkeypatch):
|
|
with pytest.raises(SystemExit, match="老库"):
|
|
run_import(monkeypatch, "--source", str(tmp_path / "ghost.db"),
|
|
"--tenant", "jiancai",
|
|
"--platform-db", str(tmp_path / "platform.db"),
|
|
"--tenants-dir", str(tmp_path / "tenants"))
|
|
|
|
|
|
# ---- 真实老库(env 指路, 缺失 skip)---------------------------------------------
|
|
|
|
def _legacy_copy(tmp_path: Path, env_var: str) -> tuple[Path, Path]:
|
|
"""(原库路径, VACUUM INTO 拷贝)——只读连原库做快照, 绝不直连写。"""
|
|
raw = os.environ.get(env_var)
|
|
if not raw:
|
|
pytest.skip(f"未设 {env_var}(机器相关: 指向真实老库 data/*.db 才测)")
|
|
src = Path(raw)
|
|
assert src.is_file(), f"{env_var} 指向的老库不存在: {src}"
|
|
dst = tmp_path / f"legacy-{env_var.split('_')[-1].lower()}.db"
|
|
ro = sqlite3.connect(f"file:{src}?mode=ro", uri=True)
|
|
try:
|
|
ro.execute("VACUUM INTO ?", (str(dst),))
|
|
finally:
|
|
ro.close()
|
|
return src, dst
|
|
|
|
|
|
def _real_rehearsal(tmp_path, monkeypatch, capsys, env_var, tenant):
|
|
src, legacy = _legacy_copy(tmp_path, env_var)
|
|
sha_before = _sha256(src)
|
|
platform_db = tmp_path / "platform.db"
|
|
run_bootstrap(monkeypatch, platform_db, tenant)
|
|
tenants_dir = tmp_path / "tenants"
|
|
tenant_db = tenants_dir / f"{tenant}.db"
|
|
|
|
run_import(monkeypatch, "--source", str(legacy), "--tenant", tenant,
|
|
"--platform-db", str(platform_db),
|
|
"--tenants-dir", str(tenants_dir))
|
|
out = capsys.readouterr().out
|
|
assert "对账结论: 全部一致" in out
|
|
|
|
# 五表对账 + 行级 + 抽查 + conversation_id 原样
|
|
for table in LEGACY_TABLES:
|
|
assert _count(legacy, table) == _count(tenant_db, table)
|
|
_assert_tables_identical(legacy, tenant_db)
|
|
assert _spot_rows(legacy) == _spot_rows(tenant_db)
|
|
|
|
# 幂等第二遍: 全部跳过, 零变化
|
|
run_import(monkeypatch, "--source", str(legacy), "--tenant", tenant,
|
|
"--platform-db", str(platform_db),
|
|
"--tenants-dir", str(tenants_dir))
|
|
out2 = capsys.readouterr().out
|
|
for table in LEGACY_TABLES:
|
|
n = _count(legacy, table)
|
|
assert _parse_report_line(out2, table) == (n, 0, n, n)
|
|
|
|
# 原库零接触
|
|
assert _sha256(src) == sha_before
|
|
return out
|
|
|
|
|
|
def test_real_jc_legacy_rehearsal(tmp_path, monkeypatch, capsys):
|
|
"""jc 真库(拷贝上)全流程: 导入→对账全绿→幂等→原库 sha 不变。"""
|
|
_real_rehearsal(tmp_path, monkeypatch, capsys,
|
|
"ALL_KEFU_TEST_LEGACY_JC", "jiancai")
|
|
|
|
|
|
def test_real_auto_legacy_rehearsal(tmp_path, monkeypatch, capsys):
|
|
"""auto 真库同理; 额外钉死 evaluations.correction_sent(新库已废弃)
|
|
在报告中显性标注, 不静默丢列。"""
|
|
out = _real_rehearsal(tmp_path, monkeypatch, capsys,
|
|
"ALL_KEFU_TEST_LEGACY_AUTO", "auto")
|
|
assert "老库多列 correction_sent" in out
|