DCU_DesktopController/scripts/test_app_lease.py

#!/usr/bin/env python3
"""test_app_lease.py — 验证 "app 全部窗口关闭 → 自动停 EtherCAT 循环" 链路。

不打真实 EtherCAT 硬件: 导入 esp32_server 后把 _ecat_post/_ecat_find_port
换成假实现, 其余(WS 握手、租约计数、GRACE 吸收、代理记账)全部走真实代码。

用法: python scripts/test_app_lease.py
"""
import base64
import hashlib
import os
import socket
import sys
import threading
import time
import urllib.request

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import esp32_server as srv  # noqa: E402
import ecat_client  # noqa: E402  # P2: _ecat_post/_ecat_find_port 实现在此模块

WS_MAGIC = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
calls = []            # 假 _ecat_post 记录到的请求路径
calls_lock = threading.Lock()


def fake_post(port, path, body, timeout, **kwargs):
    with calls_lock:
        calls.append(path)
    # 带空格的 "ok": true —— 故意与真 native 的紧凑格式不同,
    # 锁死"代理记账不能靠子串匹配"这条回归(e2e 里就是这么炸过一次)。
    return 200, b'{"ok": true, "log": ["fake"]}'


def ws_connect(port):
    """打开一条真实 TCP 上的 WebSocket, 校验 101 握手, 返回 socket。"""
    s = socket.create_connection(("127.0.0.1", port))
    key = base64.b64encode(os.urandom(16)).decode()
    s.sendall(("GET /ws/app-lease HTTP/1.1\r\n"
               f"Host: 127.0.0.1:{port}\r\n"
               "Upgrade: websocket\r\nConnection: Upgrade\r\n"
               f"Sec-WebSocket-Key: {key}\r\n"
               "Sec-WebSocket-Version: 13\r\n\r\n").encode("ascii"))
    head = b""
    while b"\r\n\r\n" not in head:
        chunk = s.recv(4096)
        if not chunk:
            raise AssertionError("握手被直接关闭")
        head += chunk
    text = head.decode("ascii", "replace")
    assert "101" in text.splitlines()[0], f"握手没回 101: {text[:120]}"
    expect = base64.b64encode(
        hashlib.sha1((key + WS_MAGIC).encode()).digest()).decode()
    assert expect in text, "Sec-WebSocket-Accept 不对"
    return s


def ws_graceful_close(s):
    """浏览器关 WS 的真实行为: 先发掩码 Close 帧, 等服务端回显后才拆 TCP。
    服务端若不回 Close(RFC 6455 要求), 这条链路的 FIN 永远到不了 → 租约不释放。"""
    s.sendall(b"\x88\x80" + b"\x00\x00\x00\x00")      # FIN+Close, masked, len 0
    s.settimeout(3.0)
    echo = s.recv(2)
    assert echo == b"\x88\x00", f"Close 回显不对: {echo!r}"
    s.close()


def post_json(port, path):
    req = urllib.request.Request(
        f"http://127.0.0.1:{port}{path}", data=b"{}",
        headers={"Content-Type": "application/json"}, method="POST")
    with urllib.request.urlopen(req, timeout=10) as r:
        return r.status, r.read()


def flag():
    with srv.g_app_lock:
        return srv.g_ecat_motor_live


def wait_flag(want, timeout=8.0):
    deadline = time.time() + timeout
    while time.time() < deadline:
        if flag() == want:
            return True
        time.sleep(0.05)
    return False


def main():
    # 加速测试: GRACE 从 3s 缩到 0.3s(重试轮次间隔同用该值, 一并缩短无妨)
    srv.APP_LEASE_GRACE_S = 0.3
    # P2 拆分后：lease 停循环路径用 esp32_server 命名空间；
    # HTTP 代理路径在 ecat_client 内部解析 _ecat_post —— 两处都要 patch。
    srv._ecat_post = fake_post
    srv._ecat_find_port = lambda: 9999
    ecat_client._ecat_post = fake_post
    ecat_client._ecat_find_port = lambda: 9999

    # 找空闲端口起真服务
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        port = probe.getsockname()[1]
    httpd, port = srv._bind_server(port)
    assert httpd is not None, "起不了测试服务"
    threading.Thread(target=httpd.serve_forever, daemon=True).start()

    # 1) 非 WS 的普通 GET 不受影响
    with urllib.request.urlopen(f"http://127.0.0.1:{port}/tabs.html", timeout=5) as r:
        assert r.status == 200 and b"tabbar" in r.read(), "普通静态页被 WS 路由挡住了"
    print("PASS 1: 静态页照常服务")

    # 2) 经代理的 motor/start 置位循环标志
    post_json(port, "/ecat/api/motor/start")
    assert flag() is True, "motor/start 经代理后没把 g_ecat_motor_live 置位"
    print("PASS 2: 代理记账 motor/start → live=True")

    # 3) 两条租约, 关一条 → 不该停
    a = ws_connect(port)
    b = ws_connect(port)
    time.sleep(0.1)
    a.close()
    time.sleep(1.0)                     # > GRACE
    assert "/api/op/stop" not in calls, "还剩一个窗口就停了循环!"
    print("PASS 3: 仍有窗口在线时不停")

    # 4) 全关 → GRACE 后自动 op/stop (走浏览器式 Close 帧回显路径,
    #    与场景 3 的裸 FIN 路径各覆盖一种断开方式)
    ws_graceful_close(b)
    deadline = time.time() + 5
    while time.time() < deadline and "/api/op/stop" not in calls:
        time.sleep(0.05)
    assert "/api/op/stop" in calls, "全部断开后没停循环"
    assert wait_flag(False), "停完没清 live 标志"
    print("PASS 4: 最后一条租约断开 → /api/op/stop")

    # 5) 刷新吸收: 断开后在 GRACE 内重连 → 不该停
    with calls_lock:
        calls.clear()
    post_json(port, "/ecat/api/motor/start")
    c = ws_connect(port)
    c.close()
    d = ws_connect(port)                # 模拟刷新重连(远快于 GRACE)
    time.sleep(1.0)
    assert "/api/op/stop" not in calls, "页面刷新被误判成关闭 app!"
    print("PASS 5: 刷新重连被 GRACE 吸收")

    # 6) 用户显式 motor/stop 后, 关窗口不该再打 op/stop
    d.close()
    time.sleep(1.0)
    with calls_lock:
        calls.clear()
    post_json(port, "/ecat/api/motor/start")
    post_json(port, "/ecat/api/motor/stop")
    assert flag() is False, "motor/stop 没清 live 标志"
    e = ws_connect(port)
    e.close()
    time.sleep(1.0)
    assert "/api/op/stop" not in calls, "循环本来就没跑还去停(多此一举)"
    print("PASS 6: 循环已停时不再重复停")

    # 7) 窗口模式竞态: 服务进程在 GRACE 定时器触发前就退出(/api/shutdown),
    #    且租约连接还挂着 —— 退出路径必须同步把循环停掉, 不能指望定时器
    #    (复用的 ecat_native 的 parent 不是本服务, 没人替它清场)。
    srv.APP_LEASE_GRACE_S = 30.0      # 定时器"永不"到点, 证明停的是退出路径
    srv.g_httpd = httpd
    post_json(port, "/ecat/api/motor/start")
    assert flag() is True
    with calls_lock:
        calls.clear()
    f = ws_connect(port)              # 租约仍在(未断开)
    post_json(port, "/api/shutdown")
    deadline = time.time() + 5
    while time.time() < deadline and "/api/op/stop" not in calls:
        time.sleep(0.05)
    assert "/api/op/stop" in calls, "服务进程退出前没停循环(窗口模式竞态)"
    assert wait_flag(False, timeout=2.0)
    f.close()
    print("PASS 7: 服务进程退出时同步停循环")

    httpd.shutdown()
    print("ALL PASS")


if __name__ == "__main__":
    main()