#!/usr/bin/env python3
"""U39：运行 ZooKeeper 3.9.6 三节点同 JVM 实验并保存证据。"""

import hashlib
import json
import os
import pathlib
import platform
import re
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import xml.etree.ElementTree as ET
from datetime import datetime

ROOT = pathlib.Path(__file__).resolve().parent
EVIDENCE = ROOT / "evidence/u39-2026-10-30.json"
MARKERS = [
    "U39 acl=PASS unauth=NOAUTH wrong=NOAUTH digest=READ",
    "U39 watch=PASS notifications=1 reread=v2",
    "U39 session=PASS reconnect=same local-expired-before-restart=true post-restart-ephemeral=deleted",
    "U39 quorum=PASS one-loss-write=majority recovered-node-read=recovered",
    "U39 cleanup=PASS tempRoot=removed",
]
EXPECTED = {
    "org.apache.zookeeper:zookeeper": "3.9.6",
    "org.apache.curator:curator-test": "5.7.0",
    "org.junit.jupiter:junit-jupiter": "6.0.3",
}


def now():
    return datetime.now().astimezone().isoformat()


def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def write_json(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", encoding="utf-8", dir=path.parent, prefix=path.name + ".", delete=False
        ) as stream:
            temporary = pathlib.Path(stream.name)
            json.dump(value, stream, ensure_ascii=False, indent=2)
            stream.write("\n")
        temporary.replace(path)
    finally:
        if temporary:
            temporary.unlink(missing_ok=True)


def artifact_digests():
    paths = [ROOT / name for name in ("pom.xml", "run.py", "README.md")]
    paths += sorted(path for path in (ROOT / "src").rglob("*") if path.is_file())
    return {str(path.relative_to(ROOT)): digest(path) for path in paths}


def maven_binary():
    supplied = os.environ.get("MAVEN_BIN")
    if supplied:
        return shutil.which(supplied) or supplied
    found = shutil.which("mvn")
    if found:
        return found
    candidates = sorted(pathlib.Path.home().glob(".m2/wrapper/dists/apache-maven-3.9.9/*/bin/mvn"))
    if candidates:
        return str(candidates[-1])
    raise RuntimeError("需要 Maven 3.9.x")


def group_members(pgid):
    result = subprocess.run(
        ["/bin/ps", "-axo", "pid=,pgid=,stat="],
        capture_output=True, text=True, check=True, timeout=5,
    )
    return [
        int(line.split()[0]) for line in result.stdout.splitlines()
        if int(line.split()[1]) == pgid and not line.split()[2].startswith("Z")
    ]


def cleanup(process):
    for signum in (signal.SIGTERM, signal.SIGKILL):
        process.poll()
        if not group_members(process.pid):
            return
        try:
            os.killpg(process.pid, signum)
        except ProcessLookupError:
            return
        deadline = time.monotonic() + 5
        while time.monotonic() < deadline:
            process.poll()
            if not group_members(process.pid):
                return
            time.sleep(0.05)
    raise RuntimeError("实验进程组清理失败: " + str(group_members(process.pid)))


def run(name, command, logs, env, timeout):
    log = logs / (name + ".log")
    process = None
    with log.open("w", encoding="utf-8") as stream:
        try:
            process = subprocess.Popen(
                command, cwd=ROOT, env=env, text=True, stdout=stream,
                stderr=subprocess.STDOUT, start_new_session=True,
            )
            code = process.wait(timeout=timeout)
        except subprocess.TimeoutExpired:
            code = 124
        finally:
            if process:
                cleanup(process)
    output = log.read_text(encoding="utf-8", errors="replace")
    result = {
        "name": name, "command": command, "exitCode": code,
        "log": str(log), "logSha256": digest(log),
    }
    if code:
        raise RuntimeError(name + " exit=" + str(code) + "\n" + output[-3000:])
    return output, result


def parse_dependencies(path):
    versions = {}
    for line in path.read_text(encoding="utf-8").splitlines():
        fields = line.strip().split(" ", 1)[0].split(":")
        if len(fields) in (5, 6) and fields[-1] in {"compile", "runtime", "test"}:
            versions[":".join(fields[:2])] = fields[-2]
    for coordinate, expected in EXPECTED.items():
        if versions.get(coordinate) != expected:
            raise ValueError(f"{coordinate}={versions.get(coordinate)}，期望 {expected}")
    return versions


def main():
    evidence = {
        "ticket": "U39 #76", "status": "running", "startedAt": now(),
        "input": "python3 labs/zookeeper/run.py", "exitCode": None,
        "environment": {"os": platform.platform(), "arch": platform.machine()},
        "observed": [], "results": [], "artifacts": {},
        "cleanup": "尚未完成",
        "limits": [
            "真实 ZooKeeper 3.9.6 三投票节点运行在同一个 JDK 17 测试 JVM 和同一台主机",
            "服务仅绑定本机随机端口；未模拟跨主机网络分区、磁盘损坏、跨地域延迟或生产容量",
            "Curator TestingCluster 只负责测试进程生命周期，不把 Curator 配方行为冒充 ZooKeeper 原生契约",
        ],
    }
    logs = None
    cluster_root = None
    exit_code = 1
    try:
        write_json(EVIDENCE, evidence)
        if len(sys.argv) != 1:
            raise ValueError("本入口不接受参数")
        evidence["artifacts"] = artifact_digests()
        logs_root = pathlib.Path.home() / ".cache/software-engineering-study/java-depth-impl/u39"
        logs_root.mkdir(parents=True, exist_ok=True)
        logs = pathlib.Path(tempfile.mkdtemp(prefix="run-", dir=logs_root))
        cluster_root = logs / "cluster-data"
        evidence["logs"] = str(logs)
        env = dict(os.environ)
        requested_java = env.get("JAVA_17_HOME")
        if requested_java:
            env["JAVA_HOME"] = requested_java
        elif sys.platform == "darwin":
            candidate = pathlib.Path("/Library/Java/JavaVirtualMachines/liberica-jdk-17.jdk/Contents/Home")
            if candidate.is_dir():
                env["JAVA_HOME"] = str(candidate)
            elif not env.get("JAVA_HOME"):
                raise RuntimeError("需要 JDK 17：设置 JAVA_HOME")
        elif not env.get("JAVA_HOME"):
            raise RuntimeError("需要 JDK 17：设置 JAVA_HOME")
        env["PATH"] = str(pathlib.Path(env["JAVA_HOME"]) / "bin") + os.pathsep + env["PATH"]
        java_output, java_result = run("jdk", ["java", "-version"], logs, env, 15)
        evidence["results"].append(java_result)
        if not re.search(r'version "17[.\"]', java_output):
            raise ValueError("需要 JDK 17")
        evidence["environment"]["jdk"] = java_output.strip().splitlines()[0]
        mvn = maven_binary()
        maven_output, maven_result = run("maven", [mvn, "-version"], logs, env, 30)
        evidence["results"].append(maven_result)
        if not re.search(r"Apache Maven 3[.]9[.]", maven_output):
            raise ValueError("需要 Maven 3.9.x")
        evidence["environment"]["maven"] = maven_output.strip().splitlines()[0]
        base = [
            mvn, "-o", "-B", "-ntp", "-q", "-f", str(ROOT / "pom.xml"),
            "-Du39.temp.root=" + str(cluster_root),
        ]
        output, test_result = run(
            "test", base + ["clean", "test", "-DfailIfNoTests=true"], logs, env, 180
        )
        evidence["results"].append(test_result)
        observed = [line for line in output.splitlines() if line.startswith("U39 ")]
        if observed != MARKERS:
            raise ValueError("观察行不完整或顺序错误: " + str(observed))
        evidence["observed"] = observed
        report = ROOT / "target/surefire-reports/TEST-study.zookeeper.ZooKeeperClusterTest.xml"
        suite = ET.parse(report).getroot()
        if any(int(suite.get(key, "0")) for key in ("failures", "errors", "skipped")):
            raise ValueError("Surefire 报告存在失败或跳过")
        evidence["testCount"] = int(suite.get("tests", "0"))
        evidence["testReport"] = {
            key: suite.get(key) for key in ("name", "tests", "failures", "errors", "skipped", "time")
        }
        evidence["testReport"]["testNames"] = [
            case.get("name") for case in suite.findall("testcase")
        ]
        if cluster_root.exists():
            raise ValueError("测试成功后仍残留集群 dataDir: " + str(cluster_root))
        dependency_file = logs / "dependencies.txt"
        _, dependency_result = run(
            "dependencies",
            base + ["dependency:list", "-DincludeScope=test",
                    "-DappendOutput=false", "-DoutputFile=" + str(dependency_file)],
            logs, env, 60,
        )
        evidence["results"].append(dependency_result)
        evidence["environment"]["dependencies"] = parse_dependencies(dependency_file)
        if artifact_digests() != evidence["artifacts"]:
            raise ValueError("运行期间实验文件发生变化")
        evidence["status"] = "passed"
        evidence["cleanup"] = "Java 断言与 runner 均确认本次进程和专用 dataDir 已删除"
        exit_code = 0
    except BaseException as exc:
        evidence["status"] = "failed"
        evidence["failure"] = {"type": type(exc).__name__, "message": str(exc)}
        print("FAILED: " + str(exc), file=sys.stderr)
    finally:
        if cluster_root is not None and cluster_root.exists():
            shutil.rmtree(cluster_root, ignore_errors=True)
            if cluster_root.exists():
                evidence["status"] = "failed"
                evidence["failure"] = {
                    "type": "CleanupError",
                    "message": "无法删除专用 dataDir: " + str(cluster_root),
                }
                exit_code = 1
            else:
                evidence["cleanup"] = "异常路径由 runner 定向删除专用 dataDir"
        evidence["finishedAt"] = now()
        evidence["exitCode"] = exit_code
        if logs:
            write_json(logs / "run.status.json", evidence)
        write_json(EVIDENCE, evidence)
    if exit_code == 0:
        print(f"PASS U39 tests={evidence['testCount']}; evidence: {EVIDENCE}")
    return exit_code


if __name__ == "__main__":
    sys.exit(main())