#!/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())