85 lines
2.0 KiB
Python
85 lines
2.0 KiB
Python
"""
|
|
天目 pytest 配置。
|
|
|
|
提供服务启停夹具，方便运行集成测试。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import time
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
API_BASE = os.environ.get("TIANMU_API_BASE", "http://localhost:8101")
|
|
TIANMU_DIR = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def _wait_for_service(url: str, timeout: float = 15.0, interval: float = 0.5) -> bool:
|
|
"""轮询等待服务就绪。
|
|
|
|
Args:
|
|
url: 健康检查 URL。
|
|
timeout: 最长等待秒数。
|
|
interval: 轮询间隔秒数。
|
|
|
|
Returns:
|
|
True 表示服务就绪，False 表示超时。
|
|
"""
|
|
start = time.monotonic()
|
|
while time.monotonic() - start < timeout:
|
|
try:
|
|
req = urllib.request.Request(url, method="GET")
|
|
with urllib.request.urlopen(req, timeout=2) as resp:
|
|
if resp.status == 200:
|
|
return True
|
|
except (urllib.error.URLError, ConnectionError, OSError):
|
|
pass
|
|
time.sleep(interval)
|
|
return False
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def tianmu_server():
|
|
"""启动天目服务（仅一次，整个 session 共用）。
|
|
|
|
如果环境变量 TIANMU_API_BASE 已设置，跳过启动（连接外部服务）。
|
|
启动后轮询 /healthz，就绪后才继续运行测试。
|
|
"""
|
|
if "TIANMU_API_BASE" in os.environ:
|
|
yield
|
|
return
|
|
|
|
proc = subprocess.Popen(
|
|
["python3", "-m", "src.main"],
|
|
cwd=str(TIANMU_DIR),
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
|
|
ready = _wait_for_service(f"{API_BASE}/healthz")
|
|
if not ready:
|
|
proc.terminate()
|
|
proc.wait()
|
|
raise RuntimeError(
|
|
f"天目服务启动超时（{API_BASE}/healthz 15s 内未就绪）"
|
|
)
|
|
|
|
yield
|
|
|
|
proc.terminate()
|
|
try:
|
|
proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
proc.kill()
|
|
proc.wait()
|
|
|
|
|
|
@pytest.fixture
|
|
def api_base() -> str:
|
|
return API_BASE
|