128 lines
4.3 KiB
Python
128 lines
4.3 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlmodel import Session
|
|
|
|
from app.core.config import settings
|
|
from app.models import TerminalDeviceState
|
|
|
|
|
|
def _seed_device_state(
|
|
db: Session, device_id: str, snr_values: list[int] | None = None
|
|
) -> None:
|
|
state = db.get(TerminalDeviceState, device_id)
|
|
if state is None:
|
|
state = TerminalDeviceState(device_id=device_id)
|
|
|
|
signal_values = snr_values if snr_values is not None else [28, 31, 36, 41, 44]
|
|
|
|
state.fix_status = "5"
|
|
state.lat = 31.359591
|
|
state.lon = 120.714489
|
|
state.alt_m = 52.1
|
|
state.satellites_used = 17
|
|
state.satellite_signal_summary = {
|
|
"satellites_in_view": 22,
|
|
"snr_values": signal_values,
|
|
"constellations": {
|
|
"GPS": {"count": 8, "used_in_navigation": 6},
|
|
"BeiDou": {"count": 10, "used_in_navigation": 8},
|
|
"Galileo": {"count": 4, "used_in_navigation": 3},
|
|
},
|
|
"satellites": [
|
|
{
|
|
"satellite_id": "G05",
|
|
"constellation": "GPS",
|
|
"azimuth_deg": 121,
|
|
"elevation_deg": 29,
|
|
"snr_dbhz": 37,
|
|
"used_in_navigation": True,
|
|
}
|
|
],
|
|
}
|
|
state.last_event_time = datetime.now(timezone.utc)
|
|
db.add(state)
|
|
db.commit()
|
|
|
|
|
|
def test_read_telemetry_latest(
|
|
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
|
) -> None:
|
|
device_id = "device-api-001"
|
|
_seed_device_state(db, device_id)
|
|
|
|
response = client.get(
|
|
f"{settings.API_V1_STR}/telemetry/latest",
|
|
headers=superuser_token_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert "generated_at" in payload
|
|
assert len(payload["devices"]) >= 1
|
|
device = next(item for item in payload["devices"] if item["device_id"] == device_id)
|
|
assert device["fix_status"] == "5"
|
|
assert device["signal_health"]["level"] == "poor"
|
|
|
|
|
|
def test_signal_health_levels_based_on_count_of_40_plus_signals(
|
|
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
|
) -> None:
|
|
_seed_device_state(db, "device-health-good", [40] * 20 + [15] * 3)
|
|
_seed_device_state(db, "device-health-fair", [40] * 10 + [20] * 3)
|
|
_seed_device_state(db, "device-health-poor", [40] * 9 + [39])
|
|
|
|
response = client.get(
|
|
f"{settings.API_V1_STR}/telemetry/latest",
|
|
headers=superuser_token_headers,
|
|
params={"limit": 200},
|
|
)
|
|
assert response.status_code == 200
|
|
devices = {item["device_id"]: item for item in response.json()["devices"]}
|
|
|
|
assert devices["device-health-good"]["signal_health"]["level"] == "good"
|
|
assert devices["device-health-fair"]["signal_health"]["level"] == "fair"
|
|
assert devices["device-health-poor"]["signal_health"]["level"] == "poor"
|
|
|
|
|
|
def test_signal_health_uses_satellite_level_values_for_ucenter_view(
|
|
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
|
) -> None:
|
|
state = db.get(TerminalDeviceState, "device-health-ucenter")
|
|
if state is None:
|
|
state = TerminalDeviceState(device_id="device-health-ucenter")
|
|
state.fix_status = "5"
|
|
state.satellite_signal_summary = {
|
|
# u-center signal bars are per-satellite bars, not per-frequency points
|
|
"snr_values": [41] * 9 + [30] * 4,
|
|
# keep raw CN0 points for diagnostics, but health should follow bar view
|
|
"cn0_values": [41] * 20 + [25] * 5,
|
|
}
|
|
db.add(state)
|
|
db.commit()
|
|
|
|
response = client.get(
|
|
f"{settings.API_V1_STR}/telemetry/latest",
|
|
headers=superuser_token_headers,
|
|
params={"device_id": "device-health-ucenter"},
|
|
)
|
|
assert response.status_code == 200
|
|
device = response.json()["devices"][0]
|
|
assert device["signal_health"]["strong_signal_count"] == 9
|
|
assert device["signal_health"]["level"] == "poor"
|
|
|
|
|
|
def test_telemetry_websocket_stream(
|
|
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
|
) -> None:
|
|
_seed_device_state(db, "device-api-002")
|
|
token = superuser_token_headers["Authorization"].split(" ", 1)[1]
|
|
|
|
with client.websocket_connect(
|
|
f"{settings.API_V1_STR}/telemetry/ws?token={token}"
|
|
) as websocket:
|
|
payload = websocket.receive_json()
|
|
|
|
assert "generated_at" in payload
|
|
assert len(payload["devices"]) >= 1
|