conftest.py

"""ndls Tavern 测试 conftest：注入动态变量到 Tavern yaml + 失败捕获（issue 04）。

通过 pytest_tavern_beta_before_every_test_run hook（beta API）往 variables dict
注入动态变量（加密密码、登录 token 等），yaml 里 {encrypted_pwd:s}/{auth_token:s} 即可引用。

加密密码 / 登录 token 首次计算后缓存（session/worker 级），避免每条 test 重复拉
nacos 策略 + RSA + doLogin。

失败捕获（issue 04）：每条 test（=1 端点 stage）跑完后把 method/url/status/body
写入 reports/.smoke/<worker>.jsonl；403 role-gated 改写 pytest outcome 为 skipped。
python -m core.smoke_report 聚合各 worker 片段 → 失败分类报告 + skipped.log。
"""
from __future__ import annotations

import json
import os
import re
import sys
import threading
from pathlib import Path

import pytest
from dotenv import load_dotenv

try:
    import allure
    from core.allure_helper import parse_deep_test_labels, parse_test_labels
    _HAS_ALLURE = True
except ImportError:  # allure 未装时仍可跑（无 attach/打标）
    _HAS_ALLURE = False

load_dotenv()

# 让 Tavern !include 能找到项目根的 common.yaml 等（无论 test 文件在哪个子目录）
_PROJECT_ROOT = str(Path(__file__).resolve().parent)
_existing = os.environ.get("TAVERN_INCLUDE", "")
if _PROJECT_ROOT not in _existing.split(":"):
    os.environ["TAVERN_INCLUDE"] = (
        f"{_existing}:{_PROJECT_ROOT}" if _existing else _PROJECT_ROOT
    )

_cache: dict = {}
# issue 18：dealer insert 会写 pending audit（按 code 锁定）+ 默认 ship_addr。固定 code
# 不可重入（AUDIT_INFO_IN_PROGRESS），故按需注入唯一 dealer_code；teardown 清默认
# ship_addr 孤儿（dealer removeById 不回收）。
_dealer_codes: dict = {}
# issue 17：priceSeries insert name 唯一性校验只查 dataStatus=IN_EFFECT（不看 delete_flag），
# soft-delete 后 name 仍占。故按需注入唯一 price_series_name；teardown 走 DB soft-delete
#（PriceSeriesController 无 removeById/disable，cleanup_price_series 设 delete_flag=1）。
_price_series_names: dict = {}
# issue 49：purchasePriceSeries insert 应用层唯一校验查重（companyCode+name, IN_EFFECT 作用域），
# 同 priceSeries。teardown 走 API removeById（updateEx -> data_status=EXPIRED，disable 非 soft-delete，
# 不动 delete_flag）。stage2 API 成功则行 expired；stage2 跳过（stage1 断言失败 / red-verify）时
# conftest 兜底 cleanup_purchase_price_series 置 expired 释名占 + dev 干净（镜像 updateEx）。
_purchase_price_series_names: dict = {}
# issue 13：customer insert 写默认 ship_addr（removeById 不回收）+ pending audit（按 code
# 锁定）。taxAccount 应用层唯一校验（IN_EFFECT），code 是业务键（audit dataKey +
# ship_addr refTypeCode），ic_code 语义唯一。三者每 test 唯一；teardown 清默认 ship_addr
# 孤儿。RUNTIME_EXE 根因：T_BI_CUSTOMER 的 SPAN_AREA_FLAG / SPAN_AREA_STORE_FLAG 是
# NOT NULL 无默认值列（dealer 表无），请求须显式传 F（见 issue 13 调研结论）。
_customer_data: dict = {}  # nodeid -> {code, tax_account, ic_code}
# issue 37：销售单 save 响应的 order code（nodeid -> code），teardown 兜底物理删该单
# 库存锁（API cancelById 失败 / 订单卡死 -> 孤儿锁 -> 物理删恢复可供量）。
_order_codes: dict = {}

# --- 失败捕获状态（per worker process）---------------------------------------
_REPORT_DIR = Path(_PROJECT_ROOT) / "reports" / ".smoke"
# per-run 子目录：并发跑（如另一 session）各自隔离，避免互相清空/覆盖。
# NDLS_SMOKE_RUN_ID 由调用方注入（如 `NDLS_SMOKE_RUN_ID=master_01 pytest ...`）。
_RUN_DIR = _REPORT_DIR / os.environ.get("NDLS_SMOKE_RUN_ID", "default")
_latest_response: dict[str, dict] = {}  # nodeid → 最新响应（1 test 1 stage 即失败 stage）
_stage_meta: dict[str, dict] = {}  # nodeid → stage 的 method/url（无响应时兜底，如路径参数格式错误）
_current = threading.local()  # 当前 test 的 nodeid（after_every_response 无 item 参数）


def _worker_id() -> str:
    return os.environ.get("PYTEST_XDIST_WORKER", "main")


def _write_record(record: dict) -> None:
    """追加一条 test 记录到本 worker 的 JSONL（per-run 子目录）。"""
    _RUN_DIR.mkdir(parents=True, exist_ok=True)
    with (_RUN_DIR / f"{_worker_id()}.jsonl").open("a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")


def _patch_tavern_error_format_keys() -> None:
    """issue 86：tavern 3.6.1 INTERNALERROR unmask patch（ADR-0011 选项 2）。

    tavern ``_core/pytest/error.py`` 生成失败报告时调 ``format_keys`` 打印 format
    variables（``_print_format_variables``）与 formatted stage（``_print_formatted_stage``）。
    对 int/float 变量用 ``{var:s}`` 格式说明符会抛 ``ValueError: Unknown format code 's'
    for object of type 'int'``，而 ``format_keys`` 的 ``except`` 只捕获
    ``MissingFormatError`` 不捕获 ``ValueError`` -> 异常崩在 ``pytest_runtest_makereport``
    hook（构造 TestReport 时 ``item.repr_failure`` -> tavern ``ReprdError`` 报告路径）
    -> pytest INTERNALERROR 级联中断整个 session，失败用例报告被掩盖。

    此 patch 替换 ``tavern._core.pytest.error`` 模块的 ``format_keys`` 绑定（错误报告
    路径），对 ``ValueError`` 降级返回原值（报告显示未 format 占位符），让失败用例
    干净报告、session 不崩。request 阶段用 ``tavern._core.dict_util.format_keys``（绑定
    不受 patch 影响），format bug 仍暴露为 test failure（不静默）。#66（issue 89）已把
    solineList 数值 ref 改 ``!int``/``!float`` cast（避 int+:s）；此 patch 留作错误报告路径
    通用安全网（ADR-0011），余下非 solineList 场景仍可能命中。
    """
    import tavern._core.pytest.error as _tavern_error

    if getattr(_tavern_error.format_keys, "_is_issue86_patched", False):
        return  # 幂等：xdist worker / 重复 configure 不重复 patch

    _orig_format_keys = _tavern_error.format_keys

    def _safe_format_keys(val, variables, *, no_double_format=True,
                         dangerously_ignore_string_format_errors=False):
        try:
            return _orig_format_keys(
                val, variables,
                no_double_format=no_double_format,
                dangerously_ignore_string_format_errors=dangerously_ignore_string_format_errors,
            )
        except ValueError:
            # 错误报告路径容忍 format spec 错误（同 dangerously_ignore 的意图），
            # 降级返回原值，避免 INTERNALERROR 级联中断 session。
            return val

    _safe_format_keys._is_issue86_patched = True  # type: ignore[attr-defined]
    _tavern_error.format_keys = _safe_format_keys


def pytest_configure(config):
    """controller（或非 xdist）启动时清空本次 run 子目录的残留 JSONL。"""
    _patch_tavern_error_format_keys()  # issue 86：unmask tavern INTERNALERROR（ADR-0011）
    if not os.environ.get("PYTEST_XDIST_WORKER") and _RUN_DIR.exists():
        for f in _RUN_DIR.glob("*.jsonl"):
            f.unlink()


def pytest_tavern_beta_after_every_response(expected, response):
    """每条响应后记录 method/url/status/body/请求头/请求体（断言前调用，无论成败）。

    allure attach 放 makereport（after_every_response 期间 allure 测试上下文未激活，
    attach 文件会写但不链接到 result）。
    """
    nodeid = getattr(_current, "nodeid", None)
    if nodeid is None:
        return
    try:
        body = response.json()
    except ValueError:
        body = None
    req = response.request
    req_body = req.body
    if isinstance(req_body, bytes):
        req_body = req_body.decode("utf-8", "replace")
    _latest_response[nodeid] = {
        "method": req.method,
        "url": req.path_url,
        "status_code": response.status_code,
        "body": body,
        "req_headers": dict(req.headers),
        "req_body": req_body,
    }

    # issue 37：销售单 save 成功 -> 提取 order code 注册，teardown 兜底物理删库存锁。
    # 仅 save 网关（POST /salesOrder/save）命中，cancel/getById 等不注册，避免误删他人锁。
    if (
        req.method == "POST"
        and req.path_url.endswith("/salesOrder/save")
        and isinstance(body, dict)
    ):
        from core.fixtures import extract_save_order_code

        code = extract_save_order_code(body)
        if code:
            _order_codes[nodeid] = code


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item):
    """记录当前 test 的 nodeid 供 after_every_response 关联。"""
    _current.nodeid = item.nodeid
    try:
        yield
    finally:
        _current.nodeid = None


def _deep_test_endpoint_label(nodeid: str) -> str:
    """从深测 yaml 提取被测端点（第一个非 GET）归一化 path，作 endpoint label。

    nodeid -> tests/{domain}/{stem}.tavern.yaml 路径 -> 读 yaml（best effort，失败返回 ""）。
    """
    from core.coverage_report import endpoint_label_from_yaml

    m = re.search(r"(tests/[a-z]+/[^/]+\.tavern\.yaml)", nodeid or "")
    if not m:
        return ""
    return endpoint_label_from_yaml(Path(_PROJECT_ROOT) / m.group(1))


def pytest_runtest_setup(item):
    """打 allure 标签：smoke 用 parse_test_labels；深测用 parse_deep_test_labels + 读 yaml 补 endpoint。"""
    if not _HAS_ALLURE:
        return
    labels = parse_test_labels(item.name)
    if labels is not None:
        allure.dynamic.feature(labels["feature"])
        allure.dynamic.story(labels["story"])
        allure.dynamic.label("tier", labels["tier"])
        allure.dynamic.label("endpoint", labels["endpoint"])
        allure.dynamic.severity(labels["severity"])
        return
    # 深测用例（tests/{domain}/*.tavern.yaml，item.name 非 smoke 三段格式）
    deep_labels = parse_deep_test_labels(item.nodeid)
    if deep_labels is None:
        return
    allure.dynamic.feature(deep_labels["feature"])
    allure.dynamic.story(deep_labels["story"])
    allure.dynamic.label("tier", deep_labels["tier"])
    allure.dynamic.label("domain", deep_labels["domain"])
    endpoint = _deep_test_endpoint_label(item.nodeid)
    if endpoint:
        allure.dynamic.label("endpoint", endpoint)
    allure.dynamic.severity("critical")  # 深测都是 write 端点


def _allure_attach_classification(stage: dict) -> None:
    """attach 失败分类（method/url/status/code/msg/category）到 allure。"""
    try:
        from core.smoke_report import StageRecord, classify_failure

        rec = StageRecord(
            method=stage.get("method", "-"),
            url=stage.get("url", "-"),
            status_code=stage.get("status_code"),
            body=stage.get("body"),
        )
        cls = classify_failure(rec)
        allure.attach(
            json.dumps(
                {"method": cls.record.method, "url": cls.record.url,
                 "status_code": cls.record.status_code, "code": cls.code,
                 "msg": cls.msg, "category": cls.category.value},
                ensure_ascii=False,
            ),
            name="失败分类",
            attachment_type=allure.attachment_type.JSON,
        )
    except Exception:  # noqa: BLE001 — attach 失败不影响测试
        pass


def _allure_attach_stage(stage: dict) -> None:
    """attach 请求（method/url/headers/body）+ 响应（status/信封原文）到 allure。"""
    try:
        allure.attach(
            json.dumps(
                {"method": stage.get("method", "-"), "url": stage.get("url", "-"),
                 "headers": stage.get("req_headers", {}), "body": stage.get("req_body")},
                ensure_ascii=False,
            ),
            name="请求",
            attachment_type=allure.attachment_type.JSON,
        )
        allure.attach(
            json.dumps(
                {"status_code": stage.get("status_code"), "body": stage.get("body")},
                ensure_ascii=False,
            ),
            name="响应",
            attachment_type=allure.attachment_type.JSON,
        )
    except Exception:  # noqa: BLE001 — attach 失败不影响测试
        pass


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
    """test 结束后写记录；403 role-gated 改写 outcome 为 skipped（不计失败）。"""
    outcome = yield
    report = outcome.get_result()
    if call.when != "call":
        return

    nodeid = item.nodeid
    test_name = item.name
    stage = _latest_response.pop(nodeid, None)
    if stage is None and report.outcome == "failed":
        # 无响应（连接错误）：用 stage meta 兜底 method/url
        meta = _stage_meta.pop(nodeid, {})
        stage = {
            "method": meta.get("method", "-"),
            "url": meta.get("url", "-"),
            "status_code": None,
            "body": None,
        }
    else:
        _stage_meta.pop(nodeid, None)

    jsonl_outcome = report.outcome
    # 403 → role-gated：pytest 标 skipped，JSONL 仍记 failed 供分类为 ROLE_GATED
    if report.outcome == "failed" and stage and stage.get("status_code") == 403:
        report.outcome = "skipped"
        report.longrepr = (str(item.fspath), 0, "role-gated 403")

    # allure attach（makereport 期间上下文激活，文件会链接到 result）
    if _HAS_ALLURE and stage:
        _allure_attach_stage(stage)
        if jsonl_outcome == "failed":
            _allure_attach_classification(stage)

    # issue 39：失败时从 longrepr 提取断言消息（DB 断言不符在此），供 run_deep_test
    # harness 提取 db_error。lazy 导入贴合 conftest 既有模式（_allure_attach_classification
    # 同样 lazy 导入 smoke_report）；passed 不提取（None）。
    failure_msg = None
    if jsonl_outcome == "failed":
        from core.smoke_report import failure_message
        failure_msg = failure_message(report)
    # issue 74: contract marker 的 passed 也记 stage（含 body.code），供 classify_mutation
    # 拆 masked（非目标先拒 *_EXISTS/RUNTIME_EXE）。failed 仍记（既有）。smoke passed 不记
    # （smoke_report 只读 failed，不膨胀 smoke JSONL）。masked 是兜底分类，#73 已挡 *_EXISTS 主场景。
    _is_contract = item.get_closest_marker("contract") is not None
    _record_stage = stage if (jsonl_outcome == "failed" or _is_contract) else None
    _write_record({"test_name": test_name, "nodeid": nodeid,
                   "outcome": jsonl_outcome,
                   "stage": _record_stage,
                   "failure_msg": failure_msg})


def pytest_tavern_beta_before_every_test_run(test_dict, variables):
    """每条 tavern test 跑前注入动态变量。首次调用算加密密码 + 登录 token 并缓存。"""
    base_url = variables.get("base_url") or os.getenv(
        "NDLS_BASE_URL", "http://api.example.com"
    )
    variables["base_url"] = base_url
    variables["login_id"] = os.getenv("NDLS_LOGIN_ID", "")
    variables["bu_code"] = os.getenv("NDLS_BU_CODE", "")

    if "encrypted_pwd" not in _cache:
        # 懒导入，避免 pytest collect 阶段触发网络请求
        from core.auth import compute_encrypted_pwd

        password = os.getenv("NDLS_PASSWORD", "")
        _cache["encrypted_pwd"] = compute_encrypted_pwd(base_url, password)
    variables["encrypted_pwd"] = _cache["encrypted_pwd"]

    if "auth_token" not in _cache:
        # 冒烟每端点 1 test：登录由 conftest 会话级（每 worker 1 次）完成，注入 {auth_token}。
        # 优先读 NDLS_AUTH_TOKEN env（NDLS_AUTH_TOKEN=$(python -m core.auth login)）：
        # 跨 xdist worker 共享单次登录，避免并发 login 触发 Sa-Token 单 session 冲突。
        # 失败缓存空串 → stage 用空 m-token → 服务端返回 ST_NOT_TOKEN，归类 AUTH。
        env_token = os.environ.get("NDLS_AUTH_TOKEN")
        if env_token:
            _cache["auth_token"] = env_token
        else:
            from core.auth import login

            login_id = os.getenv("NDLS_LOGIN_ID", "")
            try:
                _cache["auth_token"] = login(base_url, login_id, _cache["encrypted_pwd"])
            except Exception as e:  # noqa: BLE001 — 登录失败不阻断 collect，记空让测试归 AUTH
                print(f"[conftest] login failed: {e}", file=sys.stderr)
                _cache["auth_token"] = ""
    variables["auth_token"] = _cache["auth_token"]

    nodeid = getattr(_current, "nodeid", None)

    # issue 24：按需注入测试参考数据配置为 ref_* 变量（tracer yaml 引用）。仅当 test 引用
    # 任一 ref_* 时加载（无引用的冒烟测试不依赖 test_bu_ref 配置）。ref_ 前缀避免与按需注入
    # 的动态变量（{customer_code} fresh / {ship_addr_id} / {dealer_code} / {price_series_name}）
    # 冲突。顶层主数据引用存量（issue 11），不按测 create。
    if nodeid and _test_uses_ref_vars(test_dict):
        variables.update(_ref_vars())

    # issue 11/87：按需创建收货地址（仅当 test 引用 {ship_addr_id}）。
    # issue 87：原先每 nodeid 创建一个 ship_addr，xdist -n auto 下 56 条 sales_order 并发
    # 创建同客户+同固定地址 -> HAS_THE_SHIP_ADDR 重复 + PREVENT_REPEAT_SUBMIT 限流，create_ship_addr
    # 失败 -> ship_addr_id="" -> int('') ValueError。改为每 worker（进程）创建一个共享 ship_addr
    # 复用（地址带 worker 唯一戳避免跨 worker 重复），大幅减少调用数（56 -> 每 worker 1 次）避限流。
    # teardown 不再逐 test 删（共享），session finish 兜底删（dev 残留 ADR-0004 亦可接受）。
    if nodeid and _test_uses_var(test_dict, "ship_addr_id"):
        if "ship_addr_id" not in _cache:
            try:
                from core.fixtures import create_ship_addr
                ref = _test_bu_ref()
                # worker 唯一地址戳：避跨 worker 同客户同地址撞 HAS_THE_SHIP_ADDR
                import time
                addr = f"autotest-ship-{int(time.time() * 1000)}-{_worker_id()}"
                sid = create_ship_addr(
                    base_url, variables["auth_token"],
                    customer_code=ref["customer_code"],
                    dc_code=ref["dc_code"],
                    area_code=ref["area_code"],
                    address=addr,
                )
                _cache["ship_addr_id"] = sid
            except Exception as e:  # noqa: BLE001
                print(f"[conftest] create_ship_addr failed: {e}", file=sys.stderr)
                _cache["ship_addr_id"] = ""
        variables["ship_addr_id"] = _cache["ship_addr_id"]

    # issue 18：按需注入唯一 dealer_code（仅当 test 引用 {dealer_code}）。dealer insert
    # 写 pending audit 按 code 锁定，固定 code 不可重入，故每 test 一个唯一 code。teardown
    # 在 pytest_runtest_teardown 清默认 ship_addr 孤儿。
    if nodeid and _test_uses_var(test_dict, "dealer_code"):
        code = _fresh_dealer_code()
        variables["dealer_code"] = code
        _dealer_codes[nodeid] = code

    # issue 17：按需注入唯一 price_series_name（仅当 test 引用 {price_series_name}）。
    # priceSeries 无 API teardown，teardown 在 pytest_runtest_teardown 走 DB soft-delete。
    if nodeid and _test_uses_var(test_dict, "price_series_name"):
        variables["price_series_name"] = _fresh_price_series_name()
        _price_series_names[nodeid] = variables["price_series_name"]

    # issue 49：按需注入唯一 purchase_price_series_name（仅当 test 引用）。
    # purchasePriceSeries insert 应用层唯一校验（companyCode+name, IN_EFFECT 作用域），同
    # priceSeries。teardown 走 API removeById（updateEx）；conftest 兜底在 pytest_runtest_teardown。
    if nodeid and _test_uses_var(test_dict, "purchase_price_series_name"):
        variables["purchase_price_series_name"] = _fresh_purchase_price_series_name()
        _purchase_price_series_names[nodeid] = variables["purchase_price_series_name"]

    # issue 13：按需注入唯一客户三元组 code/tax_account/ic_code（仅当 test 引用任一）。
    # customer insert 同 dealer 写默认 ship_addr + pending audit，teardown 清 ship_addr 孤儿。
    if nodeid and (
        _test_uses_var(test_dict, "customer_code")
        or _test_uses_var(test_dict, "customer_tax_account")
        or _test_uses_var(test_dict, "customer_ic_code")
    ):
        cust = _fresh_customer_data()
        variables["customer_code"] = cust["code"]
        variables["customer_tax_account"] = cust["tax_account"]
        variables["customer_ic_code"] = cust["ic_code"]
        _customer_data[nodeid] = cust

    # issue 90：按需注入唯一 bank_transaction_no（仅当 test 引用 {bank_transaction_no}）。
    # paymentRecord insert bankTransactionNo 唯一校验查 dev 存量（不看 delete_flag，soft-delete
    # 仍占），固定 literal 撞 SA_PAYMENT_RECORD_TRANSACTION_NO_EXISTS -> data.id 空。故每 test
    # 唯一。teardown 走 yaml removeById（GET Long id，where_field=id 正确），无 ship_addr 孤儿，
    # 不需 conftest teardown 跟踪（同 dealer 但无孤儿清理）。
    if nodeid and _test_uses_var(test_dict, "bank_transaction_no"):
        variables["bank_transaction_no"] = _fresh_bank_transaction_no()

    # issue 92：按需注入唯一 sales_channel_code（仅当 test 引用 {sales_channel_code}）。
    # salesChannel insert 后端无 removeById（disable/{id} 逻辑删，契约层 removeById 404），
    # 入 TEARDOWN_SKIP_ENTITIES 跳过 teardown -> 每 test 残留一行（ADR-0004 可接受）。
    # code 虽非 DB 唯一约束（T_BI_SALES_CHANNEL 仅 PK ID），但固定 AUTOTEST42SC 累积 54+ 同名
    # 残留行脏 dev；fresh code 让残留可追溯 + 与 priceSeries/salesOrder（已 TEARDOWN_SKIP）一致。
    # 不需 conftest teardown 跟踪（teardown 跳过，残留 ADR-0004）。
    if nodeid and _test_uses_var(test_dict, "sales_channel_code"):
        variables["sales_channel_code"] = _fresh_sales_channel_code()

    # 记 stage 的 method/url（无响应失败兜底，如路径参数 {var} 格式错误）
    nodeid = getattr(_current, "nodeid", None)
    if nodeid:
        stages = test_dict.get("stages", [])
        if stages:
            req = stages[0].get("request", {})
            url = req.get("url", "")
            if url.startswith("{base_url}"):
                url = url[len("{base_url}"):]  # 剥 {base_url} → 路径
            _stage_meta[nodeid] = {"method": req.get("method", "-"), "url": url}


def _flatten_test_text(obj) -> str:
    """递归扁平化 test_dict 为文本（含 TypeConvertToken.value），供 {var} 占位符检测。

    issue 87：``str(test_dict)`` 对 ``!int '{ship_addr_id}'`` 加载出的 TypeConvertToken
    （IntToken/FloatToken，``tavern/_core/dict_util.py``）返回默认 ``<...object at 0x..>``
    不含字面 ``{ship_addr_id}``，导致 ``_test_uses_var`` 漏检 -> conftest 不注入
    ship_addr_id -> ``MissingFormatError``（sales_order 56 条全挂）。遍历 dict/list/tuple，
    对 TypeConvertToken 取 ``.value``（占位符原文），其余取 str()/叶子串。

    ``_walk`` 为内嵌闭包：闭包捕获 ``parts``/``seen``/``TypeConvertToken``，避免参数耦合
    （Data Clump）与每节点重复 import（TypeConvertToken 一次 import 复用）。
    """
    from tavern._core.dict_util import TypeConvertToken

    parts: list[str] = []
    seen: set = set()

    def _walk(o) -> None:
        # 循环引用/自引用保护（tavern 内部包装的 test 结构可能有环）；标量/叶子直接处理
        if not isinstance(o, (str, int, float, bool, type(None))):
            oid = id(o)
            if oid in seen:
                return
            seen.add(oid)

        if isinstance(o, str):
            parts.append(o)
        elif isinstance(o, TypeConvertToken):
            # !int/!float '{var}' -> token.value = '{var}'（占位符原文，含字面 {var}）
            parts.append(str(getattr(o, "value", "")))
        elif isinstance(o, dict):
            for v in o.values():
                _walk(v)
        elif isinstance(o, (list, tuple)):
            for v in o:
                _walk(v)
        else:
            parts.append(str(o))

    try:
        _walk(obj)
    except Exception as e:  # noqa: BLE001 - 防御：walk 异常不阻断 before_every_test_run hook
        # 退回 str() 旧路径（漏检 !int token，但不崩 test setup -> 不致整 test 缺变量）
        print(f"[conftest] _flatten_test_text walk failed: {type(e).__name__}: {e}", file=sys.stderr)
        return str(obj)
    return "\n".join(parts)


def _test_uses_var(test_dict: dict, var: str) -> bool:
    """test 的 stages 是否引用 {var}（粗检：扁平文本含 '{var}' 或 '{var:'）。

    用 ``_flatten_test_text`` 而非 ``str(test_dict)``--后者对 ``!int '{var}'`` 加载出的
    TypeConvertToken 返回 ``<...object at 0x..>`` 不含字面 ``{var}``，会漏检（issue 87）。
    test_dict 含 Tavern !anything 等 sentinel，不可 JSON 序列化，故仍走自定义扁平化。
    """
    text = _flatten_test_text(test_dict)
    return f"{{{var}}}" in text or f"{{{var}:" in text


def _test_uses_ref_vars(test_dict: dict) -> bool:
    """test 是否引用任一 ref_* 参考数据变量（无引用的冒烟测试不依赖 test_bu_ref 配置）。

    issue 87：同样用 ``_flatten_test_text``--纯 ``!int '{ref_x}'``（无字符串 ref）时
    ``str()`` 漏检，ref_* 不注入。
    """
    return "{ref_" in _flatten_test_text(test_dict)


def _test_bu_ref() -> dict:
    """加载 data/test_bu_ref.json（缓存）。供 conftest 注入 ref_* 变量 + create_ship_addr 取值。"""
    if "test_bu_ref" not in _cache:
        from core.test_data import load_test_bu_ref

        _cache["test_bu_ref"] = load_test_bu_ref()
    return _cache["test_bu_ref"]


def _ref_vars() -> dict:
    """从 test_bu_ref 生成 ref_* tavern 变量（header + 扁平化 solines），缓存。

    ref_ 前缀避免与按需注入的动态变量冲突。header 字段照搬为 ref_{key}；solines 扁平为
    ref_sol{i}_{field}（i 从 1 起），供 tracer solineList 用 !int/!float/\"{var}\" 引用。

    issue 89（#66 根治）：soline 数值字段注入原生 int/float（不再 str cast）。generator
    ``gen_contract_yaml`` 对 solineList 嵌套体 per-field 类型 cast（数值 -> ``!int``/``!float``，
    string -> ``{var:s}``），tavern ``IntToken/FloatToken.constructor`` 接原生数值 -> int/float。
    旧 conftest ``str()`` cast（issue 87 兜底 ``int+:s`` ValueError）已冗余，删之，类型 cast
    完全归 generator。
    """
    if "ref_vars" not in _cache:
        ref = _test_bu_ref()
        v = {f"ref_{k}": val for k, val in ref.items() if k != "solines"}
        for i, sol in enumerate(ref["solines"], start=1):
            for k in (
                "sku_code", "order_qty_big", "order_qty_small",
                "conversion_ratio", "sales_amount", "sales_price",
            ):
                # 原样注入：str 字段（sku_code）保 str，数值字段保 int/float（generator !int/!float cast）
                v[f"ref_sol{i}_{k}"] = sol[k]
        _cache["ref_vars"] = v
    return _cache["ref_vars"]


_dealer_counter = 0


def _fresh_dealer_code() -> str:
    """生成唯一 dealer code：DLR<ms><counter>（≤20 位，dealer 编码长度上限 1～20）。

    dealer insert 写 pending audit 按 code 锁定，固定 code 不可重入，故每 test 唯一。
    timestamp 用 wall clock（conftest 普通代码，time 可用）；counter 防同毫秒并发碰撞。
    """
    import time

    global _dealer_counter
    _dealer_counter += 1
    return f"DLR{int(time.time() * 1000)}{_dealer_counter:03d}"


_price_series_counter = 0


def _fresh_price_series_name() -> str:
    """生成唯一价格套 name：autotest-ps-<ms><counter>。

    priceSeries insert name 唯一性校验只查 dataStatus=IN_EFFECT，soft-delete（delete_flag=1）
    不释放 name，故每 test 唯一 name 避免撞历史 soft-delete 行。timestamp wall clock +
    counter 防同毫秒并发碰撞（同 _fresh_dealer_code 模式）。
    """
    import time

    global _price_series_counter
    _price_series_counter += 1
    return f"autotest-ps-{int(time.time() * 1000)}{_price_series_counter:03d}"


_purchase_price_series_counter = 0


def _fresh_purchase_price_series_name() -> str:
    """生成唯一采购价格套 name：autotest-pps-<ms><counter>。

    purchasePriceSeries insert 应用层唯一校验查重（companyCode+name, IN_EFFECT 作用域），
    同 priceSeries（issue 17）：disable 后名可复用，但 red-verify stage1 失败致 stage2 teardown
    跳过时残留 in_effect 行占名 -> 同名重跑撞 NAME_EXISTS。故每 test 唯一名避撞。timestamp
    wall clock + counter 防同毫秒并发碰撞（同 _fresh_price_series_name 模式）。
    """
    import time

    global _purchase_price_series_counter
    _purchase_price_series_counter += 1
    return f"autotest-pps-{int(time.time() * 1000)}{_purchase_price_series_counter:03d}"


_customer_counter = 0


def _fresh_customer_data() -> dict:
    """生成唯一客户三元组：code / tax_account / ic_code（均 ≤20 位 @PsValidCode）。

    customer insert：taxAccount 应用层唯一校验（IN_EFFECT，非 DB 约束），code 是业务键
    （audit dataKey + ship_addr refTypeCode），ic_code 语义唯一。soft-delete（delete_flag=1）
    不改 data_status（仍 wait_audit ≠ IN_EFFECT），taxAccount 理论可复用，但三者每 test 唯一
    最稳（仿 _fresh_dealer_code）。timestamp wall clock + counter 防同毫秒并发碰撞。

    code 传请求体（@PsAutoCodeType 检测到非空跳过自动生成），便于 teardown 按 code 清
    ship_addr 孤儿（同 dealer 模式）。
    """
    import time

    global _customer_counter
    _customer_counter += 1
    ms = int(time.time() * 1000)
    return {
        # 4 + 13 + 3 = 20
        "code": f"CUST{ms}{_customer_counter:03d}",
        # 9 + 6 + 3 = 18（社会信用代码样式，@PsValidCode 仅校验长度 1～20）
        "tax_account": f"91330110M{ms % 1000000:06d}{_customer_counter:03d}",
        # 2 + 13 + 3 = 18
        "ic_code": f"IC{ms}{_customer_counter:03d}",
    }


_bank_transaction_no_counter = 0


def _fresh_bank_transaction_no() -> str:
    """生成唯一收款银行流水号：BT<ms><counter>（18 位，BANK_TRANSACTION_NO VARCHAR2(64)）。

    paymentRecord insert bankTransactionNo 唯一校验查 dev 存量（不看 delete_flag，soft-delete
    仍占名），固定 literal 撞 SA_PAYMENT_RECORD_TRANSACTION_NO_EXISTS -> data.id 空 -> DB 断言
    无法定位行。故每 test 唯一避撞 dev 存量 + 自身历史残留。timestamp wall clock + counter 防同
    毫秒并发碰撞（同 _fresh_dealer_code 模式）。teardown 走 yaml removeById，不需 conftest 跟踪。
    """
    import time

    global _bank_transaction_no_counter
    _bank_transaction_no_counter += 1
    return f"BT{int(time.time() * 1000)}{_bank_transaction_no_counter:03d}"


_sales_channel_counter = 0


def _fresh_sales_channel_code() -> str:
    """生成唯一 salesChannel code：SC<ms><counter>（18 位，CODE VARCHAR2(64)）。

    salesChannel insert 后端无 removeById（disable/{id} 逻辑删，契约层 removeById 404），
    入 TEARDOWN_SKIP_ENTITIES 跳过 teardown -> 每 test 残留一行（ADR-0004 可接受）。code 虽
    非唯一约束（仅 PK ID），固定 AUTOTEST42SC 累积同名残留脏 dev；fresh code 让残留可追溯 +
    与 priceSeries/salesOrder（已 TEARDOWN_SKIP）一致。timestamp wall clock + counter 防同毫秒
    并发碰撞（同 _fresh_bank_transaction_no 模式）。不需 conftest teardown 跟踪（teardown 跳过）。
    """
    import time

    global _sales_channel_counter
    _sales_channel_counter += 1
    return f"SC{int(time.time() * 1000)}{_sales_channel_counter:03d}"


def pytest_runtest_teardown(item, nextitem):
    """teardown：清本 test 创建的 dealer/customer/price_series/purchase_price_series/order。

    issue 87：ship_addr 改为每 worker 共享（``_cache["ship_addr_id"]``），不再逐 nodeid 创建，
    故此处不再删 ship_addr（共享 addr 由 ``pytest_sessionfinish`` 兜底删）。
    """
    # issue 18：清 dealer insert 触发的默认 ship_addr 孤儿（dealer removeById 不回收）。
    dealer_code = _dealer_codes.pop(item.nodeid, None)
    if dealer_code is not None:
        try:
            from core.fixtures import cleanup_entity_ship_addrs
            base = os.getenv("NDLS_BASE_URL", "http://api.example.com").rstrip("/")
            cleanup_entity_ship_addrs(base, _cache.get("auth_token", ""), "DEALER", dealer_code)
        except Exception as e:  # noqa: BLE001
            print(f"[conftest] teardown cleanup_entity_ship_addrs(DEALER, {dealer_code}) failed: {e}", file=sys.stderr)

    # issue 13：清 customer insert 触发的默认 ship_addr 孤儿（customer removeById 不回收）。
    cust = _customer_data.pop(item.nodeid, None)
    if cust is not None:
        try:
            from core.fixtures import cleanup_entity_ship_addrs
            base = os.getenv("NDLS_BASE_URL", "http://api.example.com").rstrip("/")
            cleanup_entity_ship_addrs(base, _cache.get("auth_token", ""), "CUSTOMER", cust["code"])
        except Exception as e:  # noqa: BLE001
            print(f"[conftest] teardown cleanup_entity_ship_addrs(CUSTOMER, {cust['code']}) failed: {e}", file=sys.stderr)

    # issue 17：priceSeries 无 API teardown，走 DB soft-delete（delete_flag=1）。
    ps_name = _price_series_names.pop(item.nodeid, None)
    if ps_name is not None:
        try:
            from core.fixtures import cleanup_price_series
            cleanup_price_series(name=ps_name)
        except Exception as e:  # noqa: BLE001
            print(f"[conftest] teardown cleanup_price_series({ps_name}) failed: {e}", file=sys.stderr)

    # issue 49：purchasePriceSeries 走 API removeById（updateEx -> data_status=EXPIRED）teardown；
    # 兜底：stage2 API 未跑（stage1 断言失败 / red-verify）时，把残留 in_effect 行置 expired
    # 释名占（镜像 updateEx disable 语义，区别 priceSeries 的 delete_flag=1 soft-delete）。
    pps_name = _purchase_price_series_names.pop(item.nodeid, None)
    if pps_name is not None:
        try:
            from core.fixtures import cleanup_purchase_price_series
            cleanup_purchase_price_series(name=pps_name)
        except Exception as e:  # noqa: BLE001
            print(f"[conftest] teardown cleanup_purchase_price_series({pps_name}) failed: {e}", file=sys.stderr)

    # issue 37：兜底物理删本 test save 的订单库存锁（API cancelById 失败 / 订单卡死时
    # 恢复可供量）。cancel 成功则后端 doUnLock 已删锁 -> cleanup 命中 0 行 no-op；
    # save-only 无锁（锁仅 submit 创建）-> no-op。仅 submit-卡死孤儿锁被清（正中兜底）。
    order_code = _order_codes.pop(item.nodeid, None)
    if order_code is not None:
        try:
            from core.fixtures import cleanup_inventory_locks
            cleanup_inventory_locks(order_code=order_code)
        except Exception as e:  # noqa: BLE001
            print(f"[conftest] teardown cleanup_inventory_locks({order_code}) failed: {e}", file=sys.stderr)


def pytest_sessionfinish(session, exitstatus):
    """session 结束兜底：删每 worker 共享的 ship_addr（issue 87 改为 per-worker 复用后，
    逐 test teardown 不再删，此处兜底；best-effort，失败仅日志，dev 残留 ADR-0004 可接受）。

    xdist 每 worker 是独立进程，各自 sessionfinish 删自己的 ``_cache["ship_addr_id"]``。
    """
    sid = _cache.get("ship_addr_id")
    if sid:
        try:
            from core.fixtures import remove_ship_addr
            base = os.getenv("NDLS_BASE_URL", "http://api.example.com").rstrip("/")
            remove_ship_addr(base, _cache.get("auth_token", ""), sid)
        except Exception as e:  # noqa: BLE001
            print(f"[conftest] sessionfinish remove_ship_addr({sid}) failed: {e}", file=sys.stderr)