STM32F4-Base/test/host/common/unit_test_harness.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""unit_test_harness.py - 主机侧单元测试（L0.5）的通用运行器 harness。

职责（**不认识任何具体模块**）：
    探测本机 MSVC 工具链 -> 拼 INCLUDE/LIB -> 编译测试单元 -> 运行 -> 判定退出码。
    被测模块的差异（源文件、桩目录、额外 include、断言汇总行特征）全部由调用方
    （各模块自己的 `run_unit_test.py`）传入。

为什么要 harness（2026-10-01 测试架构评审 T-3）：
    看门狗单测的运行器（`test/host/watchdog/run_unit_test.py`，170 行）把工具链
    探测、`INCLUDE`/`LIB` 拼接、GBK 解码、`/wd4819` 抑制、退出码契约**全部写死在
    模块目录内**。第二个 L0.5 单测出现时须整段复制——工具链细节一旦修正（新增 VS
    版本、SDK 布局变化、编码处理调整）就会发散。抽到这里后，各模块运行器只剩
    "声明式"配置。

实现要点（为何不用 vcvars64.bat）：
    vcvars64.bat 内部调用 reg.exe 探测环境；本机策略禁止 reg.exe。故这里**直接
    拼 INCLUDE/LIB 环境变量**并绝对路径调用 cl.exe——只需 3 个已知目录
    （MSVC 版本 / Windows SDK include / SDK lib），无副作用、可重复。

退出码契约（各模块运行器统一遵循）：
    0 = PASS（编译 + 运行成功，且日志满足 summary_marker 要求）
    1 = FAIL（有测试用例未通过）
    2 = 环境错误（找不到编译器 / 编译失败 / 未产生汇总行）

用法（模块侧 `run_unit_test.py`）：

    from unit_test_harness import run_host_unit_test, TestBuild

    if __name__ == "__main__":
        raise SystemExit(run_host_unit_test(TestBuild(
            name="Task Watchdog Tests",
            src=<测试单元 .c 的绝对路径>,
            include_dirs=[<桩目录>, <生产源码目录>],
            summary_marker="=== Task Watchdog Tests:",
        )))

设计依据：docs/任务规划/登记簿.md T-3；docs/测试说明/测试开发标准流程.md §4.2
创建日期：2026-10-01
"""

import glob
import hashlib
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass, field

# ----------------------------- 工具链探测位置 -----------------------------
# 显式覆盖（环境变量，见 docs/环境与部署/本地工具路径环境变量方案.md）：
#   DTU_MSVC_CL           x64 Host cl.exe 完整路径（最高优先级；无效直接报错）
#   DTU_WINDOWS_KITS_ROOT Windows Kits\10 根目录
# 未设置时用标准布局自动探测：默认安装根由系统变量 %ProgramFiles% /
# %ProgramFiles(x86)% 推导（不写死盘符）；`_first` 取每个根目录下最大的版本子目录。


def _default_vs_roots():
    """标准 VS 安装根（2019 -> Program Files (x86)；2022/2026 -> Program Files）。"""
    pf = os.environ.get("ProgramFiles", r"C:\Program Files")
    pf86 = os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)")
    roots = []
    for year, base in (("2019", pf86), ("2022", pf), ("2026", pf)):
        for edition in ("BuildTools", "Community", "Professional", "Enterprise"):
            roots.append(os.path.join(base, "Microsoft Visual Studio", year, edition))
    # VS2026 的实际安装目录名是 "18"（2026-10 台架实测：vswhere → ...\18\Community，
    # MSVC 14.51 + SDK 10.0.26100 布局与 2022 一致，cl.exe 路径模式不变）；
    # 上面 "2026" 目录名在本机并不存在，保留以防将来布局回改
    for edition in ("BuildTools", "Community", "Professional", "Enterprise"):
        roots.append(os.path.join(pf, "Microsoft Visual Studio", "18", edition))
    return roots


def _default_sdk_roots():
    r"""Windows SDK 默认安装根（%ProgramFiles(x86)%\Windows Kits\10）。"""
    pf86 = os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)")
    return [os.path.join(pf86, "Windows Kits", "10")]


def _sdk_dirs(sdk):
    """返回某 Windows Kits 根下的 (include 子目录列表, lib 子目录列表)。"""
    sdk_inc = _first(os.path.join(sdk, r"Include\*"))
    sdk_lib = _first(os.path.join(sdk, r"Lib\*"))
    inc = []
    if sdk_inc:
        inc = [os.path.join(sdk_inc, "ucrt"),
               os.path.join(sdk_inc, "shared"),
               os.path.join(sdk_inc, "um")]
    lib = []
    if sdk_lib:
        lib = [os.path.join(sdk_lib, "ucrt", "x64"),
               os.path.join(sdk_lib, "um", "x64")]
    return inc, lib

# 编译开关（各模块共用；模块特有开关见 TestBuild.extra_flags）
_BASE_FLAGS = [
    "/nologo", "/W4", "/std:c11", "/utf-8",
    # C4819（"源代码包含不能在当前代码页中表示的字符"）是**纯编码诊断**：
    # 测试单元确为 UTF-8（带 BOM，已核验），且已显式传 /utf-8，字节解码无误、
    # 编译产物正确。但在"只设 INCLUDE/LIB、不调 vcvars"的调用方式下，
    # MSVC 的代码页探测会退化为 `代码页(0)` → 恒定误报。
    # 测试单元仅主机侧编译（ARMCC 从不编译 test/），故显式屏蔽，保持门禁输出干净。
    "/wd4819",
]


@dataclass
class TestBuild:
    """一个主机侧单测的构建描述（声明式；harness 只读这些字段）。"""

    name: str                       # 人类可读名称（打印用），如 "Task Watchdog Tests"
    src: str                        # 测试单元 .c 的绝对路径（内部 #include 生产源）
    include_dirs: list = field(default_factory=list)   # /I 目录（桩目录需在前，覆盖真实头）
    summary_marker: str = ""        # 运行输出中必须出现的汇总行前缀；空 = 不校验
    extra_flags: list = field(default_factory=list)    # 追加的 cl.exe 开关
    exe_name: str = ""              # 产物 exe 名；空 = 由 name 派生
    obj_tag: str = ""               # 中间 obj 目录名；空 = 由 name 派生

    def resolved_exe(self):
        if self.exe_name:
            return self.exe_name
        tag = "".join(c if c.isalnum() else "_" for c in self.name).strip("_").lower()
        return tag + ".exe"

    def resolved_obj_tag(self):
        if self.obj_tag:
            return self.obj_tag
        tag = "".join(c if c.isalnum() else "_" for c in self.name).strip("_").lower()
        return tag + "_obj"


# ----------------------------- 产物目录归属（跨工作树并行） -----------------------------
# 现象（2026-10-25 三条道并行开发实测）：全量 ci_local 的 L0 里 test-cppcheck /
# test-coverage 偶发 "files were modified by this hook"，单独重跑即绿、工作树零改动。
# 根因是旧写法把 obj/exe 直接落在**全机共用**的 %TEMP%，名字由套件名派生
# （%TEMP%\net_socket_tests.exe 与 %TEMP%\net_socket_obj）。多棵工作树并行跑**同名
# 套件**就会互相覆盖，而最坏的不是编不过：A 树编译完，B 树把同名 exe 又盖一遍，
# A 树随后执行的是 **B 树编出来的二进制**——两棵树的 summary_marker 一模一样，
# harness 分辨不出来，于是拿着别人的运行结果给自己下结论（比红更难查）。
# 覆盖率侧同源：%TEMP%\dtu_test_cov 也是固定共享名（docs/测试说明/主机单测覆盖率
# 口径.md 里"并行期每人只跑 --module 自己模块"就是为绕开它写的规避；本处修好后
# 那条规避不再必要）。
#
# 口径：**按工作树分家**。产物根 = %TEMP%\dtu_host\<仓库根路径哈希>\，跨树天然不撞，
# 同一棵树内仍复用同一份目录（33 套的编译缓存没必要白白重编）。**不用锁**——锁会把
# L0.5 变成和 build/bench 一样的全局串行，而这里并没有需要互斥的独占资源，只是名字
# 撞了。显式覆盖：DTU_HOST_BUILD=<目录>（CI/容器要固定路径时用）。


def repo_root():
    """本 harness 所在工作树的根（test/host/common/ 往上三级）。

    每棵工作树都有自己的 test/host/（工作树之间只共享 .venv 与模拟器 SDK 两个目录
    联接），所以"harness 文件自身的位置"就是可靠的树身份，不依赖 cwd 或环境变量。
    """
    here = os.path.dirname(os.path.abspath(__file__))   # <root>/test/host/common
    root = here
    for _ in range(3):                                      # common -> host -> test -> root
        root = os.path.dirname(root)
    return os.path.normpath(root)


def host_build_root():
    """主机单测产物根目录（按工作树分家；判据见上方"产物目录归属"）。"""
    env = os.environ.get("DTU_HOST_BUILD")
    if env:
        return os.path.abspath(os.path.expanduser(env))
    key = hashlib.sha1(os.path.normcase(repo_root()).encode("utf-8")).hexdigest()[:10]
    return os.path.join(tempfile.gettempdir(), "dtu_host", key)


def _write_owner(root, build):
    """在产物根里落一份归属凭据（<exe>.owner.json）：并发/偶发问题时一眼看出这份
    产物属于哪棵树、哪个进程、什么时候跑的。写失败不阻塞测试（只是诊断信息）。"""
    info = {"suite": build.name, "repo_root": repo_root(), "pid": os.getpid(),
            "started": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
            "build_root": root}
    try:
        with open(os.path.join(root, build.resolved_exe() + ".owner.json"), "w",
                  encoding="utf-8") as f:
            json.dump(info, f, ensure_ascii=True, sort_keys=True)
            f.write("\n")
    except OSError:
        pass


def _first(pattern):
    """glob 命中里取最大者（版本号目录的字符串序即数值序，前提是位数一致）。"""
    hits = sorted(glob.glob(pattern))
    return hits[-1] if hits else None


def decode(b):
    """把 subprocess 字节输出解码为字符串。

    本机 MSVC 输出为 GBK（代码页 936），测试程序自身输出为 UTF-8；
    按 UTF-8 优先、失败回退 GBK 处理。
    """
    if isinstance(b, str):
        return b
    try:
        return b.decode("utf-8")
    except UnicodeDecodeError:
        return b.decode("gbk", errors="replace")


def locate_toolchain():
    """返回 (cl.exe, include_dirs, lib_dirs)；找不到返回 (None, [], [])。

    优先级（见方案文档）：DTU_MSVC_CL / DTU_WINDOWS_KITS_ROOT（显式设置但
    无效 → 立即报错，**不静默回退**自动探测）→ 标准布局自动探测
    （_default_vs_roots / _default_sdk_roots）。
    """
    # ---- cl.exe ----
    explicit_cl = os.environ.get("DTU_MSVC_CL", "").strip()
    if explicit_cl:
        if not os.path.isfile(explicit_cl):
            print("[ENV] DTU_MSVC_CL 已设置但不存在: {}（请修正变量值，不做自动探测回退）"
                  .format(explicit_cl), file=sys.stderr)
            return None, [], []
        tail = os.path.normpath(explicit_cl).lower()
        if not tail.endswith(os.path.join("bin", "hostx64", "x64", "cl.exe")):
            print("[ENV] DTU_MSVC_CL 必须是 x64 Host 编译器（路径以 bin\\Hostx64\\x64\\cl.exe 结尾）: {}"
                  .format(explicit_cl), file=sys.stderr)
            return None, [], []
        cl = explicit_cl
    else:
        cl = None
        for vs in _default_vs_roots():
            c = _first(os.path.join(vs, r"VC\Tools\MSVC\*\bin\Hostx64\x64\cl.exe"))
            if c:
                cl = c
                break
    if not cl:
        return None, [], []

    # cl.exe = <vs>\VC\Tools\MSVC\<ver>\bin\Hostx64\x64\cl.exe
    msvc_ver_dir = os.path.dirname(os.path.dirname(os.path.dirname(
        os.path.dirname(cl))))  # -> ...\MSVC\<ver>
    inc = [os.path.join(msvc_ver_dir, "include")]
    lib = [os.path.join(msvc_ver_dir, "lib", "x64")]

    # ---- Windows SDK ----
    explicit_sdk = os.environ.get("DTU_WINDOWS_KITS_ROOT", "").strip()
    if explicit_sdk:
        if not os.path.isdir(explicit_sdk):
            print("[ENV] DTU_WINDOWS_KITS_ROOT 已设置但不存在: {}（请修正变量值，不做自动探测回退）"
                  .format(explicit_sdk), file=sys.stderr)
            return None, [], []
        sdk_inc, sdk_lib = _sdk_dirs(explicit_sdk)
        if not sdk_inc and not sdk_lib:
            print("[ENV] DTU_WINDOWS_KITS_ROOT 下未找到 Include\\<ver> 或 Lib\\<ver>: {}".format(explicit_sdk),
                  file=sys.stderr)
            return None, [], []
    else:
        sdk_inc, sdk_lib = [], []
        for sdk in _default_sdk_roots():
            sdk_inc, sdk_lib = _sdk_dirs(sdk)
            if sdk_inc or sdk_lib:
                break
    inc += sdk_inc
    lib += sdk_lib
    return cl, inc, lib


def build_and_run(build, out=None):
    """编译并运行一个测试单元。返回退出码 0/1/2（见模块 docstring 契约）。"""
    if out is None:
        out = sys.stdout
    err = sys.stderr

    cl, inc, lib = locate_toolchain()
    if not cl:
        print("[ENV] 找不到 MSVC cl.exe（设置 DTU_MSVC_CL，或安装 VS2019/2022/2026 的 C++ 工具链）", file=err)
        return 2

    # 产物落**按工作树分家**的根目录（判据见上方"产物目录归属"）；旧写法用
    # tempfile.gettempdir() 把 obj/exe 摊在全机共用的 %TEMP%，跨树同名即互踩。
    # 而且**每次跑独占一个子目录**（obj 与 exe 都在里面）——这一步才真正消灭竞态：
    # 两个进程哪怕同树并发、哪怕有人把 DTU_HOST_BUILD 指到同一处，也拿不到同一个
    # exe 路径（旧写法会出现"A 编完 B 覆盖、A 跑的是 B 的二进制"），也拿不到同一个
    # .obj（并发 cl.exe 写同名 obj 会直接编不过）。这里不牺牲什么：每个测试单元就一个
    # .c，cl 每次都是整体重编，"共享 obj 省编译"本来就是错觉。跑完默认删掉该子目录；
    # 调试要保留就用 DTU_HOST_BUILD_KEEP=1（会打印路径）。
    root = host_build_root()
    rundir = os.path.join(root, "runs", "{}-{}".format(os.getpid(), time.time_ns()))
    objdir = os.path.join(rundir, build.resolved_obj_tag())
    os.makedirs(objdir, exist_ok=True)   # 顺带把 rundir 建出来
    exe = os.path.join(rundir, build.resolved_exe())
    _write_owner(rundir, build)
    keep = (os.environ.get("DTU_HOST_BUILD_KEEP") or "").strip().lower() not in (
        "", "0", "false", "no")
    env = dict(os.environ)
    env["INCLUDE"] = os.pathsep.join(inc)
    env["LIB"] = os.pathsep.join(lib)
    env["PATH"] = os.path.dirname(cl) + os.pathsep + env.get("PATH", "")

    cmd = [cl] + _BASE_FLAGS
    for d in build.include_dirs:
        cmd += ["/I", d]
    cmd += list(build.extra_flags)
    cmd += [build.src, "/Fe:" + exe, "/Fo:" + objdir + os.sep]

    print("=== 编译并运行 {} ===（产物 {}）".format(build.name, rundir), file=out)
    try:
        cp = subprocess.run(cmd, env=env, capture_output=True)
        cout, cerr = decode(cp.stdout), decode(cp.stderr)
        if cout:
            out.write(cout)
        if cerr:
            err.write(cerr)
        if cp.returncode != 0:
            print("[ENV] 编译失败", file=err)
            return 2

        # 只跑"本次编译确实产出的"exe：路径是本次独占的，所以只要存在就必须是本次
        # 写的；不存在 = 编译虽返回 0 但没落地（磁盘/杀软/并发清理），一律按环境错误
        # 处理，绝不拿着来路不明的二进制出结论。
        if not os.path.isfile(exe):
            print("[ENV] 编译返回 0 但没有产物 exe: {}".format(exe), file=err)
            return 2

        run = subprocess.run([exe], env=env, capture_output=True)
        rout = decode(run.stdout)
        out.write(rout)
        rerr = decode(run.stderr)
        if rerr:
            err.write(rerr)

        if build.summary_marker and build.summary_marker not in rout:
            print("[ENV] 未产生测试汇总行（找不到 {!r}）".format(build.summary_marker),
                  file=err)
            return 2
        if run.returncode == 0:
            print("[OK] 单元测试全部通过", file=out)
            return 0
        print("[FAIL] 有测试用例未通过", file=err)
        return 1
    finally:
        # 跑完（含异常/Ctrl-C）就把本次独占目录收掉：残留的半截 exe 永远不会被下一
        # 次运行当作自己的产物。调试场景用 DTU_HOST_BUILD_KEEP=1 保留（会打印路径）。
        if keep:
            print("[KEEP] 本次产物保留在 {}".format(rundir), file=out)
        else:
            shutil.rmtree(rundir, ignore_errors=True)


def run_host_unit_test(build):
    """模块侧入口薄封装：固定控制台 UTF-8 后调用 build_and_run。"""
    for stream in (sys.stdout, sys.stderr):
        try:
            stream.reconfigure(encoding="utf-8", errors="replace")
        except Exception:
            pass
    return build_and_run(build)