"""
父进程死亡防护（PR_SET_PDEATHSIG）测试

背景与语义见 gateway/src/chongming/app/core/parent_death.py：进程树 onefile 引导 →
granian master → worker（fork）。上游被 SIGKILL 时 worker 会成孤儿（继续监听
端口、挂在服务发现里，其它网关派发到"半死副本"每次白吃 2 秒超时）。守卫让内核
在父进程死亡时给本进程发 SIGTERM（走既有优雅停机摘除注册）。

验证（三层 fork 复刻真实进程树：本测试(引导) → 中间进程(master) → 孙进程(worker)）：
  1. 关闭开关 CHONGMING_NO_PARENT_GUARD=1 → 不生效（返回 False）
  2. 中间进程 SIGKILL 自己（模拟 master 被 -9，内核不可捕获）→ 已设守卫的孙进程
     收到 SIGTERM 并退出
  3. 对照：同样结构但不设守卫 → 孙进程收不到信号（证明信号确实来自 PDEATHSIG，
     而不是别的东西清理孤儿）
  4. 幂等：重复调用不重复 prctl

运行: uv run python test/test_parent_death.py
"""
import importlib.util
import os
import signal
import sys
import tempfile
import time
from pathlib import Path

ROOT = Path(__file__).parent.parent
_PD_PATH = ROOT / "gateway" / "src" / "chongming" / "app" / "core" / "parent_death.py"

# 独立加载（parent_death 无包内依赖，避免连带导入整个 app 包）
_spec = importlib.util.spec_from_file_location("chongming.app.core.parent_death", _PD_PATH)
pd = importlib.util.module_from_spec(_spec)
sys.modules["chongming.app.core.parent_death"] = pd
_spec.loader.exec_module(pd)  # type: ignore[union-attr]

failures = []


def check(name, cond, detail=""):
    status = "PASS" if cond else "FAIL"
    print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
    if not cond:
        failures.append(name)


# ========== 1. 关闭开关 ==========
os.environ[pd.GUARD_OFF_ENV] = "1"
check("关闭开关打开时不生效", pd.guard_parent_death() is False)
os.environ.pop(pd.GUARD_OFF_ENV, None)

# ========== 2/3. 三层 fork：守卫组 / 对照组 ==========
tmp = Path(tempfile.mkdtemp(prefix="parent_death_test_"))


def scenario(name: str, use_guard: bool) -> dict:
    """中间进程 SIGKILL 自己，观察孙进程是否收到 SIGTERM。"""
    d = tmp / name
    d.mkdir()
    ready, dead, pidf = d / "ready", d / "dead", d / "gc.pid"

    middle = os.fork()
    if middle == 0:  # 中间进程（master 角色）
        gc = os.fork()
        if gc == 0:  # 孙进程（worker 角色）
            def _on_term(*_):
                dead.write_text("term", encoding="utf-8")
                os._exit(0)

            signal.signal(signal.SIGTERM, _on_term)
            if use_guard:
                pd.guard_parent_death()
            pidf.write_text(str(os.getpid()), encoding="utf-8")
            ready.write_text("1", encoding="utf-8")
            time.sleep(15)  # 等信号；超时说明没收到
            os._exit(1)
        for _ in range(250):  # 等孙进程就绪，再硬杀自己
            if ready.exists():
                break
            time.sleep(0.02)
        os.kill(os.getpid(), signal.SIGKILL)
        os._exit(1)  # 不会走到

    os.waitpid(middle, 0)  # 中间进程必死于 SIGKILL
    gc_pid = int(pidf.read_text(encoding="utf-8"))
    for _ in range(100):  # 最多等 5s
        if dead.exists():
            break
        time.sleep(0.05)
    got = dead.exists()
    try:  # 清理（对照组孙进程仍活着）
        os.kill(gc_pid, signal.SIGKILL)
    except ProcessLookupError:
        pass
    return {"got_term": got, "gc_pid": gc_pid}


guarded = scenario("guarded", use_guard=True)
check(
    "master(SIGKILL) 死后，已设守卫的 worker 收到 SIGTERM 退出",
    guarded["got_term"],
    f"gc={guarded['gc_pid']}",
)

control = scenario("control", use_guard=False)
check(
    "对照组（未设守卫）收不到信号 —— 证明退出确实来自 PDEATHSIG",
    not control["got_term"],
    f"gc={control['gc_pid']}",
)

# ========== 4. 幂等（在子进程里做：避免给测试进程本身设 PDEATHSIG） ==========
pid = os.fork()
if pid == 0:
    a = pd.guard_parent_death()
    b = pd.guard_parent_death()
    os._exit(0 if (a is True and b is True) else 1)
_, status = os.waitpid(pid, 0)
check("重复调用幂等（不重复 prctl）", os.waitstatus_to_exitcode(status) == 0)

# ========== 5. fork 继承回归（master 先设 → worker 仍必须自己设上） ==========
# 背景：若用模块级"已设置"标志位做幂等，fork 的子进程会继承 True 而短路——
# 但内核在 fork 时恰好清零 PDEATHSIG，于是 worker 其实没设上（本轮实测到过）。
# 这里父进程先真实设置，再 fork，断言子进程调用守卫时**内核状态确实被设上**。
libc5 = pd._load_libc()
pd.guard_parent_death()  # 父进程（本测试进程）真实设置一次
pid = os.fork()
if pid == 0:
    cleared = pd._current_pdeathsig(libc5) == 0  # fork 后内核已清零（不继承）
    ok = pd.guard_parent_death() is True  # 回归点：不能因父进程设过而短路
    set_now = pd._current_pdeathsig(libc5) == int(signal.SIGTERM)
    os._exit(0 if (cleared and ok and set_now) else 1)
_, status = os.waitpid(pid, 0)
libc5.prctl(pd._PR_SET_PDEATHSIG, 0)  # 清除测试进程自己的设置，避免副作用
check(
    "父进程设过后，fork 的子进程仍会真实设置 PDEATHSIG",
    os.waitstatus_to_exitcode(status) == 0,
)

import shutil

shutil.rmtree(tmp, ignore_errors=True)

print("\n" + ("=" * 40))
if failures:
    print(f"❌ {len(failures)} 项未通过: {failures}")
    sys.exit(1)
print("✅ 全部通过")