448 lines
20 KiB
Python
448 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M1b 真实库集成测试（QC#5 整改）。
|
||
|
||
设计原则（对应 QC 退回意见 #5）：
|
||
* 不再只有「纯内存 FakeDB 单测」。本文件在**可连库环境真实执行** DDL + CRUD，
|
||
覆盖 4 张表的建表幂等、唯一键 tenant_key 打头去重、平台公共模板 tenant_id IS NULL、
|
||
子对象扩展 KV 读写、关联表软引用判定（Q-OPEN-3：零基表写入）。
|
||
* 环境受限时（无 MySQL/MariaDB 或未提供连接参数）**显式 SKIP 并打印移交说明**，
|
||
绝不把 SKIP 伪装成 PASS；移交部署验证的清单见 docs/M1b-qc-rework-evidence.md §5。
|
||
|
||
运行方式（三选一）：
|
||
1) 环境变量：
|
||
PBL_TEST_DB_HOST=127.0.0.1 PBL_TEST_DB_PORT=3306 \
|
||
PBL_TEST_DB_USER=root PBL_TEST_DB_PASSWORD=xxx PBL_TEST_DB_NAME=pbl_test \
|
||
python3 tests/test_m1b_realdb.py
|
||
2) 配置文件：PBL_TEST_DB_DSN=mysql://user:pass@host:3306/dbname
|
||
3) 无参数：自动探测 127.0.0.1:3306，连不上则 SKIP（退出码 0，但打印 SKIPPED）。
|
||
|
||
退出码：0 = 全部通过 或 环境受限 SKIP；1 = 有真实 FAIL；2 = 依赖缺失且未 SKIP 判定失败。
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
import _m1b_loader # noqa: F401
|
||
import traceback
|
||
from datetime import datetime
|
||
|
||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
DDL_PATH = os.path.join(REPO, "sql", "m1b_ddl.sql")
|
||
|
||
TENANT_A = "T_M1B_A"
|
||
TENANT_B = "T_M1B_B"
|
||
PLATFORM_KEY = "" # tenant_id NULL -> tenant_key ''
|
||
NOW = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
RESULTS = []
|
||
|
||
|
||
def log(ok, name, detail=""):
|
||
RESULTS.append((name, bool(ok), detail))
|
||
print("[%s] %s :: %s" % ("PASS" if ok else "FAIL", name, detail))
|
||
|
||
|
||
def skip(name, detail):
|
||
RESULTS.append((name, None, detail))
|
||
print("[SKIP] %s :: %s" % (name, detail))
|
||
|
||
|
||
# ------------------------------------------------------------------ 连接
|
||
def resolve_dsn():
|
||
dsn = os.environ.get("PBL_TEST_DB_DSN")
|
||
if dsn:
|
||
m = re.match(r"mysql(?:\+\w+)?://([^:]+):([^@]*)@([^:/]+):?(\d+)?/(.+)", dsn)
|
||
if not m:
|
||
raise ValueError("PBL_TEST_DB_DSN 格式非法: %s" % dsn)
|
||
return dict(host=m.group(3), port=int(m.group(4) or 3306),
|
||
user=m.group(1), password=m.group(2), database=m.group(5))
|
||
if os.environ.get("PBL_TEST_DB_HOST"):
|
||
return dict(host=os.environ["PBL_TEST_DB_HOST"],
|
||
port=int(os.environ.get("PBL_TEST_DB_PORT", 3306)),
|
||
user=os.environ.get("PBL_TEST_DB_USER", "root"),
|
||
password=os.environ.get("PBL_TEST_DB_PASSWORD", ""),
|
||
database=os.environ.get("PBL_TEST_DB_NAME", "pbl_test"))
|
||
return dict(host="127.0.0.1", port=3306, user="root", password="",
|
||
database="pbl_test")
|
||
|
||
|
||
def connect(cfg):
|
||
"""返回 (conn, driver_name)；驱动缺失或连不上抛异常。"""
|
||
try:
|
||
import pymysql # type: ignore
|
||
conn = pymysql.connect(host=cfg["host"], port=cfg["port"], user=cfg["user"],
|
||
password=cfg["password"], database=cfg["database"],
|
||
charset="utf8mb4", autocommit=True,
|
||
connect_timeout=5)
|
||
return conn, "pymysql"
|
||
except ImportError:
|
||
pass
|
||
try:
|
||
import MySQLdb # type: ignore
|
||
conn = MySQLdb.connect(host=cfg["host"], port=cfg["port"], user=cfg["user"],
|
||
passwd=cfg["password"], db=cfg["database"],
|
||
charset="utf8mb4")
|
||
conn.autocommit(True)
|
||
return conn, "MySQLdb"
|
||
except ImportError:
|
||
pass
|
||
try:
|
||
import mariadb # type: ignore
|
||
conn = mariadb.connect(host=cfg["host"], port=cfg["port"], user=cfg["user"],
|
||
password=cfg["password"], database=cfg["database"])
|
||
conn.autocommit = True
|
||
return conn, "mariadb"
|
||
except ImportError:
|
||
raise RuntimeError("NO_DRIVER: 未安装 pymysql / MySQLdb / mariadb 任一驱动")
|
||
|
||
|
||
def split_ddl(sql_text):
|
||
body = re.sub(r"/\*.*?\*/", " ", sql_text, flags=re.S)
|
||
stmts, buf = [], []
|
||
for line in body.splitlines():
|
||
i = line.find("--")
|
||
if i >= 0:
|
||
line = line[:i]
|
||
buf.append(line)
|
||
if line.rstrip().endswith(";"):
|
||
s = "\n".join(buf).strip()
|
||
if s and s != ";":
|
||
stmts.append(s.rstrip(";"))
|
||
buf = []
|
||
return [s for s in stmts if s.strip()]
|
||
|
||
|
||
# ------------------------------------------------------------------ 用例
|
||
def t_ddl_idempotent(cur):
|
||
"""T1: DDL 执行两遍均成功（CREATE TABLE IF NOT EXISTS 幂等）。"""
|
||
with open(DDL_PATH, "r", encoding="utf-8") as f:
|
||
stmts = split_ddl(f.read())
|
||
for rnd in (1, 2):
|
||
for s in stmts:
|
||
cur.execute(s)
|
||
log(len(stmts) == 4, "T1_DDL幂等",
|
||
"执行 %d 条语句 × 2 轮无异常" % len(stmts))
|
||
|
||
|
||
def t_tables_exist(cur):
|
||
"""T2: 4 张表全部存在，且引擎/字符集符合 mariadb 方言。"""
|
||
cur.execute("""SELECT TABLE_NAME, ENGINE, TABLE_COLLATION
|
||
FROM information_schema.TABLES
|
||
WHERE TABLE_SCHEMA = DATABASE()
|
||
AND TABLE_NAME IN ('pbl_blueprint_template','pbl_blueprint_ref',
|
||
'pbl_subobject_ext','pbl_ext_field_def')""")
|
||
rows = cur.fetchall()
|
||
names = sorted(r[0] for r in rows)
|
||
expect = sorted(['pbl_blueprint_template', 'pbl_blueprint_ref',
|
||
'pbl_subobject_ext', 'pbl_ext_field_def'])
|
||
engines = set(r[1] for r in rows)
|
||
log(names == expect and engines == {"InnoDB"}, "T2_四表存在",
|
||
"表=%s 引擎=%s" % (names, engines))
|
||
|
||
|
||
def t_no_foreign_key(cur):
|
||
"""T3: 4 张表零 FOREIGN KEY（Q-OPEN-3）。"""
|
||
cur.execute("""SELECT COUNT(*) FROM information_schema.KEY_COLUMN_USAGE
|
||
WHERE TABLE_SCHEMA = DATABASE()
|
||
AND REFERENCED_TABLE_NAME IS NOT NULL
|
||
AND TABLE_NAME LIKE 'pbl\\_%'""")
|
||
n = cur.fetchone()[0]
|
||
log(n == 0, "T3_零外键", "information_schema 外键约束数=%d" % n)
|
||
|
||
|
||
def t_unique_key_tenant_first(cur):
|
||
"""T4: 每个 UNIQUE 索引首列均为 tenant_key。"""
|
||
cur.execute("""SELECT TABLE_NAME, INDEX_NAME, COLUMN_NAME, SEQ_IN_INDEX
|
||
FROM information_schema.STATISTICS
|
||
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME LIKE 'pbl\\_%'
|
||
AND NON_UNIQUE = 0 AND INDEX_NAME <> 'PRIMARY'
|
||
ORDER BY TABLE_NAME, INDEX_NAME, SEQ_IN_INDEX""")
|
||
bad = []
|
||
seen = {}
|
||
for t, ix, col, seq in cur.fetchall():
|
||
if int(seq) == 1:
|
||
seen[(t, ix)] = col
|
||
for (t, ix), col in seen.items():
|
||
if col != "tenant_key":
|
||
bad.append("%s.%s 首列=%s" % (t, ix, col))
|
||
log(seen and not bad, "T4_唯一键tenant打头",
|
||
"唯一索引 %d 个，违规 %s" % (len(seen), bad or "无"))
|
||
|
||
|
||
def t_platform_template_null_tenant(cur):
|
||
"""T5: 平台公共模板 tenant_id IS NULL 可插入，且 tenant_key='' 唯一去重生效。"""
|
||
cur.execute("DELETE FROM pbl_blueprint_template WHERE template_code LIKE 'M1B_T5%%'")
|
||
cur.execute(
|
||
"""INSERT INTO pbl_blueprint_template
|
||
(tenant_key, tenant_id, template_code, template_name, scope, status,
|
||
version, is_default, usage_count, created_at, updated_at)
|
||
VALUES (%s, NULL, %s, %s, 'platform', 'published', '1.0.0', 0, 0, %s, %s)""",
|
||
(PLATFORM_KEY, "M1B_T5_PLATFORM", "平台公共模板", NOW, NOW))
|
||
cur.execute(
|
||
"""SELECT id, tenant_id, scope FROM pbl_blueprint_template
|
||
WHERE tenant_key = %s AND template_code = 'M1B_T5_PLATFORM'""",
|
||
(PLATFORM_KEY,))
|
||
row = cur.fetchone()
|
||
ok_null = row is not None and row[1] is None and row[2] == "platform"
|
||
|
||
dup = False
|
||
try:
|
||
cur.execute(
|
||
"""INSERT INTO pbl_blueprint_template
|
||
(tenant_key, tenant_id, template_code, template_name, scope, status,
|
||
version, is_default, usage_count, created_at, updated_at)
|
||
VALUES (%s, NULL, %s, %s, 'platform', 'published', '1.0.0', 0, 0, %s, %s)""",
|
||
(PLATFORM_KEY, "M1B_T5_PLATFORM", "平台公共模板重复", NOW, NOW))
|
||
except Exception:
|
||
dup = True
|
||
log(ok_null and dup, "T5_平台模板NULL租户",
|
||
"tenant_id IS NULL 插入=%s；同 tenant_key+code 重复插入被唯一键拒绝=%s"
|
||
% (ok_null, dup))
|
||
|
||
|
||
def t_tenant_isolation(cur):
|
||
"""T6: 同 template_code 在不同租户下可共存（tenant_key 打头唯一键）。"""
|
||
cur.execute("DELETE FROM pbl_blueprint_template WHERE template_code = 'M1B_T6_SAME'")
|
||
for tk, tid in ((TENANT_A, TENANT_A), (TENANT_B, TENANT_B)):
|
||
cur.execute(
|
||
"""INSERT INTO pbl_blueprint_template
|
||
(tenant_key, tenant_id, template_code, template_name, scope, status,
|
||
version, is_default, usage_count, created_at, updated_at)
|
||
VALUES (%s, %s, 'M1B_T6_SAME', %s, 'tenant', 'draft', '1.0.0', 0, 0, %s, %s)""",
|
||
(tk, tid, "租户模板-" + tk, NOW, NOW))
|
||
cur.execute("SELECT tenant_key FROM pbl_blueprint_template "
|
||
"WHERE template_code = 'M1B_T6_SAME' ORDER BY tenant_key")
|
||
got = [r[0] for r in cur.fetchall()]
|
||
log(got == [TENANT_A, TENANT_B], "T6_租户隔离",
|
||
"同 code 跨租户共存，tenant_key=%s" % got)
|
||
|
||
|
||
def t_subobject_ext_kv(cur):
|
||
"""T7: 子对象扩展 KV 写入/读取/唯一键覆盖（基表零改动）。"""
|
||
cur.execute("DELETE FROM pbl_subobject_ext WHERE subobject_id LIKE 'M1B_T7%%'")
|
||
cur.execute(
|
||
"""INSERT INTO pbl_subobject_ext
|
||
(tenant_key, tenant_id, blueprint_id, subobject_type, subobject_id,
|
||
ext_key, ext_value, ext_value_type, source, status, created_at, updated_at)
|
||
VALUES (%s, %s, 900001, 'driving_question', 'M1B_T7_OBJ',
|
||
'difficulty', '3', 'number', 'manual', 'active', %s, %s)""",
|
||
(TENANT_A, TENANT_A, NOW, NOW))
|
||
cur.execute(
|
||
"""SELECT ext_value, ext_value_type FROM pbl_subobject_ext
|
||
WHERE tenant_key = %s AND subobject_type = 'driving_question'
|
||
AND subobject_id = 'M1B_T7_OBJ' AND ext_key = 'difficulty'""",
|
||
(TENANT_A,))
|
||
row = cur.fetchone()
|
||
ok_read = row is not None and row[0] == "3" and row[1] == "number"
|
||
|
||
dup = False
|
||
try:
|
||
cur.execute(
|
||
"""INSERT INTO pbl_subobject_ext
|
||
(tenant_key, tenant_id, blueprint_id, subobject_type, subobject_id,
|
||
ext_key, ext_value, ext_value_type, source, status, created_at, updated_at)
|
||
VALUES (%s, %s, 900001, 'driving_question', 'M1B_T7_OBJ',
|
||
'difficulty', '5', 'number', 'manual', 'active', %s, %s)""",
|
||
(TENANT_A, TENANT_A, NOW, NOW))
|
||
except Exception:
|
||
dup = True
|
||
log(ok_read and dup, "T7_子对象扩展KV",
|
||
"读回=%s；同 (tenant,type,obj,key) 重复插入被拒=%s" % (ok_read, dup))
|
||
|
||
|
||
def t_ext_field_def_platform_override(cur):
|
||
"""T8: 平台公共字段定义 + 租户级覆盖（同 field_key 两条并存，读取顺序 tenant->platform）。"""
|
||
cur.execute("DELETE FROM pbl_ext_field_def WHERE field_key LIKE 'm1b_t8%%'")
|
||
cur.execute(
|
||
"""INSERT INTO pbl_ext_field_def
|
||
(tenant_key, tenant_id, subobject_type, field_key, field_label, field_type,
|
||
required, scope, sort_no, status, created_at, updated_at)
|
||
VALUES ('', NULL, 'driving_question', 'm1b_t8_difficulty', '难度(平台)',
|
||
'number', 0, 'platform', 10, 'active', %s, %s)""", (NOW, NOW))
|
||
cur.execute(
|
||
"""INSERT INTO pbl_ext_field_def
|
||
(tenant_key, tenant_id, subobject_type, field_key, field_label, field_type,
|
||
required, scope, sort_no, status, created_at, updated_at)
|
||
VALUES (%s, %s, 'driving_question', 'm1b_t8_difficulty', '难度(租户覆盖)',
|
||
'number', 1, 'tenant', 10, 'active', %s, %s)""",
|
||
(TENANT_A, TENANT_A, NOW, NOW))
|
||
cur.execute(
|
||
"""SELECT tenant_key, field_label, required FROM pbl_ext_field_def
|
||
WHERE subobject_type = 'driving_question' AND field_key = 'm1b_t8_difficulty'
|
||
AND status = 'active'
|
||
ORDER BY CASE WHEN tenant_key = %s THEN 0 ELSE 1 END""", (TENANT_A,))
|
||
rows = cur.fetchall()
|
||
ok = len(rows) == 2 and rows[0][0] == TENANT_A and int(rows[0][2]) == 1
|
||
log(ok, "T8_字段定义平台+租户覆盖",
|
||
"命中 %d 条，优先级首条 tenant_key=%s required=%s"
|
||
% (len(rows), rows[0][0] if rows else None, rows[0][2] if rows else None))
|
||
|
||
|
||
def t_blueprint_ref_softlink(cur):
|
||
"""T9: 关联表软引用（ref_table/ref_id 字符串），且对基表零写入。"""
|
||
cur.execute("DELETE FROM pbl_blueprint_ref WHERE ref_id LIKE 'M1B_T9%%'")
|
||
before = _base_table_fingerprint(cur)
|
||
cur.execute(
|
||
"""INSERT INTO pbl_blueprint_ref
|
||
(tenant_key, tenant_id, blueprint_id, ref_type, ref_table, ref_id, ref_name,
|
||
relation_kind, sort_no, status, created_at, updated_at)
|
||
VALUES (%s, %s, 900001, 'world', 'world', 'M1B_T9_WORLD', '测试世界',
|
||
'reference', 1, 'active', %s, %s)""",
|
||
(TENANT_A, TENANT_A, NOW, NOW))
|
||
cur.execute(
|
||
"""SELECT ref_type, ref_table, ref_id, relation_kind FROM pbl_blueprint_ref
|
||
WHERE tenant_key = %s AND ref_id = 'M1B_T9_WORLD'""", (TENANT_A,))
|
||
row = cur.fetchone()
|
||
after = _base_table_fingerprint(cur)
|
||
ok = (row is not None and row[0] == "world" and row[1] == "world"
|
||
and row[3] == "reference" and before == after)
|
||
log(ok, "T9_关联表软引用零基表写入",
|
||
"软引用行=%s；基表指纹前后一致=%s" % (row is not None, before == after))
|
||
|
||
|
||
def _base_table_fingerprint(cur):
|
||
"""基表存在性+行数指纹（若基表不存在则记为 absent，不影响判定）。"""
|
||
fp = {}
|
||
for t in ("world", "scene", "entity", "script"):
|
||
try:
|
||
cur.execute(
|
||
"SELECT COUNT(*) FROM information_schema.TABLES "
|
||
"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s", (t,))
|
||
if cur.fetchone()[0] == 0:
|
||
fp[t] = "absent"
|
||
continue
|
||
cur.execute("SELECT COUNT(*) FROM `%s`" % t)
|
||
fp[t] = cur.fetchone()[0]
|
||
except Exception:
|
||
fp[t] = "error"
|
||
return json.dumps(fp, sort_keys=True)
|
||
|
||
|
||
def t_ref_unique_dedup(cur):
|
||
"""T10: 同 (tenant, blueprint, ref_type, ref_id) 重复关联被唯一键拒绝。"""
|
||
dup = False
|
||
try:
|
||
cur.execute(
|
||
"""INSERT INTO pbl_blueprint_ref
|
||
(tenant_key, tenant_id, blueprint_id, ref_type, ref_table, ref_id,
|
||
ref_name, relation_kind, sort_no, status, created_at, updated_at)
|
||
VALUES (%s, %s, 900001, 'world', 'world', 'M1B_T9_WORLD', '重复',
|
||
'reference', 2, 'active', %s, %s)""",
|
||
(TENANT_A, TENANT_A, NOW, NOW))
|
||
except Exception:
|
||
dup = True
|
||
log(dup, "T10_关联唯一去重", "重复关联被唯一键拒绝=%s" % dup)
|
||
|
||
|
||
def t_template_instantiate_payload(cur):
|
||
"""T11: 模板实例化——payload 深拷贝到新蓝图关联条目（JSON 可解析）。"""
|
||
payload = json.dumps({"driving_question": [{"title": "如何减少碳排", "difficulty": 3}],
|
||
"artifact_def": [{"code": "A1"}]}, ensure_ascii=False)
|
||
cur.execute("DELETE FROM pbl_blueprint_template WHERE template_code = 'M1B_T11_TPL'")
|
||
cur.execute(
|
||
"""INSERT INTO pbl_blueprint_template
|
||
(tenant_key, tenant_id, template_code, template_name, scope, status, version,
|
||
is_default, usage_count, blueprint_payload, subobject_payload, ref_payload,
|
||
created_at, updated_at)
|
||
VALUES (%s, %s, 'M1B_T11_TPL', '实例化模板', 'tenant', 'published', '1.0.0',
|
||
0, 0, %s, %s, %s, %s, %s)""",
|
||
(TENANT_A, TENANT_A, payload, payload, "[]", NOW, NOW))
|
||
cur.execute("SELECT blueprint_payload, subobject_payload, usage_count "
|
||
"FROM pbl_blueprint_template WHERE tenant_key = %s "
|
||
"AND template_code = 'M1B_T11_TPL'", (TENANT_A,))
|
||
bp, sp, usage = cur.fetchone()
|
||
ok_json = isinstance(json.loads(bp), dict) and isinstance(json.loads(sp), dict)
|
||
cur.execute("UPDATE pbl_blueprint_template SET usage_count = usage_count + 1, "
|
||
"updated_at = %s WHERE tenant_key = %s AND template_code = 'M1B_T11_TPL'",
|
||
(NOW, TENANT_A))
|
||
cur.execute("SELECT usage_count FROM pbl_blueprint_template WHERE tenant_key = %s "
|
||
"AND template_code = 'M1B_T11_TPL'", (TENANT_A,))
|
||
new_usage = cur.fetchone()[0]
|
||
log(ok_json and int(usage) == 0 and int(new_usage) == 1, "T11_模板实例化payload",
|
||
"payload JSON 可解析=%s；usage_count %s -> %s" % (ok_json, usage, new_usage))
|
||
|
||
|
||
def t_cleanup(cur):
|
||
"""T12: 清理测试数据（只删本测试写入的 pbl_ 行，绝不触碰基表）。"""
|
||
n = 0
|
||
for sql in (
|
||
"DELETE FROM pbl_blueprint_template WHERE template_code LIKE 'M1B_T%'",
|
||
"DELETE FROM pbl_subobject_ext WHERE subobject_id LIKE 'M1B_T%'",
|
||
"DELETE FROM pbl_ext_field_def WHERE field_key LIKE 'm1b_t8%'",
|
||
"DELETE FROM pbl_blueprint_ref WHERE ref_id LIKE 'M1B_T%'",
|
||
):
|
||
rc = cur.execute(sql) # 只执行一次，返回受影响行数
|
||
n += int(rc or 0)
|
||
log(True, "T12_测试数据清理", "清理 %d 行（仅 pbl_ 自有表，零基表操作）" % n)
|
||
|
||
|
||
CASES = [
|
||
t_ddl_idempotent, t_tables_exist, t_no_foreign_key, t_unique_key_tenant_first,
|
||
t_platform_template_null_tenant, t_tenant_isolation, t_subobject_ext_kv,
|
||
t_ext_field_def_platform_override, t_blueprint_ref_softlink, t_ref_unique_dedup,
|
||
t_template_instantiate_payload, t_cleanup,
|
||
]
|
||
|
||
|
||
def main():
|
||
print("=" * 72)
|
||
print("M1b 真实库集成测试 DDL=%s" % DDL_PATH)
|
||
print("=" * 72)
|
||
if not os.path.exists(DDL_PATH):
|
||
print("FATAL: 未找到 sql/m1b_ddl.sql")
|
||
return 2
|
||
|
||
try:
|
||
cfg = resolve_dsn()
|
||
conn, driver = connect(cfg)
|
||
except Exception as e:
|
||
msg = str(e)
|
||
print("-" * 72)
|
||
print("SKIPPED（环境受限，未连真实库）：%s" % msg)
|
||
print("连接参数尝试：%s" % json.dumps(
|
||
{k: v for k, v in (resolve_dsn() or {}).items() if k != "password"},
|
||
ensure_ascii=False))
|
||
print("移交部署验证：本文件 12 个用例（T1~T12）须在部署环境执行，"
|
||
"清单见 docs/M1b-qc-rework-evidence.md §5。")
|
||
print("复跑命令：PBL_TEST_DB_DSN=mysql://user:pass@host:3306/db "
|
||
"python3 tests/test_m1b_realdb.py")
|
||
for c in CASES:
|
||
skip(c.__name__, "环境受限未执行（不视为已完成自测）")
|
||
print("-" * 72)
|
||
print("结果：0 PASS / 0 FAIL / %d SKIP" % len(CASES))
|
||
return 0
|
||
|
||
print("已连接真实库 driver=%s host=%s:%s db=%s"
|
||
% (driver, cfg["host"], cfg["port"], cfg["database"]))
|
||
print("-" * 72)
|
||
cur = conn.cursor()
|
||
failed = 0
|
||
for case in CASES:
|
||
try:
|
||
case(cur)
|
||
except Exception as e:
|
||
failed += 1
|
||
log(False, case.__name__, "异常 %s\n%s" % (e, traceback.format_exc()))
|
||
try:
|
||
cur.close()
|
||
conn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
total = len(RESULTS)
|
||
passed = sum(1 for _, ok, _ in RESULTS if ok is True)
|
||
fails = sum(1 for _, ok, _ in RESULTS if ok is False)
|
||
print("-" * 72)
|
||
print("结果：%d PASS / %d FAIL / 共 %d" % (passed, fails, total))
|
||
for name, ok, detail in RESULTS:
|
||
if ok is False:
|
||
print(" FAIL %s :: %s" % (name, detail))
|
||
return 1 if fails else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|