socks-manager/tests/e2e_stop_semantics.py

"""测试手动停止语义: stop_instance 后 sync loop 不得自动重启。

这是 eba7ae8 引入的回归: sync loop 周期调 sync_instances, 而
sync_instances 只看 DB 里 enabled=True 的实例。如果 stop_instance
只停进程不改 enabled, 30 秒内实例会被自动拉回来, 用户手动停止失效。

流程:
  1. create_app (sync loop interval=2s)
  2. 插 enabled 实例, 等 loop 拉起
  3. 调 stop_instance(iid)
  4. 等 5s (2 个 loop 周期), 断言端口始终不回来 + DB.enabled=False
  5. 调 start_instance(iid), 断言端口回来 + DB.enabled=True
"""
import asyncio
import logging
import os
import socket
import sys
import tempfile
import time

sys.path.insert(0, os.getcwd())

_tmp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False, dir="/tmp")
_tmp_db.close()
os.environ["SM_DB_URI"] = f"sqlite:///{_tmp_db.name}"
os.environ["SM_ADMIN_PASSWORD"] = "test123"
os.environ["SM_SECRET_KEY"] = "test-secret-key"
os.environ["SM_LOG_LEVEL"] = "WARNING"
os.environ["SM_SYNC_INTERVAL"] = "2"
os.environ["SM_SYNC_LOOP_ENABLE"] = "true"

from app import create_app
from database import db as _db
from models import Instance


def find_free_port():
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    p = s.getsockname()[1]
    s.close()
    return p


def is_listening(port):
    try:
        s = socket.socket(); s.settimeout(0.5)
        s.connect(("127.0.0.1", port)); s.close()
        return True
    except (ConnectionRefusedError, socket.timeout, OSError):
        return False


def wait_listening(port, timeout_s=8.0, step=0.3):
    t0 = time.time()
    while time.time() - t0 < timeout_s:
        if is_listening(port):
            return True, time.time() - t0
        time.sleep(step)
    return False, time.time() - t0


def main():
    logging.basicConfig(level=logging.WARNING,
                        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
    port = find_free_port()
    flask_app = create_app()
    mgr = flask_app.instance_manager  # type: ignore[attr-defined]

    # 插入 enabled 实例
    with flask_app.app_context():
        inst = Instance(
            name="semtest", listen_host="127.0.0.1", listen_port=port,
            timeout=5, enabled=True, auth_method="none",
            bandwidth_down=0, bandwidth_up=0, max_concurrent=10,
        )
        _db.session.add(inst)
        _db.session.commit()
        iid = inst.id
        print(f"[setup] inserted instance id={iid} port={port}")

    # 等 sync loop 拉起
    ok, elapsed = wait_listening(port, timeout_s=8.0)
    assert ok, f"sync loop 没在 8s 内拉起 port {port}"
    print(f"[1] sync loop 初次拉起 OK ({elapsed:.1f}s)")

    # 手动停止
    assert mgr.stop_instance(iid), "stop_instance 返回 False"
    time.sleep(0.5)
    assert not is_listening(port), "stop_instance 后端口仍 listen"
    print(f"[2] stop_instance OK, port {port} 已停")

    # 等 2 个 sync loop 周期, 断言不被自动拉回
    t0 = time.time()
    while time.time() - t0 < 5.0:
        assert not is_listening(port), (
            f"REGRESSION: stop_instance 后 {time.time()-t0:.1f}s 端口被 sync loop 自动拉回!"
        )
        time.sleep(0.3)
    with flask_app.app_context():
        i = Instance.query.get(iid)
        assert i.enabled is False, f"stop_instance 后 DB.enabled={i.enabled}, 应为 False"
    print(f"[3] 5s (2 个 loop 周期) 内未被自动重启, DB.enabled=False ✓")

    # 手动启动, 应恢复
    assert mgr.start_instance(iid), "start_instance 返回 False"
    ok, elapsed = wait_listening(port, timeout_s=5.0)
    assert ok, "start_instance 后端口未恢复"
    with flask_app.app_context():
        i = Instance.query.get(iid)
        assert i.enabled is True, f"start_instance 后 DB.enabled={i.enabled}, 应为 True"
        assert i.running is True, f"start_instance 后 DB.running={i.running}, 应为 True"
    print(f"[4] start_instance OK, port {port} 恢复, DB.enabled=True ✓")

    # 清理
    mgr.stop_instance(iid)
    try: os.unlink(_tmp_db.name)
    except Exception: pass
    print("\nPASS: 手动停止不被 sync loop 自动重启; 手动启动正常恢复")
    return True


if __name__ == "__main__":
    rc = main()
    sys.exit(0 if rc else 1)