unknown
afc55ab910
- 新建 device 设备表(设备/系统信息 + 网络/时区/模拟器/位置 + 关联用户 + 状态/时间戳), 以客户端 device_id 唯一标识;登录后回填 user_id,游客态可空 - 原 device 模型改名 device_liveness(实为无障碍存活表),腾出 device 名给新表并保持 表名与实体名一致;同步更新 __init__ / repositories / admin queries / 文档引用 - 新增 POST /api/v1/device/report:软鉴权(带 JWT 绑用户、游客也收),按 device_id upsert; 设备信息非空才覆盖,位置 latitude/longitude 整字段覆盖含清空(以本次上报实际为准) - 新增 alembic migration bb47051068c8 建 device 表(裁掉 autogenerate 带出的无关表漂移) - 补 /device/report 端点与 upsert 单元测试 4 例
99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
"""设备档案上报端点 /api/v1/device/report + upsert 测试。"""
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.core.security import create_token
|
|
from app.db.session import SessionLocal
|
|
from app.models.device import Device
|
|
from app.models.user import User
|
|
|
|
|
|
def _payload(**over) -> dict:
|
|
base = {
|
|
"device_id": "dev-report-1",
|
|
"platform": "android",
|
|
"oem": "Xiaomi",
|
|
"model": "PJF110",
|
|
"os_version": "Android 14",
|
|
"app_version": "0.2.12(62)",
|
|
"channel": "yingyongbao",
|
|
"screen": "1080x2400",
|
|
"network": "wifi",
|
|
"timezone": "Asia/Shanghai",
|
|
"is_emulator": False,
|
|
"latitude": 31.23,
|
|
"longitude": 121.47,
|
|
}
|
|
base.update(over)
|
|
return base
|
|
|
|
|
|
def _get(device_id: str) -> Device | None:
|
|
with SessionLocal() as db:
|
|
return db.execute(
|
|
select(Device).where(Device.device_id == device_id)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def test_report_creates_device_as_guest(client) -> None:
|
|
r = client.post("/api/v1/device/report", json=_payload(device_id="dev-guest"))
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["ok"] is True
|
|
d = _get("dev-guest")
|
|
assert d is not None
|
|
assert d.user_id is None # 游客态:未绑用户
|
|
assert d.oem == "Xiaomi"
|
|
assert d.is_emulator is False
|
|
assert d.latitude == 31.23
|
|
assert d.last_active_at is not None
|
|
|
|
|
|
def test_report_binds_user_when_authed(client) -> None:
|
|
with SessionLocal() as db:
|
|
u = User(phone="13800000001", username="20000000001")
|
|
db.add(u)
|
|
db.commit()
|
|
uid = u.id
|
|
token, _ = create_token(user_id=uid, token_type="access")
|
|
r = client.post(
|
|
"/api/v1/device/report",
|
|
json=_payload(device_id="dev-user"),
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
assert _get("dev-user").user_id == uid
|
|
|
|
|
|
def test_report_clears_location_when_absent(client) -> None:
|
|
# 首次带位置
|
|
client.post(
|
|
"/api/v1/device/report",
|
|
json=_payload(device_id="dev-loc", latitude=10.0, longitude=20.0),
|
|
)
|
|
assert _get("dev-loc").latitude == 10.0
|
|
# 再次上报拿不到位置(lat/lng=None)→ 必须清空,不能保留旧坐标
|
|
r = client.post(
|
|
"/api/v1/device/report",
|
|
json=_payload(device_id="dev-loc", latitude=None, longitude=None, oem="HONOR"),
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
d = _get("dev-loc")
|
|
assert d.latitude is None # 位置以实际为准 → 清空
|
|
assert d.longitude is None
|
|
assert d.oem == "HONOR" # 设备信息正常更新
|
|
|
|
|
|
def test_report_sticky_fields_not_wiped_by_missing(client) -> None:
|
|
# 首次全量
|
|
client.post("/api/v1/device/report", json=_payload(device_id="dev-sticky"))
|
|
# 再次只带 device_id + 位置(不带 oem/model)→ 设备信息保留旧值(非空才覆盖)
|
|
client.post(
|
|
"/api/v1/device/report",
|
|
json={"device_id": "dev-sticky", "latitude": 1.0, "longitude": 2.0},
|
|
)
|
|
d = _get("dev-sticky")
|
|
assert d.oem == "Xiaomi" # 未被漏带的 None 冲掉
|
|
assert d.model == "PJF110"
|
|
assert d.latitude == 1.0 # 位置按本次
|