87 lines
2.6 KiB
Python
87 lines
2.6 KiB
Python
import sqlite3
|
|
from typing import Any, Dict, List
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from exporter.api import create_app
|
|
from exporter.config import DeviceConfig, GlobalConfig
|
|
from exporter.metrics import TransceiverCollector
|
|
from exporter.models import DeviceHealthState, DeviceMetricsSnapshot
|
|
from exporter.registry import DeviceRegistry
|
|
|
|
|
|
VALID_FERNET_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
|
|
|
|
|
|
class DummyStore:
|
|
"""简单的测试用存储，实现 API 需要的接口，并在指定操作上抛 OperationalError。"""
|
|
|
|
def __init__(self, fail_on: str | None = None) -> None:
|
|
self.fail_on = fail_on
|
|
|
|
# 与 SQLiteDeviceStore 接口对齐
|
|
def init_db(self) -> None: # pragma: no cover - 在这些测试中不会调用
|
|
return
|
|
|
|
def load_runtime_devices(self) -> List[DeviceConfig]:
|
|
return []
|
|
|
|
def save_device(self, cfg: DeviceConfig) -> None:
|
|
if self.fail_on == "save":
|
|
raise sqlite3.OperationalError("database is locked")
|
|
|
|
def delete_device(self, name: str) -> None:
|
|
if self.fail_on == "delete":
|
|
raise sqlite3.OperationalError("database is locked")
|
|
|
|
def close(self) -> None: # pragma: no cover - 这里无需验证
|
|
return
|
|
|
|
|
|
def _build_app_with_dummy_store(fail_on: str | None) -> TestClient:
|
|
gc = GlobalConfig()
|
|
gc.api_token = "token"
|
|
gc.runtime_db_path = ":memory:"
|
|
gc.password_secret = VALID_FERNET_KEY
|
|
|
|
registry = DeviceRegistry(global_scrape_interval=gc.scrape_interval_seconds)
|
|
cache: Dict[str, DeviceMetricsSnapshot] = {}
|
|
health: Dict[str, DeviceHealthState] = {}
|
|
collector = TransceiverCollector(cache, health)
|
|
|
|
store = DummyStore(fail_on=fail_on)
|
|
app = create_app(registry, store, collector, gc)
|
|
return TestClient(app)
|
|
|
|
|
|
def test_post_device_sqlite_operational_error_returns_503():
|
|
client = _build_app_with_dummy_store(fail_on="save")
|
|
|
|
payload = {
|
|
"name": "dev-save-error",
|
|
"host": "192.0.2.10",
|
|
"port": 830,
|
|
"username": "u",
|
|
"password": "p",
|
|
"enabled": True,
|
|
}
|
|
resp = client.post(
|
|
"/api/v1/devices",
|
|
headers={"X-API-Token": "token"},
|
|
json=payload,
|
|
)
|
|
assert resp.status_code == 503
|
|
assert "database is locked" in resp.json()["detail"]
|
|
|
|
|
|
def test_delete_device_sqlite_operational_error_returns_503():
|
|
client = _build_app_with_dummy_store(fail_on="delete")
|
|
|
|
resp = client.delete(
|
|
"/api/v1/devices/nonexistent",
|
|
headers={"X-API-Token": "token"},
|
|
)
|
|
assert resp.status_code == 503
|
|
assert "database is locked" in resp.json()["detail"]
|
|
|