baskbuddy
/
zhangshun
13
0
Fork 1
Code Issues Pull Requests Projects Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
69 Commits
11 Branches
0 Tags
7.8 MiB
 
 
 
Tree: 2770d4c544
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '2770d4c544'
${ noResults }
zhangshun/docs/tools/b01_demo_display_data_test.py

594 lines
26 KiB
Raw Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""B01 演示数据工具的**归属安全**测试（隔离库 + 合成数据 + 哨兵数据）。
自包含：启动**仅本测试使用**的 MySQL 容器（b01-demo-test-mysql），建库
``mydreamjob_b01_demo_test``、合成业务表结构 + 哨兵数据，逐项验证工具的归属安全。
结束时只销毁本测试容器（不触碰其他容器/数据卷）。
不能只用全空库证明安全：冲突与哨兵数据是本项验收重点。
用法：python docs/tools/b01_demo_display_data_test.py
退出码：0=全部通过；1=存在失败；2=环境不可用。
"""
import argparse
import json
import os
import subprocess
import sys
import time
import pymysql
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import b01_demo_display_data as tool # noqa: E402
CONTAINER = "qsp-final-demo-1019"
IMAGE = "mysql:8.4"
HOST = "127.0.0.1"
PORT = 3308
DB = "mydreamjob_b01_demo_test"
APP_USER = "b01demo"
APP_PASS = "b01-demo-only"
ROOT_PASS = "b01-root-test"
DEMO_PHONES = [c["phone"] for c in tool.COMPANIES]
DEMO_NAMES = [c["name"] for c in tool.COMPANIES]
DEMO_TITLES = [(j["phone"], j["title"]) for j in tool.JOBS]
RESULTS = []
OWN_CONTAINER_ID = None
def check(case, ok, detail=""):
RESULTS.append(dict(case=case, ok=bool(ok), detail=str(detail)[:500]))
print("[%s] %-60s %s" % ("PASS" if ok else "FAIL", case, str(detail)[:120]))
return bool(ok)
# ── Docker / 连接 ───────────────────────────────────────────────────
def _run(cmd):
return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8",
errors="replace")
def start_container():
global OWN_CONTAINER_ID
if _run(["docker", "inspect", CONTAINER]).returncode == 0:
raise RuntimeError("同名容器存在，拒绝删除或接管：" + CONTAINER)
r = _run(["docker", "run", "-d", "--name", CONTAINER,
"-e", "MYSQL_ROOT_PASSWORD=" + ROOT_PASS,
"-e", "MYSQL_ROOT_HOST=%",
"-e", "MYSQL_DATABASE=" + DB,
"-p", "127.0.0.1:%d:3306" % PORT, IMAGE,
"--character-set-server=utf8mb4", "--collation-server=utf8mb4_unicode_ci"])
if r.returncode != 0:
raise RuntimeError("docker run 失败: " + (r.stderr or "")[:400])
OWN_CONTAINER_ID = r.stdout.strip()
for _ in range(90):
try:
c = root_conn()
c.close()
return
except Exception:
time.sleep(2)
raise RuntimeError("MySQL 容器未就绪")
def stop_container():
if OWN_CONTAINER_ID:
_run(["docker", "rm", "-f", OWN_CONTAINER_ID])
def root_conn(db=DB, autocommit=True):
return pymysql.connect(host=HOST, port=PORT, user="root", password=ROOT_PASS,
database=db, charset="utf8mb4", autocommit=autocommit)
def app_conn():
return pymysql.connect(host=HOST, port=PORT, user=APP_USER, password=APP_PASS,
database=DB, charset="utf8mb4", autocommit=False)
class Args:
def __init__(self):
self.host, self.port, self.database = HOST, PORT, DB
def identity_of(cur):
return tool.target_identity(cur, Args())
# ── 合成表结构 + 哨兵数据 ───────────────────────────────────────────
DDL = [
"DROP TABLE IF EXISTS applications",
"DROP TABLE IF EXISTS b01_demo_ledger",
"DROP TABLE IF EXISTS jobs",
"DROP TABLE IF EXISTS companies",
"DROP TABLE IF EXISTS user_roles",
"DROP TABLE IF EXISTS users",
"""CREATE TABLE users (
id BIGINT AUTO_INCREMENT PRIMARY KEY, phone VARCHAR(20) NOT NULL UNIQUE,
email VARCHAR(100), nickname VARCHAR(50), avatar_url VARCHAR(500),
password_hash VARCHAR(255), role VARCHAR(30) NOT NULL DEFAULT 'STUDENT',
is_active TINYINT(1) NOT NULL DEFAULT 1, is_verified TINYINT(1) NOT NULL DEFAULT 0,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""",
"""CREATE TABLE user_roles (
user_id BIGINT NOT NULL, role VARCHAR(30) NOT NULL, PRIMARY KEY (user_id, role),
CONSTRAINT fk_ur FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""",
"""CREATE TABLE companies (
id BIGINT AUTO_INCREMENT PRIMARY KEY, owner_user_id BIGINT NOT NULL,
name VARCHAR(120) NOT NULL, logo_url VARCHAR(500), description TEXT,
industry VARCHAR(80), website VARCHAR(255), address VARCHAR(255),
company_size VARCHAR(40) NULL, verified TINYINT(1) NOT NULL DEFAULT 0,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
CONSTRAINT uk_owner UNIQUE (owner_user_id),
CONSTRAINT fk_co FOREIGN KEY (owner_user_id) REFERENCES users(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""",
"""CREATE TABLE jobs (
id BIGINT AUTO_INCREMENT PRIMARY KEY, company_id BIGINT NOT NULL,
title VARCHAR(120) NOT NULL, description TEXT NOT NULL, requirements TEXT,
city VARCHAR(80) NOT NULL, employment_type VARCHAR(30) NOT NULL,
salary_min INT, salary_max INT, status VARCHAR(20) NOT NULL DEFAULT 'DRAFT',
published_at DATETIME, experience VARCHAR(40), education VARCHAR(40),
benefits_json TEXT, salary_months INT, work_location VARCHAR(255),
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
CONSTRAINT fk_job_co FOREIGN KEY (company_id) REFERENCES companies(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""",
"""CREATE TABLE applications (
id BIGINT AUTO_INCREMENT PRIMARY KEY, job_id BIGINT NOT NULL,
student_user_id BIGINT NOT NULL,
CONSTRAINT fk_app_job FOREIGN KEY (job_id) REFERENCES jobs(id),
CONSTRAINT fk_app_stu FOREIGN KEY (student_user_id) REFERENCES users(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""",
]
def reset_schema(cur):
for stmt in DDL:
cur.execute(stmt)
def seed_sentinels(cur):
cur.execute("INSERT INTO users(phone, email, nickname, role, is_active, is_verified) "
"VALUES('13900000001','s1@x.com','哨兵甲','STUDENT',1,0)")
s1 = cur.lastrowid
cur.execute("INSERT INTO user_roles(user_id, role) VALUES(%s,'STUDENT')", (s1,))
cur.execute("INSERT INTO users(phone, nickname, role) VALUES('13900000002','哨兵乙','STUDENT')")
s2 = cur.lastrowid
cur.execute("INSERT INTO companies(owner_user_id, name, description, industry, company_size, "
"verified) VALUES(%s,'哨兵企业（非本批）','哨兵简介','其它','10-19人',0)", (s2,))
sc = cur.lastrowid
cur.execute("INSERT INTO jobs(company_id,title,description,city,employment_type,status,"
"experience,education,benefits_json,salary_months,work_location) "
"VALUES(%s,'哨兵岗位','哨兵描述','北京','FULL_TIME','PUBLISHED','1-3年','本科',"
"'[\"哨兵福利\"]',12,'北京市海淀区')", (sc,))
sj = cur.lastrowid
return dict(user1=s1, user2=s2, company=sc, job=sj)
# ── 快照工具 ────────────────────────────────────────────────────────
def table_rows(cur, table, order="1"):
cur.execute("SELECT * FROM `%s` ORDER BY %s" % (table, order))
cols = [d[0] for d in cur.description]
return [dict(zip(cols, row)) for row in cur.fetchall()]
def norm(rows):
return json.dumps([{k: (None if v is None else str(v)) for k, v in r.items()} for r in rows],
ensure_ascii=False, sort_keys=True)
def snapshot(cur, tables=("users", "user_roles", "companies", "jobs")):
return {t: norm(table_rows(cur, t)) for t in tables}
def count(cur, sql, params=None):
cur.execute(sql, params if params else None)
return int(cur.fetchone()[0])
def demo_counts(cur):
fmt_p = ",".join(["%s"] * len(DEMO_PHONES))
fmt_n = ",".join(["%s"] * len(DEMO_NAMES))
users = count(cur, "SELECT COUNT(*) FROM users WHERE phone IN (%s)" % fmt_p, tuple(DEMO_PHONES))
comps = count(cur, "SELECT COUNT(*) FROM companies WHERE name IN (%s)" % fmt_n, tuple(DEMO_NAMES))
jobs = 0
for phone, title in DEMO_TITLES:
jobs += count(cur, "SELECT COUNT(*) FROM jobs j JOIN companies c ON c.id=j.company_id "
"JOIN users u ON u.id=c.owner_user_id WHERE u.phone=%s AND j.title=%s",
(phone, title))
return dict(users=users, companies=comps, jobs=jobs)
def ledger_state(cur, identity):
if not tool.table_exists(cur, tool.LEDGER_TABLE):
return None
rows = tool.read_ledger(cur, identity)
types = {}
for r in rows:
types[r["resource_type"]] = types.get(r["resource_type"], 0) + 1
return dict(total=len(rows), types=types)
# ══ 测试用例 ════════════════════════════════════════════════════════
def t01_first_import(conn, cur):
reset_schema(cur); seed_sentinels(cur)
ident = identity_of(cur)
created, conflicts = tool.apply_import(conn, cur, ident)
check("T1 空库首次导入无冲突", conflicts == [], conflicts)
check("T1 新建 3 企业/12 岗位(含 3 用户)", created == dict(users=3, companies=3, jobs=12), created)
dc = demo_counts(cur)
check("T1 库内本批计数 = 3/3/12", dc == dict(users=3, companies=3, jobs=12), dc)
ls = ledger_state(cur, ident)
check("T1 账本登记 18 行(3/3/12)", ls and ls["total"] == 18
and ls["types"] == dict(USER=3, COMPANY=3, JOB=12), ls)
def t02_repeat_import_idempotent(conn, cur):
ident = identity_of(cur)
before = snapshot(cur)
before_ledger = ledger_state(cur, ident)
created, conflicts = tool.apply_import(conn, cur, ident)
check("T2 重复导入无冲突且不重复创建",
conflicts == [] and created == dict(users=0, companies=0, jobs=0), (created, conflicts))
check("T2 既有业务资料逐字段不变", snapshot(cur) == before)
check("T2 账本不变", ledger_state(cur, ident) == before_ledger, ledger_state(cur, ident))
def t03_occupied_phone_refused(conn, cur):
reset_schema(cur)
cur.execute("INSERT INTO users(phone,nickname,role,is_active,is_verified) "
"VALUES(%s,'他人账号','STUDENT',1,0)", (DEMO_PHONES[0],))
occ = cur.lastrowid
cur.execute("INSERT INTO user_roles(user_id,role) VALUES(%s,'STUDENT')", (occ,))
conn.commit()
before_u = norm(table_rows(cur, "users")); before_r = norm(table_rows(cur, "user_roles"))
ident = identity_of(cur)
created, conflicts = tool.apply_import(conn, cur, ident)
check("T3 固定手机号被非本批占用 → 拒绝", conflicts and created is None, conflicts[:1])
check("T3 占用账号完全不变(users)", norm(table_rows(cur, "users")) == before_u)
check("T3 占用账号完全不变(user_roles)", norm(table_rows(cur, "user_roles")) == before_r)
check("T3 未创建任何本批业务数据", count(cur, "SELECT COUNT(*) FROM companies") == 0
and count(cur, "SELECT COUNT(*) FROM jobs") == 0)
def t04_nonbatch_company_and_title_refused(conn, cur):
# 4a 同名企业被非本批占用
reset_schema(cur)
cur.execute("INSERT INTO users(phone,nickname,role) VALUES('13911112222','他人','COMPANY')")
other = cur.lastrowid
cur.execute("INSERT INTO companies(owner_user_id,name,description,verified) "
"VALUES(%s,%s,'他人企业',0)", (other, DEMO_NAMES[0]))
oc = cur.lastrowid
conn.commit()
b_co = norm(table_rows(cur, "companies"))
ident = identity_of(cur)
created, conflicts = tool.apply_import(conn, cur, ident)
check("T4a 同名企业被非本批占用 → 拒绝", conflicts and created is None, conflicts[:1])
check("T4a 他方企业不被覆盖/认领", norm(table_rows(cur, "companies")) == b_co)
check("T4a 未新增本批企业/岗位", count(cur, "SELECT COUNT(*) FROM companies") == 1
and count(cur, "SELECT COUNT(*) FROM jobs") == 0)
# 4b 同名岗位：本批企业下已存在同名岗位但账本缺失 → 拒绝，不认领
reset_schema(cur)
ident = identity_of(cur)
tool.apply_import(conn, cur, ident)
cur.execute("DELETE FROM b01_demo_ledger WHERE resource_type='JOB' ORDER BY id LIMIT 1")
conn.commit()
b_j = norm(table_rows(cur, "jobs")); b_led = norm(table_rows(cur, "b01_demo_ledger"))
created, conflicts = tool.apply_import(conn, cur, ident)
check("T4b 同名岗位无本批账本归属 → 拒绝", conflicts and created is None, conflicts[:1])
check("T4b 岗位与账本均不被改写",
norm(table_rows(cur, "jobs")) == b_j and norm(table_rows(cur, "b01_demo_ledger")) == b_led)
def t05_demo_text_without_ledger_not_deleted(conn, cur):
reset_schema(cur)
# 只有演示文字标记、没有账本
cur.execute("INSERT INTO users(phone,nickname,role) VALUES(%s,%s,'COMPANY')",
(DEMO_PHONES[0], tool.DEMO_MARK))
u = cur.lastrowid
cur.execute("INSERT INTO companies(owner_user_id,name,description,verified) "
"VALUES(%s,%s,%s,1)", (u, DEMO_NAMES[0], tool.DEMO_MARK))
c = cur.lastrowid
cur.execute("INSERT INTO jobs(company_id,title,description,city,employment_type,status) "
"VALUES(%s,%s,%s,'上海','FULL_TIME','PUBLISHED')", (c, DEMO_TITLES[0][1], tool.DEMO_MARK))
conn.commit()
before = snapshot(cur)
ident = identity_of(cur)
removed, conflicts = tool.apply_cleanup(conn, cur, ident)
check("T5 无账本 → 拒绝清理", removed is None and conflicts, conflicts[:1])
check("T5 演示文字记录逐字段不变", snapshot(cur) == before)
check("T5 未创建账本表", not tool.table_exists(cur, tool.LEDGER_TABLE))
def t06_ledger_lost_or_identity_changed(conn, cur):
ident = identity_of(cur)
# 6a 账本整表丢失后清理 → 拒绝
reset_schema(cur)
tool.apply_import(conn, cur, ident)
cur.execute("DROP TABLE b01_demo_ledger"); conn.commit()
before = snapshot(cur)
removed, conflicts = tool.apply_cleanup(conn, cur, ident)
check("T6a 账本表丢失 → 拒绝清理", removed is None and conflicts, conflicts[:1])
check("T6a 本批数据未被删除", snapshot(cur) == before)
# 6b 资源身份变化（企业改名）→ 拒绝清理
reset_schema(cur)
tool.apply_import(conn, cur, ident)
cur.execute("UPDATE companies SET name=%s WHERE name=%s", ("被改名企业", DEMO_NAMES[0]))
conn.commit()
b1 = snapshot(cur)
removed, conflicts = tool.apply_cleanup(conn, cur, ident)
check("T6b 企业身份变化 → 拒绝清理", removed is None and conflicts, conflicts[:1])
check("T6b 改名后数据未被删除", snapshot(cur) == b1)
# 6c 账本 ID 指向异常（把某条 JOB 账本指向哨兵岗位）→ 拒绝
reset_schema(cur); seed_sentinels(cur)
tool.apply_import(conn, cur, ident)
sentinel_job = count(cur, "SELECT id FROM jobs WHERE title='哨兵岗位'")
cur.execute("UPDATE b01_demo_ledger SET resource_id=%s WHERE resource_type='JOB' LIMIT 1",
(sentinel_job,)); conn.commit()
b2 = snapshot(cur)
removed, conflicts = tool.apply_cleanup(conn, cur, ident)
check("T6c 账本 ID 指向异常 → 拒绝清理", removed is None and conflicts, conflicts[:1])
check("T6c 哨兵与业务数据均未被删除", snapshot(cur) == b2)
def t07_external_dependency_refused(conn, cur):
reset_schema(cur); seed_sentinels(cur)
ident = identity_of(cur)
tool.apply_import(conn, cur, ident)
# 新增非本批依赖：学生投递到本批岗位
cur.execute("INSERT INTO users(phone,nickname,role) VALUES('13922223333','外部学生','STUDENT')")
stu = cur.lastrowid
job = count(cur, "SELECT j.id FROM jobs j JOIN companies c ON c.id=j.company_id "
"JOIN users u ON u.id=c.owner_user_id WHERE u.phone=%s AND j.title=%s",
(DEMO_TITLES[0][0], DEMO_TITLES[0][1]))
cur.execute("INSERT INTO applications(job_id, student_user_id) VALUES(%s,%s)", (job, stu)); conn.commit()
before = snapshot(cur)
removed, conflicts = tool.apply_cleanup(conn, cur, ident)
check("T7 存在本批之外依赖 → 拒绝清理", removed is None and conflicts, conflicts[:1])
check("T7 本批数据与外部依赖全部保留", snapshot(cur) == before)
def t08_midway_error_rolls_back(conn, cur):
reset_schema(cur)
ident = identity_of(cur)
calls = {"n": 0}
real = tool.insert_ledger
def boom(*a, **k):
calls["n"] += 1
if calls["n"] == 3: # 已在事务中间
raise RuntimeError("injected failure")
return real(*a, **k)
tool.insert_ledger = boom
try:
raised = False
try:
tool.apply_import(conn, cur, ident)
except Exception:
raised = True
check("T8 中途异常被抛出", raised and calls["n"] >= 3, calls)
finally:
tool.insert_ledger = real
check("T8 本批业务数据整体回滚(users)", count(cur, "SELECT COUNT(*) FROM users") == 0)
check("T8 本批业务数据整体回滚(companies)", count(cur, "SELECT COUNT(*) FROM companies") == 0)
check("T8 本批业务数据整体回滚(jobs)", count(cur, "SELECT COUNT(*) FROM jobs") == 0)
ls = ledger_state(cur, ident)
check("T8 归属记账一并回滚(账本 0 行)", ls is not None and ls["total"] == 0, ls)
def t09_dry_run_and_preview_zero_write(conn, cur):
reset_schema(cur); seed_sentinels(cur); conn.commit()
before = snapshot(cur)
cmd = [sys.executable, os.path.join(os.path.dirname(os.path.abspath(__file__)),
"b01_demo_display_data.py"),
"--host", HOST, "--port", str(PORT), "--database", DB,
"--user", APP_USER, "--password", APP_PASS]
r = _run(cmd)
cur2 = conn.cursor()
check("T9 导入 dry-run 退出码 0", r.returncode == 0, r.stdout[-200:])
check("T9 导入 dry-run 零写入且不建账本表",
snapshot(cur2) == before and not tool.table_exists(cur2, tool.LEDGER_TABLE))
# 导入后清理预览
ident = identity_of(cur2)
tool.apply_import(conn, cur2, ident)
b2 = snapshot(cur2)
ls_before = ledger_state(cur2, ident)
r2 = _run(cmd + ["--cleanup"])
check("T9 清理预览退出码 0", r2.returncode == 0, r2.stdout[-200:])
check("T9 清理预览零写入", snapshot(cur2) == b2 and ledger_state(cur2, ident) == ls_before)
def t10_cleanup_only_batch_and_sentinels_intact(conn, cur):
reset_schema(cur)
sent = seed_sentinels(cur)
ident = identity_of(cur)
tool.apply_import(conn, cur, ident)
sent_ids = {"users": [sent["user1"], sent["user2"]], "companies": [sent["company"]],
"jobs": [sent["job"]]}
sent_before = {
"users": norm([r for r in table_rows(cur, "users") if r["id"] in sent_ids["users"]]),
"companies": norm([r for r in table_rows(cur, "companies") if r["id"] in sent_ids["companies"]]),
"jobs": norm([r for r in table_rows(cur, "jobs") if r["id"] in sent_ids["jobs"]]),
"user_roles": norm([r for r in table_rows(cur, "user_roles")
if r["user_id"] in sent_ids["users"]]),
}
removed, conflicts = tool.apply_cleanup(conn, cur, ident)
check("T10 正常清理无冲突", conflicts == [], conflicts)
check("T10 仅删除本批 12 岗位/3 企业/3 用户/18 账本",
removed == dict(jobs=12, companies=3, users=3, ledger=18), removed)
dc = demo_counts(cur)
check("T10 本批业务数据已清空", dc == dict(users=0, companies=0, jobs=0), dc)
check("T10 账本已清空", ledger_state(cur, ident)["total"] == 0)
sent_after = {
"users": norm([r for r in table_rows(cur, "users") if r["id"] in sent_ids["users"]]),
"companies": norm([r for r in table_rows(cur, "companies") if r["id"] in sent_ids["companies"]]),
"jobs": norm([r for r in table_rows(cur, "jobs") if r["id"] in sent_ids["jobs"]]),
"user_roles": norm([r for r in table_rows(cur, "user_roles")
if r["user_id"] in sent_ids["users"]]),
}
check("T10 哨兵记录逐字段不变(users/companies/jobs/user_roles)", sent_after == sent_before)
def t11_second_cleanup_noop(conn, cur):
ident = identity_of(cur)
before = snapshot(cur)
removed, conflicts = tool.apply_cleanup(conn, cur, ident)
check("T11 再次清理安全无操作", conflicts == [] and removed == dict(jobs=0, companies=0,
users=0, ledger=0), (removed, conflicts))
check("T11 库内无变化", snapshot(cur) == before)
def t12_non_whitelist_target_refused_before_connect(conn, cur):
ok = True
for host, port, db in [("127.0.0.1", 3308, "mydreamjob_b01_demo"),
("127.0.0.1", 3308, "mydreamjob_b01_demo_test")]:
try:
tool.assert_safe_target(host, port, db)
except SystemExit:
ok = False
check("T12 白名单目标通过", ok)
for host, port, db, why in [
("127.0.0.1", 3307, "mydreamjob_b01_demo", "非白名单端口"),
("10.0.0.9", 3308, "mydreamjob_b01_demo", "非回环 host"),
("127.0.0.1", 3308, "some_other_db", "非白名单库"),
("127.0.0.1", 3308, "mydreamjob_prod", "疑似线上库")]:
try:
tool.assert_safe_target(host, port, db)
check("T12 拒绝 %s" % why, False, "未拒绝")
except SystemExit as e:
check("T12 拒绝 %s" % why, "REFUSED" in str(e), str(e)[:60])
# 进程级：错误目标必须在**建立连接前**立即拒绝（不等待连接超时）
cmd = [sys.executable, os.path.join(os.path.dirname(os.path.abspath(__file__)),
"b01_demo_display_data.py"),
"--host", "127.0.0.1", "--port", "3307", "--database", "mydreamjob_b01_demo",
"--user", "nobody", "--password", "nope"]
t0 = time.time()
r = _run(cmd)
elapsed = time.time() - t0
check("T12 非白名单目标非零退出且快速拒绝",
r.returncode != 0 and "REFUSED" in (r.stderr + r.stdout) and elapsed < 10,
"rc=%s elapsed=%.1fs" % (r.returncode, elapsed))
def assert_refusal_preserves_rows(conn, cur, ident, label):
before = snapshot(cur, ("users", "user_roles", "companies", "jobs", tool.LEDGER_TABLE))
_, conflicts = tool.apply_cleanup(conn, cur, ident)
check(label + " 清理拒绝", bool(conflicts), conflicts)
check(label + " 清理未改业务或账本", snapshot(cur, tuple(before)) == before)
_, conflicts = tool.apply_import(conn, cur, ident)
check(label + " 导入拒绝", bool(conflicts), conflicts)
check(label + " 导入未改业务或账本", snapshot(cur, tuple(before)) == before)
def t13_real_user_identity_changed(conn, cur):
reset_schema(cur); seed_sentinels(cur)
ident = identity_of(cur)
tool.apply_import(conn, cur, ident)
cur.execute("UPDATE users SET phone='18800009991' WHERE phone=%s", (DEMO_PHONES[0],))
conn.commit()
assert_refusal_preserves_rows(conn, cur, ident, "T13 真实用户身份改变")
def t14_job_parent_changed(conn, cur):
reset_schema(cur); seed_sentinels(cur)
ident = identity_of(cur)
tool.apply_import(conn, cur, ident)
ledger = tool.read_ledger(cur, ident)
job = next(r for r in ledger if r['resource_type'] == 'JOB')
cur.execute("SELECT company_id FROM jobs WHERE id=%s", (job['resource_id'],))
old = cur.fetchone()[0]
other = next(r['resource_id'] for r in ledger if r['resource_type'] == 'COMPANY' and r['resource_id'] != old)
cur.execute("UPDATE jobs SET company_id=%s WHERE id=%s", (other, job['resource_id']))
conn.commit()
assert_refusal_preserves_rows(conn, cur, ident, "T14 岗位父企业改变")
def t15_company_name_changed(conn, cur):
reset_schema(cur); seed_sentinels(cur)
ident = identity_of(cur)
tool.apply_import(conn, cur, ident)
cur.execute("UPDATE companies SET name='changed-identity' WHERE name=%s", (DEMO_NAMES[0],))
conn.commit()
assert_refusal_preserves_rows(conn, cur, ident, "T15 企业真实名字改变")
TESTS = [
("T1", t01_first_import),
("T2", t02_repeat_import_idempotent),
("T3", t03_occupied_phone_refused),
("T4", t04_nonbatch_company_and_title_refused),
("T5", t05_demo_text_without_ledger_not_deleted),
("T6", t06_ledger_lost_or_identity_changed),
("T7", t07_external_dependency_refused),
("T8", t08_midway_error_rolls_back),
("T9", t09_dry_run_and_preview_zero_write),
("T10", t10_cleanup_only_batch_and_sentinels_intact),
("T11", t11_second_cleanup_noop),
("T12", t12_non_whitelist_target_refused_before_connect),
("T13", t13_real_user_identity_changed),
("T14", t14_job_parent_changed),
("T15", t15_company_name_changed),
]
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--keep-container", action="store_true")
parser.add_argument("--out")
args = parser.parse_args()
try:
start_container()
except Exception as exc:
print("[ENV] 无法启动隔离 MySQL 容器：%s" % exc)
return 2
try:
admin = root_conn()
with admin.cursor() as ac:
ac.execute("CREATE USER IF NOT EXISTS '%s'@'%%' IDENTIFIED BY '%s'" % (APP_USER, APP_PASS))
ac.execute("GRANT ALL PRIVILEGES ON `%s`.* TO '%s'@'%%'" % (DB, APP_USER))
ac.execute("FLUSH PRIVILEGES")
admin.close()
for name, fn in TESTS:
conn = root_conn(autocommit=False)
try:
with conn.cursor() as cur:
fn(conn, cur)
except Exception as exc:
import traceback
check("%s 执行异常" % name, False, traceback.format_exc()[-400:])
finally:
try:
conn.rollback()
except Exception:
pass
conn.close()
finally:
if not args.keep_container:
stop_container()
bad = [r for r in RESULTS if not r["ok"]]
print("\n== DEMO-TOOL SAFETY SUMMARY: total=%d fail=%d ==" % (len(RESULTS), len(bad)))
if args.out:
with open(args.out, "w", encoding="utf-8") as fh:
json.dump(RESULTS, fh, ensure_ascii=False, indent=1)
return 1 if bad else 0
if __name__ == "__main__":
sys.exit(main())