| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273 |
- """Mock API 内嵌启停的真实端口生命周期回归（缺陷回归守卫）。
- 背景：uvicorn 0.30+ 的 Server 已无 startup_error 属性，旧实现访问它会抛 AttributeError，
- 使 POST /api/mock-api/start 恒定返回 500（而服务其实已启动），端口被占用时也无法如实报错。
- 既有测试全部走进程内 TestClient，覆盖不到真实绑定路径——本文件用真实端口补齐该盲区。
- """
- from __future__ import annotations
- import asyncio
- import socket
- import pytest
- import app # noqa: E402 conftest 已把模拟器根目录加入 sys.path
- def _free_port() -> int:
- """取一个当前空闲的端口（绑定 0 由 OS 分配后立即释放）。"""
- with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
- s.bind(("127.0.0.1", 0))
- return s.getsockname()[1]
- @pytest.fixture()
- def mock_port(monkeypatch):
- """把 mockApiPort 指向空闲端口；启停在用例内完成，夹具只复位全局句柄。"""
- port = _free_port()
- monkeypatch.setitem(app.CONFIG, "mockApiPort", port)
- yield port
- app.MOCK_HANDLE.server = None
- app.MOCK_HANDLE.task = None
- app.MOCK_HANDLE.last_error = ""
- def test_start_binds_real_port_and_is_idempotent(mock_port):
- """启动成功须返回 running=true、端口真实可连通、重复调用幂等，且不抛异常（不再 500）。"""
- async def scenario():
- result = await app._start_mock_api()
- try:
- assert result["running"] is True, result
- assert result["port"] == mock_port
- assert app._mock_running() is True
- assert app.MOCK_HANDLE.last_error == ""
- # 端口必须真实可连通，不能只是内部状态为真
- with socket.create_connection(("127.0.0.1", mock_port), timeout=2):
- pass
- again = await app._start_mock_api()
- assert again.get("alreadyRunning") is True, again
- finally:
- await app._stop_mock_api()
- asyncio.run(scenario())
- assert app._mock_running() is False
- def test_start_on_occupied_port_reports_error_not_raise(mock_port):
- """端口被占用须返回 {running:false, error} 并写入 lastError，而不是抛 AttributeError。"""
- blocker = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- blocker.bind(("127.0.0.1", mock_port))
- blocker.listen(1)
- async def scenario():
- try:
- return await app._start_mock_api()
- finally:
- blocker.close()
- result = asyncio.run(scenario())
- assert result["running"] is False, result
- assert str(mock_port) in result["error"]
- assert app.MOCK_HANDLE.last_error # /api/status 的 lastError 诊断字段恢复有效
- assert app._mock_running() is False
|
