122 lines
3.7 KiB
Python
122 lines
3.7 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from typing import Any, Dict
|
||
|
||
import pytest
|
||
import yaml
|
||
from cryptography.fernet import Fernet
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_main_starts_and_stops_cleanly(tmp_path, monkeypatch) -> None:
|
||
"""
|
||
集成级测试：
|
||
- 使用最小 config.yaml；
|
||
- 替换 SQLiteDeviceStore / ConnectionManager / scraper_loop / uvicorn.Server；
|
||
- 验证 async_main 能够启动并在 server 结束后优雅关闭 Scraper 和连接。
|
||
"""
|
||
# 延迟导入以便 monkeypatch 作用于 exporter.main 模块对象
|
||
from exporter import main as exporter_main
|
||
|
||
# 准备配置文件
|
||
key = Fernet.generate_key().decode()
|
||
config_path = tmp_path / "config.yaml"
|
||
config_data: Dict[str, Any] = {
|
||
"global": {
|
||
"http_listen": "127.0.0.1:0",
|
||
"runtime_db_path": str(tmp_path / "devices.db"),
|
||
"password_secret": key,
|
||
# 其他字段使用默认值
|
||
},
|
||
"devices": [],
|
||
}
|
||
config_path.write_text(yaml.safe_dump(config_data), encoding="utf-8")
|
||
|
||
created_store: Dict[str, Any] = {}
|
||
created_cm: Dict[str, Any] = {}
|
||
seen: Dict[str, Any] = {}
|
||
|
||
class DummyStore:
|
||
def __init__(self, db_path: str, encryptor, timeout: float = 5.0) -> None:
|
||
self.db_path = db_path
|
||
self.encryptor = encryptor
|
||
self.timeout = timeout
|
||
self.init_db_called = False
|
||
self.closed = False
|
||
created_store["instance"] = self
|
||
|
||
def init_db(self) -> None:
|
||
self.init_db_called = True
|
||
|
||
def load_runtime_devices(self):
|
||
return []
|
||
|
||
def save_device(self, cfg) -> None: # pragma: no cover - not used here
|
||
raise NotImplementedError
|
||
|
||
def delete_device(self, name: str) -> None: # pragma: no cover - not used here
|
||
raise NotImplementedError
|
||
|
||
def close(self) -> None:
|
||
self.closed = True
|
||
|
||
class DummyConnectionManager:
|
||
def __init__(self, global_cfg) -> None:
|
||
self.global_cfg = global_cfg
|
||
self.closed = False
|
||
created_cm["instance"] = self
|
||
|
||
def acquire_session(self, cfg): # pragma: no cover - Scraper 不会真正调用
|
||
raise NotImplementedError
|
||
|
||
def close_all(self) -> None:
|
||
self.closed = True
|
||
|
||
def dummy_scraper_loop(
|
||
stop_event,
|
||
registry,
|
||
connection_manager,
|
||
netconf_get_rpc,
|
||
cache,
|
||
health,
|
||
global_cfg,
|
||
) -> None:
|
||
# 记录 stop_event，并等待它被置位后退出
|
||
seen["stop_event"] = stop_event
|
||
stop_event.wait(timeout=1.0)
|
||
|
||
class DummyServer:
|
||
def __init__(self, config) -> None:
|
||
self.config = config
|
||
self.served = False
|
||
|
||
async def serve(self) -> None:
|
||
# 立即返回以触发 shutdown 流程
|
||
await asyncio.sleep(0)
|
||
self.served = True
|
||
|
||
# 应用 monkeypatch
|
||
monkeypatch.setattr(exporter_main, "SQLiteDeviceStore", DummyStore)
|
||
monkeypatch.setattr(exporter_main, "ConnectionManager", DummyConnectionManager)
|
||
monkeypatch.setattr(exporter_main, "scraper_loop", dummy_scraper_loop)
|
||
monkeypatch.setattr(exporter_main.uvicorn, "Server", DummyServer)
|
||
|
||
# 运行 async_main（应当能够在短时间内返回）
|
||
await exporter_main.async_main(["--config", str(config_path)])
|
||
|
||
store = created_store["instance"]
|
||
cm = created_cm["instance"]
|
||
|
||
# 验证 DB 初始化与关闭被调用
|
||
assert store.init_db_called is True
|
||
assert store.closed is True
|
||
|
||
# 验证连接被关闭
|
||
assert cm.closed is True
|
||
|
||
# 验证 Scraper 收到 stop_event 并退出
|
||
assert "stop_event" in seen
|
||
assert seen["stop_event"].is_set()
|
||
|